Learn to combine RxJs sequences with super intuitive interactive diagrams

We have a whole reference section on RxJS operators with lots of useful diagrams.

In any reasonably complex application, you will typically have data coming from more than one source. This could be several external data points, such as Firebase backends, or multiple UI widgets that interact with the user. Sequence composition is a powerful technique that lets you create complex queries across these multiple data sources by combining relevant streams into a single one. RxJs offers a range of operators designed for this purpose, and in this article, we'll explore the most commonly used ones.

To make the differences between these operators clear, I've even taken on the role of a part-time animation specialist to design intuitive data flow diagrams. However, since these diagrams are embedded as animated GIFs, they may take some time to load. Please bear with us.

In the code snippets that follow, I'll be using lettable operators. If you're not familiar with them, you can learn more here. I'll also rely on a custom stream operator that emits values asynchronously, with the first item delivered synchronously upon subscription.

Here’s the legend for the types of diagrams used throughout this article:

Merging multiple sequences concurrently

The first operator we'll examine is merge. This operator takes a number of observable streams and concurrently emits all values from each input stream. As soon as any combined sequence produces a value, that value is passed into the resulting sequence. In the documentation, this process is often called flattening.

The resulting stream completes only when all input streams complete. If any of the input streams throws an error, the result will also throw an error. Conversely, if any input stream never completes, the resulting sequence will never finish either.

You should choose this operator when the order of emissions doesn't matter and you simply want to aggregate all values from multiple streams as if they were one single stream.

Look at the diagram below to see how the merge operator combines two streams, A and B, each yielding three items. The values fall through to the resulting sequence as they occur.

The following code example illustrates the setup from the diagram:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 200, 3, 'partial');
merge(a, b).subscribe(fullObserver('merge'));
// can also be used as an instance operator
a.pipe(merge(b)).subscribe(fullObserver('merge'));

You can also try an editable demo on Stackblitz:

Concatenating multiple sequences sequentially

The next method is concat. This operator processes streams sequentially. It subscribes to each input stream one at a time, emitting values only from the current stream until it completes. Only then does it move on to the next stream, passing its values through to the result.

Like merge, the resulting stream finishes when all input streams complete, and it will throw an error if any stream errors. If one of the later streams never gets to run because an earlier one never completes, it will remain unsubscribed.

Use this operator when the order of emissions is critical and you need to see values from the streams passed in first before any other values. For instance, you might have an observable that delivers cached values and another that fetches data from a remote server. If you combine them with concat, you ensure the cached value appears first.

In the diagram below, concat combines two streams, A and B, each producing three items. Values flow to the result first from A and then from B.

Below is the code example matching the diagram above:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 200, 3, 'partial');
concat(a, b).subscribe(fullObserver('concat'));
// can also be used as an instance operator
a.pipe(concat(b)).subscribe(fullObserver('concat'));

And an editable demo on Stackblitz:

Combining sequences ambiguously

The race operator introduces an interesting concept. It doesn't combine streams in the traditional sense; instead, it selects the observable sequence that produces its first value first. Once one sequence starts emitting, the other sequences are unsubscribed and entirely ignored.

The resulting stream completes when this chosen input stream completes, and it will throw an error if that selected stream errors out. Likewise, it will never complete if the selected stream does not complete.

This operator is handy when you have multiple resources that could provide the same data, such as servers distributed globally, where network latency is unpredictable. Using race, you can dispatch the same request to multiple sources and use the result from whichever responds first.

The diagram shows race combining two streams, A and B, each with three items. Only values from A are emitted because it starts emitting first.

Here is the code that models the above diagram:

const a = intervalProducer('a', 200, 3, 'partial');
const b = intervalProducer('b', 500, 3, 'partial');
race(a, b).subscribe(fullObserver('race'));
// can also be used as an instance operator
a.pipe(race(b)).subscribe(fullObserver('race'));

And an editable demo on Stackblitz:

Combing unknown number of sequences with higher-order observables

