मैं एक आरएक्स अवलोकन योग्य कैसे बना सकता हूं जो अंतिम पर्यवेक्षक सदस्यता रद्द करते समय प्रकाशन ईवेंट रोकता है?

Question

मैं एक अवलोकन योग्य (विभिन्न माध्यमों के माध्यम से) बनाउंगा और इसे इच्छुक पार्टियों को वापस कर दूंगा, लेकिन जब वे सुनते हैं, तो मैं अवलोकन को फाड़ना चाहता हूं ताकि संसाधनों का उपभोग जारी न रहे। एक पब सब सिस्टम में विषयों को बनाने के रूप में इसके बारे में सोचने का एक और तरीका। जब कोई भी विषय किसी और विषय की सदस्यता नहीं लेता है, तो आप विषय और उसके फ़िल्टरिंग को और अधिक नहीं पकड़ना चाहते हैं।

Answer 1

आरएक्स पहले से ही आपकी आवश्यकताओं के अनुरूप करने के लिए एक ऑपरेटर है - अच्छी तरह से दो वास्तव में - Publish & RefCount।

यहाँ उनके उपयोग का तरीका बताया गया है:

IObservable xs = ... 

var rxs = xs.Publish().RefCount(); 

var sub1 = rxs.Subscribe(x => { }); 
var sub2 = rxs.Subscribe(x => { }); 

//later 
sub1.Dispose(); 

//later 
sub2.Dispose(); 

//The underlying subscription to `xs` is now disposed of. 

सरल।

Answer 2

यदि मैं आपका प्रश्न समझ गया हूं तो आप इस तरह के अवलोकन योग्य बनाना चाहते हैं कि जब सभी ग्राहकों ने अपनी सदस्यता का निपटारा किया हो यानी कोई और ग्राहक नहीं है, तो आप एक साफ अप फ़ंक्शन निष्पादित करना चाहते हैं जो उत्पादन से आगे के मूल्यों को देखने योग्य बनाएगा ।

//Wrap a disposable 
public class WrapDisposable : IDisposable 
    { 
     IDisposable disp; 
     Action act; 
     public WrapDisposable(IDisposable _disp, Action _act) 
     { 
      disp = _disp; 
      act = _act; 
     } 
     void IDisposable.Dispose() 
     { 
      act(); 
      disp.Dispose(); 
     } 
    } 

    //Observable that we want to clean up after all subs are done 
    public static IObservable<long> GenerateObs(out Action cleanup) 
    { 
     cleanup =() => 
     { 
      Console.WriteLine("All subscribers are done. Do clean up"); 
     }; 
     return Observable.Interval(TimeSpan.FromSeconds(1)); 
    } 
    //Wrap the observable 
    public static IObservable<T> WrapToClean<T>(IObservable<T> obs, Action onAllDone) 
    { 
     int count = 0; 
     return Observable.CreateWithDisposable<T>(ob => 
     { 
      var disp = obs.Subscribe(ob); 
      Interlocked.Increment(ref count); 
      return new WrapDisposable(disp,() => 
      { 
       if (Interlocked.Decrement(ref count) == 0) 
       { 
        onAllDone();             
       } 
      }); 
     }); 
    } 

// प्रयोग उदाहरण: इस यदि आप तो चाहते हैं कि आप नीचे की तरह कुछ कर सकते हैं

Action cleanup; 
var obs = GenerateObs(out cleanup); 
var newObs = WrapToClean(obs, cleanup); 
newObs.Take(6).Subscribe(Console.WriteLine); 
newObs.Take(5).Subscribe(Console.WriteLine);