RX IObserver 订阅时间

Question

我是 .net 中的 Rx 新手，但已经开始使用它并成功完成了一些网络通信

非常简单的例子：

    IObservable<Result> SendRequest(Request request)
    {
        return Observable.Create<Result>(observer =>
        {
            NetworkComms.SendReqeust(request,
                result =>
                {
                    observer.OnNext(result);
                    observer.OnCompleted();
                });
            return Disposable.Empty;
        }).SubscribeOn(_commsScheduler);
    }

我遇到的问题是，在调用者订阅返回的 IObservable 之前，实际上并未发送命令 (NetworkComms.SendRequest)。在某些情况下，请求是即发即弃的命令，结果毫无意义，因此调用者实际订阅返回的 IObservable 毫无意义。

我需要的功能是：

1. 命令会立即发送，即使调用者从未订阅 IObservable
2. 当客户端订阅时，即使延迟订阅也会得到结果
3. 该命令只发送一次，但所有订阅都应该得到相同的结果。

我尝试使用 .Replay().RefCount() 执行此操作，并在返回 IObservable 之前在内部执行 Subscribe()。这几乎可以工作，但是如果客户端在收到结果后订阅（因此在完成时自动处理序列之后），它会导致再次调用订阅代码，发送命令第二次。

是否有一个简单的 Rx 扩展可以为我处理这种情况，还是我需要自己动手？

Answer 1

您似乎想使用AsyncSubject<T>：

IObservable<Result> SendRequest(Request request)
{
    var subject = new AsyncSubject<Result>();
    NetworkComms.SendReqeust(request, result =>
    {
        subject.OnNext(result);
        subject.OnCompleted();
    });
    return subject.AsObservable();
}

我应该补充一点，您想要的界面有点奇怪。如果语义是“在我调用此方法的那一刻，就发出了一个请求，任何想要得到响应的人都可以在以后得到响应”，那么请考虑只使用Task<Result> 而不是IObservable<Result>。

Task<Result> SendRequestAsync(Request request)
{
    var tcs = new TaskCompletionSource<Result>();
    NetworkComms.SendReqeust(request, result => tcs.SetResult(result));
    return tcs.Task;
}

此外，考虑将此Task<Result> SendRequestAsync(Request request) 方法直接放在您的NetworkComms 类中。