RXJS - start timer only in standby mode?

Question

I use a stream that is throttled when I view the window.
When throttling (before scrolling), it issues values ​​to the console.

However, when the stream is inactive (the user does not scroll the window) - I want the timer to set. However - if the user starts scrolling again - I do not want this timer to display values.

I am currently doing this:

const observable = Rx.Observable.fromEvent(window, 'scroll'); const subscriber = observable .throttleTime(300 ) .map(() => 'throttle') .merge(Rx.Observable.interval(1000).map(() => 'tick') ) .subscribe( (x) => { console.log('Next: event!', x); }, (err) => { console.log('Error: %s', err); }, () => { console.log('Completed'); }); 

The problem is that when scrolling - I see as "throttle" AND "tick" (I should see only the "throttle")

Think about it from another POV. Work must always be done. If I scroll - this throttle scroll - should trigger a job. If I do not scroll, the timer should begin work and begin work. (and stops if the user starts scrolling again).

Question:
How to start the timer after idle without scrolling?

Plnkr

Answer 1

I would do it like this:

 const scroll$ = Rx.Observable.fromEvent(window, 'scroll') .throttleTime(300 /* ms */) .publish(); scroll$.connect(); const subscriber = scroll$ .map(() => 'throttle') .race(Rx.Observable.interval(1000).map(() => 'tick')) .take(1) .repeat() .subscribe( (x) => { console.log('Next: event!', x); }, (err) => { console.log('Error: %s', err); }, () => { console.log('Completed'); }); 

This uses the race() operator to subscribe only to an Observable, which emits the first, which is a 1s interval or scroll event. Right after that, I want to run this again at a different interval, so I use take(1).repeat() .

I also had to include scroll$ Observable in the Hot Observable to keep throttleTime() between recurring subscriptions.

Your updated demo: https://plnkr.co/edit/sWzSm32uoOQ1hOKigo4s?p=preview

Answer 2

You can use debounceTime to define periods without scrolling.

 const scroll = Rx.Observable.fromEvent(window, 'scroll') .throttleTime(300) .mapTo(false); const noscroll = Rx.Observable.fromEvent(window, 'scroll') .startWith(0) // init with no scroll. .debounceTime(300) // detect no scroll after 300 ms. .mapTo(true); scroll.merge(noscroll) .switchMap(e => e ? Rx.Observable.interval(1000).mapTo("Tick!") : Rx.Observable.of("Scroll!")) // start the interval if there was no scroll. Stop the interval if there was a scroll. .subscribe(updateTimer) 

Another problem with your code is to use merge , which will support both sources on the subscription, instead I use switchMap (sibling mergeMap ), which will subscribe to the internal observable every time a new event is switchMap , but also unsubscribe the previous internal source. if another event is selected from the source.

Re: "another POV" part of the question: you can replace Rx.Observable.interval(1000) in switchMap with a job. Scrolling cancels / unsubscribes from the task (when leaving empty ), if there is no more scrolling, the task will start again.

Live demo