Understanding concatAll
concatAll merges multiple inner observable streams and emits values from each input stream one after another. While it behaves similarly to concat, the key distinction is that concatAll receives its streams not as direct arguments, but rather from an outer observable that emits other observables — known as inner streams.
At any given moment, this operator maintains just one active subscription. Values from that active inner stream flow downstream to the observer. When that active stream completes, concatAll shifts its subscription to the next inner observable in line. As each inner sequence produces values, those values become part of the final output stream — a technique commonly described as flattening in RxJS documentation.
The mechanics of the operator unfold as follows:
- Begin by subscribing to a higher-order observable (the source).
- Upon receiving a new inner observable from the source, establish a subscription to it.
- Whenever a value emerges from this inner observable, deliver it downstream to the observer.
- If additional inner observables arrive from the source, add them to a waiting queue.
- After the currently active inner observable sends its completion notification, move on to subscribe to the next queued observable.
- Only when every queued inner observable finishes can the final completion signal be sent to the observer.
- Should any inner observable throw an error, an error notification is immediately dispatched to the observer.
Keep in mind that if any of the input streams never completes, concatAll will never finish its work. Consequently, any streams waiting further down the queue will remain unsubscribed indefinitely.
The visual below illustrates the H higher-order stream emitting two inner streams, A and B. Through concatAll, values from both streams are relayed to the output as they appear. Unlike mergeAll, which juggles multiple subscriptions, concatAll processes inner observables in strict sequence: it subscribes to A first, waits for A to terminate, and only then turns its attention to B.
The diagram demonstrates this sequential behavior:
When to reach for it
This operator shines when the order of emitted values matters and you need to prioritize streams that arrive earlier. Imagine you have one observable providing data from a local cache and another fetching data from a remote API. If you want the cached value to always surface first, concatAll allows you to combine these sources while preserving that priority ordering.
Consider the following example that uses concatAll to join two streams together:
const a = interval(500).pipe(map((v) => 'a' + v), take(3));
const b = interval(500).pipe(map((v) => 'b' + v), take(3));
const higherOrderObservable = of(a, b);
higherOrderObservable.pipe(concatAll()).subscribe((value) => console.log(value));