पर्यवेक्षकों में (संख्याओं) पर्यवेक्षकों को ट्रैक करें?

Question

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

क्या यह पता लगाने का कोई तरीका है कि कितने पर्यवेक्षकों ने एक अवलोकन पर सब्सक्राइब किया है? या शायद यह जानने का तरीका है कि पर्यवेक्षक बुला रहे हैं या सदस्यता लें?

Answer 1

IObservable<T> एक interface है जिसे आप कार्यान्वित कर सकते हैं। इंटरफ़ेस की सदस्यता विधि में आप आंतरिक रूप से एक सूची बनाए रखकर पर्यवेक्षकों का ट्रैक रख सकते हैं।

निम्नलिखित कोड स्निपेट एमएसडीएन से है।

private List<IObserver<Location>> observers; 

public IDisposable Subscribe(IObserver<Location> observer) 
{ 
    if (! observers.Contains(observer)) 
     observers.Add(observer); 

    // ------- If observers.Count == 1 create connection. ------- 

    return new Unsubscriber(observers, observer); 
} 
private class Unsubscriber : IDisposable 
{ 
    private List<IObserver<Location>>_observers; 
    private IObserver<Location> _observer; 

    public Unsubscriber(List<IObserver<Location>> observers, IObserver<Location> observer) 
    { 
     this._observers = observers; 
     this._observer = observer; 
    } 

    public void Dispose() 
    { 
     if (_observer != null && _observers.Contains(_observer)) 
     _observers.Remove(_observer); 
     // ----------- if observers.Count == 0 close connection ----------- 
    } 
} 

Answer 2

मैं बस रेफकाउंट/प्रकाशित का उपयोग करता हूं। मैं हमेशा ऐसा महसूस करता हूं कि मैं IObservable को कार्यान्वित कर रहा हूं, मैं बहुत मेहनत कर रहा हूं।

myColdObservable.Publish().RefCount(); 

इससे डिस्कनेक्ट होने के बाद यह आपके अवलोकन रोकने योग्य हो जाएगा। यहां एक नमूना है:

var coldObservable = Observable 
    .Interval(TimeSpan.FromSeconds(1)) 
    .ObserveOn(Scheduler.TaskPool) 
    .Select(_ => DoSomething()); 

var refCountObs = coldObservable.Publish().RefCount(); 

CompositeDisposable d = new CompositeDisposable(); 
d.Add(refCountObs.Subscribe(n => Console.WriteLine("First got: " + n))); 
d.Add(refCountObs.Subscribe(n => Console.WriteLine("Second got: " + n))); 
d.Add(refCountObs.Subscribe(n => Console.WriteLine("Third got: " + n))); 

//Wait a bit for work to happen 
System.Threading.Thread.Sleep(10000); 

//Everyone unsubscribes 
d.Dispose(); 

//Observe that DoSomething is not called. 
System.Threading.Thread.Sleep(3000); 

इस मामले को कवर नहीं करता है जहाँ आप वास्तव में ग्राहकों की संख्या जानना चाहता हूँ, लेकिन मैं इस काम को रोकने की अपनी आवश्यकताओं के साथ फिट बैठता है अगर कोई ग्राहक हैं लगता है।

Answer 3

सामान्य रूप से, IObservable लागू नहीं करें; आम तौर पर आरएक्स में पहले से ही सोमिंग है जो सीधे या रचना के माध्यम से आपकी मदद कर सकती है। यदि आपको कभी भी IOervervable लागू करना है, तो पर्यवेक्षक का उपयोग करें। पर्यवेक्षक अनुबंध आदि के लिए आवश्यक सभी गारंटी प्राप्त करने के लिए,

आपकी समस्या के लिए - प्रकाशन और रेफकाउंट का उपयोग करने का सुझाव बिल्कुल रचना है आप देख रहे हैं। यदि आप किसी कारण से खुद को गिनना चाहते हैं, तो पर्यवेक्षण का उपयोग करें। संभवतः पर्यवेक्षण के साथ सदस्यता को अवरुद्ध करने के लिए देखें। अंततः अनुक्रम समाप्ति को रोकने के लिए। या, स्रोत को एक अवलोकन के साथ लपेटें। बनाएं, पर्यवेक्षक को लिपटे अनुक्रम में अग्रेषित करें, और तर्क को गिनती के साथ लौटाए गए IDISposable को लपेटें (Disposable.Create का उपयोग करके)।

चीयर्स,

-Bart (आरएक्स टीम) एक पुराने एक की

Answer 4

बिट, लेकिन मैं के रूप में मैं एक समस्या है जहाँ मैं ग्राहकों की संख्या पता करने के लिए आवश्यक था इस पोस्ट भर में आया था। बार्ट के सुझाव का उपयोग करके मैं इस विस्तार के साथ आया था।

public static IObservable<T> CountSubscribers<T>(this IObservable<T> source, Action<int> countChanged) 
{ 
int count = 0; 

return Observable.Defer(() => 
{ 
    count = Interlocked.Increment(ref count); 
    countChanged(count); 
    return source.Finally(() => 
    { 
     count = Interlocked.Decrement(ref count); 
     countChanged(count); 
    }); 
}); 
}