IObserver को async/onEext/OnError/OnCompleted विधियों का इंतजार कैसे करें?

Question

मैं System.Net.WebSocket के लिए एक एक्सटेंशन विधि लिखने की कोशिश कर रहा हूं जो इसे प्रतिक्रियाशील एक्सटेंशन (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") 
    ); 
} 

इस कोड काम करता है, लेकिन मैं async के उपयोग के बारे में चिंतित हूं/onNext, onError के अंदर का इंतजार है, और lambdas onCompleted। मुझे पता है कि यह एसिंक शून्य लैम्ब्डा लौटाता है, जो कि (और कभी-कभी ऐसे मुद्दों का कारण बनता है जिन्हें मैंने पहले ही चलाया है)।

मैं आरएक्स.नेट दस्तावेज के साथ-साथ इंटरनेट पर ब्लॉग पोस्ट पर पढ़ रहा हूं और मुझे IObserver में एसिंक/प्रतीक्षा विधि का उपयोग करने के लिए उचित तरीका (यदि कोई है) नहीं मिल रहा है। क्या ऐसा करने का कोई सही तरीका है? यदि नहीं, तो मैं इस समस्या के आसपास कैसे काम कर सकता हूं?

Answer 1

सदस्यता लें async विधियों को नहीं लेता है। तो यहां क्या होता है कि आप async void से अग्नि-और-भूल तंत्र का उपयोग कर रहे हैं। समस्या यह है कि आगे के संदेशों को क्रमबद्ध नहीं किया जाएगा।

सदस्यता के अंदर 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); 
    } 

पहले हम एक Observable में async onNext विधि परिवर्तित, दो async दुनिया ब्रिजिंग। इसका मतलब है onNext के अंदर async/प्रतीक्षा करें सम्मानित किया जाएगा। इसके बाद हम विधि को Defer में लपेटें ताकि यह बनाया जाने पर सही प्रारंभ न हो। इसके बजाय Concat अगले डिलीवर किए जाने पर केवल अगले डिलीवर किए जाने योग्य कॉल को कॉल करेगा।

Ps। मुझे उम्मीद है कि आरएक्सनेट की अगली रिलीज में async सब्सक्राइब में समर्थन होगा।