Reactive Extensions (Rx) - Probe mit letzten bekannten Wert, wenn kein Wert in vorhanden ist Intervall

Question

Ich habe einen beobachtbaren Strom, den Wert in inkonsistenten Abständen wie folgt erzeugt:

------1---2------3----------------4--------------5--- 

Und ich möchte dies probieren, aber ohne leere Proben, sobald der einen Wert erzeugt wurde:

------1---2------3----------------4--------------5----- 

----_----1----2----3----3----3----4----4----4----5----5 

ich dachte offensichtlich Replay().RefCount() hier verwendet werden könnte, den letzten bekannten Wert zu Sample() zu schaffen, aber da es an die Quelle nicht, es erneut abonnieren streamen didn Ich werde nicht trainieren.

Irgendwelche Gedanken darüber, wie ich das tun kann?

Answer 1

Quellstrom Angenommen ist IObservable<int> xs dann und Ihre Abtastintervall Timespan duration dann:

xs.Publish(ps => 
    Observable.Interval(duration) 
     .Zip(ps.MostRecent(0), (x,y) => y) 
     .SkipUntil(ps)) 

für eine generische Lösung, ersetzen den 0 Parameter MostRecent mit default(T) wo IObservable<T> ist der Quellenstrom-Typ.

Der Zweck Publish ist Abonnement Nebenwirkungen zu verhindern, da wir zweimal an die Quelle anmelden müssen - einmal für MostRecent und einmal für SkipUntil. Der Zweck der letzteren besteht darin, Abtastwerte bis zum ersten Ereignis des Quellenstroms zu verhindern.

Sie können dies vereinfachen, wenn Sie nicht vor dem Quellenstrom des ersten Ereignisse zu bekommen Standardwerte egal:

Observable.Interval(duration) 
    .Zip(xs.MostRecent(0), (x,y) => y) 

Ein verwandter Betreiber WithLatestFrom auch von Interesse sein könnte; Das kommt in der nächsten Version zu Rx. Details finden Sie unter here.