Understanding exhaustMap

The exhaustMap operator merges the behavior of two distinct operators: exhaust and map. The map component transforms each value originating from a higher-order source observable into an inner observable. The exhaust component then subscribes to that inner observable, forwarding its emissions to the observer, but only if no other subscription is currently active. If a subscription is already active, the newly emitted inner observable is disregarded.

At any given moment, exhaustMap maintains at most one active subscription whose values are relayed downstream. When the higher-order observable emits a fresh inner observable while the current one is still running, the new arrival is discarded. After the active inner observable completes, the operator resumes waiting for the next emission from the higher-order source, while ignoring any inner observables that were previously skipped.

The operational flow of this operator can be broken down as follows:

  1. Initiate a subscription to the higher-order source observable.
  2. Upon receiving a new inner observable, verify whether an active subscription already exists.
  3. If a subscription is active, drop the incoming inner observable.
  4. If no subscription is active, apply the mapping function to obtain the inner observable and subscribe to it.
  5. Emit any values from that inner observable to the observer.
  6. Send the complete notification to the observer only after both the higher-order source and the current inner subscription have finished.
  7. In case the inner observable emits an error, transmit the error notification to the observer.

It is important to note that exhaustMap will not finalize if any of the involved streams fail to complete. This also implies that some emitted inner observables may never get subscribed to at all.

Practical Application

This operator proves particularly valuable in scenarios where multiple triggers may attempt to start a long-running operation, such as a login request over HTTP. By leveraging exhaustMap, you can ensure that consecutive triggers are ignored until the ongoing task concludes.

The following example demonstrates how to use exhaustMap for exactly this purpose:

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));

Interactive Demo

Further Reading