用 Observable 替换 TaskCompletionSource

Question

在我的 .NET 4.0 库中，我有一段代码通过网络发送数据并等待响应。为了不阻塞调用代码，该方法返回一个Task<T>，它在收到响应时完成，以便代码可以像这样调用该方法：

// Send the 'message' to the given 'endpoint' and then wait for the response
Task<IResult> task = sender.SendMessageAndWaitForResponse(endpoint, message);
task.ContinueWith(
    t => 
    {
        // Do something with t.Result ...
    });

底层代码使用TaskCompletionSource，这样它就可以等待响应消息，而不必启动一个线程，只是让它闲置在那里直到响应到来：

private readonly Dictionary<int, TaskCompletionSource<IResult>> m_TaskSources
    = new Dictionary<int, TaskCompletionSource<IResult>>();

public Task<IResult> SendMessageAndWaitForResponse(int endpoint, object message)
{
    var source = new TaskCompletionSource<IResult>(TaskCreationOptions.None);
    m_TaskSources.Add(endpoint, source);

    // Send the message here ...

    return source.Task;
}

收到响应后，将按如下方式处理：

public void CompleteWaitForResponseResponse(int endpoint, IResult value)
{
    if (m_TaskSources.ContainsKey(endpoint))
    {
        var source = m_TaskSources[endpoint];
        source.SetResult(value);
        m_TaskSources.Remove(endpoint);
    }
}

现在我想添加一个超时，这样调用代码就不会无限期地等待响应。但是在 .NET 4.0 上是 somewhat messy，因为没有简单的方法可以让任务超时。所以我想知道 Rx 是否能够更容易地做到这一点。所以我想出了以下内容：

private readonly Dictionary<int, Subject<IResult>> m_SubjectSources
    = new Dictionary<int, Subject<IResult>>();

private Task<IResult> SendMessageAndWaitForResponse(int endpoint, object message, TimeSpan timeout)
{
    var source = new Subject<IResult>();
    m_SubjectSources.Add(endpoint, source);

    // Send the message here ...

    return source.Timeout(timeout).ToTask();
}

public void CompleteWaitForResponseResponse(int endpoint, IResult value)
{
    if (m_SubjectSources.ContainsKey(endpoint))
    {
        var source = m_SubjectSources[endpoint];
        source.OnNext(value);
        source.OnCompleted();
        m_SubjectSources.Remove(endpoint);
    }
}

这一切似乎都没有问题，但是我看到几个问题指出 Subject 应该是 avoided 所以现在我想知道是否有更多的 Rx-y 方式来实现我的目标。

Answer 1

避免在 Rx 中使用 Subject 的建议通常被夸大了。 Rx 中必须有事件源，它可以是Subject。

Subject 的问题通常是在两个 Rx 查询之间使用，否则可能会被连接，或者已经有明确定义的转换为 IObservable<T>（例如 Observable.FromEventXXX 或 Observable.FromAsyncXXX 等。

如果您愿意，您可以使用以下方法取消Dictionary 和多个Subjects。这使用单个主题并向客户端返回过滤后的查询。

这不是“更好”本身，这是否有意义将取决于您的场景的具体情况，但它可以节省大量的主题，并为您提供了一个很好的选择来监控所有结果单个流。如果您是串行分派结果（比如从消息队列中），这可能是有意义的。

// you only need to synchronize if you are receiving results in parallel
private readonly ISubject<Tuple<int,IResult>, Tuple<int,IResult>> results =
    Subject.Synchronize(new Subject<Tuple<int,IResult>>());

private Task<IResult> SendMessageAndWaitForResponse(
    int endpoint, object message, TimeSpan timeout)
{           
    // your message processing here, I'm just echoing a second later
    Task.Delay(TimeSpan.FromSeconds(1)).ContinueWith(t => {
        CompleteWaitForResponseResponse(endpoint, new Result { Value = message }); 
    });

    return results.Where(r => r.Item1 == endpoint)
                  .Select(r => r.Item2)
                  .Take(1)
                  .Timeout(timeout)
                  .ToTask();
}

public void CompleteWaitForResponseResponse(int endpoint, IResult value)
{
    results.OnNext(Tuple.Create(endpoint,value));
}

我为这样的结果定义了一个类：

public class Result : IResult
{
    public object Value { get; set; }
}

public interface IResult
{
    object Value { get; set; }
}

编辑 - 回应 cmets 中的其他问题。

• 无需处理单个主题 - 它不会泄漏，并且会在超出范围时被垃圾回收。

• ToTask 确实接受取消令牌 - 但这确实是为了从客户端取消。

• 如果远程端断开连接，您可以使用results.OnError(exception); 向所有客户端发送错误 - 您需要同时实例化一个新的主题实例。

类似：

private void OnRemoteError(Exception e)
{
    results.OnError(e);        
}

这将以预期的方式向所有客户端显示为错误任务。

这也是相当线程安全的，因为订阅先前发送 OnError 的主题的客户端将立即收到错误消息 - 从那时起它就死了。然后准备好后，您可以使用以下命令重新初始化：

private void OnInitialiseConnection()
{
    // ... your connection logic

    // reinitialise the subject...
    results = Subject.Synchronize(new Subject<Tuple<int,IResult>>());
}

对于个别客户端错误，您可以考虑：

• 扩展您的 IResult 界面以将错误作为数据包含在内

• 然后，您可以通过扩展SendMessageAndWaitForResponse 中的 Rx 查询，选择性地将其投射到该客户端的故障中。例如，IResult 的 Exception 和 HasError 属性，以便您可以执行以下操作：

  return results.Where(r => r.Item1 == endpoint)
              .SelectMany(r => r.Item2.HasError
                  ? Observable.Throw<IResult>(r.Item2.Exception)
                  : Observable.Return(r.Item2))
              .Take(1)
              .Timeout(timeout)
              .ToTask();