zip – RxJS Reference

zip

Q: What is the difference between zip and combineLatest in RxJS?

zip() pairs values by their index, emitting only when every source has a new value that has not been used yet, while combineLatest() emits whenever any source emits, combining it with the latest value from the others. Because zip() waits for the slower stream, its output rate is dictated by the slowest source. Use zip() when you need strictly matched pairs, and combineLatest() when you need the most recent value from each stream.

Q: How does zip() decide when to emit values in RxJS?

zip() caches each value from every source observable under its index and emits a tuple only once every source has a cached value for that same index. If one source emits faster than another, the extra values are buffered until their counterparts arrive. As a result, the publishing rate of the combined stream is limited by the slowest input stream.

Q: When does a zip() observable complete in RxJS?

The stream returned by zip() completes when any inner observable completes and no more cached pairs remain to be emitted. It does not complete if all inner streams are still active, and if any inner stream errors, the error is forwarded to the subscriber immediately. Make sure the shorter source completes only after its pending values have been matched, or you may lose expected output.

Q: How do I combine two observables into pairs with zip in RxJS?

Pass the observables directly to zip(), for example zip(a, b).subscribe(observer), and each emission will be an array containing one value from a and one from b at the same position. To transform the pair, use the map() operator after zip(). The subscription receives values only when both streams have produced their next value, so no manual synchronization is needed.

zip operator combines observable streams in a way that resembles the mechanics of a zipper on clothing or a bag. It brings together two or more sequences of corresponding values as a tuple (a pair in case of two input streams). It waits for the corresponding value to be emitted from all input streams, then transforms them into a single value using a projection function and emits the result. It will only emit the value once it has a pair of fresh values from each source sequence. So if one of the source observables emits values faster than the other sequence, the rate of publishing will be dictated by the slower of the two sequences.

The resulting stream completes when any of the inner streams complete and the corresponding matched pairs are emitted from other streams. It will never complete if any of the inner streams doesn’t complete and will throw an error if any of the inner streams errors out.

The operator works in the following way:

Subscribe to all input observables
When a source observable emits a value, add to the cache for this source observable under certain index
If there’s a cached value for each observable in the cache for this index, emit the combined values to the observer
As long as one of the source observables complete, and no more cached pairs to emit, send the complete notification to the observer, zip complete.
If any of the source observables throws an error, send the error notification to the observer.

In the diagram below you can see the zip operator combining two streams A and B. As soon as a corresponding pair is matched the resulting sequence produces a combined value:

zip 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);

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

And the editable demo:

Additional resources

Official documentation

Frequently asked questions

What is the difference between zip and combineLatest in RxJS?

zip() pairs values by their index, emitting only when every source has a new value that has not been used yet, while combineLatest() emits whenever any source emits, combining it with the latest value from the others. Because zip() waits for the slower stream, its output rate is dictated by the slowest source. Use zip() when you need strictly matched pairs, and combineLatest() when you need the most recent value from each stream.

How does zip() decide when to emit values in RxJS?

zip() caches each value from every source observable under its index and emits a tuple only once every source has a cached value for that same index. If one source emits faster than another, the extra values are buffered until their counterparts arrive. As a result, the publishing rate of the combined stream is limited by the slowest input stream.

When does a zip() observable complete in RxJS?

The stream returned by zip() completes when any inner observable completes and no more cached pairs remain to be emitted. It does not complete if all inner streams are still active, and if any inner stream errors, the error is forwarded to the subscriber immediately. Make sure the shorter source completes only after its pending values have been matched, or you may lose expected output.

How do I combine two observables into pairs with zip in RxJS?

Pass the observables directly to zip(), for example zip(a, b).subscribe(observer), and each emission will be an array containing one value from a and one from b at the same position. To transform the pair, use the map() operator after zip(). The subscription receives values only when both streams have produced their next value, so no manual synchronization is needed.

Share this post

zip – RxJS Reference

zip

Additional resources

See also

Frequently asked questions

Sign up for our newsletter