Reactive Extensions (Rx) - sample with last known value when no value is present in interval

Question

I have an observable stream that produces values at inconsistent intervals like this:

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

And I would like to sample this but without any empty samples once the a value has been produced:

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

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

I obviously thought Replay().RefCount() could be used here to provide the last known value to Sample() but as it doesn't re-subscribe to the source stream it didn't work out.

Any thoughts on how I can do this?

Answer 1

Assuming your source stream is IObservable<int> xs then and your sampling interval is Timespan duration then:

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

For a generic solution, replace the 0 parameter to MostRecent with default(T) where IObservable<T> is the source stream type.

The purpose of Publish is to prevent subscription side effects since we need to subscribe to the source twice - once for MostRecent and once for SkipUntil. The purpose of the latter is to prevent sampling values until the source stream's first event.

You can simplify this if you don't care about getting default values before the source stream's first event:

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

A related operator WithLatestFrom might also be of interest; this is coming to Rx in the next release. See here for details.