mergeAll – RxJS Reference

mergeAll

Q: What is the difference between mergeAll and merge in RxJS?

merge() takes a fixed set of observables as arguments, while mergeAll() is an operator that takes a higher-order observable, one that emits other observables, and subscribes to each inner stream as it arrives. Both emit values from all streams concurrently without preserving order. Use mergeAll() when the inner streams are produced dynamically by a source, and merge() when you already have the streams as separate variables.

Q: How does mergeAll flatten a higher-order observable in RxJS?

mergeAll() subscribes to the higher-order source and, for every inner observable it emits, subscribes to that inner observable immediately. Values from every active inner stream are passed to the subscriber as they occur, which is why this process is called flattening. The resulting stream completes only after the source and all inner observables have completed, and any inner error is forwarded as an error notification.

Q: When should I use mergeAll instead of concat or switchMap in RxJS?

Use mergeAll() when you don't care about the order of emissions and want all values from all inner streams as they happen. concat() waits for each inner stream to complete before starting the next, so ordering is preserved but work is sequential. Choose mergeAll() for concurrent work where every result matters, and avoid it when only the latest result should be kept.

Q: Does mergeAll complete when the inner observables complete in RxJS?

No, mergeAll() only sends the complete notification after the higher-order source and all inner observables have completed. If the source completes but some inner stream is still emitting, the output stays open. If any inner observable throws an error, the error is sent to the observer immediately, so handle errors on inner streams to avoid terminating the entire merged sequence.

Q: How do I use mergeAll with mergeMap in RxJS?

mergeMap() is equivalent to calling map() to project each source value into an inner observable and then mergeAll() to flatten the result. Use mergeAll() directly when your source already emits observables, and use mergeMap() when you need to create an inner observable from each value. Both subscribe to inner streams concurrently and emit their values without preserving order.

mergeAll combines a number of inner observable streams and concurrently emits all values from every input stream. It’s similar to merge, but instead of taking a set of streams directly as input, it takes an observable source that produces other streams (observables). Those streams are often referred to as inner stream (observable) and the stream that emits them is known as higher-order observable.

As values from any combined sequence are produced, those values are emitted as part of the resulting sequence. Such process is often referred to as flattening in documentation.

Use this operator if you’re not concerned with the order of emissions and is simply interested in all values coming out from multiple combined streams as if they were produced by one stream.

The operator works in the following way:

Subscribe to the higher-order source observable
When a new inner observable is emitted from the source observable, subscribe to it
When a new value arrives from an inner source observable, pass it down to an observer
Only after all inner source observables 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.

In the diagram below you can see the H higher-order stream that produces two inner streams A and B. The mergeAll operator combines values from these two streams and then passes them through to the resulting sequence as they occur:

mergeAll operator diagram

Usage

Use this operator if you’re not concerned with the order of emissions and is simply interested in all values coming out from multiple combined streams as if they were produced by one stream.

Here’s an example of using merge on multiple streams:

const a = interval(500).pipe(map((v) => 'a' + v), take(3));
const b = interval(500).pipe(map((v) => 'b' + v), take(3));
merge(a, b).subscribe((value) => console.log(value));

Playground

Additional resources

Official documentation

Frequently asked questions

What is the difference between mergeAll and merge in RxJS?

merge() takes a fixed set of observables as arguments, while mergeAll() is an operator that takes a higher-order observable, one that emits other observables, and subscribes to each inner stream as it arrives. Both emit values from all streams concurrently without preserving order. Use mergeAll() when the inner streams are produced dynamically by a source, and merge() when you already have the streams as separate variables.

How does mergeAll flatten a higher-order observable in RxJS?

mergeAll() subscribes to the higher-order source and, for every inner observable it emits, subscribes to that inner observable immediately. Values from every active inner stream are passed to the subscriber as they occur, which is why this process is called flattening. The resulting stream completes only after the source and all inner observables have completed, and any inner error is forwarded as an error notification.

When should I use mergeAll instead of concat or switchMap in RxJS?

Use mergeAll() when you don't care about the order of emissions and want all values from all inner streams as they happen. concat() waits for each inner stream to complete before starting the next, so ordering is preserved but work is sequential. Choose mergeAll() for concurrent work where every result matters, and avoid it when only the latest result should be kept.

Does mergeAll complete when the inner observables complete in RxJS?

No, mergeAll() only sends the complete notification after the higher-order source and all inner observables have completed. If the source completes but some inner stream is still emitting, the output stays open. If any inner observable throws an error, the error is sent to the observer immediately, so handle errors on inner streams to avoid terminating the entire merged sequence.

How do I use mergeAll with mergeMap in RxJS?

mergeMap() is equivalent to calling map() to project each source value into an inner observable and then mergeAll() to flatten the result. Use mergeAll() directly when your source already emits observables, and use mergeMap() when you need to create an inner observable from each value. Both subscribe to inner streams concurrently and emit their values without preserving order.

Share this post