Reaktive Erweiterungen: Drossel/Probe mit variierendem Intervall

Question

Ich habe eine IObservable, die Werte in zufälligen Intervallen produziert, und ich möchte diese Sequenz drosseln. Eine Sache, die ich herausgefunden habe, ist, dass dieDefinition von "Drosselung" nicht die gleiche wie meine ist.

Throttle produziert nur Werte nach Ablauf des angegebenen Intervalls mit Stille (es erzeugt den zuletzt gesehenen Wert). Ich dachte Drosselung würde bedeuten, Werte in dem angegebenen Intervall zu produzieren (es sei denn, es gibt natürlich Stille).

Say, ich erwartete Observable.Interval(100).Select((_,i) => i).Throttle(200) zu produzieren (modulo irgendwelche Performance/Timing-Probleme) die geraden Zahlen, da ich es auf "Halbgeschwindigkeit" Drosselung bin. Diese Sequenz erzeugt jedoch überhaupt keinen Wert, da es nie eine Ruheperiode von 200 gibt.

So entdeckte ich, dass Sample tatsächlich das "Drosselungs" -Verhalten macht, das ich möchte. Observable.Interval(100).Select((_,i) => i).Sample(200) erzeugt (wieder, modulo irgendwelche Performance/Timing-Probleme) die Folge der geraden Zahlen.

Allerdings habe ich ein anderes Problem: das Intervall variiert, abhängig vom letzten "abgetasteten" Wert. Was ich will, ist einen Operator zu schreiben, die wie folgt aussieht:

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector); 

Der intervalSelector Parameter das Intervall für die nächste Probe erzeugt, und die erste Probe ... wird entweder auf dem ersten Wert oder von einem zusätzlichen Parameter genommen Es ist mir egal.

Ich habe versucht, dies zu schreiben, aber ich endete mit einer großen gewundenen Konstruktion, die nicht ganz richtig funktionierte. Meine Frage ist, kann ich dies mit den vorhandenen Operatoren (aka, mit einem Einzeiler) bauen?

Answer 1

Viele Stunden später, und mit ein wenig Schlaf, habe ich es. Diese

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector) 
{ 
    return source.TimeInterval() 
       .Scan(Tuple.Create(TimeSpan.Zero, false, default(T)), (acc, v) => 
       { 
        if(v.Interval >= acc.Item1) 
        { 
         return Tuple.Create(intervalSelector(v.Value), true, v.Value); 
        } 
        return Tuple.Create(acc.Item1 - v.Interval, false, v.Value); 
       }) 
       .Where(t => t.Item2) 
       .Select(x => x.Item3); 
} 

funktioniert, wie ich will: jedes Mal erzeugt einen Wert x, stoppt er Werte bis intervalSelector(x) Zeit vergeht zu erzeugen.

Answer 2

Nicht das, was Sie suchen for Observable.BufferWithTime?