What are use cases for mergeMap operator?
tl;dr; mergeMap
is way more powerful than map
. Understanding mergeMap
is the necessary condition to access full power of Rx.
similarities
both
mergeMap
andmap
acts on a single stream (vs.zip
,combineLatest
)both
mergeMap
andmap
can transform elements of a stream (vs.filter
,delay
)
differences
map
cannot change size of the source stream (assumption:
map
itself does notthrow
); for each element from source exactly onemapped
element is emitted;map
cannot ignore elements (like for examplefilter
);in case of the default scheduler the transformation happens synchronously; to be 100% clear: the source stream may deliver its elements asynchronously, but each next element is immediately
mapped
and re-emitted further;map
cannot shift elements in time like for exampledelay
no restrictions on return values
id
:x => x
mergeMap
can change size of the source stream; for each element there might be arbitrary number (0, 1 or many) of new elements created/emitted
it offers full control over asynchronicity - both when new elements are created/emitted and how many elements from the source stream should be processed concurrently; for example assume source stream emitted 10 elements but
maxConcurrency
is set to 2 then two first elements will be processed immediately and the rest 8 buffered; once one of the processedcomplete
d the next element from source stream will be processed and so on - it is bit tricky, but take a look at the example belowall other operators can be implemented with just
mergeMap
andObservable
constructormay be used for recursive async operations
return values has to be of Observable type (or Rx has to know how to create observable out of it - e.g. promise, array)
id
:x => Rx.Observable.of(x)
array analogy
let array = [1,2,3]
fn map mergeMap
x => x*x [1,4,9] error /*expects array as return value*/
x => [x,x*x] [[1,1],[2,4],[3,9]] [1,1,2,4,3,9]
The analogy does not show full picture and it basically corresponds to .mergeMap
with maxConcurrency
set to 1. In such a case elements will be ordered as above, but in general case it does not have to be so. The only guarantee we have is that emission of new elements will be order by their position in the underlying stream. For example: [3,1,2,4,9,1]
and [2,3,1,1,9,4]
are valid, but [1,1,4,2,3,9]
is not (since 4
was emitted after 2
in the underlying stream).
A couple of examples using mergeMap
:
// implement .map with .mergeMap
Rx.Observable.prototype.mapWithMergeMap = function(mapFn) {
return this.mergeMap(x => Rx.Observable.of(mapFn(x)));
}
Rx.Observable.range(1, 3)
.mapWithMergeMap(x => x * x)
.subscribe(x => console.log('mapWithMergeMap', x))
// implement .filter with .mergeMap
Rx.Observable.prototype.filterWithMergeMap = function(filterFn) {
return this.mergeMap(x =>
filterFn(x) ?
Rx.Observable.of(x) :
Rx.Observable.empty()); // return no element
}
Rx.Observable.range(1, 3)
.filterWithMergeMap(x => x === 3)
.subscribe(x => console.log('filterWithMergeMap', x))
// implement .delay with .mergeMap
Rx.Observable.prototype.delayWithMergeMap = function(delayMs) {
return this.mergeMap(x =>
Rx.Observable.create(obs => {
// setTimeout is naive - one should use scheduler instead
const token = setTimeout(() => {
obs.next(x);
obs.complete();
}, delayMs)
return () => clearTimeout(token);
}))
}
Rx.Observable.range(1, 3)
.delayWithMergeMap(500)
.take(2)
.subscribe(x => console.log('delayWithMergeMap', x))
// recursive count
const count = (from, to, interval) => {
if (from > to) return Rx.Observable.empty();
return Rx.Observable.timer(interval)
.mergeMap(() =>
count(from + 1, to, interval)
.startWith(from))
}
count(1, 3, 1000).subscribe(x => console.log('count', x))
// just an example of bit different implementation with no returns
const countMoreRxWay = (from, to, interval) =>
Rx.Observable.if(
() => from > to,
Rx.Observable.empty(),
Rx.Observable.timer(interval)
.mergeMap(() => countMoreRxWay(from + 1, to, interval)
.startWith(from)))
const maxConcurrencyExample = () =>
Rx.Observable.range(1,7)
.do(x => console.log('emitted', x))
.mergeMap(x => Rx.Observable.timer(1000).mapTo(x), 2)
.do(x => console.log('processed', x))
.subscribe()
setTimeout(maxConcurrencyExample, 3100)
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.1.1/Rx.min.js"></script>
.mergeMap()
lets you flatten a higher-order Observable into a single stream. For instance:
Rx.Observable.from([1,2,3,4])
.map(i => getFreshApiData())
.subscribe(val => console.log('regular map result: ' + val));
//vs
Rx.Observable.from([1,2,3,4])
.mergeMap(i => getFreshApiData())
.subscribe(val => console.log('mergeMap result: ' + val));
function getFreshApiData() {
return Rx.Observable.of('retrieved new data')
.delay(1000);
}
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.1.0/Rx.js"></script>
See my answer at this other question for an in-dept explanation of the .xxxMap()
operators: Rxjs - How can I extract multiple values inside an array and feed them back to the observable stream synchronously