reset count above max time interval, in Rx count-based aggregation

Question

Count-based filtering without a time constraint

IObservable filteredStream = changes.Buffer(3);

How to introduce inactivity reset?

But how to introduce a timeout TimeSpan tooLong so that the counting would restart from zero whenever the interval between two values exceeds this maximum?

Answer 1

I think this is what you're after.

var longGap = source.Throttle(tooLong);
var filtered = source
  .Window(() => { return longGap; })  // Gives a window between every longGap
  .Select(io => io.Buffer(maxItems).Where(l => l.Count == maxItems))
  .Switch();  // Flattens the IObservable<IObservable<IList>> to IObservable<IList>