आरएक्स

Question

में एक कॉलबैक मैं एक सुंदर तरीका Rx के साथ एक सादे कॉलबैक प्रतिनिधि से एक Observable बनाने के लिए की तलाश में हूँ, Observable.FromEventPattern को कुछ इसी तरह के लिए प्रत्यक्ष?

कहें, मैं Win32 EnumWindows एपीआई लपेट रहा हूं जो EnumWindowsProc मैं प्रदान करता हूं।

मैं जानता हूँ कि मैं इस कॉलबैक के लिए एक अस्थायी सी # घटना एडाप्टर बनाने और इसे FromEventPattern दे सकते हैं। इसके अलावा, मैं शायद IObservable मैन्युअल रूप से कार्यान्वित कर सकता हूं, इसलिए यह EnumWindowsProc कॉलबैक से IObserver.OnNext पर कॉल करेगा।

Rx में एक कॉलबैक है कि मैं याद कर रहा हूँ लपेटकर के लिए वहाँ एक मौजूदा पैटर्न है?

Answer 1

आप एक Subject<T> उपयोग कर सकते हैं आरएक्स के कार्यात्मक दुनिया में जरूरी प्रोग्रामिंग दुनिया से ले जाने के लिए इस्तेमाल किया जा सकता है।

Subject<T> दोनों IObservable<T> और IObserver<T> लागू करता है, तो आप अपने OnNext, OnError और OnCompleted तरीकों कॉल कर सकते हैं और ग्राहकों को सूचित कर दिया जाएगा।

आप को बेनकाब करने के Subject<T> एक संपत्ति के रूप में तो तुम इतनी .AsObservable() का उपयोग कर के रूप में इस तथ्य यह है कि IObservable<T> तथ्य एक Subject<T> में है छुपाता करना चाहिए चाहते हैं। यह ((Subject<string>) obj.Event).OnNext("Foo") जैसी चीजें असंभव बनाता है।

Answer 2

ध्यान रखें कि EnumWindows में प्रयोग किया कॉलबैक आरएक्स से आसानी से अलग हैं। विशेष रूप से, कॉलबैक कॉलर को अपने रिटर्न वैल्यू के माध्यम से वापस संवाद कर सकता है। आरएक्स पर्यवेक्षक ऐसा नहीं कर सकते हैं। साथ ही, कॉलबैक एकाधिक पैरामीटर प्राप्त कर सकते हैं, लेकिन आरएक्स पर्यवेक्षकों को एक ही मूल्य प्राप्त होता है। तो आपको एकाधिक पैरामीटर को एक ऑब्जेक्ट में लपेटने की आवश्यकता है।

इस बात को ध्यान में रखते हुए, Subject का उपयोग करने का विकल्प Observable.Create का उपयोग करना है। इस तरह आप वास्तव में एक पर्यवेक्षक होने पर केवल कॉलबैक पंजीकृत करते हैं और यदि वह पर्यवेक्षक सदस्यता रद्द करता है तो आप इसे अनधिकृत करते हैं।

तुल्यकालिक एपीआई आपको एक उदाहरण का उपयोग किया है के लिए, आप कुछ इस तरह कर सकते हैं। इस उदाहरण में नोट वास्तव में कॉलबैक मिड-स्ट्रीम को अनधिकृत करने का कोई तरीका नहीं है क्योंकि इससे पहले कि हम सदस्यता रद्द करने योग्य डिस्पोजेबल वापस लौट सकें, इससे पहले कि यह सब सिंक्रनाइज़ हो जाए।

public static IObservable<Foo> WrapFooApi(string arg1, string arg2) 
{ 
    return Observable.Create<Foo>(observer => 
    { 
     FooApi.enumerate(arg1, arg2, e => 
     { 
      observer.OnNext(new Foo(e)); 
      return true; 
     }); 

     // In your case, FooApi.enumerate is actually synchronous 
     // so when we get to this line of code, we know 
     // the stream is complete. 
     observer.OnCompleted(); 
     return Disposable.Empty; 
    }); 
} 

// Usage 
WrapFooApi("a", "b").Take(1).Subscribe(...); // only takes first item 

हम एक छोटे से asynchronicity, जो पर्यवेक्षक समय एक डिस्पोजेबल है कि यह आपको सूचित करने के निपटान कर सकते हैं पाने के लिए दे देंगे शुरू करने से जल्दी रोकने में असमर्थ होने के साथ समस्या को ठीक कर सकते हैं। CancellationToken प्राप्त करने के लिए हम CreateAsync का उपयोग कर सकते हैं जो पर्यवेक्षक सदस्यता रद्द करते समय रद्द कर देगा। और हम Task.Run अंदर FooApi कोड चला सकते हैं:

public static IObservable<Foo> WrapFooApi(string args) 
{ 
    return Observable.Create<Foo>(observer => 
    { 
     FooToken token = default(FooToken); 
     var unsubscribe = Disposable.Create(() => FooApi.Unregister(token)); 
     token = FooApi.Register(args, e => 
     { 
      observer.OnNext(new Foo(e)); 
     }); 

     return unsubscribe; 
    }); 
} 

:

public static IObservable<Foo> WrapFooApi(string arg1, string arg2) 
{ 
    return Observable.CreateAsync<Foo>(async (observer, ct) => 
    { 
     await Task.Run(() => FooApi.register_callback(arg1, arg2, e => 
     { 
      observer.OnNext(e); 

      // Returning false will stop the enumeration 
      return !ct.IsCancellationRequested; 
     })); 
     observer.OnCompleted(); 
    }); 
} 

अधिक परंपरागत अतुल्यकालिक कॉलबैक एपीआई, जहाँ आप कुछ बिंदु पर रजिस्टर और कुछ अन्य बिंदु पर अपंजीकृत में, आप कुछ इस तरह अधिक हो सकता है