आरएक्स बफर ग्राहक को खाली कॉल के बिना

Question

.NET 4.6 का उपयोग कर मेरे डब्ल्यूपीएफ एप्लिकेशन में मेरे पास एक ऐसा ईवेंट है जो उच्च दर (कई सौ प्रति सेकेंड) पर नए डेटा पॉइंट चलाता है, लेकिन हर समय नहीं। यह डेटा एक चार्ट में प्रदर्शित होता है।

मैं प्रत्येक 50 एमएस चार्ट को अपडेट करना चाहता हूं और प्रत्येक नए डेटा बिंदु के बाद नहीं।
यह प्राप्त करने के लिए कि मैं आरएक्स से Buffer(TimeSpan.FromMilliseconds(50)) का उपयोग करने के बावजूद, सिद्धांत में ठीक काम करता है। लेकिन मेरे ग्राहक को हर 50 एमएस भी कहा जाता है यदि कोई नया डेटा पॉइंट नहीं बनाया गया है जो कि मैं बिल्कुल नहीं चाहता हूं।

मुझे लगता है कि बाहर का परीक्षण करने के लिए एक छोटे नमूना आवेदन बनाया: https://dotnetfiddle.net/TV5tD4

वहाँ:

using System; 
using System.Reactive.Linq; 

namespace RxTester 
{ 
    public class Program 
    { 
     private static event EventHandler TheEvent; 

     static void Main(string[] args) 
     { 
      var observable = Observable.FromEvent<EventHandler, EventArgs>(h => (s, e) => h(e), h => TheEvent += h, h => TheEvent -= h); 
      var subscriber = observable.Buffer(TimeSpan.FromMilliseconds(1000)) 
       .Subscribe(e => Console.WriteLine($"{DateTime.Now.ToLongTimeString()}: {e.Count} elements received...")); 

      var random = new Random(); 
      var timer = new System.Timers.Timer(2000) 
       { 
        AutoReset = true, 
        Enabled = true 
       }; 
      timer.Elapsed += (s, e) => 
       { 
        var amount = random.Next(1, 10); 
        for (int i = 0; i < amount; ++i) 
         TheEvent?.Invoke(null, null); 
       }; 

      Console.ReadLine(); 

      timer.Enabled = false; 
      subscriber.Dispose(); 
     } 
    } 
} 

इसे चलाने के लिए करने के लिए या का उपयोग निम्नलिखित फिडल आप "आरएक्स-Linq" NuGet पैकेज जोड़ने की जरूरत आप कई "0 तत्व प्राप्त हुए" देखते हैं जो मैं टालना चाहता हूं। मुझे पता है कि मैं e.Count == 0 के लिए सरल जांच कर सकता हूं, लेकिन जैसे ही मैं ऐसे बफरों का उपयोग करता हूं, यह मेरे लिए अनुकूल नहीं लगता है।

तत्व उपलब्ध होने पर केवल तत्व के नए buffered ब्लॉक बनाने का कोई तरीका है?
मैं समय-समय पर बैचिंग घटनाओं की अपनी समस्या को हल करने के लिए अन्य दृष्टिकोणों के लिए भी खुला हूं - मैंने पहले ही टीपीएल डेटाफ्लो BatchBlock में देखा है, लेकिन ऐसा लगता है कि केवल गिनती आधारित ब्लॉक आकारों का समर्थन करता है।

Answer 1

एक बार फिर हम इस विस्तार

public static IObservable<IList<TSource>> BufferWhenAvailable<TSource> 
              (this IObservable<TSource> source, 
              TimeSpan threshold) 
{ 
    return source.Publish(sp => 
        sp.GroupByUntil(_ => true, _ => Observable.Timer(threshold)) 
         .SelectMany(i => i.ToList())); 

} 

Answer 2

ऐसा करने का मानक तरीका बनाने के लिए शक्तिशाली GroupByUntil विधि का उपयोग कर सकते हैं बस

.Buffer(period) 
.Where(buffer=>buffer.Any()) 

तो प्रभावी ढंग से कर रही है कि तुम क्या (count==0) से बचने के लिए चाहते हैं। हालांकि, यह चेक बहुत सस्ता है और मैं कल्पना करता हूं कि यदि अन्य लागत शामिल है तो शेड्यूलिंग से कहीं ज्यादा सस्ता है। एकमात्र चिंता हो रही आवंटन राशि (प्रत्येक 50 एमएमएस List<T>) और उसके बाद आने वाले जीसी जेन 0 दबाव का निर्माण हो सकता है।