signature: mergeMap(project: function: Observable, resultSelector: function: any, concurrent: number): Observable
💡 flatMap is an alias for mergeMap!
💡 If only one inner subscription should be active at a time, try
switchMap
!
💡 If the order of emission and subscription of inner observables is
important, try concatMap
!
This operator is best used when you wish to flatten an inner observable but want to manually control the number of inner subscriptions.
For instance, when using switchMap
each inner subscription is completed when the source emits, allowing only one active inner subscription. In contrast, mergeMap
allows for multiple inner subscriptions to be active at a time. Because of this, one of the most common use-case for mergeMap
is requests that should not be canceled, think writes rather than reads. Note that if order must be maintained concatMap
is a better option.
Be aware that because mergeMap
maintains multiple active inner subscriptions at once it's possible to create a memory leak through long-lived inner subscriptions. A basic example would be if you were mapping to an observable with an inner timer, or a stream of dom events. In these cases, if you still wish to utilize mergeMap
you may want to take advantage of another operator to manage the completion of the inner subscription, think take
or takeUntil
. You can also limit the number of active inner subscriptions at a time with the concurrent
parameter, seen in example 4.
//emit 'Hello'
const source = Rx.Observable.of('Hello');
//map to inner observable and flatten
const example = source.mergeMap(val => Rx.Observable.of(`${val} World!`));
//output: 'Hello World!'
const subscribe = example.subscribe(val => console.log(val));
//emit 'Hello'
const source = Rx.Observable.of('Hello');
//mergeMap also emits result of promise
const myPromise = val =>
new Promise(resolve => resolve(`${val} World From Promise!`));
//map to promise and emit result
const example = source.mergeMap(val => myPromise(val));
//output: 'Hello World From Promise'
const subscribe = example.subscribe(val => console.log(val));
/*
you can also supply a second argument which receives the source value and emitted
value of inner observable or promise
*/
//emit 'Hello'
const source = Rx.Observable.of('Hello');
//mergeMap also emits result of promise
const myPromise = val =>
new Promise(resolve => resolve(`${val} World From Promise!`));
const example = source.mergeMap(
val => myPromise(val),
(valueFromSource, valueFromPromise) => {
return `Source: ${valueFromSource}, Promise: ${valueFromPromise}`;
}
);
//output: "Source: Hello, Promise: Hello World From Promise!"
const subscribe = example.subscribe(val => console.log(val));
//emit value every 1s
const source = Rx.Observable.interval(1000);
const example = source.mergeMap(
//project
val => Rx.Observable.interval(5000).take(2),
//resultSelector
(oVal, iVal, oIndex, iIndex) => [oIndex, oVal, iIndex, iVal],
//concurrent
2
);
/*
Output:
[0, 0, 0, 0] <--1st inner observable
[1, 1, 0, 0] <--2nd inner observable
[0, 0, 1, 1] <--1st inner observable
[1, 1, 1, 1] <--2nd inner observable
[2, 2, 0, 0] <--3rd inner observable
[3, 3, 0, 0] <--4th inner observable
*/
const subscribe = example.subscribe(val => console.log(val));
- mergeMap 📰 - Official docs
- map vs flatMap 📹 💵 - Ben Lesh
- Async requests and responses in RxJS 📹 💵 - André Staltz
- Use RxJS mergeMap to map and merge higher order observables 📹 💵 - André Staltz
- Use RxJS mergeMap for fine grain custom behavior 📹 💵 - André Staltz
📁 Source Code: https://github.com/ReactiveX/rxjs/blob/master/src/operator/mergeMap.ts