Catch and Switch: Flexible Error Handling with `catchError`

Hello! Welcome to the next lesson in our course on building a modern JavaScript framework.

In previous modules, we've laid a robust foundation for our reactive library. We've implemented the core Observable architecture and built a suite of powerful operators for transformation, filtering, and timing. Now, we'll turn our attention to a crucial aspect of any real-world application: error handling.

This lesson focuses on implementing the catchError operator. You will learn how to intercept errors in an observable stream and decide whether to suppress the error by providing a fallback value or to propagate it after performing some action. We'll pay special attention to creating a type-safe and flexible operator signature using TypeScript's advanced features, specifically conditional types.

1. The Fragility of a Stream: The Observable Contract

Before we build an error-handling operator, we must first understand why it's so essential. The behavior of observables is governed by a strict set of rules known as the "Observable Contract."

To understand this contract, please read the first section of the following article.

As you've just read, an observable stream is terminal. It can emit many next notifications, but once it sends an error (or complete) notification, it's finished. It cannot emit any more values.

Consider a stream that listens for button clicks, where each click triggers an API request. If one of those API requests fails and emits an error, the entire stream terminates. Subsequent button clicks would do nothing. This is rarely the desired behavior. We need a way to "catch" the error and prevent it from destroying our stream.

2. Introducing catchError: Rescuing the Stream

The catchError operator is our tool for this job. It intercepts an error notification from its source observable and, instead of letting it propagate, it invokes a function you provide. This function receives the error and must return a new observable, which will be used as a replacement for the one that just failed.

This allows for two primary strategies:

1. Catch and Replace: Your function returns an observable that emits a default or fallback value. The main stream continues, unaware that an error ever occurred.
2. Catch and Rethrow: Your function performs a side effect (like logging the error) and then returns an observable that immediately errors out. This allows you to inspect an error without stopping its propagation to the final subscriber.

Let's explore these strategies in more detail.

To see a practical demonstration of handling errors in nested streams—a very common scenario—the following video is helpful.

3. Implementing the catchError Operator

Now it's time to implement our own catchError operator. It will be a pipeable operator, meaning it's a function that returns another function that transforms a source observable.

The structure will be: catchError(selector) => (source) => new Observable(...).

• The selector function is what you, the user of the operator, will provide. It takes the error as an argument and must return a new Observable.
• The operator subscribes to the source observable.
• If the source emits a value or completes, the operator passes it along.
• If the source errors, the operator calls your selector function, gets the replacement observable, and subscribes to it, piping its emissions to the final destination.

A Flexible and Type-Safe Signature

A key part of this task is to define a flexible return signature using TypeScript. The output observable should be typed as a union of the source's value type (T) and the replacement observable's value type (R). The signature will look like this:

export function catchError<T, R>(selector: (err: any) => Observable<R>): OperatorFunction<T, T | R>

Here, OperatorFunction<A, B> is an alias for (source: Observable<A>) => Observable<B>.

To make this even more robust and align with modern RxJS, we can define a utility type using conditional types to extract the value type from the observable returned by the selector.

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

With this, our signature can be even more generic:

export function catchError<T, O extends Observable<any>>(selector: (err: any) => O): OperatorFunction<T, T | ObservedValueOf<O>>

This signature correctly infers the type of the replacement stream and creates the appropriate union type for the output.

Implementation Steps

Here is the skeleton for your catchError.ts file. Your task is to fill in the logic inside the new Observable's producer function.

import { Observable, Observer } from "../observable";
import { OperatorFunction } from "../types";

// This is the core implementation.
export function catchError<T, R>(
  selector: (err: any) => Observable<R>
): OperatorFunction<T, T | R> {
  return (source: Observable<T>): Observable<T | R> => {
    return new Observable((subscriber: Observer<T | R>) => {
      const sourceSubscription = source.subscribe({
        next(value) {
          // TODO: What happens when the source emits a value?
          subscriber.next(value);
        },
        error(err) {
          // TODO: This is the main logic.
          // 1. Call the selector function with the error.
          // 2. Subscribe to the observable returned by the selector.
          // 3. Pipe the replacement's notifications to the subscriber.
          const replacement$ = selector(err);
          const replacementSubscription = replacement$.subscribe({
            next(value) { subscriber.next(value); },
            error(err) { subscriber.error(err); },
            complete() { subscriber.complete(); }
          });
          // Don't forget to include this new subscription in the teardown logic!
          subscriber.add(replacementSubscription);
        },
        complete() {
          // TODO: What happens when the source completes?
          subscriber.complete();
        },
      });

      // The main subscription to the source stream.
      subscriber.add(sourceSubscription);
    });
  };
}

Notice the use of subscriber.add(). In our previous lessons, we implemented the Subscription class with an add method to handle teardown logic. It's crucial that when the final subscription is torn down, both the subscription to the source and any potential subscription to the replacement observable are cleaned up.

4. Testing Your catchError Operator

With the implementation complete, you need to verify its correctness with unit tests. Create a catchError.test.ts file and use Vitest to cover the following scenarios.

Key Test Cases:

1. No Error: Create a source observable that emits values and completes. The catchError operator should pass all values and the completion notification through without modification.
2. Catch and Replace:
   • Create a source that emits an error (e.g., throwError('Oops')).
   • Use catchError to return a new observable with a fallback value (e.g., of('Fallback Value')).
   • Assert that the subscriber receives the fallback value and a complete notification, but not the error.
3. Catch and Rethrow:
   • Create a source that emits an error.
   • Use catchError to return throwError('New Error').
   • Assert that the subscriber's error callback is invoked with 'New Error'.
4. Asynchronous Error:
   • Ensure your logic works when the source observable errors asynchronously (e.g., inside a setTimeout). Vitest's fake timers will be useful here.

Here is an example of a "Catch and Replace" test to get you started:

import { expect, test, vi } from "vitest";
import { of, throwError } from "../creation";
import { catchError } from "./catchError";

test("should catch an error and switch to a new observable", () => {
  const errorSpy = vi.fn();
  const nextSpy = vi.fn();
  const completeSpy = vi.fn();

  const source$ = throwError<string>("Original Error");
  const fallback$ = of("Fallback Value");

  source$.pipe(catchError((err) => {
    expect(err).toBe("Original Error");
    return fallback$;
  })).subscribe({
    next: nextSpy,
    error: errorSpy,
    complete: completeSpy,
  });

  expect(nextSpy).toHaveBeenCalledWith("Fallback Value");
  expect(nextSpy).toHaveBeenCalledTimes(1);
  expect(errorSpy).not.toHaveBeenCalled();
  expect(completeSpy).toHaveBeenCalled();
});

Conclusion

In this lesson, you've implemented one of the most important operators in a reactive library. You've learned:

• The Observable Contract dictates that a stream terminates upon an error.
• The catchError operator intercepts errors and allows you to substitute a replacement observable, saving the stream from termination.
• The two main error handling strategies are Catch and Replace (providing a fallback) and Catch and Rethrow (propagating the error).
• Using TypeScript generics and conditional types enables the creation of a flexible and type-safe operator signature that correctly infers the output type.

This robust error handling is fundamental. In our next lesson, we will build upon this by implementing the retry operator, which, instead of switching to a new observable, gives the original source observable another chance to succeed by resubscribing to it.