Buffer/throttle element drag events with RxJS

Question

I'd like to implement a scenario when certain things happen on a continuous source of events (dragging an element) - but with some buffering/throttling. I'd like to receive a notification let's say

• every 400 milliseconds
• but only if there were new items in the source (dragging actually happened)

My closest idea with the throttle operator outlined below just waits for a 400 ms pause and then provides the sequence - it does not provide a value on continuous drag:

Rx.Observable
      .fromEvent(element, "drag")
      .throttle(400);

I guess some source of a timer would be needed, but in that case how can I combine a timer source and a drag source with the aforementioned criteria?

Answer 1

Rx uses the notion of Schedulers (Rx.Scheduler instances to be specific). The throttle method takes an optional second argument which is the scheduler to use. If you do not supply a 2nd argument, then Rx.Scheduler.timeout is used. This scheduler uses setTimeout to schedule things in the future.

In your example, that means whenever a drag event occurs, throttle will store the event and will not tell you about it. Then it schedule an action 400ms from now (via the the scheduler, which ultimately means via setTimeout) to notify you of the event. If another drag event arrives before this timeout expires, then it will cancel the timeout and start a new one. This is because throttle will only notify you once the incoming events pause for at least 400ms. That means if you are dragging really fast, then you will not get any events until you finally slow down your dragging.

Judging by your description, you may actually prefer to use sample instead of throttle. Sample will give you an event every n ms if any event occurred during that interval, e.g.

Rx.Observable
    .interval(500)
    .sample(1500)
    .take(5)
    .subscribe(function (x) {
        console.log('x: ' + x);
    });

<script src='https://rawgit.com/Reactive-Extensions/RxJS/v.2.5.3/dist/rx.all.js'></script>

This will produce values:

"x: 1"
"x: 4"
"x: 7"
"x: 10"
"x: 13"

Where each value is an average of the intermittent value sum, i.e.:

• (0 + 1 + 2) / 3 = 1
• (3 + 4 + 5) / 3 = 4
• ...

You'd use it like so:

Rx.Observable
    .fromEvent(element, 'drag')
    .sample(400)
    .subscribe(function (e) {
        // ...
    });