exhaustMap – RxJS Reference

exhaustMap

Q: How does exhaustMap work in RxJS?

exhaustMap() maps each value from a source observable to an inner observable and subscribes to it only when no inner subscription is currently active. While an inner observable is running, any new inner observables emitted by the source are dropped, and values from the active inner stream are passed to the observer. Once it completes, the operator waits for the next inner observable to arrive.

Q: When should I use exhaustMap to ignore repeated actions in Angular?

Use exhaustMap() when a user action starts a long-running task, such as a login HTTP request, and similar actions should be ignored until that task finishes. For example, a button click stream piped through exhaustMap() into an HTTP call ignores extra clicks while the first request is pending. Once that request completes, the next click starts a new one.

Q: What is the difference between exhaustMap and exhaust in RxJS?

exhaustMap() is a combination of map() and exhaust(): it projects each source value to an inner observable and then applies the exhaust strategy. The standalone exhaust() operator works on a higher-order observable that already emits inner observables, subscribing to one only when no other is active and ignoring the rest. Use exhaustMap() when you need to project values into inner streams.

Q: Why does my exhaustMap stream never complete in RxJS?

An exhaustMap() stream completes only after both the source observable and the currently active inner observable have completed. If the source or an inner stream never completes, such as an interval() without a limit, the output never completes and some inner observables are never subscribed to. Add take() to bound the streams so the operator can finish.

Q: How does exhaustMap handle errors thrown by inner observables?

If any inner observable throws an error, exhaustMap() sends the error notification to the observer, so the error reaches the subscriber through the same stream as values. To keep the outer stream running after a failed request, catch the error inside the inner observable with catchError() before it reaches exhaustMap().

exhaustMap operator is basically a combination of two operators – exhaust and map. The map part lets you map a value from a higher-order source observable to an inner observable stream. The exhaust part than subscribes to an inner observable and passes values down to an observer if there’s no active subscription already, otherwise it just ignores new inner observables.

exhaustMap has only one active subscription at a time from which the values are passed down to an observer. When the higher-order observable emits a new inner observable stream, if the current stream hasn’t completed, this new inner observable is dropped. Once the current active stream completes, the operator waits for another inner observable to subscribe ignoring previous inner observables.

The operator works in the following way:

Subscribe to a higher-order source observable
When a new inner observable arrives from a source observable, check if there’s no active subscription
If there’s an active subscription, ignore the observable
If there’s no active subscription, execute a map function that returns an inner observable and subscribe to it
When a new value arrives from this inner observable, pass it down to an observer
Only after a higher-order source observable and current active subscriptions complete, send the complete notification to the observer.
If any of the inner source observables throws an error, send the error notification to the observer.

Please note that exhaustMap will never complete if some input streams don't complete. This also means that some streams will never be subscribed to.

Usage

exhaustMap is useful when your system can produce multiple actions that trigger a long lasting task, like a login HTTP request, and all other similar actions should be ignored until the current task is complete.

Here’s an example of using exhaustMap to do exactly this:

const request = (index) => timer(500).pipe(take(1), mapTo({index}));

const a = request(1);
const b = request(2);
const actions = interval(100).pipe(take(2), map((v, i) => [a, b][i]));

// logs 1
actions.pipe(exhaustMap((request) => request)).subscribe((res: any) => console.log(res.index));

Playground

Additional resources

Official documentation

Frequently asked questions

How does exhaustMap work in RxJS?

exhaustMap() maps each value from a source observable to an inner observable and subscribes to it only when no inner subscription is currently active. While an inner observable is running, any new inner observables emitted by the source are dropped, and values from the active inner stream are passed to the observer. Once it completes, the operator waits for the next inner observable to arrive.

When should I use exhaustMap to ignore repeated actions in Angular?

Use exhaustMap() when a user action starts a long-running task, such as a login HTTP request, and similar actions should be ignored until that task finishes. For example, a button click stream piped through exhaustMap() into an HTTP call ignores extra clicks while the first request is pending. Once that request completes, the next click starts a new one.

What is the difference between exhaustMap and exhaust in RxJS?

exhaustMap() is a combination of map() and exhaust(): it projects each source value to an inner observable and then applies the exhaust strategy. The standalone exhaust() operator works on a higher-order observable that already emits inner observables, subscribing to one only when no other is active and ignoring the rest. Use exhaustMap() when you need to project values into inner streams.

Why does my exhaustMap stream never complete in RxJS?

An exhaustMap() stream completes only after both the source observable and the currently active inner observable have completed. If the source or an inner stream never completes, such as an interval() without a limit, the output never completes and some inner observables are never subscribed to. Add take() to bound the streams so the operator can finish.

How does exhaustMap handle errors thrown by inner observables?

If any inner observable throws an error, exhaustMap() sends the error notification to the observer, so the error reaches the subscriber through the same stream as values. To keep the outer stream running after a failed request, catch the error inside the inner observable with catchError() before it reaches exhaustMap().

Share this post