使用iObservable实现异步方法

Question

使用返回 IObservable 的方法来实现标准 Begin/End 异步模式的替代方案是否合理？在下面的代码中，我使用 Rx 包装旧版 API，以为其提供更好的接口。

API 上的 Messages 属性是一个热门 IObservable。我的问题是，我认为如果我在返回过滤响应的 IObservable 之前收到响应消息（即 m.RequestId == requestId ），订阅者将看不到它并且不会知道操作已完成。

有没有办法正确地做到这一点？

    public IObservable<bool> DoAsyncRequest()
    {
        Observable.Defer(delegate
        {
            int requestId = GenerateRequestId();

            this.api.DoApiRequest(requestId);

            return this.api.Messages
                .Where(m => m.RequestId == requestId)
                .Take(1)
                .Timeout(DefaultTimeout);

        });
    }

Answer 1

首先，返回 IObservable 来实现异步方法绝对没问题。

其次，如果是 DoApiRequest 方法启动通过 Messages 发出值的进程，您可以先订阅消息，然后再使用 CreateWithDisposable 调用该方法>

public IObservable<bool> DoAsyncRequest()
{
    return Observable.CreateWithDisposable<bool>(observer =>
        {    
            var disposable = this.api.Messages
                .Where(m => m.RequestId == requestId)
                .Take(1)
                .Timeout(DefaultTimeout);
                .Subscribe(observer);

            int requestId = GenerateRequestId();

            this.api.DoApiRequest(requestId);

            return disposable;

        });
}

或者，由于您似乎可以访问 api 对象的任何类，因此您也可以更改 DoApiRequest 以返回冷可观察值，即可能废弃 Messages 属性，并在 DoApiRequest 方法上返回 IObservable。

注意

如果我正确地假设 DoApiRequest 方法确实启动了通过消息发出值的过程，那么如果多次调用此方法，您将得到奇怪的结果。一个调用可能会收到原本要为另一调用返回的消息。如果可以的话，我会亲自研究替代解决方案。