Rx IObservable produce value only when value has changed by a certain margin

Question

I have a feeling this might be a very simple Extension method that I have missed but I can't see to find it...

I basically want to take a stream that produces a stream where the value slowly incrementing upon each new value. I want to Throttle/Sample this, not by time, but by "tolerance". e.g.

var ob = Enumerable.Range(0, 30).ToObservable(); // 0, 1, 2, 3, 4, 5,....., 30
var largeMovingOb = ob.WhenChangedBy(10); // 0, 10, 20, 30

when I have sequences such as [1, 4, 20, 33] and I want to output when value has changed by more than 15 of the last one - which would result in: [1, 20]. Where as a change by value of 12 would result in: [1, 20, 33]

Is there a built-in Rx extension for this? Ideally it would work on all numeric types without writing an overload for each.

Answer 1

I think this is a good fit for Observable.Scan

var ob = Enumerable.Range(0, 30).ToObservable();
var largeMovingOb = ob.Scan((acc, i) => acc + 10 > i ? acc : i)
  .DistinctUntilChanged();