Type-Safe Scan Operator Implementation

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

In our last lesson, we implemented the filter operator and explored how to use TypeScript's type predicates and function overloads to create a type-safe operator that can narrow the type of an observable stream. This was a great example of a stateless operator—each emission is handled independently without any memory of past values.

Today, we're taking a significant step forward by implementing our first stateful operator: scan. This operator maintains an internal state, or "memory," that it updates with each new value from the source. It's the reactive equivalent of Array.prototype.reduce, but with a key difference: it emits the accumulated value after each source emission, giving you a running stream of the accumulated state.

Lesson Goal: By the end of this lesson, you will be able to implement and test a type-safe scan operator function, ensuring the accumulator's type is correctly inferred and maintained.

1. Understanding the scan Operator

The scan operator is fundamental for managing state in reactive programming. It works by applying an accumulator function to each value from the source observable. This function takes two arguments: the previously accumulated value and the current value. The result becomes the new accumulated value, which is then emitted by the scan operator.

To get started, scan also requires a seed or initial value for the accumulator.

Here is a visual representation of how scan works:

To solidify this concept, let's watch a short video that explains the core mechanics of the scan operator.

As the video demonstrated, scan is incredibly useful for tasks like collecting user events, calculating running totals, or managing any form of state that evolves over time based on an input stream.

2. Designing a Type-Safe scan Operator

One of the most powerful features of scan is its ability to transform the type of the stream. The accumulated value does not need to be the same type as the source values. For example, you can have a source of numbers (Observable<number>) and use scan to accumulate them into an array (Observable<number[]>).

To achieve this with type safety, we'll need two generic type parameters:

• T: The type of the values from the source observable.
• R: The type of the accumulated value (and thus the output observable's value type).

This leads us to the following function signature:

scan<T, R>(
  accumulator: (acc: R, value: T) => R,
  seed: R
): (source: Observable<T>) => Observable<R>

Let's break this down:

• accumulator: (acc: R, value: T) => R: A function that takes the last accumulated value (R) and the new source value (T) and returns the new accumulated value (R).
• seed: R: The initial value for the accumulator, of type R.
• The operator itself returns a function that transforms a source Observable<T> into an output Observable<R>.

3. Implementation

Now, let's implement the operator. The logic is a direct translation of the concepts we've discussed. We'll maintain a state variable inside the operator and update it with each emission.

Create a new file at src/operators/scan.ts:

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

/**
 * Applies an accumulator function over the source Observable, and returns each
 * intermediate result, with an optional seed value.
 *
 * @param accumulator The accumulator function.
 * @param seed The initial accumulator value.
 * @returns A function that returns an Observable that emits the accumulated
 * values.
 */
export function scan<T, R>(
  accumulator: (acc: R, value: T) => R,
  seed: R
): (source: Observable<T>) => Observable<R> {
  return (source: Observable<T>): Observable<R> => {
    return new Observable<R>(observer => {
      // Initialize the accumulated state with the seed value.
      let acc = seed;

      const sourceSubscription = source.subscribe({
        next: (value) => {
          try {
            // Compute the new accumulated value.
            acc = accumulator(acc, value);
            // Emit the new accumulated value.
            observer.next(acc);
          } catch (err) {
            // If the accumulator function throws, propagate the error.
            observer.error(err);
          }
        },
        error: (err) => {
          observer.error(err);
        },
        complete: () => {
          observer.complete();
        },
      });

      // Return the teardown logic.
      return () => {
        sourceSubscription.unsubscribe();
      };
    });
  };
}

This implementation is clean and robust. It initializes the state, updates it on each next call, and correctly handles errors and completion.

4. Testing the scan Operator

A stateful operator requires thorough testing. We need to verify that it accumulates correctly, handles different types, and propagates lifecycle events properly.

For inspiration on how scan is used and tested in a real-world scenario, you can look at the article "Unit Testing RxJS Observables - A Practical Guide". It shows scan being used to manage a stateful price lookup table.

Now, let's write our own tests. Create a new test file at src/operators/scan.test.ts:

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

describe('scan', () => {
  it('should accumulate values with a seed', () => {
    const source$ = of(1, 2, 3);
    const observer = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    source$.pipe(scan((acc, curr) => acc + curr, 10)).subscribe(observer);

    expect(observer.next).toHaveBeenCalledTimes(3);
    expect(observer.next).toHaveBeenNthCalledWith(1, 11); // 10 + 1
    expect(observer.next).toHaveBeenNthCalledWith(2, 13); // 11 + 2
    expect(observer.next).toHaveBeenNthCalledWith(3, 16); // 13 + 3
    expect(observer.error).not.toHaveBeenCalled();
    expect(observer.complete).toHaveBeenCalled();
  });

  it('should handle a different accumulator type', () => {
    const source$ = of(1, 2, 3);
    const observer = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    // Accumulating numbers into an array of strings
    source$
      .pipe(scan((acc, curr) => [...acc, `item ${curr}`], [] as string[]))
      .subscribe(observer);

    expect(observer.next).toHaveBeenCalledTimes(3);
    expect(observer.next).toHaveBeenNthCalledWith(1, ['item 1']);
    expect(observer.next).toHaveBeenNthCalledWith(2, ['item 1', 'item 2']);
    expect(observer.next).toHaveBeenNthCalledWith(3, ['item 1', 'item 2', 'item 3']);
    expect(observer.complete).toHaveBeenCalled();
  });

  it('should propagate errors from the source', () => {
    const error = new Error('Source Error');
    const source$ = new Observable(observer => {
      observer.next(1);
      observer.error(error);
    });
    const observer = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    source$.pipe(scan((acc, curr) => acc + curr, 0)).subscribe(observer);

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

  it('should handle errors thrown by the accumulator function', () => {
    const error = new Error('Accumulator Error');
    const source$ = of(1, 2, 3);
    const observer = { next: vi.fn(), error: vi.fn(), complete: vi.fn() };

    source$
      .pipe(
        scan((acc, curr) => {
          if (curr > 2) {
            throw error;
          }
          return acc + curr;
        }, 0)
      )
      .subscribe(observer);

    expect(observer.next).toHaveBeenCalledTimes(2);
    expect(observer.next).toHaveBeenCalledWith(1);
    expect(observer.next).toHaveBeenCalledWith(3);
    expect(observer.error).toHaveBeenCalledWith(error);
    expect(observer.complete).not.toHaveBeenCalled();
  });
});

Run your tests (npm test or vitest) to confirm that everything is working as expected. These tests cover the core functionality, type transformation, and error handling paths.

Conclusion

Excellent work! You have now implemented scan, your first stateful operator. This is a major milestone, as scan is the cornerstone of state management in many reactive applications.

Key Takeaways:

• The scan operator is a stateful operator that works like Array.prototype.reduce for streams, emitting each intermediate result.
• It requires an accumulator function and a seed (initial) value.
• Using two generic types, <T, R>, allows scan to safely transform the stream into a new type, such as accumulating numbers (T) into an array (R).
• scan is a fundamental building block for managing complex state in a declarative, reactive way. The patterns shown in the resources, like using merge with scan to handle different actions, are common in real-world applications.

Next Lesson Preview:
In our next lesson, we will implement the tap operator. Unlike map, filter, or scan, tap does not modify the stream. Instead, it's used for performing side effects, such as logging values for debugging, without altering the data passing through. It's a simple but indispensable tool for developing and debugging reactive code.