The operators I've covered so far, whether used as static or instance versions, require a known number of sequences. But what if you don't know all the sequences upfront and need to combine them dynamically at runtime? This scenario is common with asynchronous code — for example, a network request for a resource might trigger a set of subsequent requests determined by the initial response.

RxJs provides variations of the operators we've seen that operate on a stream of streams, referred to as higher-order observables or Observable-of-Observables. These operators assume the emissions are sequences and handle them according to the same rules described earlier.

All such operators emit an error if any inner stream errors, and they can only be used as instance operators. Let's go through them one by one.

MergeAll

This operator combines all emitted inner streams, producing values concurrently from each, much like the plain merge.

In the diagram, H is a higher-order stream that yields two inner streams, A and B. The mergeAll operator merges the values from these streams, passing them through to the result as they arrive.

Here's the code example that demonstrates the diagram above:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(mergeAll()).subscribe(fullObserver('mergeAll'));

And an editable demo on Stackblitz:

ConcatAll

This operator mirrors the concat behavior, producing values from each inner stream sequentially.

The diagram below shows the higher-order stream H producing two inner streams, A and B. The concatAll operator first emits values from A and then from B.

Here is the code that corresponds to the diagram:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(concatAll()).subscribe(fullObserver('concatAll'));

And an editable demo on Stackblitz:

SwitchAll

Sometimes, receiving values from all inner observables is not desirable. In certain scenarios, we only need the values from the latest inner sequence. A classic example is a search feature. As a user types, we send a request to a server, and the result comes back as an observable. If the user types more characters before the response returns, another request is sent — now we have two searches in flight. However, we're no longer interested in the result of the very first search. Merging both results would confuse the user, so we need a way to focus only on the most recent stream. That's exactly what the switchAll operator does. It subscribes only to the latest inner sequence and ignores all previous ones.

The diagram below shows the higher-order stream H producing two inner streams, A and B. The switchAll operator emits values first from A and then from B.

Here's the code example for the diagram:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(switchAll()).subscribe(fullObserver('switchAll'));

And an editable demo on Stackblitz:

concatMap, mergeMap and switchMap

Interestingly, the mapping operators concatMap, mergeMap, and switchMap are used far more frequently than their corresponding flattening operators concatAll, mergeAll, and switchAll. Yet, when you think about it, they are nearly identical. Each *Map operator is simply a combination of two steps: first, mapping a value to an observable stream, and then applying the flattening logic to the resulting higher-order stream.

Let's look at a familiar code snippet showing how mergeAll works:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(mergeAll()).subscribe(fullObserver('mergeAll'));

Here, the map operator produces a higher-order stream, and mergeAll combines the values from the inner observables. This is why we can replace map and mergeAll with the single mergeMap operator, like this:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), mergeMap(i => [a, b][i]));

h.subscribe(fullObserver('mergeMap'));

The outcome is exactly the same. The same principle applies to both concatMap and switchMap — feel free to try it out yourself.

Combing sequences by pairing their values

All the operators we've discussed so far flatten multiple sequences, passing the original values through unchanged as if they came from a single stream. The following set of operators still accepts multiple input sequences but differs in one key way: they pair up values from each sequence to produce a single, combined output.

Each operator can take an optional projection function as its final parameter, which defines how values are combined in the result. For the examples, I'll rely on the default projection function, which simply joins the values with a comma. I'll show a custom projection function at the end of this section.

CombineLatest

The first operator here is combineLatest. It takes the most recent value from each input sequence and transforms them into a single output value. RxJs caches the latest value from every sequence. Once all sequences have emitted at least one value, it computes a combined result using the provided projection function, and emits it to the result stream.

The resulting stream completes when all input streams complete. If any stream throws an error, the result errors out as well. Should any input stream never complete, the result will also remain incomplete. Additionally, if any stream emits no value but completes, the resulting stream completes at that moment without emitting anything — because there is no value to include from that stream. Similarly, if a stream emits nothing and never completes, combineLatest will neither emit nor complete, as it waits indefinitely for all streams to provide a value.

