combineLatest – RxJS Reference

combineLatest

Q: Why doesn't combineLatest emit anything until every input observable has emitted a value?

combineLatest waits because it needs a cached value from each input before it can build a combined result. It stores the latest value from every source, and only after all sources have emitted at least once does it emit the combined values, usually as an array. After that, any new emission from one source overrides its cached value and produces a new combined emission.

Q: How does combineLatest handle an error thrown by one of its source observables?

If any source observable throws an error, combineLatest sends the error notification to the observer and the resulting stream errors. The error is propagated from the failing inner stream, so the subscriber receives it through its error handler. Plan for this when subscribing, because one failing source stops the combined stream.

Q: When should I use combineLatest to compute state that depends on several streams?

Use combineLatest when a combined value must stay up to date whenever any part of the state changes. A typical case is a monitoring system where each service is a stream emitting a Boolean for its availability. The projection function can perform a logical AND, so the overall status is green only when all services are available.

combineLatest allows to merge several streams by taking the most recent value from each input observable and emitting those values to the observer as a combined output (usually as an array). The operators caches the last value for each input observable and only once all input observables produced at least one value it emits the combined cached values to the observer.

The resulting stream completes when all inner streams complete and will throw an error if any of the inner streams throws an error. It will never complete if any of the inner streams doesn’t complete. On the other hand, if any stream does not emit value but completes, resulting stream will complete at the same moment without emitting anything, since it will be now impossible to include value from completed input stream in resulting sequence. Also, if some input stream does not emit any value and never completes, combineLatest will also never emit and never complete, since, again, it will wait for all streams to emit some value.

The operator works in the following way:

Subscribe to all input observables
When a source observable emits a value, override any existing value in the cache for this observable
If here’s a cached value for each obsevable in the cache, emit the combined values to the observer
Only after all source observables complete, send the complete notification to the observer.
If any of the source observables throws an error, send the error notification to the observer.

In the diagram below you can see the combineLatest operator combining two streams A and B. As soon as all streams have emitted at least one value each new emission produces a combined value through the result stream:

combineLatest operator diagram

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

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

combineLatest(a, b).subscribe(fullObserver(operator));

And the editable demo:

Usage

This operator can be useful if you need to evaluate some combination of state which needs to be kept up-to-date when part of the state changes. A simple example would be a monitoring system. Each service is represented by a sequence that returns a Boolean indicating the availability of a said service. The monitoring status is green if all services are available so the projection function should simply perform a logical AND.

Additional resources

Official documentation

Frequently asked questions

Why doesn't combineLatest emit anything until every input observable has emitted a value?

combineLatest waits because it needs a cached value from each input before it can build a combined result. It stores the latest value from every source, and only after all sources have emitted at least once does it emit the combined values, usually as an array. After that, any new emission from one source overrides its cached value and produces a new combined emission.

When does the stream returned by combineLatest complete in RxJS?

The resulting stream completes only after all input observables have completed. If any input never completes, combineLatest never completes either, even when the other inputs have finished. Completion is sent to the observer at the moment the last source completes.

What happens in combineLatest if one input completes without emitting a value?

The resulting stream completes immediately without emitting anything. Because combineLatest needs a value from every input to produce a combined result, an input that completes without ever emitting makes that result impossible. By contrast, an input that never emits and never completes leaves combineLatest neither emitting nor completing, since it waits indefinitely.

How does combineLatest handle an error thrown by one of its source observables?

If any source observable throws an error, combineLatest sends the error notification to the observer and the resulting stream errors. The error is propagated from the failing inner stream, so the subscriber receives it through its error handler. Plan for this when subscribing, because one failing source stops the combined stream.

When should I use combineLatest to compute state that depends on several streams?

Use combineLatest when a combined value must stay up to date whenever any part of the state changes. A typical case is a monitoring system where each service is a stream emitting a Boolean for its availability. The projection function can perform a logical AND, so the overall status is green only when all services are available.

Share this post