如何使用 async/await OnNext/OnError/OnCompleted 方法实现 IObserver？

Question

我正在尝试为 System.Net.WebSocket 编写一个扩展方法，该方法将使用 Reactive Extensions (Rx.NET) 将其转换为 IObserver。可以看下面的代码：

public static IObserver<T> ToObserver<T>(this WebSocket webSocket, IWebSocketMessageSerializer<T> webSocketMessageSerializer)
{
    // Wrap the web socket in an interface that's a little easier to manage
    var webSocketMessageStream = new WebSocketMessageStream(webSocket);

    // Create the output stream to the client
    return Observer.Create<T>(
        onNext:      async message => await webSocketMessageStream.WriteMessageAsync(webSocketMessageSerializer.SerializeMessage(message)),
        onError:     async error   => await webSocketMessageStream.CloseAsync(WebSocketCloseStatus.InternalServerError, string.Format("{0}: {1}", error.GetType(), error.Message)),
        onCompleted: async ()      => await webSocketMessageStream.CloseAsync(WebSocketCloseStatus.NormalClosure, "Server disconnected")
    );
}

此代码有效，但我担心在 onNext、onError 和 onCompleted lambda 中使用 async/await。我知道这会返回一个 async void lambda，这是不受欢迎的（有时会导致我自己已经遇到的问题）。

我一直在阅读 Rx.NET 文档以及互联网上的博客文章，但我似乎找不到在 IObserver 中使用 async/await 方法的正确方法（如果有的话）。有没有合适的方法来做到这一点？如果没有，那我该如何解决这个问题？

Answer 1

订阅不采用async 方法。因此，这里发生的情况是您正在使用来自async void 的即发即弃机制。问题是 onNext 消息将不再被序列化。

您应该将其包装到管道中以允许 Rx 等待它，而不是在订阅中调用 async 方法。

在onError 和onCompleted 上使用即发即弃是可以的，因为它们保证是从Observable 调用的最后一件事。请记住，与Observable 关联的资源可以在onError 和onCompleted 返回之后，在它们完成之前清理。

我写了一个小的扩展方法来完成这一切：

    public static IDisposable SubscribeAsync<T>(this IObservable<T> source, Func<T, Task> onNext, Action<Exception> onError, Action onCompleted)
    {
        return source.Select(e => Observable.Defer(() => onNext(e).ToObservable())).Concat()
            .Subscribe(
            e => { }, // empty
            onError,
            onCompleted);
    }

首先，我们将async onNext 方法转换为Observable，连接两个异步世界。这意味着 onNext 中的 async/await 将受到尊重。接下来我们将该方法包装到Defer 中，这样它就不会在创建时立即启动。相反，Concat 只会在前一个完成时调用下一个延迟的 observable。

附言。我希望 Rx.Net 的下一个版本将在订阅中支持async。