Given a class:
class Foo { DateTime Timestamp {get; set;} }
...and an IObservable<Foo>
, with guaranteed monotonically increasing Timestamp
s, how can I generate an IObservable<IList<Foo>>
chunked into Lists based on those Timestamp
s?
I.e. each IList<Foo>
should have five seconds of events, or whatever. I know I can use Buffer
with a TimeSpan
overload, but I need to take the time from the events themselves, not the wall clock. (Unless there a clever way of providing an IScheduler
here which uses the IObservable
itself as the source of .Now
?)
If I try to use the Observable.Buffer(this IObservable<Foo> source, IObservable<Foo> bufferBoundaries)
overload like so:
IObservable<Foo> foos = //...;
var pub = foos.Publish();
var windows = pub.Select(x => new DateTime(
x.Ticks - x.Ticks % TimeSpan.FromSeconds(5).Ticks)).DistinctUntilChanged();
pub.Buffer(windows).Subscribe(x => t.Dump())); // linqpad
pub.Connect();
...then the IList
instances contain the item that causes the window to be closed, but I really want this item to go into the next window/buffer.
E.g. with timestamps [0, 1, 10, 11, 15]
you will get blocks of [[0], [1, 10], [11, 15]]
instead of [[0, 1], [10, 11], [15]]