Static Merge Operator with Type Inference

Hello! Welcome back to our course on building a modern JavaScript framework.

In the previous modules, you've built a robust foundation for our observable library, including the core classes, creation functions like of and interval, and essential transformation operators like map and filter.

Today, we move into Module 5: Pipeable Operators: Combination & Timing. We'll start by implementing our first combination operator, which, as the name suggests, combines multiple source observables into a single output stream.

Our goal for this lesson is to implement and test a static merge operator that combines multiple observables, inferring the output type as a union of the input observable types. This operator is fundamental for handling concurrent events from different sources, like user inputs from multiple buttons or data from parallel API calls.

1. Understanding the merge Operator

Before we write any code, let's get a clear conceptual understanding of what merge does. It takes multiple input observables and creates a single output observable that emits all the values from all the inputs, in the order they arrive.

To get a good overview, please read the introduction and first example from the RxJS documentation on merge.

As you saw, merge acts like a funnel. It doesn't care which observable a value comes from; it just pushes it into the output stream as soon as it's emitted. The resulting stream completes only when all of the source observables have completed. If any source stream errors, the merged stream will immediately error as well.

2. The Type-Safe Signature: Inferring Union Types

A key requirement of our implementation is type safety. If we merge an Observable<string> and an Observable<number>, the resulting observable should be correctly typed as Observable<string | number>. This is achieved using TypeScript's union types.

To achieve this, we'll define a static merge function that uses generics and rest parameters. The function signature is a bit advanced, but it's a powerful pattern in TypeScript.

// src/operators/merge.ts

import { Observable } from '../observable';
import { Subscription } from '../subscription';
import { Observer } from '../observer';

// This is the type we want to achieve. Let's break it down.
type ObservableValue<T> = T extends Observable<infer U> ? U : never;

export function merge<T extends Observable<any>[]>(
  ...observables: T
): Observable<ObservableValue<T[number]>> {
  // Implementation will go here
}

Let's dissect that return type: Observable<ObservableValue<T[number]>>.

• T extends Observable<any>[]: T is a tuple type representing the array of observables passed to the function (e.g., [Observable<string>, Observable<number>]).
• T[number]: This is an indexed access type. It gets the types of the elements in the array T, resulting in a union. For our example, this would be Observable<string> | Observable<number>.
• ObservableValue<...>: This is a conditional type we defined. It takes an observable type and "extracts" its inner value type.
  • When TypeScript applies this to the union Observable<string> | Observable<number>, it distributes the conditional type over the union.
  • It evaluates ObservableValue<Observable<string>> which results in string.
  • It evaluates ObservableValue<Observable<number>> which results in number.
  • The final result is the union of these types: string | number.

This elegant piece of type-level programming ensures our merge function is perfectly type-safe.

3. Implementation Strategy

Now for the runtime logic. Our merge function must return a new Observable. The core logic resides within the producer function passed to the Observable constructor.

Here's the plan:

1. Initialization: Inside the producer, we need a counter to track how many of the input observables are still active (i.e., have not completed). Initialize it to the total number of input observables.
2. Subscription Management: We'll need an array to hold the Subscription objects for each input observable. This is crucial for cleanup.
3. Subscribe to All Sources: Iterate over the input observables. For each one, subscribe to it.
4. Forward Emissions:
   • next: When any input observable emits a value, pass it directly to the observer of our merged stream.
   • error: If any input observable errors, immediately call error on the merged stream's observer. This will terminate the whole process.
   • complete: When an input observable completes, decrement the active observables counter. If the counter reaches zero, it means all sources are done, so we can call complete on the merged stream's observer.
5. Teardown Logic: The producer function must return a Subscription whose unsubscribe method will clean up everything. This method should iterate through the array of inner subscriptions and unsubscribe from all of them. This prevents memory leaks if a consumer unsubscribes before the merged stream completes naturally.

4. Your Task: Implement merge

It's time to put this all together. Create a new file src/operators/merge.ts and implement the merge function.

Here is a skeleton to get you started.

// src/operators/merge.ts

import { Observable } from '../observable';
import { Subscription } from '../subscription';
import { Observer } from '../observer';

type ObservableValue<T> = T extends Observable<infer U> ? U : never;

export function merge<T extends Observable<any>[]>(
  ...observables: T
): Observable<ObservableValue<T[number]>> {
  return new Observable(observer => {
    let activeSubscriptions = observables.length;
    const subscriptions = new Subscription();

    if (activeSubscriptions === 0) {
      observer.complete();
      return subscriptions;
    }

    const sources = observables.map(source$ => {
      const innerSub = source$.subscribe({
        next: value => {
          observer.next(value);
        },
        error: err => {
          observer.error(err);
        },
        complete: () => {
          activeSubscriptions--;
          if (activeSubscriptions === 0) {
            observer.complete();
          }
        },
      });
      return innerSub;
    });

    sources.forEach(sub => subscriptions.add(sub));

    return subscriptions;
  });
}

A quick note on the skeleton:
I've used the Subscription class's add method, which you implemented in Module 2. It allows grouping multiple subscriptions together so they can all be unsubscribed with a single call. This is a common and clean pattern for operators that manage multiple inner subscriptions.

5. Testing Your Implementation

With your merge function implemented, the next step is to verify its behavior with unit tests. Create a new test file src/operators/merge.test.ts.

The article on testing RxJS provides excellent patterns for testing complex observable chains, even though it uses a slightly different operator (mergeWith). The key takeaway is using Subjects or simple of/interval observables to control the sources and assert the output.

Here are the test cases you should cover:

1. Merging Different Types:

   • Create of('a', 'b') and of(1, 2).
   • Merge them and collect the results into an array.
   • Assert that the final array contains ['a', 'b', 1, 2] (or a similar interleaved order, since of is synchronous).
   • The TypeScript compiler should correctly infer the output type as Observable<string | number>.

2. Asynchronous Merging:

   • Use vitest.useFakeTimers().
   • Create two interval observables with different periods (e.g., 100ms and 150ms). You can use pipe with map to give their outputs distinct values.
   • Merge them and advance the timers.
   • Assert that the values are emitted in the correct interleaved time order.

3. Completion Logic:

   • Merge two finite observables (e.g., created with of or a take operator on an interval).
   • Assert that the complete callback is only called after both source observables have completed.

4. Error Handling:

   • Create one observable that emits values and another that immediately errors (e.g., new Observable(obs => obs.error('test error'))).
   • Merge them.
   • Assert that the error callback of the merged subscription is called and that no values from the other observable are emitted.

5. Unsubscription:

   • Create two interval observables.
   • Merge them and subscribe.
   • Immediately (or after a short delay) call unsubscribe() on the merged subscription.
   • Assert that no values are received. This implicitly tests that the inner subscriptions were torn down.

Conclusion

In this lesson, you've implemented merge, your first combination operator. This is a significant step, moving from single-stream transformations to orchestrating multiple streams concurrently.

Key Takeaways:

• merge combines multiple observables into one, emitting values as they arrive from any source.
• The merged stream completes only when all sources complete, and errors if any source errors.
• A robust implementation requires careful management of inner subscriptions to prevent memory leaks.
• Advanced TypeScript generics (like conditional and indexed access types) are essential for creating truly type-safe operators that correctly infer complex output types like unions.

In our next lesson, we will implement another powerful combination operator: combineLatest. We'll explore how it differs from merge by combining the latest values from each stream into a single emission, and how its typing requirements introduce new challenges.