Làm thế nào để thực hiện một IObserver với async/await OnNext/OnError/OnCompleted phương pháp?

Question

Tôi đang cố gắng để viết một phương pháp mở rộng cho System.Net.WebSocket mà sẽ biến nó thành một IObserver bằng cách sử dụng phần mở rộng phản ứng (Rx.NET). Bạn có thể xem mã bên dưới:

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") 
    ); 
} 

Mã này làm việc, nhưng tôi lo ngại về việc sử dụng async/chờ đợi bên trong onNext, onerror, và onCompleted lambdas. Tôi biết rằng điều này trả về một số không đồng bộ async lambda, được cau mày (và đôi khi gây ra các vấn đề mà tôi đã chạy vào bản thân mình).

Tôi đã đọc trên tài liệu Rx.NET cũng như các bài đăng blog trên internet và tôi dường như không tìm được cách thích hợp (nếu có) để sử dụng phương thức async/await trong IObserver. Có cách nào thích hợp để làm việc này không? Nếu không, thì làm thế nào tôi có thể giải quyết vấn đề này?

Answer 1

Đăng ký không thực hiện các phương thức async. Vì vậy, những gì xảy ra ở đây là bạn đang sử dụng một cơ chế lửa và quên từ async void. Vấn đề là các tin nhắn onNext sẽ không còn được đăng.

Thay vì gọi phương thức async bên trong Đăng ký, bạn nên bọc nó vào đường ống để cho phép Rx đợi.

Được phép sử dụng lửa và quên trên onError và onCompleted vì chúng được đảm bảo là điều cuối cùng được gọi là từ Observable. Hãy nhớ rằng các tài nguyên liên quan đến Observable có thể được dọn sạch sau khi onError và onCompleted được trả về, trước khi chúng hoàn tất.

Tôi đã viết một phương pháp khuyến nông nhỏ mà làm tất cả điều này:

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); 
    } 

Đầu tiên chúng tôi chuyển đổi phương pháp async onNext thành một Observable, cầu nối hai thế giới async. Điều này có nghĩa là không đồng bộ/chờ đợi bên trong onNext sẽ được tôn trọng. Tiếp theo chúng ta bọc phương thức này vào Defer để nó không bắt đầu ngay khi được tạo. Thay vào đó, Concat sẽ gọi lệnh tiếp theo bị trì hoãn chỉ khi kết thúc trước đó.

Ps. Tôi hy vọng bản phát hành tiếp theo của Rx.Net sẽ có hỗ trợ async khi đăng ký.