Implementing and Testing the `tap` Operator

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

In our previous lesson, we implemented scan, our first stateful operator. We saw how it could accumulate values over time, effectively acting as a state management tool within the observable stream itself. This was an example of an operator that transforms the data passing through it.

Today, we shift our focus to a different but equally important type of operator: one that doesn't modify the stream at all. We will be implementing the tap operator. Its purpose is to allow you to "tap into" the observable stream to perform side effects, like logging data for debugging, without altering the values, errors, or completion signals that are passed along.

Lesson Goal: By the end of this lesson, you will be able to implement and test a type-safe tap operator for performing side effects without altering the stream.

---

1. Understanding Side Effects and the tap Operator

In functional and reactive programming, a "side effect" is any interaction with the outside world that occurs within a function, which is not part of its return value. Common examples include:

• Logging to the console (console.log)
• Writing to local storage
• Making a network request
• Modifying a variable outside the function's scope

While we generally try to minimize side effects in reactive code, they are often necessary for debugging or interacting with non-reactive parts of an application. The tap operator is the designated tool for this job. It acts as a transparent window into the stream, letting you observe notifications (next, error, complete) as they pass by without changing them.

To see why this is so useful in practice, let's watch a video that explains the role of tap and provides several real-world use cases.

As the video highlights, tap is your go-to tool for debugging observable chains. Instead of subscribing early just to see a value, you can insert a tap(console.log) anywhere in your pipe to inspect the stream at that specific point.

2. Designing a Flexible tap Operator

A robust tap operator should be able to observe all three types of notifications an observable can produce: next, error, and complete. To accommodate this, we can design our tap operator to accept two kinds of arguments:

1. A single function, which will be treated as the next handler.
2. An object with optional next, error, and complete methods, similar to the Observer interface we created earlier.

This provides both a convenient shorthand for simple logging and a powerful way to react to the full lifecycle of the stream.

Let's read a short article that explains this design and the flexibility of tap.

Based on this, we can define the signature for our operator. It will be a "mono-type" operator, meaning the output observable will have the same value type as the source.

// In a new file: src/operators/tap.ts

import { Observable } from '../core/observable';

// This type defines the observer-like object that tap can accept.
export type TapObserver<T> = {
  next?: (value: T) => void;
  error?: (err: any) => void;
  complete?: () => void;
};

export function tap<T>(
  observerOrNext: TapObserver<T> | ((value: T) => void)
): (source: Observable<T>) => Observable<T> {
  // Implementation will go here
}

This signature uses a union type to allow for both a full TapObserver object and a simple next function.

3. Implementation

The implementation of tap involves creating a new Observable that wraps the source. Inside this new observable, we subscribe to the source and, for each notification, we first execute the side effect (the corresponding tap handler) and then pass the original notification down to the final observer.

It's crucial to wrap the side-effect calls in a try...catch block. If a tap handler throws an error, the stream should error out, and the subscription should be terminated.

Here is the full implementation. Create a new file at src/operators/tap.ts:

import { Observable } from '../core/observable';

export type TapObserver<T> = {
  next?: (value: T) => void;
  error?: (err: any) => void;
  complete?: () => void;
};

/**
 * Perform a side effect for every emission on the source Observable, but return
 * an Observable that is identical to the source.
 *
 * @param observerOrNext An object with next, error, and complete handlers, or
 * a single function for the next handler.
 * @returns A function that returns an Observable that is identical to the source.
 */
export function tap<T>(
  observerOrNext: TapObserver<T> | ((value: T) => void)
): (source: Observable<T>) => Observable<T> {
  const tapObserver: TapObserver<T> =
    typeof observerOrNext === 'function'
      ? { next: observerOrNext }
      : observerOrNext;

  return (source: Observable<T>): Observable<T> => {
    return new Observable<T>(observer => {
      const sourceSubscription = source.subscribe({
        next: (value) => {
          try {
            tapObserver.next?.(value);
          } catch (err) {
            observer.error(err);
            return;
          }
          observer.next(value);
        },
        error: (err) => {
          try {
            tapObserver.error?.(err);
          } catch (e) {
            observer.error(e);
            return;
          }
          observer.error(err);
        },
        complete: () => {
          try {
            tapObserver.complete?.();
          } catch (err) {
            observer.error(err);
            return;
          }
          observer.complete();
        },
      });

      return () => {
        sourceSubscription.unsubscribe();
      };
    });
  };
}

