CODEWITHSUNDEEP
c#.netlinqsystem.reactive
Cel
Cel

Reputation: 6689

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

Count-based filtering without a time constraint

IObservable filteredStream = changes.Buffer(3);

How to introduce inactivity reset?

enter image description here

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?

Upvotes: 10

Views: 471

Answers (1)

Matthew Finlay
Matthew Finlay

Reputation: 3464

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>

Upvotes: 3

Related Questions