Combining Streams with RxJS Operators

Hello! Welcome back to our in-depth exploration of RxJS.

In our previous lesson, we honed our ability to control emissions from a single stream using operators like distinctUntilChanged and sampleTime. We learned to filter out noise and redundant data, a crucial step in building efficient, reactive applications.

Today, we take the next logical step: moving from managing a single stream to orchestrating multiple streams at once. This lesson addresses the learning outcome: Combine multiple streams declaratively using combineLatest, merge, concat, and zip. Mastering these operators is fundamental to solving complex asynchronous problems, such as coordinating multiple API requests, merging different user input sources, or calculating derived state in a React application.

1. A Tale of Two Strategies: Flattening vs. Pairing

Before we dive into the specific operators, it's helpful to categorize them based on their primary strategy. When combining streams, operators generally do one of two things:

1. Flattening (merge, concat): These operators take multiple streams and flatten their emissions into a single output stream. The values themselves are passed through unchanged, as if they all came from one source. Think of it as merging several lanes of traffic into one.
2. Pairing (zip, combineLatest): These operators also take multiple streams, but instead of just passing values through, they wait for specific conditions to be met and then emit a new value that is a combination (usually an array or tuple) of values from the source streams. Think of it as assembling a product from parts that arrive on different conveyor belts.

We will explore each of these categories.

2. Flattening Operators: merge and concat

These operators are your go-to tools when you want to combine events from multiple sources into a single, unified stream.

merge: The Concurrent Combiner

The merge operator subscribes to all input observables simultaneously and simply passes on the values as they are emitted, in the order they arrive. It's like mixing several audio channels into one—you hear every sound as it happens.

The resulting stream completes only when all of the input streams have completed.

When to use merge:
Use merge when you have multiple streams of the same type of event and you want to react to them in the same way, regardless of which stream they came from. The order of emissions between streams is not guaranteed, only the temporal order of their arrival.

• Example: In a UI, you might want to trigger a "show activity" notification in response to several different user actions. You can merge a stream of clicks$, keyStrokes$, and mouseMoves$ into a single userActivity$ stream.

concat: The Sequential Combiner

The concat operator also flattens multiple streams into one, but it does so sequentially. It subscribes to the first observable, lets it emit all its values and complete, and only then subscribes to the second observable, and so on. It's a queue for observables.

When to use concat:
Use concat when the order of operations is critical. It guarantees that one stream's work is finished before the next one begins.

• Example: A classic use case is handling cached data. You can concat a stream that gets data from a cache (fromCache$) with a stream that gets it from an API (fromApi$). The fromApi$ stream will only be subscribed to if fromCache$ completes (e.g., emits the cached value and finishes, or just completes without a value).

A word of caution: If you concat an observable that never completes (like a stream of button clicks), any subsequent observables in the chain will never be subscribed to.

3. Pairing Operators: zip and combineLatest

These operators are used when you need to create a new state based on the values from multiple sources.

zip: The Synchronized Pairer

The zip operator waits until it has received one new emission from each of its input observables. It then "zips" them together into an array and emits that array. It maintains an internal queue for each stream, so if one stream emits faster than another, the extra values are buffered until their counterparts arrive.

This video provides a fantastic real-world analogy that makes zip intuitive.

When to use zip:
Use zip when you have streams whose values are related by their emission index. You want to pair the first value from stream A with the first from stream B, the second with the second, and so on.

• Example: Imagine you have two streams: one emitting user IDs (userIds$) and another emitting user avatars (userAvatars$) in the same order. You can zip them to create a stream of [userId, userAvatar] pairs.

combineLatest: The State Combiner

The combineLatest operator is one of the most frequently used combination operators in UI development. It also combines values into an array, but its logic is different from zip:

1. It waits for all input observables to have emitted at least one value.
2. After this initial emission, it emits a new array of the latest values whenever any of the input observables emits a new value.

The same video you just watched has an excellent segment comparing zip directly to combineLatest.

When to use combineLatest:
Use combineLatest whenever you have a piece of state that depends on the latest values from several independent sources.

• Example (highly relevant to React): A classic form validation scenario. You have a username$ stream and a password$ stream. A isFormValid$ stream can be created by using combineLatest([username$, password$]). The combined stream will emit a new validity status whenever the username or the password changes, allowing you to enable/disable a submit button in real-time.

4. Summary and Pitfalls

To solidify your understanding, let's turn to a comprehensive article that summarizes these operators and discusses their performance characteristics and common pitfalls.

Practice Thought-Exercise

Consider a simple e-commerce UI with the following features:

• A dropdown to select a currency (USD, EUR).
• A slider to select a maximum price.
• A text input for a product search query.

You need to fetch a list of products from an API whenever any of these controls change. The API endpoint is api/products?currency=...&maxPrice=...&query=....

Which combination operator would you use to combine the streams from these three UI controls to trigger the API call? Why?

Conclusion

You have now learned the four fundamental operators for combining streams. Understanding their distinct behaviors is a major step toward writing sophisticated reactive code.

Key Takeaways:

• merge: Combines streams concurrently. Use when you want to react to events from multiple sources as they happen.
• concat: Combines streams sequentially. Use when the order of execution is critical.
• zip: Pairs emissions one-to-one. Use when you need to synchronize streams based on their emission index.
• combineLatest: Combines the latest values from all streams. Use when computing derived state that depends on multiple sources.

In our next lesson, we will complete our tour of the primary combination operators by looking at forkJoin. We'll see how it's used to execute a group of observables in parallel and collect their results, much like Promise.all, making it perfect for handling multiple, finite async operations like HTTP requests.