आरएक्स फ्रेमवर्क: मूल नमूदार अनुक्रम

Question

एक नमूदार स्रोत को देखते हुए मतदान हुए निम्न-स्तर डिवाइस के राज्य (एक के परिवर्तन) द्वारा उत्पन्न को बाधित किए बिना समय समाप्ति पर एक कार्रवाई को अंजाम ...

// observable source metacode: 
IObservable<DeviceState> source = Observable.Interval(TimeSpan.FromSeconds(0.5)) 
    .Select(tick => new DeviceState(_device.ReadValue())) 
    .DistinctUntilChanged(); 

.. । और एक उपभोक्ता है कि यूआई अद्यतन करता है ...

// UI metacode: 
service.GetObservableDeviceStates() 
    .Subscribe(state => viewModel.CurrentState = state.ToString()); 

... मैं स्रोत के लिए सदस्यता को बाधित किए बिना, स्रोत की 'निष्क्रियता' के एक्स सेकंड के बाद एक कस्टम क्रिया को निष्पादित करने की जरूरत है। कुछ ऐसा:

// UI metacode: 
service.GetObservableDeviceStates() 
    .DoOnTimeout(TimeSpan.FromSeconds(x),() => viewModel.CurrentState = "Idle") 
    .Subscribe(state => viewModel.CurrentState = state.ToString()); 

सर्वोत्तम प्रथाओं क्या हैं? संभावित समाधान जो मन में आ रहे हैं (मैं एक आरएक्स noob हूँ):

1. Buffer (भले ही वह इतना पठनीय नहीं है)
2. आसपास this Timeout overload बजाना;

3. कुछ खास "सेवा-साइड" जब कुछ भी नहीं (DistinctUntilChanged उपयोग करने के बजाय) बदल जाता रिटर्निंग और यूआई कोड पर यह के साथ काम:

   service.GetObservableDeviceStates() .Subscribe (राज्य => viewModel.CurrentState = राज्य। विशेष? "निष्क्रिय": राज्य.ToString());

संपादित करें: के रूप में in the answer सूचना दी, समाधान है:

 service.GetObservableDeviceStates() 
      .Do(onNext) 
      .Throttle(TimeSpan.FromSeconds(x)) 
      .Subscribe(onTimeout); 

EDIT2 (चेतावनी)

तो onNext और onTimeout अद्यतन UI घटक, CrossThreadExceptions से बचने के लिए दो ObserveOn (uiSynchronizationContext) जरूरी है, क्योंकि थ्रॉटल एक और थ्रेड पर काम करता है!

 service.GetObservableDeviceStates() 
      .ObserveOn(uiSynchronizationContext) 
      .Do(onNext) 
      .Throttle(TimeSpan.FromSeconds(x)) 
      .ObserveOn(uiSynchronizationContext) 
      .Subscribe(onTimeout); 

Answer 1

समय समाप्त कम या ज्यादा observables जो एक अतुल्यकालिक संचालन का प्रतिनिधित्व करने के लिए है -, या जैसे के लिए एक डिफ़ॉल्ट मान देने के लिए OnError अगर नमूदार समय की एक निश्चित राशि में आपको सूचित नहीं किया गया है ने कहा।

ऑपरेटर जो आप खोज रहे हैं Throttle है, भले ही यह पहले ऐसा प्रतीत न हो। Throttle(p) आपको एक स्ट्रीम देता है जो एक स्रोत बनाता है जब स्रोत स्ट्रीम ने p अवधि के लिए कोई मान नहीं बनाया है।

अपने मौजूदा कोड के समानांतर, आप source.Throttle(period).Do(...side effect) का उपयोग कर सकते हैं।

Answer 2

मैं व्यक्तिगत रूप से इसके लिए विधि विधि से बचूंगा। यह इस उदाहरण में कोड को काफी आसान बनाता है, लेकिन मुझे लगता है कि 'डू' का उपयोग कोड बेस में आपके पास जल्द ही स्पेगेटी होता है।

आप जो परिणाम ढूंढ रहे हैं और अभी भी मोनाड * को बनाए रखने के लिए आप Amb, Timer, TakeUntil, Throttle आदि के संयोजनों का उपयोग करने पर विचार कर सकते हैं। या सरल शब्दों में, मुझे लगता है कि आप आदर्श रूप से आने वाले स्टेटस मानों का अनुक्रम चाहते हैं और आपके कोड में टाइमर डालने की आवश्यकता नहीं है (यानी इसे सेवा में लोड करें)।

public IObservable<DeviceStatus> GetObservableDeviceStates(TimeSpan silencePeriod) 
{ 
    return Observable.Create<DeviceStatus>(
    o=> 
    { 
     var idle = Observable.Timer(silencePeriod).Select(_=>new DeviceStatus("Idle")); 

     var polledStatus = Observable.Interval(TimeSpan.FromSeconds(0.5)) 
         .Select(tick => new DeviceStatus(_device.ReadValue())) 
         .DistinctUntilChanged() 
         .Publish(); 

     var subscription = (from status in polledStatus 
          from cont in Observable.Return(status).Concat(idle.TakeUntil(polledStatus)) 
          select cont) 
        .Subscribe(o); 

     return new CompositeDisposable(subscription, polledStatus.Connect()); 
    }); 
} 

इस कोड को अब सेवा एक निष्क्रिय स्थिति मान लौट रहा एक बार परिवर्तन चुप्पी की निर्दिष्ट अवधि उत्पन्न हुई है।

यह आपके यूआई मेटा कोड सरल रहता है और DeviceStatus से संबंधित तर्क रहता है जहां से वह संबंधित

// UI metacode: 
service.GetObservableDeviceStates(TimeSpan.FromSeconds(2)) 
    .Subscribe(state => viewModel.CurrentState = state.ToString());