Reactive Extensions (Rx) - próbka z ostatniej wartości wiadomo, kiedy żadna wartość jest obecny w przedział

Question

Mam zaobserwowania strumień, który wytwarza wartości na niespójnych odstępach tak:

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

i chciałbym aby spróbować tego, ale bez pustych próbek gdy wartość zostało ono wyprodukowane:

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

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

I oczywiście myśleli Replay().RefCount() mogłyby zostać wykorzystane tutaj, aby zapewnić ostatnią wartość znane Sample() ale ponieważ nie ponownie zapisać się do źródła strumienia to nie zrobił nie wyszło.

Jakieś przemyślenia na temat tego, w jaki sposób mogę to zrobić?

Answer 1

Zakładając, że strumień źródłowy jest IObservable<int> xs a następnie interwał próbkowania wynosi Timespan duration następnie:

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

Dla ogólnego rozwiązania, zastąpić parametr MostRecent0 z default(T) gdzie IObservable<T> jest rodzaj strumienia źródła.

Celem Publish jest zapobieganie efektom ubocznym subskrypcji, ponieważ musimy dwukrotnie zasubskrybować źródło - raz dla MostRecent i raz dla SkipUntil. Celem tego ostatniego jest zapobieganie próbkowaniu wartości aż do pierwszego zdarzenia strumienia źródłowego.

Można uprościć tę opcję, jeśli nie dbają o uzyskanie wartości domyślne przed strumienia materiałów wsadowych w pierwszym przypadku:

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

Podobnym operator WithLatestFrom może być również przedmiotem zainteresowania; to zbliża się do Rx w następnym wydaniu. Szczegóły: here.