Implementing and Testing the RxJS `retry` Operator

Hello! Let's continue our journey in building a reactive library.

In our last lesson, we built the catchError operator, a powerful tool for handling errors by replacing a failing observable stream with a new one. This prevents errors from terminating our stream unexpectedly. Today, we'll explore another crucial error-handling strategy: giving a failing stream a second chance.

This lesson addresses the learning outcome: Implement and test a retry operator to resubscribe to a failing source observable a specified number of times. Instead of replacing the stream, retry will resubscribe to the original source, hoping it will succeed on a subsequent attempt. This is particularly useful for transient issues, like temporary network failures.

1. The Concept of Retrying

When an error occurs, sometimes the best course of action is simply to try again. The retry operator formalizes this concept within the world of observables. It intercepts an error notification and, instead of propagating it to the subscriber, it resubscribes to the source observable.

To see a quick demonstration of this in action, let's watch a short clip.

As you saw, retry is straightforward in its basic form: you tell it how many times to retry, and it handles the resubscription logic for you.

2. The Evolution of the retry API

In modern reactive libraries like RxJS, the retry operator has evolved. For a developer building a new framework, understanding this evolution provides valuable context for API design. Initially, complex retry logic (like adding a delay between attempts) required a more complicated operator called retryWhen. This has since been simplified.

Let's look at how the modern API compares to the old one.

The key takeaway is that a modern, developer-friendly API favors a declarative configuration object ({ count, delay }) over the more complex, imperative approach of retryWhen. For our lesson, we will implement the core retry(count) logic, which is the foundation for this more advanced configuration.

3. Implementing the retry Operator

Now, let's build our retry operator. It will be a pipeable operator that takes a count and returns a new observable.

The core logic is as follows:

1. Subscribe to the source observable.
2. If the source emits a next or complete notification, pass it through to the subscriber.
3. If the source emits an error:
   • Check if there are any retries left.
   • If yes, decrement the retry count and resubscribe to the source.
   • If no, propagate the error to the subscriber.

A clean way to implement the resubscription logic is with a function that handles the subscription process, which can call itself upon failure. Crucially, we must leverage our Subscription architecture to ensure that all internal subscriptions are properly cleaned up when the user unsubscribes.

Here is the skeleton for your retry.ts file. Your task is to implement the logic within the producer function.

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

export function retry<T>(count: number): OperatorFunction<T, T> {
  return (source: Observable<T>): Observable<T> => {
    return new Observable((subscriber) => {
      let retriesLeft = count;

      const subscribeToSource = () => {
        // Create a subscription to the source observable.
        const innerSubscription = source.subscribe({
          next(value) {
            // TODO: What happens when the source emits a value?
            subscriber.next(value);
          },
          error(err) {
            // TODO: This is the core retry logic.
            // 1. Check if retriesLeft > 0.
            // 2. If so, decrement retriesLeft and call subscribeToSource() again.
            // 3. If not, propagate the error to the main subscriber.
            if (retriesLeft > 0) {
              retriesLeft--;
              subscribeToSource();
            } else {
              subscriber.error(err);
            }
          },
          complete() {
            // TODO: What happens when the source completes successfully?
            subscriber.complete();
          },
        });

        // Add the subscription to the main subscriber's teardown logic.
        // This ensures that if the user unsubscribes, this inner subscription is cleaned up.
        subscriber.add(innerSubscription);
      };

      // Initial subscription
      subscribeToSource();
    });
  };
}

This structure elegantly handles resource management. Each time subscribeToSource is called, a new innerSubscription is created and added to the main subscriber. When the final consumer unsubscribes, the main subscriber's teardown logic is triggered, which will in turn call unsubscribe on all added innerSubscription instances, preventing memory leaks.

4. Testing Your retry Operator

Thorough testing is essential to confirm our operator behaves as expected. Create a retry.test.ts file and write tests for the following scenarios.

Key Test Cases:

1. No Error: The source completes without error. The retry operator should not interfere, and all values should be passed through.
2. Retries and Succeeds: The source fails once but succeeds on the second attempt. The subscriber should receive the values from the successful attempt.
3. Retries Exhausted: The source fails more times than the allowed retry count. The subscriber should receive the final error after all retries are exhausted.
4. retry(0): The source fails. With retry(0), the error should be propagated immediately without any retry attempts.
5. Unsubscription During Retry: Ensure that if a subscriber unsubscribes while the operator is in a retry sequence, all underlying subscriptions are terminated. You can test this by creating a source that errors asynchronously.

Here is an example test for the "Retries and Succeeds" scenario to get you started:

import { expect, test, vi } from "vitest";
import { Observable } from "../observable";
import { retry } from "./retry";

test("should retry and succeed on the second attempt", () => {
  const nextSpy = vi.fn();
  const errorSpy = vi.fn();
  const completeSpy = vi.fn();

  let attempt = 0;
  const source$ = new Observable<string>((subscriber) => {
    attempt++;
    if (attempt === 1) {
      subscriber.error("First attempt failed");
    } else {
      subscriber.next("A");
      subscriber.next("B");
      subscriber.complete();
    }
  });

  source$.pipe(retry(1)).subscribe({
    next: nextSpy,
    error: errorSpy,
    complete: completeSpy,
  });

  expect(attempt).toBe(2);
  expect(nextSpy).toHaveBeenCalledTimes(2);
  expect(nextSpy).toHaveBeenCalledWith("A");
  expect(nextSpy).toHaveBeenCalledWith("B");
  expect(errorSpy).not.toHaveBeenCalled();
  expect(completeSpy).toHaveBeenCalled();
});

For more inspiration on how to structure these kinds of tests, the article "RxJS: Retry With Delay" provides a link to a full test file for a similar operator.

Conclusion

In this lesson, you have added another essential error-handling operator to our library. You've learned:

• retry provides a mechanism to resubscribe to a failing observable, giving it another chance to succeed.
• The implementation requires careful management of a retry counter and the resubscription process.
• Leveraging our existing Subscription architecture with subscriber.add() is key to ensuring proper resource cleanup and preventing memory leaks.
• Modern reactive APIs favor simple, declarative configurations (e.g., retry({ count, delay })) for common tasks.

You now have two complementary error-handling strategies: catchError to replace a stream and retry to repeat it.

In our next lesson, we will shift our focus from error handling to controlling the lifecycle of a stream. You will implement the take operator, which allows you to limit the number of emissions from a source before it automatically completes.