Understanding exhaust

exhaust operates on a source observable whose emitted values are themselves observables. This type of source is known as a higher-order observable, and the observables it emits are referred to as inner observables. When these inner streams produce values, those values are forwarded to the subscriber as part of the output sequence. In official documentation, this mechanism is commonly described as flattening.

At any given moment, exhaust maintains only a single active subscription whose emissions are relayed downstream. If the higher-order observable pushes a new inner stream while the current one is still running, that new stream gets discarded. Only after the active stream finishes does the operator become receptive to another inner observable — but previously emitted ones that were ignored remain ignored.

Here is how the operator behaves step by step:

  1. Begin by subscribing to the higher-order source observable.
  2. Whenever the source emits an inner observable, establish a subscription to it.
  3. Each value arriving from this inner observable is forwarded to the observer.
  4. If another inner observable appears while one is still active, check whether an active subscription exists.
  5. When a subscription is already in place, the new observable is skipped; otherwise, it gets subscribed to.
  6. The complete notification is sent to the observer only after both the higher-order source and the active inner subscription have finished.
  7. If any inner observable throws an error, the error notification is passed along to the observer.

It is important to keep in mind that exhaust will not emit a complete signal if any of the involved streams never terminate. Consequently, some inner observables may never be subscribed to at all.

Practical Application

This operator proves handy in scenarios where multiple events can trigger an expensive, long-running operation — such as a login request — and it is desirable to ignore all subsequent triggers until the current one concludes.

Below is a concise example demonstrating how exhaust can implement this pattern:

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

Interactive Demo

Further Reading