Implementing `combineLatest` with Tuple Types

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

In our last lesson, we implemented the merge operator, which acts like a funnel, emitting values from multiple source observables as they arrive. This was our first foray into combining streams.

Today, we'll continue in Module 5: Combination & Timing by implementing another fundamental combination operator: combineLatest. While its name sounds similar to merge, its behavior is quite different and serves a distinct purpose: synchronizing the latest state from multiple streams.

Our learning goal is to implement and test a static combineLatest operator, using tuple types to correctly type the combined array of latest values. This will require us to dive deeper into TypeScript's type system to ensure our operator is not only functional but also fully type-safe.

1. Understanding combineLatest

combineLatest creates an observable that emits an array containing the latest value from each of its source observables. It has two key characteristics:

1. It will not emit any value until every source observable has emitted at least once.
2. After the initial emission, it will emit a new array whenever any of the source observables emits a new value. This new array will contain the newly emitted value along with the last-emitted values from all other sources.

To visualize this, let's start with the official RxJS documentation.

As you can see, combineLatest is perfect for scenarios where you need a combined view of the state of several independent streams. For example, calculating a value based on the latest inputs from several form fields.

2. The Typing Challenge: From Union Types to Tuples

In the merge operator, the output was Observable<string | number> because the stream could emit either a string or a number at any time. With combineLatest, the output is more structured: it's an array where the element at index 0 always corresponds to the first input observable, the element at index 1 to the second, and so on.

If we combine Observable<string> and Observable<number>, we don't want an Observable<(string | number)[]>. This type is too loose; it allows [1, 'hello'], which is incorrect. We need an Observable<[string, number]>. This is where TypeScript tuples become essential.

A tuple is an array with a fixed number of elements whose types are known. To get a quick refresher on them, please watch the following video.

This exact typing problem is a common point of confusion when using combineLatest. The following Stack Overflow thread perfectly illustrates the issue and why tuples are the correct solution.

3. A Type-Safe Signature with Mapped Types

To achieve our desired Observable<[string, number]> output, we need a sophisticated generic signature that can transform a tuple of Observable types into a tuple of their inner value types. We can do this with a mapped type.

// A helper to extract the inner value type from an Observable
type ObservedValue<O> = O extends Observable<infer T> ? T : never;

// A mapped type that transforms a tuple of Observables into a tuple of their values
type ObservedValues<T extends readonly Observable<any>[]> = {
  [K in keyof T]: ObservedValue<T[K]>;
};

export function combineLatest<T extends readonly Observable<any>[]>(
  sources: [...T]
): Observable<ObservedValues<T>> {
  // Implementation goes here
}

Let's break down this signature:

• T extends readonly Observable<any>[]: We constrain T to be a tuple of observables. Using readonly helps TypeScript preserve the tuple structure.
• sources: [...T]: Using a rest parameter in the arguments also helps TypeScript infer T as a tuple type rather than widening it to a regular array type (e.g., Observable<string | number>[]).
• ObservedValues<T>: This is the core of our solution. It's a mapped type that iterates over each key K in the tuple T (which are the indices '0', '1', etc.) and applies our ObservedValue helper to each observable in the tuple.
  • Input: [Observable<string>, Observable<number>]
  • Output: [string, number]

This ensures that our combineLatest function will have a perfectly typed output tuple.

4. Implementation Strategy

The implementation of combineLatest is more stateful than merge. The producer function returned by our operator will need to manage several pieces of state:

1. latestValues: An array to store the most recent value from each source observable.
2. hasEmitted: A boolean array to track which sources have emitted at least once.
3. emittedCount: A number to track how many sources have emitted. We can only start producing output when this equals the total number of sources.
4. completedCount: A number to track how many sources have completed. The output stream completes when this number equals the total number of sources.

The logic for the inner observer provided to each source will be:

• next(value):
  1. Check if the source has emitted before. If not, mark it as having emitted and increment emittedCount.
  2. Update the latestValues array at the source's index with the new value.
  3. If emittedCount equals the total number of sources, it means all streams are now active. Emit a copy of the latestValues array to the main observer.