Notice how in the error handler, if tapObserver.error itself throws an error, we propagate that new error. Otherwise, we propagate the original error from the source. This mirrors the behavior of established libraries like RxJS.

4. Testing the tap Operator

Testing tap requires us to verify two main things:

1. That the side-effect functions are called with the correct notifications.
2. That the output stream is an exact mirror of the source stream (unless a tap handler itself throws an error).

Let's create a test file at src/operators/tap.test.ts to cover these scenarios.

import { describe, it, expect, vi } from 'vitest';
import { of } from '../creation/of';
import { tap } from './tap';
import { Observable } from '../core/observable';

describe('tap', () => {
  it('should call the next handler and pass values through', () => {
    const source$ = of(1, 2, 3);
    const tapNext = vi.fn();
    const finalObserver = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    source$.pipe(tap(tapNext)).subscribe(finalObserver);

    // Check tap handler
    expect(tapNext).toHaveBeenCalledTimes(3);
    expect(tapNext).toHaveBeenNthCalledWith(1, 1);
    expect(tapNext).toHaveBeenNthCalledWith(2, 2);
    expect(tapNext).toHaveBeenNthCalledWith(3, 3);

    // Check final observer
    expect(finalObserver.next).toHaveBeenCalledTimes(3);
    expect(finalObserver.next).toHaveBeenNthCalledWith(1, 1);
    expect(finalObserver.next).toHaveBeenNthCalledWith(2, 2);
    expect(finalObserver.next).toHaveBeenNthCalledWith(3, 3);
    expect(finalObserver.complete).toHaveBeenCalled();
  });

  it('should work with an observer object for next, error, and complete', () => {
    const error = new Error('Source Error');
    const source$ = new Observable(observer => {
      observer.next(1);
      observer.error(error);
    });
    
    const tapObserver = {
      next: vi.fn(),
      error: vi.fn(),
      complete: vi.fn(),
    };
    const finalObserver = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    source$.pipe(tap(tapObserver)).subscribe(finalObserver);

    expect(tapObserver.next).toHaveBeenCalledWith(1);
    expect(tapObserver.error).toHaveBeenCalledWith(error);
    expect(tapObserver.complete).not.toHaveBeenCalled();

    expect(finalObserver.next).toHaveBeenCalledWith(1);
    expect(finalObserver.error).toHaveBeenCalledWith(error);
    expect(finalObserver.complete).not.toHaveBeenCalled();
  });

  it('should propagate an error if the tap next handler throws', () => {
    const source$ = of(1, 2);
    const error = new Error('Tap Error');
    const tapNext = vi.fn(value => {
      if (value === 2) {
        throw error;
      }
    });
    const finalObserver = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    source$.pipe(tap(tapNext)).subscribe(finalObserver);

    expect(finalObserver.next).toHaveBeenCalledOnce();
    expect(finalObserver.next).toHaveBeenCalledWith(1);
    expect(finalObserver.error).toHaveBeenCalledWith(error);
    expect(finalObserver.complete).not.toHaveBeenCalled();
  });

  it('should call the complete handler', () => {
    const source$ = of(1);
    const tapComplete = vi.fn();
    const finalObserver = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    source$.pipe(tap({ complete: tapComplete })).subscribe(finalObserver);

    expect(tapComplete).toHaveBeenCalledOnce();
    expect(finalObserver.complete).toHaveBeenCalledOnce();
  });
});

Run your tests (npm test or vitest) to ensure the implementation is correct. These tests confirm that tap correctly invokes the side-effect handlers for all notification types and maintains the integrity of the stream.

---

Conclusion

Great job! You've successfully implemented tap, a seemingly simple but incredibly useful operator. It's an essential part of the reactive developer's toolkit, providing a clean, declarative way to introduce side effects for debugging and other interactions.

Key Takeaways:

• The tap operator is used for performing side effects without modifying the observable stream.
• It is a "mono-type" operator, meaning pipe(tap(...)) results in an Observable of the same type.
• A flexible tap implementation can accept either a single function for next notifications or an observer-like object to handle next, error, and complete.
• tap is invaluable for debugging, allowing you to inspect values at any stage of a pipe chain with tap(console.log).

Next Lesson Preview:
So far, we've built operators that transform (map), filter (filter), accumulate (scan), and observe (tap) a single stream. In the next module, we'll move on to a new class of operators: combination operators. We'll start by implementing merge, an operator that takes multiple source observables and combines them into a single output stream, emitting values from any of the sources as they arrive.