Learn RxJS
Search…
mergeAll
signature: mergeAll(concurrent: number): Observable
mergeAll(concurrent: number): ObservableCollect and subscribe to all observables.
Examples
Example 1: mergeAll with promises
1
// RxJS v6+
2
import { map, mergeAll } from 'rxjs/operators';
3
import { of } from 'rxjs';
4
5
const myPromise = val =>
6
new Promise(resolve => setTimeout(() => resolve(`Result: ${val}`), 2000));
7
//emit 1,2,3
8
const source = of(1, 2, 3);
9
10
const example = source.pipe(
11
//map each value to promise
12
map(val => myPromise(val)),
13
//emit result from source
14
mergeAll()
15
);
16
17
/*
18
output:
19
"Result: 1"
20
"Result: 2"
21
"Result: 3"
22
*/
23
const subscribe = example.subscribe(val => console.log(val));
Copied!
Example 2: mergeAll with concurrent parameter
1
// RxJS v6+
2
import { take, map, delay, mergeAll } from 'rxjs/operators';
3
import { interval } from 'rxjs';
4
5
const source = interval(500).pipe(take(5));
6
7
/*
8
interval is emitting a value every 0.5s. This value is then being mapped to interval that
9
is delayed for 1.0s. The mergeAll operator takes an optional argument that determines how
10
many inner observables to subscribe to at a time. The rest of the observables are stored
11
in a backlog waiting to be subscribe.
12
*/
13
const example = source
14
.pipe(
15
map(val => source.pipe(delay(1000), take(3))),
16
mergeAll(2)
17
)
18
.subscribe(val => console.log(val));
19
/*
20
The subscription is completed once the operator emits all values.
21
*/
Copied!
Additional Resources
- 🎥 💵 - André Staltz
Last modified 9mo ago