Was ist der beste Weg, um einen Observable "Rate Limit" zu verbrauchen?

Question

Ich habe eine Reihe von Ereignissen kommen und ich muss alle von ihnen ohne Verlust ausführen, aber ich möchte sicherstellen, dass sie zu den entsprechenden Zeitfenstern gepuffert und konsumiert werden. Jeder hat eine Lösung?

Ich kann keine Operatoren in Rx finden, die das ohne den Verlust der Ereignisse tun können (Throttle - verliert Ereignisse). Ich habe auch über Pufferung, Verzögerung usw. nachgedacht. Kann keine gute Lösung finden.

Ich habe versucht, einen Timer in der Mitte zu setzen, aber irgendwie funktioniert es nicht funktionieren:

GetInitSequence() 
      .IntervalThrottle(TimeSpan.FromSeconds(5)) 
      .Subscribe(
       item => 
        { 
         Console.WriteLine(DateTime.Now); 
         // Process item 
        } 
      ); 

public static IObservable<T> IntervalThrottle<T>(this IObservable<T> source, TimeSpan dueTime) 
    { 
     return Observable.Create<T>(o => 
      { 
       return source.Subscribe(x => 
        { 
         new Timer(state => 
          o.OnNext((T)state), x, dueTime, TimeSpan.FromMilliseconds(-1)); 
        }, o.OnError, o.OnCompleted); 
     }); 
    } 

Answer 1

Die Frage ist nicht 100% klar ist, so dass ich einige Vermutungen mache.

Observable.Delay ist nicht das, was Sie wollen, denn das wird eine Verzögerung von dem Zeitpunkt an, zu dem jedes Ereignis eintrifft, erstellen, anstatt nur Zeitintervalle für die Verarbeitung zu erstellen.

Observable.Buffer ist nicht das, was Sie wollen, weil das dazu führt, dass alle Ereignisse in jedem gegebenen Intervall an Sie weitergegeben werden und nicht einzeln nacheinander.

Also ich glaube, dass Sie nach einer Lösung suchen, die eine Art Metronom erzeugt, die abhebt und Ihnen ein Ereignis pro Zecke gibt. Dies kann naiver mit Observable.Interval für das Metronom konstruiert und Zip dafür Ihre Quelle zu verbinden:

var source = GetInitSequence(); 
var trigger = Observable.Interval(TimeSpan.FromSeconds(5));  
var triggeredSource = source.Zip(trigger, (s,_) => s); 
triggeredSource.Subscribe(item => Console.WriteLine(DateTime.Now)); 

Diese alle 5 Sekunden löst (im Beispiel oben), und geben Sie die ursprünglichen Elemente der Reihe nach.

Das einzige Problem mit dieser Lösung ist, dass, wenn Sie keine Quellenelemente für (sagen wir) 10 Sekunden haben, wenn die Quellenelemente ankommen, werden sie sofort ausgesendet, da einige der 'Trigger'-Ereignisse sitzen Dort warten auf sie. Marmordiagramm für dieses Szenario:

source: -a-b-c----------------------d-e-f-g 
trigger: ----o----o----o----o----o----o----o 
result: ----a----b----c-------------d-e-f-g 

Dies ist ein sehr vernünftiges Problem.Es gibt zwei Fragen hier schon, dass es angehen:

Rx IObservable buffering to smooth out bursts of events

A way to push buffered events in even intervals

Die Lösung wird auch eine Haupt Drain Erweiterungsmethode und sekundäre Buffered Erweiterung. Ich habe diese geändert, um viel einfacher zu sein (keine Notwendigkeit für Drain, verwenden Sie einfach Concat). Die Nutzung ist:

var bufferedSource = source.StepInterval(TimeSpan.FromSeconds(5)); 

Die Erweiterung Methode StepInterval:

public static IObservable<T> StepInterval<T>(this IObservable<T> source, TimeSpan minDelay) 
{ 
    return source.Select(x => 
     Observable.Empty<T>() 
      .Delay(minDelay) 
      .StartWith(x) 
    ).Concat(); 
} 

Answer 2

Wie wäre es Observable.Buffer? Dies sollte alle Ereignisse im 1s-Fenster als einzelnes Ereignis zurückgeben.

var xs = Observable.Interval(TimeSpan.FromMilliseconds(100)); 
var bufferdStream = xs.Buffer(TimeSpan.FromSeconds(5)); 
bufferdStream.Subscribe(item => { Console.WriteLine("Number of events in window: {0}", item.Count); }); 

Es könnte sein, was Sie fragen, ist nicht klar. Was soll dein Code tun? Es sieht so aus, als würden Sie nur zögern, indem Sie für jedes Ereignis einen Timer erstellen. Es bricht auch die Semantik des Observablen, da das Nächste und das Letzte vor dem nächsten auftreten können.

Beachten Sie, dass dies auch nur bei der verwendeten Zeitschaltuhr so ​​genau ist. Typischerweise sind die Timer auf maximal 16ms genau.

Edit:

Ihr Beispiel wird, und Artikel über alle Ereignisse im Fenster:

GetInitSequence() 
      .Buffer(TimeSpan.FromSeconds(5)) 
      .Subscribe(
       item => 
        { 
         Console.WriteLine(DateTime.Now); 
         // Process item 
        } 
      ); 

Answer 3

Ich weiß, das ist einfach zu einfach sein könnte, aber würde das funktionieren?

var intervaled = source.Do(x => { Thread.Sleep(100); }); 

Im Grunde legt dies nur eine minimale Verzögerung zwischen den Werten. Zu einfach?

Answer 4

Nach dem Vorbild der Enigmativity Antwort, wenn alles, was Sie tun wollen, nur Verzögerung von einem Timespan alle Werte ist, ich kann nicht sehen, warum Delay nicht der Betreiber wollen Sie

GetInitSequence() 
     .Delay(TimeSpan.FromSeconds(5)) //ideally pass an IScheduler here 
     .Subscribe(
      item => 
       { 
        Console.WriteLine(DateTime.Now); 
        // Process item 
       } 
     );