This operator is useful when you need to track a combination of state variables that must be kept up-to-date as individual parts change. For instance, imagine a monitoring system where each service emits a Boolean indicating availability. A green status might be shown only when all services are available, so the projection function would perform a logical AND over those values.

The diagram below shows combineLatest combining two streams, A and B. Once each has emitted at least one value, every subsequent emission creates a combined value on the result stream:

Here is the code example for the above diagram:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');
combineLatest(a, b).subscribe(fullObserver('latest'));

And an editable demo on Stackblitz:

Zip

This operator presents another intriguing merging feature that shares some similarities with how a zipper works on a jacket or a bag. It pairs together two or more sequences of matching values into a tuple (a pair when dealing with two input streams). It holds off until the matching value is emitted from every input stream, then applies a projection function to convert them into a single value and pushes that result downstream. It only emits once it has a fresh pair of values from each source sequence, meaning that if one source produces values at a faster rate than the other, the overall emission pace will follow the slower of the two streams.

The output stream finalizes when any of the inner streams finish and all the corresponding matched pairs have been emitted from the other streams. It will keep running indefinitely if any inner stream never completes, and it will propagate an error if any inner stream fails.

This operator proves quite handy when you need a stream that generates values across a certain range with a regular interval. Here's a straightforward example featuring a projection function that pulls values exclusively from the range stream:

zip(range(3, 5), interval(500), v => v).subscribe();

In the visual below, you can observe the zip operator merging two streams A and B. Whenever a pair of matching values aligns, the resulting sequence delivers a combined value:

Here is the code example that illustrates the configuration shown in the diagram above:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

zip(a, b).subscribe(fullObserver('zip'));

And stackblitz editable demo:

forkJoin

At times you may have a collection of streams but only care about the last value each one produces. These sequences frequently emit only a single item. For instance, you might need to fire off several network requests and only proceed once responses have arrived for all of them. This bears resemblance to Promise.all behavior. However, if a stream happens to emit more than one value, everything except the final one gets discarded.

The resulting stream emits just once, and only when all of the inner streams have completed. It will run indefinitely if any inner stream never finishes, and it will throw an error if any inner stream fails.

In the diagram below you can see the forkJoin operator bringing together two streams A and B. As soon as a matching pair is found the resulting sequence produces a combined value:

Here is the code example that demonstrates the setup illustrated by the above diagram:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

forkJoin(a, b).subscribe(fullObserver('forkJoin'));

And stackblitz editable demo:

WithLatestFrom

The final operator we'll examine in this article is withLatestFrom. This one is useful when you have a primary controlling stream but also need the current values from other streams. Whereas the related combineLatest operator fires a new value any time there's an emission from any of the input streams, withLatestFrom only fires when the controlling stream produces a new value.

Similar to combineLatest, it still requires at least one value from each stream and may terminate without emitting anything if the controlling stream completes early. It will never complete unless the controlling stream ends, and it will throw an error if any of the inner streams errors out.

In the diagram below you can see the withLatestFrom operator merging two streams A and B, where stream B acts as the controlling stream. Whenever stream B emits a fresh value, the output sequence produces a combined value using the most recent value from stream A:

Here is the code example that mirrors the behavior shown in the above diagram:

const a = stream('a', 3000, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

b.pipe(withLatestFrom(a)).subscribe(fullObserver('latest'));

And stackblitz editable demo:

Projection function

As noted at the start of this section, all operators that pair up values accept an optional projection function. This function determines how the final output is shaped. With this function you can decide to emit just a value from a single input stream, or you can merge values in any manner you see fit:

// return value from the second sequence
zip(s1, s2, s3, (v1, v2, v3) => v2)

// join values using dash as a separator
zip(s1, s2, s3, (v1, v2, v3) => `${v1}-${v2}-${v3}`)

// return single boolean result
zip(s1, s2, s3, (v1, v2, v3) => v1 && v2 && v3)