RxJS: Change throttleTime after X emissions

Question

I'm building an RxJS slideshow where, if user holds the right arrow key, I want to navigate to the next tile at every 500 ms. I'm using throttleTime like below:

const forwardNavigation$ = fromEvent(document, 'keydown').pipe(
    filter(event => event.keyCode === KEY_CODE.arrowRight),
    throttleTime(500)
);

What I would like to do now is to reduce the throttleTime to 100ms after I have navigated to the 5th tile without releasing the arrow key.

Is that possible, how would one go about implementing that?

District

Answer 1

If you want a more declarative approach using only Observables, the following will work as well:

const source$ = fromEvent(document, 'keydown').pipe(
  filter((event: KeyboardEvent) => event.code === "ArrowRight")
);
const slow$ = source$.pipe(throttleTime(500), take(5));
const fast$ = source$.pipe(throttleTime(100));

const forwardNavigation$ = concat(slow$, fast$);
// subscribe to forwardNavigation$ and execute navigation code

What this does is create two Observables from the keyboard events, with different throttleTimes. It then uses concat to merge the results of both, which will ignore the second until the first completes. Adding take(5) to the first observable means it completes after 5 emissions, at which point the fast$ Observable takes over.

Working StackBlitz.

Kilometer

Thanks, that's precisely what I was looking for! Being a beginner in RxJs I couldn't even imagine how to solve this problem, your code and explanation gave me great insights into it! – District

By the way, since you mention you are new to rxjs, this pattern of splitting a source Observable, doing different operations on each “branch” and then merging the results is one I have used over and over. Here is a good article explaining this pattern. – Kilometer

Answer 2

You may use throttle operator and return appropriate interval observable instead. Refer to the following example:

import {fromEvent, interval} from 'rxjs';
import {filter, throttle} from 'rxjs/operators';

let iteration = 1;

const forwardNavigation$ = fromEvent(document, 'keydown').pipe(
    filter(event => event.keyCode === 13),
    throttle(() => iteration++ % 5 === 0 ? interval(100) : interval(500))
);

forwardNavigation$.subscribe(console.log)

A working demo can be found here. Observe the console while pressing the Enter key.

Gilroy

Thank you, that is helpful! I am just wondering, is it possible to avoid the global var iteration? – District

iteration here essentially points to the number of current slide, and 5 is the total number of slides. If you can devise those with another set of variables, it should also work. – Gilroy

Answer 3

For the sake of completeness, here's the final code based on @dmcgrandle answer:

import { fromEvent, concat } from 'rxjs';
import { filter, throttleTime, take, skip, startWith, switchMap } from 'rxjs/operators';

const keyboardRightKey$ = fromEvent(document, 'keydown').pipe(
    filter(e => e.keyCode === 39)
);

fromEvent(document, 'keyup').pipe(
    startWith(true),
    switchMap(() => concat(
      keyboardRightKey$.pipe(throttleTime(500), take(5)),
      keyboardRightKey$.pipe(throttleTime(100), skip(1))
    ))
);

I added another bit to reset the stream on every keyup and then used RxVIZ to create a visualization of it.

District

Lots of ways to skin this cat. Here's another using higher-order Observables: of( source$.pipe(throttleTime(500), take(5)), source$.pipe(throttleTime(100)) ).pipe(concatAll()). StackBlitz here – Kilometer