आरएक्स और कार्य - नया कार्य शुरू होने पर चल रहे कार्य को रद्द करें?

Question

मेरे पास एक उपयोगकर्ता इंटरैक्शन परिदृश्य है जिसे मैं आरएक्स के साथ संभालना चाहता हूं।

परिदृश्य विहित के समान है "जब उपयोगकर्ता के टाइपिंग बंद कुछ काम करना" (आम तौर पर, क्या उपयोगकर्ता द्वारा अब तक टाइप किया है के लिए खोज) (1) - लेकिन मैं यह भी करने की जरूरत है:

• (2) केवल नवीनतम "ऐसा कुछ काम" इकाइयों के परिणामों की मिल (नीचे देखें)
• (3) जब काम की एक नई इकाई शुरू होता है, रद्द प्रगति में किसी भी काम (मेरे मामले में यह सीपीयू है गहन)

के लिए (1) मैं उपयोगकर्ता घटनाओं के लिए IObservable का उपयोग करता हूं, .Throttle() के साथ थ्रॉटल किया गया ताकि केवल ईवेंट ("उपयोगकर्ता टाइपिंग रोकता है") के बीच विरामों को ट्रिगर कर सके।

उस से, मैं .Select(_ => CreateMyTask(...).ToObservable())।

यह मुझे IObservable<IObservable<T>> देता है जहां प्रत्येक आंतरिक अवलोकन एक ही कार्य को लपेटता है।

प्राप्त करने के लिए (2) अंततः मैं काम की नवीनतम इकाई से परिणाम प्राप्त करने के लिए .Switch() लागू करता हूं।

(3) - लंबित कार्यों को रद्द करने के बारे में क्या?

पिछले एक (रों) से

अगर मैं सही ढंग से समझ, जब भी वहाँ एक नया आंतरिक IObservable<T> है, .Switch() विधि यह की सदस्यता और सदस्यता समाप्त करने पर, उन्हें Dispose() के कारण।
शायद इसे रद्द करने के लिए कार्य को ट्रिगर करने के लिए किसी भी तरह से तारित किया जा सकता है?

Answer 1

क्या आपको कार्य के साथ काम करना है?

यदि आप पूरी तरह से पर्यवेक्षकों के साथ काम करने में प्रसन्न हैं तो आप इसे अच्छी तरह से कर सकते हैं।

कुछ इस तरह कर रही है कोशिश:

var query = 
    Observable.Create<int>(o => 
    { 
     var cancelling = false; 
     var cancel = Disposable.Create(() => 
     { 
      cancelling = true; 
     }); 
     var subscription = Observable.Start(() => 
     { 
      for (var i = 0; i < 100; i++) 
      { 
       Thread.Sleep(10); //1000 ms in total 
       if (cancelling) 
       { 
        Console.WriteLine("Cancelled on {0}", i); 
        return -1; 
       } 
      } 
      Console.WriteLine("Done"); 
      return 42; 
     }).Subscribe(o); 
     return new CompositeDisposable(cancel, subscription); 
    }); 

यह नमूदार Thread.Sleep(10); साथ पाश के लिए कुछ कड़ी मेहनत कर रहा है, लेकिन जब नमूदार निपटान किया जाता है पाश से बाहर निकल गया है और गहन सीपीयू काम रहता है। फिर आप प्रगति कार्य को रद्द करने के लिए Switch के साथ मानक Rx Dispose का उपयोग कर सकते हैं।

आपको अपनी पसंद की विधि में बंडल कि चाहते हैं, तो इस कोशिश:

public static IObservable<T> Start<T>(Func<Func<bool>, T> work) 
{ 
    return Observable.Create<T>(o => 
    { 
     var cancelling = false; 
     var cancel = Disposable 
      .Create(() => cancelling = true); 
     var subscription = Observable 
      .Start(() => work(() => cancelling)) 
      .Subscribe(o); 
     return new CompositeDisposable(cancel, subscription); 
    }); 
} 

और फिर इस तरह एक समारोह से कॉल करने की:

Func<Func<bool>, int> work = cancelling => 
{ 
    for (var i = 0; i < 100; i++) 
    { 
     Thread.Sleep(10); //1000 ms in total 
     if (cancelling()) 
     { 
      Console.WriteLine("Cancelled on {0}", i); 
      return -1; 
     } 
    } 
    Console.WriteLine("Done"); 
    return 42; 
}; 

यहाँ मेरी कोड है कि इस से साबित कर दिया है काम किया:

var disposable = 
    ObservableEx 
     .Start(work) 
     .Subscribe(x => Console.WriteLine(x)); 

Thread.Sleep(500); 
disposable.Dispose(); 

मुझे अपने आउटपुट के रूप में "50 पर रद्द किया गया" (कभी-कभी "51 पर रद्द किया गया") मिला।

Answer 2

तुम बस Observable.FromAsync उपयोग कर सकते हैं जो टोकन कि रद्द कर दिया उत्पन्न होगा जब पर्यवेक्षक unsubcribes:

input.Throttle(...) 
    .Select(_ => Observable.FromAsync(token => CreateMyTask(..., token))) 
    .Switch() 
    .Subscribe(...); 

इस काम की प्रत्येक इकाई के लिए एक नया टोकन बनाएं और यह नया करने के लिए हर बार रद्द Switch स्विच होगा ।