Rx Tampon benim WPF uygulamasında

Question

her zaman ben yüksek oranda (saniye başına birkaç yüz) yeni veri noktaları patlar bir olay .Net 4.6 kullanarak değil, abonede. Bu veriler bir grafikte görüntülenir.

Ben grafiği her 50 ms güncelleme ve sonra değil her yeni veri noktası istiyoruz.
Bunu başarmak için, teoride iyi çalışan Rx'den Buffer(TimeSpan.FromMilliseconds(50))'u kullanıyorum. ANCAK Aboneye ayrıca istediğim gibi olmayan yeni veri noktaları oluşturulmadığında her 50 ms'de bir çağrılır. Orada https://dotnetfiddle.net/TV5tD4

: aşağıdaki Fiddle çalıştırmak veya kullanmak için Sen "Rx-Linq" Nuget paketi eklemeniz gerekir

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(); 
     } 
    } 
} 

:

Bunu test etmek biraz örnek uygulama yarattı kaçınmak istediğim birkaç "0 öğe alındı" görüyorsunuz. e.Count == 0 için basit bir kontrol yapabileceğimi biliyorum, ancak bu tamponların çoğunu kullandığım için bu bana uygun görünmüyor.

Öğeler mevcutsa yalnızca yeni arabellek blokları oluşturmanın bir yolu var mı?
Ayrıca, zaman zaman toplu iş olayları sorunumu çözmek için diğer yaklaşımlar için açık - Ben zaten TPL Dataflows BatchBlock içine baktım, ancak bu yalnızca sayma tabanlı blok boyutlarını destekliyor gibi görünüyor.

Answer 1

Bir kez daha bu uzantıyı

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

Bunu yapmanın standart bir yol oluşturmak için güçlü GroupByUntil yöntemi kullanabilirsiniz basitçe

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

Yani etkili bir (count==0) kaçınmak istediğini yapıyor. Ancak, bu kontrol çok ucuzdur ve eğer diğer maliyetlerden daha ucuzsa hayal edeceğim. Tek endişe, gerçekleşen miktar tahsisleri olabilir (her 50 ms'de bir List<T> oluşturur) ve daha sonra inşa edebilecek olan yaklaşan GC Gen0 basıncı olabilir.