exhaust – RxJS Reference

exhaust

Q: What does the exhaust operator do in RxJS?

exhaust() flattens a higher-order observable by keeping only one active inner subscription at a time and emitting its values to the observer. When a new inner observable arrives while the current one has not completed, it is dropped. Once the active inner stream completes, the operator waits for the next inner observable to subscribe, ignoring those that arrived earlier. Values from the currently active stream are passed down as they are produced.

Q: How do I ignore new HTTP requests while one is still running in RxJS?

Use exhaust() on a higher-order observable whose inner observables are long-lasting tasks, such as a login HTTP request. The operator subscribes to the first inner observable and drops every other inner observable emitted while it is still active. In the usage example, a second request emitted 100 ms later, while a 500 ms timer was running, was dropped, so only 1 was logged.

Q: Why does exhaust() never complete in RxJS?

exhaust() does not complete until the higher-order source observable and the currently active inner subscription have both completed. If any input stream never completes, the resulting observable never completes either. Because the operator only subscribes to an inner observable when no other one is active, some inner observables may never be subscribed to at all. Make sure the source and each inner stream finish, for example by adding take(1) to each request stream.

Q: What happens when an inner observable throws an error in exhaust()?

If any inner source observable throws an error, exhaust() sends the error notification to the observer, so the observer's error callback receives it. Errors from inner streams are therefore propagated through the flattened result just like values are. Check the error handling on each inner request stream when a single failed task should not stop the whole sequence.

exhaust takes a source observable that produces other observable streams. Such observable is called higher-order observable and the streams emitted as values are called inner observables. As values from inner sequences are produced, those values are emitted as part of the resulting sequence to an observer. Such process is often referred to as flattening in documentation.

exhaust only has 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 is emitted from the higher-order observable, subscribe to it
When a new value arrives from this inner observable, pass it down to an observer
When a new inner observable is emitted from the higher-order observable, check if there’s any active subscription
If there’s an active subscription, ignore the observable, otherwise subscribe to it
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 exhaust will never complete if some of the input streams don’t complete. This also means that some streams will never be subscribed to.

Usage

exhaust 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 exhaust 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(exhaust()).subscribe((res: any) => console.log(res.index));

Playground

Additional resources

Official documentation

Frequently asked questions

What does the exhaust operator do in RxJS?

exhaust() flattens a higher-order observable by keeping only one active inner subscription at a time and emitting its values to the observer. When a new inner observable arrives while the current one has not completed, it is dropped. Once the active inner stream completes, the operator waits for the next inner observable to subscribe, ignoring those that arrived earlier. Values from the currently active stream are passed down as they are produced.

How do I ignore new HTTP requests while one is still running in RxJS?

Use exhaust() on a higher-order observable whose inner observables are long-lasting tasks, such as a login HTTP request. The operator subscribes to the first inner observable and drops every other inner observable emitted while it is still active. In the usage example, a second request emitted 100 ms later, while a 500 ms timer was running, was dropped, so only 1 was logged.

Why does exhaust() never complete in RxJS?

exhaust() does not complete until the higher-order source observable and the currently active inner subscription have both completed. If any input stream never completes, the resulting observable never completes either. Because the operator only subscribes to an inner observable when no other one is active, some inner observables may never be subscribed to at all. Make sure the source and each inner stream finish, for example by adding take(1) to each request stream.

What happens when an inner observable throws an error in exhaust()?

If any inner source observable throws an error, exhaust() sends the error notification to the observer, so the observer's error callback receives it. Errors from inner streams are therefore propagated through the flattened result just like values are. Check the error handling on each inner request stream when a single failed task should not stop the whole sequence.

Share this post