• error(err): Immediately propagate the error to the main observer and clean up.
• complete():
  1. Increment completedCount.
  2. If completedCount equals the total number of sources, complete the main observer.
• Teardown: As with merge, ensure that unsubscribing from the main combineLatest observable unsubscribes from all inner source observables.

5. Your Task: Implement and Test combineLatest

Now it's time to implement the operator. Create a new file src/operators/combineLatest.ts and use the following skeleton.

// src/operators/combineLatest.ts

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

// Helper types from section 3
type ObservedValue<O> = O extends Observable<infer T> ? T : never;
type ObservedValues<T extends readonly Observable<any>[]> = {
  [K in keyof T]: ObservedValue<T[K]>;
};

export function combineLatest<T extends readonly Observable<any>[]>(
  ...sources: T
): Observable<ObservedValues<T>> {
  return new Observable(observer => {
    if (sources.length === 0) {
      observer.complete();
      return new Subscription();
    }

    const subscriptions = new Subscription();
    const latestValues = new Array(sources.length);
    const hasEmitted = new Array(sources.length).fill(false);
    let emittedCount = 0;
    let completedCount = 0;

    sources.forEach((source$, i) => {
      const innerSub = source$.subscribe({
        next: value => {
          latestValues[i] = value;
          if (!hasEmitted[i]) {
            hasEmitted[i] = true;
            emittedCount++;
          }
          if (emittedCount === sources.length) {
            // Important: emit a copy of the array
            observer.next([...latestValues] as ObservedValues<T>);
          }
        },
        error: err => {
          observer.error(err);
        },
        complete: () => {
          completedCount++;
          if (completedCount === sources.length) {
            observer.complete();
          }
        },
      });
      subscriptions.add(innerSub);
    });

    return subscriptions;
  });
}

Testing Your Implementation

Create a corresponding test file src/operators/combineLatest.test.ts. Your tests should verify all aspects of the operator's behavior.

Test Cases to Cover:

1. Type Safety (Compile-time check):

   • const result$ = combineLatest(of('A'), of(1));
   • Verify that the type of result$ is inferred as Observable<[string, number]>.

2. Initial Emission Logic:

   • Use vitest.useFakeTimers() and two delayed observables (e.g., using a custom observable with setTimeout).
   • Assert that no value is emitted until both sources have emitted their first value.
   • Assert that the first emitted value is an array containing the first value from each source.

3. Subsequent Emissions:

   • Continue from the previous test. Trigger a new emission from one of the sources.
   • Assert that a new array is emitted immediately, containing the new value from the triggered source and the old value from the other.

4. Completion Logic:

   • Combine two finite observables (e.g., of('a', 'b') and of(1, 2)).
   • Assert that the complete callback is called only after both sources have completed.

5. Error Handling:

   • Combine a valid observable with one that errors.
   • Assert that the error callback is triggered on the combined stream as soon as the source errors.

6. Unsubscription:

   • Combine two interval observables.
   • Subscribe and then immediately unsubscribe.
   • Use spies or advance timers to ensure no values are ever emitted to the observer's next method, proving the inner subscriptions were torn down.

For inspiration on combining streams in a component, you can review this short video. While it uses Angular, the core RxJS pattern of combining streams and then mapping the result to a more convenient object is a universal and useful technique.

Conclusion

Excellent work! You've now implemented combineLatest, a powerful operator for managing and synchronizing state from multiple sources. You've also tackled a significant typing challenge, using TypeScript's mapped and tuple types to create a fully type-safe implementation.

Key Takeaways:

• combineLatest waits for all sources to emit once, then emits an array of the latest values whenever any source changes.
• It is fundamentally different from merge, which simply passes through all values as they arrive.
• Tuple types are the correct way to type the output of combineLatest to preserve positional type information.
• Advanced generic patterns, like mapped types, allow us to create flexible and type-safe function signatures that automatically infer these complex tuple types.

In our next lesson, we will shift our focus from combining streams to controlling their timing. We'll dive into implementing operators like delay, debounceTime, and throttleTime, making extensive use of Vitest's fake timers to test their asynchronous behavior deterministically.