RxJS Mastery – RxJS Operator Patterns: concat, merge, switch and exhaust

Many RxJS operators follow similar patterns. You must have noticed until now that there are multiple operators for the family of:

• concat: concat, concatMap, concatAll
• merge: mergeScan, mergeMap, mergeAll
• switch: switchScan, switchMap, switchAll
• exhaust: exhaustMap, exhaustAll

Let us have a look at each of the families and the principle behind it. Understanding it is going to help you decide which operator to use when.

The RxJS concat principle

The concat operators should be used if the order is important. Because concat makes sure that the operator is applied to one Observable at a time until that Observable completes. Once the work on one Observable is done, the next one is taken into account.

Below there is a basic example with three Observables. The concat principle makes sure that all values of the first Observable are emitted before continuing with the second one, and so on.

Let’s have this example running in a test to give a second view:

const { expectObservable, cold } = helpers;

const mapToIndexedString = () => {
    return map((v, i) => `${v}${i + 1}`);
};

const source$ = of(
    cold('-aaa|').pipe(mapToIndexedString()),
    cold('-bbb|').pipe(mapToIndexedString()),
    cold('-ccc|').pipe(mapToIndexedString()),
);
const result$ = source$.pipe(concatAll());

expectObservable(result$).toBe(
    '-012-345-678|',
    ['a1', 'a2', 'a3', 'b1', 'b2', 'b3', 'c1', 'c2', 'c3']
);

The mapToIndexedString function is simply appending the index + 1 to the value coming from the Observable, e.g., a at index 0 is mapped to a1. As we can see the result$ Observable in the above test contains all values. Furthermore, the values are emitted in sequence. Values from the b Observable only arrive once we are done with a, i.e. the first inner Observable has completed.

Please note the following when using concat.

• If the order and especially the sequential handling is important, concat is a good choice.
• All values will be considered, i.e. nothing is ignored or cancelled
• Concat relies on the completion of Observables. Watch out for Observables that don’t complete.

The RxJS merge principle

The merge operators are the way to go if you want to merge multiple independent Observables into a single stream without any special preferences. So, in this case, throughput is good and all values from all Observables are just merged.

Compared to concat, merge works parallelly on multiple Observables. In fact, concat is a special case of merge with concurrency of 1. In the below example, merge is applied on the same Observables as in the previous case for concat.

const source$ = of(
    cold('-aaa|').pipe(mapToIndexedString()),
    cold('-bbb|').pipe(mapToIndexedString()),
    cold('-ccc|').pipe(mapToIndexedString()),
);
const result$ = source$.pipe(mergeAll());

result$.subscribe({ next: console.log });

// outputs: a1, b1, c1, a2, b2, c2, a3, b3, c3

Merge is probably the easiest principle among the four mentioned:

• If you would like to consider all values and you do not care about the order, merge is a good choice.
• Merge can result in a lot of backpressure though. So, if your further processing of the values is resource-intensive, additional measures should be taken, e.g., throttling or decreasing the concurrency parameter of merge.

The RxJS switch principle

Those operators switch to the newest Observable available. Values from other Observables that were started before are not considered anymore. In fact, switch calls unsubscribe on the current Observable which might even cancel running operations behind that Observable, e.g. an HTTP request.

You would apply a switch operator compared to the others if:

• you want to limit the execution to one inner Observable
• running operations should be canceled and new ones taken into account
• old results are probably not usable and therefore switching to the most recent Observable is the goal.

The RxJS exhaust principle

Exhaust only works on one Observable at a time and ignores new arrivals during that time. Only Observables arriving after the operator is not busy anymore, are again taken into account. This protects running Observables from being canceled. You could say that the operator is exhausted quickly with already one operation and will decline approaches during that time.

Exhaust is a good choice for the following cases:

• There is an HTTP request, e.g. for saving something, that you do not want to have canceled. But at the same time, you also want to prevent too many requests from piling up if the user desperately hits the save button multiple times.

This post is part of the RxJS mastery series. As always the code examples can be found on GitHub.