RxJS take first then throttle and wait

Question

I want to observer the mousewheel event using RxJS-DOM so that when the first event fires, I forward that on and then drop any and all values until the delay between subsequent values passes a previously specified duration.

The operator I'm imagining might look something like:

Rx.DOM.fromEvent(window, 'mousewheel', (e) => e.deltaY)
.timegate(500 /* ms */)

Imagine the following data stream:

0 - (200 ms) - 1 - (400ms) - 2 - (600ms) - 3

where the value being sent is the number and the time describes how long the next value takes to arrive. Since 0 is the first value, it would be emitted and then all values until 3 would be dropped because the individual delays between subsequent values are not greater than 500ms.

Unlike throttle, the time delay between values is calculated whether or not the last received value is emitted or not. With throttle, 0 would be send, 200 ms would elapse, 1 would be evaluated and fail, 400 ms would elapse and 2 would be evaluated and PASS because the time elapse between the last emitted value (0) and the currently received one (2) is 600 ms whereas with my operator, it would evaluate relative to the 1 and the time elapse would be 400 ms, therefore failing the test.

And this operator isn't debounce either. Instead of waiting until the interval has elapsed to emit, it first sends the first value then evaluates against all future values and so on.

Does an operator like this already exist? And if not, how would I go about making one?

Answer 1

You can achieve this relatively easily using the timeInterval operator which computes precisely the time interval between successive values. Here is a sample code you can adapt to your guise.

http://jsfiddle.net/a7uusL6t/

var Xms = 500;
var click$ = Rx.Observable.fromEvent(document, 'click').timeInterval()
var firstClick$ = click$.first().map(function(x){return x.value});

var res$ = firstClick$
  .concat(click$
    .filter(function (x) {
      console.log('interval', x.interval);
      return x.interval > Xms;})
    .map(function(x){return x.value})
   );

res$.subscribe(function(x){console.log(x)})

Answer 2

I solved my problem with something similar but more refined than @user3743222's answer:

const events = Rx.Observable.fromEvent(window, 'mousewheel', (e) => e.deltaY);
const firstEventObservable = events.take(1);
const remainingEventsObservable = events.skip(1)
    .timeInterval()
    .filter(x => x.interval >= this.props.delayDuration)
    .map(x => x.value);
const pageChangeObservable = firstEventObservable.concat(remainingEventsObservable);