Implementing and Testing `takeUntil`

Hello! Let's dive into our next lesson.

In our previous session, we implemented the take operator, which completes a stream after a fixed number of emissions. This was our first step in programmatically managing the lifecycle of an observable. Today, we'll build on that by exploring a more dynamic way to complete a stream.

This lesson addresses the learning outcome: Implement and test a takeUntil operator that completes the source stream based on an emission from a notifier observable. This operator is a cornerstone of reactive programming, especially for managing subscriptions tied to UI component lifecycles or other asynchronous events. It introduces the important concept of coordinating between two separate observable streams.

1. Understanding the takeUntil Operator

The takeUntil operator mirrors a source observable, passing its values along. However, it also subscribes to a second observable, called the "notifier." The moment the notifier emits its first value, takeUntil completes the main stream. Any subsequent emissions from the notifier are ignored.

This behavior is perfectly captured by a marble diagram.

To get a formal definition and see its signature, let's consult the official RxJS documentation.

Now, let's see a practical demonstration of this concept. The most common use case involves a long-lived stream (like an interval or a WebSocket connection) and a Subject that acts as the notifier.

The pattern you just saw is fundamental for preventing memory leaks in component-based frameworks like Angular, React, or Vue, where you might trigger the notifier subject when a component is unmounted.

2. Implementing the takeUntil Operator

Now it's time to build our own takeUntil. This operator is more complex than take because it involves managing two subscriptions simultaneously: one to the source and one to the notifier.

Here's the implementation strategy:

1. The takeUntil function accepts a notifier observable and returns a pipeable operator.
2. The returned operator creates and returns a new Observable.
3. Inside the producer function of this new observable, we subscribe to the source. Any next, error, or complete notifications from the source are passed directly to our main subscriber.
4. Crucially, we also subscribe to the notifier observable.
5. When the notifier emits its first next value, we immediately call subscriber.complete(). We don't need to listen for any more values from the notifier.
6. Resource Management: Both the source and notifier subscriptions must be cleaned up properly. The Subscription object we built is perfect for this. We will add both internal subscriptions to the main subscriber's subscription. When the stream terminates (either by completion, error, or unsubscription), our Subscription teardown logic will automatically unsubscribe from both the source and the notifier.

Here is the skeleton for your takeUntil.ts file. Your task is to set up and manage the two subscriptions.

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

export function takeUntil<T>(notifier: Observable<any>): OperatorFunction<T, T> {
  return (source: Observable<T>): Observable<T> => {
    return new Observable((subscriber) => {
      // Subscribe to the source observable.
      // Pass through next, error, and complete events.
      const sourceSubscription = source.subscribe({
        next(value) {
          subscriber.next(value);
        },
        error(err) {
          subscriber.error(err);
        },
        complete() {
          subscriber.complete();
        },
      });

      // Add the source subscription to the main subscription for cleanup.
      subscriber.add(sourceSubscription);

      // TODO: Implement the notifier logic.
      // 1. Subscribe to the `notifier` observable.
      // 2. In the `next` handler of the notifier's subscription, call `subscriber.complete()`.
      //    This will trigger the teardown of the whole chain.
      // 3. You don't need to handle the notifier's `error` or `complete` cases,
      //    as they don't cause the source to complete.
      // 4. Add the notifier subscription to the main subscriber for proper cleanup.
      const notifierSubscription = notifier.subscribe({
        next() {
          subscriber.complete();
        },
        // No need to handle error or complete from the notifier
      });

      subscriber.add(notifierSubscription);
    });
  };
}

This structure ensures that no matter how the stream terminates—the notifier fires, the source completes, the source errors, or the consumer unsubscribes—all internal subscriptions are properly disposed of.

3. Testing Your takeUntil Operator

A robust test suite is essential. For takeUntil, we need to verify its behavior in several key scenarios.

Create a takeUntil.test.ts file and add tests for the following cases:

1. Notifier emits before source completes: The source should stop emitting values and complete as soon as the notifier emits.
2. Source completes before notifier emits: The output stream should complete normally when the source completes.
3. Source errors before notifier emits: The error should be propagated to the subscriber.
4. Notifier completes without emitting: The source stream should be unaffected and continue emitting values.
5. Unsubscription: If the consumer unsubscribes, ensure both the source and notifier subscriptions are torn down. You can use vi.spyOn on the unsubscribe methods of mock observables to confirm this.

Here’s a sample test for the primary success case to get you started. You will need an implementation of Subject from a previous lesson to write this test.

import { expect, test, vi } from "vitest";
import { interval } from "../creation/interval"; // Assuming you have this
import { Subject } from "../subject"; // Assuming you have this
import { takeUntil } from "./takeUntil";
import { take } from "./take"; // Useful for making the test deterministic

test("should emit values until the notifier emits", () => {
  const nextSpy = vi.fn();
  const completeSpy = vi.fn();
  const errorSpy = vi.fn();

  const source$ = interval(10); // Emits 0, 1, 2, ... every 10ms
  const notifier$ = new Subject<void>();

  // We use take(5) to ensure the test terminates even if takeUntil fails.
  source$.pipe(takeUntil(notifier$), take(5)).subscribe({
    next: nextSpy,
    complete: completeSpy,
    error: errorSpy,
  });

  // Let some values pass through
  setTimeout(() => {
    // At 25ms, source should have emitted 0 and 1.
    expect(nextSpy).toHaveBeenCalledWith(0);
    expect(nextSpy).toHaveBeenCalledWith(1);
    
    // Now, trigger the notifier
    notifier$.next();
  }, 25);

  // Wait for the stream to complete
  return new Promise((resolve) => {
    setTimeout(() => {
      expect(nextSpy).toHaveBeenCalledTimes(2); // Only 0 and 1 should have been emitted
      expect(completeSpy).toHaveBeenCalledOnce();
      expect(errorSpy).not.toHaveBeenCalled();
      resolve();
    }, 50);
  });
});

Note: This test uses setTimeout for simplicity. In a more advanced setup, you would use fake timers (vi.useFakeTimers()) for more reliable and faster asynchronous tests.

4. An Important Pitfall: Operator Order

Your experience as a lead developer means you know that subtle implementation details can lead to major bugs. With takeUntil, there is a classic pitfall related to operator ordering that causes memory leaks.

It is crucial to place takeUntil at or near the end of your pipe chain.

The reason for this is that when takeUntil completes the stream, it triggers the teardown logic for all operators that came before it in the chain. If an operator like mergeMap or switchMap comes after takeUntil, its inner subscription will never be told to unsubscribe, creating a memory leak.

Conclusion

Congratulations! You've implemented takeUntil, a sophisticated and highly practical operator.

Key Takeaways:

• takeUntil completes a source observable when a second notifier observable emits a value.
• Its implementation requires managing two subscriptions concurrently, making robust resource management (via our Subscription class) essential.
• The operator is a primary tool for preventing memory leaks by tying an observable's lifetime to another event, such as a component's destruction.
• The placement of takeUntil in an operator chain is critical; it should almost always be the last operator to ensure all other operators in the chain are properly torn down.

This lesson concludes our module on core pipeable operators for error handling and completion. You now have a solid set of tools for managing the observable lifecycle.

In our next lesson, we will begin a new module and shift our focus from building individual operators to creating a higher-level abstraction. We will start designing a type-safe Event Bus, which will leverage the observable foundation you've built to provide a powerful application-level communication pattern.