प्रतिक्रियाशील एक्सटेंशन: बफर जब तक कि ग्राहक निष्क्रिय नहीं होता

Question

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

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

मैंने Func<IObservable<TBufferClosing>> या IObservable<TBufferOpening> and Func<TBufferOpening, IObservable<TBufferClosing>> लेते हुए अधिक जटिल बफर ओवरलोड का उपयोग करने पर ध्यान दिया, लेकिन मुझे इनका उपयोग करने के उदाहरणों के बारे में उदाहरण नहीं मिल पा रहे हैं, जो मुझे करने की कोशिश कर रहे हैं, उन्हें लागू करने के तरीके के बारे में बहुत कम आंकड़े नहीं हैं।

Answer 1

क्या यह वही है जो आप चाहते हैं?

var xs = new Subject<int>(); 
var ys = new Subject<Unit>(); 

var zss = 
    xs.Buffer(ys); 

zss 
    .ObserveOn(Scheduler.Default) 
    .Subscribe(zs => 
    { 
     Thread.Sleep(1000); 
     Console.WriteLine(String.Join("-", zs)); 
     ys.OnNext(Unit.Default); 
    }); 

ys.OnNext(Unit.Default); 
xs.OnNext(1); 
Thread.Sleep(200); 
xs.OnNext(2); 
Thread.Sleep(600); 
xs.OnNext(3); 
Thread.Sleep(400); 
xs.OnNext(4); 
Thread.Sleep(300); 
xs.OnNext(5); 
Thread.Sleep(900); 
xs.OnNext(6); 
Thread.Sleep(100); 
xs.OnNext(7); 
Thread.Sleep(1000); 

मेरे परिणाम:

1-2-3 
4-5 
6-7 

Answer 2

आपको क्या करना होगा मूल्यों बफ़र होना और फिर जब कार्यकर्ता यह वर्तमान बफर के लिए पूछता है के लिए तैयार है कुछ है और फिर इसे रीसेट करता है। यह अपने कार्यकर्ता में

var tictac = new TicTac<double>(); 

IObservable<double> source = .... 

source.Subscribe(x=>tictac.Push(x)); 

फिर

class TicTac<Stuff> { 

    private TaskCompletionSource<List<Stuff>> Items = new TaskCompletionSource<List<Stuff>>(); 

    List<Stuff> in = new List<Stuff>(); 

    public void push(Stuff stuff){ 
     lock(this){ 
      if(in == null){ 
       in = new List<Stuff>(); 
       Items.SetResult(in); 
      } 
      in.Add(stuff); 
     } 
    } 

    private void reset(){ 
     lock(this){ 
      Items = new TaskCompletionSource<List<Stuff>>(); 
      in = null; 
     } 
    } 

    public async Task<List<Stuff>> Items(){ 
     List<Stuff> list = await Items.Task; 
     reset(); 
     return list; 
    } 
} 

तो RX और कार्य के संयोजन के साथ किया जा सकता है

while(true){ 

    var items = await tictac.Items(); 

    Thread.Sleep(100); 

    for each (item in items){ 
     Console.WriteLine(item); 
    } 

} 

Answer 3

तरह से मैं पहले यह किया है ऊपर खींचने के लिए है डॉटपीक/परावर्तक में विधि का निरीक्षण करें और उस कतार की अवधारणा को लें जो इसे हमारे आवश्यकताओं के अनुरूप है और अनुकूलित करता है। उदाहरण के लिए, तेजी से टिकने वाले डेटा (जैसे वित्त) के साथ यूआई अनुप्रयोगों में यूआई थ्रेड घटनाओं के साथ बाढ़ आ सकती है और कभी-कभी यह पर्याप्त तेज़ी से अद्यतन नहीं कर सकती है। इन मामलों में हम अंतिम घटना को छोड़कर सभी घटनाओं को छोड़ना चाहते हैं (किसी विशेष उपकरण के लिए)। इस मामले में हमने ObserveOn की आंतरिक कतार को टी के एक ही मान में बदल दिया (ObserveLatestOn (IScheduler) के लिए देखो)। आपके मामले में आप कतार चाहते हैं, हालांकि आप पूरी कतार को केवल पहले मान को धक्का नहीं देना चाहते हैं। इससे आप कार्य शुरू कर पाएंगे।