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?

Drainage

I believe you are looking for debounce. – Undertenant

No, debounce sends the value AFTER the time interval, I want the value sent before and then to drop all values that aren't longer than the duration after the first value has been sent. – Drainage

From the docs: Emits an item from the source Observable after a particular timespan has passed without the Observable omitting any other items. --OR-- Ignores values from an observable sequence which are followed by another value within a computed debounced duration. – Drainage

So debounce will receive a value, wait for the duration to elapse, then forward that value and otherwise drop it OR it will ignore a value that has values sent after it after the computed duration. I don't want to ignore that value, I want to send that first value and skip all others that don't have the interval between the nth-1 and nth. – Drainage

@Undertenant in the example I gave, using debounce, 0 would never get sent, only 2 would because the delay between 2 and 3 is greater than the interval provided of 500ms. – Drainage

Ah, I see. throttle sounds more like it then. – Undertenant

It's also not throttle. – Drainage

Throttle will send the first value in the time interval then wait until the next time interval and send the first received value and so on. So in my example stream, using throttle would result in an output of 0, 2, 3, not 0 3 because throttle looks at the accumulated time difference from the last emitted value to the current value. – Drainage

So for example, using throttle, 0 will be send then 200ms elapses, 1 is evaluated and fails, then 400ms elapses, and 2 is evaluated and passes because the duration between the last emitted value (0) and 2, is 600ms, which is >= 500ms. What I'm describing should evaluate the time elapse from the last value received, not the last value emitted, if that makes sense. Ill clarify more in my post. – Drainage

have a look at #37146581 – Congener

I should have specified, I'm using github.com/Reactive-Extensions/RxJS not github.com/ReactiveX/rxjs because I'm not going to integrate a beta library into my production code. – Drainage

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)})

Congener

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);

Drainage

more refined in what sense? I don't see so much difference. I don't need to use skip as the concat does it already. Apart from that, it is all the same – Congener