mergeAll
This operator is best used when you have multiple, short-lived observables that produce values independently and you want to flatten them into a single output stream. Real-world examples of this can be seen in scenarios like processing multiple requests simultaneously, where each request is an observable and the responses need to be combined into a single stream.
Consider a web page that displays live news updates from different sources. Each source produces updates as separate observables, and you want to merge them all together to show a single feed of news updates. In this case, mergeAll can come to the rescue.
It's essential to note that
mergeAll will start emitting values as soon as any of the inner observables emit a value. This is different from combineLatest, which waits for each observable to emit at least one value before producing an output.When deciding between mergeAll and other operators, keep in mind the following:
- If you need to merge multiple observables that rely on each other for calculations or decisions,
combineLatestmay be more suitable. - If you're working with observables that only emit one value or you only require the last value of each before completion,
forkJoinis likely a better choice. - If you need to merge observables that produce values independently and are short-lived,
mergeAllis the operator to reach for.
Example 1: mergeAll with promises
// RxJS v6+
import { map, mergeAll } from 'rxjs/operators';
import { of } from 'rxjs';
const myPromise = val =>
new Promise(resolve => setTimeout(() => resolve(`Result: ${val}`), 2000));
//emit 1,2,3
const source = of(1, 2, 3);
const example = source.pipe(
//map each value to promise
map(val => myPromise(val)),
//emit result from source
mergeAll()
);
/*
output:
"Result: 1"
"Result: 2"
"Result: 3"
*/
const subscribe = example.subscribe(val => console.log(val));
Example 2: mergeAll with concurrent parameter
// RxJS v6+
import { take, map, delay, mergeAll } from 'rxjs/operators';
import { interval } from 'rxjs';
const source = interval(500).pipe(take(5));
/*
interval is emitting a value every 0.5s. This value is then being mapped to interval that
is delayed for 1.0s. The mergeAll operator takes an optional argument that determines how
many inner observables to subscribe to at a time. The rest of the observables are stored
in a backlog waiting to be subscribe.
*/
const example = source
.pipe(
map(val => source.pipe(delay(1000), take(3))),
mergeAll(2)
)
.subscribe(val => console.log(val));
/*
The subscription is completed once the operator emits all values.
*/
Last modified 5mo ago