Testing `throttleTime` with Vitest Fake Timers

Hello! Welcome back.

In our previous lesson, we implemented the debounceTime operator, mastering how to delay emissions until a stream of values has paused. This gave us valuable practice with Vitest's fake timers to test asynchronous logic in a deterministic way.

Today, we continue our exploration of timing operators in Module 5 by implementing throttleTime. While it also controls the rate of emissions, its strategy is fundamentally different from debouncing. Understanding both is essential for a robust reactive toolkit.

Our learning goal is to use Vitest's fake timers to implement and test a throttleTime operator to discard emissions for a specified duration. We'll see how a small change in logic—from waiting for a pause to enforcing a cooldown—results in very different behavior.

1. Understanding Throttling vs. Debouncing

debounceTime is like a patient listener: it waits for you to finish talking before responding. throttleTime is different. It's like a gatekeeper that lets one person through, then closes the gate for a fixed period. It doesn't wait for a pause; it enforces a mandatory waiting period after an event has occurred.

• Debounce: Emits a value only after a specified time has passed without any new values. Useful for "autocomplete" search fields.
• Throttle: Emits a value immediately, then ignores all subsequent values for a specified duration. Useful for preventing spam-clicks on a button or handling rapid-fire events like scrolling or window resizing.

Let's look at the official RxJS documentation for throttleTime to solidify this concept.

As you can see from the diagram, throttleTime is about guaranteeing a minimum time gap between emissions.

2. Designing the throttleTime Operator

The default behavior of throttleTime, which we will implement, is often called "leading" throttle. It emits the first value that starts the throttled period.

Here's the logic:

1. State: We need to track whether we are currently in a "throttling" period. A simple boolean flag, let's call it isThrottling, will work.
2. On a next emission:
   • Check the isThrottling flag.
   • If false, it means the gate is open. Emit the value to the observer, set isThrottling to true, and start a setTimeout for the specified duration.
   • If true, do nothing. The value is simply ignored.
3. When the setTimeout completes:
   • Set isThrottling back to false, opening the gate for the next value.
4. Error/Completion: Errors and completion notifications should be passed through immediately, without being throttled.
5. Unsubscription: The teardown logic must clear any pending timeout to prevent memory leaks.

This logic is simpler than debounceTime as we don't need to store the last value, but it requires careful management of the isThrottling state and the timer.

3. Your Task: Implement and Test throttleTime

Now, let's put this design into practice.

3.1. Implementation

Create a new file at src/operators/throttleTime.ts. Your implementation will be a pipeable operator that returns a new Observable.

Here is the code based on our design. I've added comments to guide you through the state management.

// src/operators/throttleTime.ts

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

export function throttleTime<T>(duration: number): (source: Observable<T>) => Observable<T> {
  return (source: Observable<T>) =>
    new Observable<T>(observer => {
      let isThrottling = false;
      let timeoutId: ReturnType<typeof setTimeout> | null = null;

      const sourceSub = source.subscribe({
        next: value => {
          // If we are not currently throttling, emit the value and start the timer.
          if (!isThrottling) {
            observer.next(value);
            isThrottling = true;

            timeoutId = setTimeout(() => {
              isThrottling = false;
              // Note: We don't need to do anything with the last value here for a leading throttle.
            }, duration);
          }
          // If we are throttling, the value is ignored.
        },
        error: err => {
          // Propagate errors immediately.
          observer.error(err);
        },
        complete: () => {
          // Propagate completion immediately.
          observer.complete();
        },
      });

      // Teardown logic: clear any pending timer and unsubscribe from the source.
      return () => {
        if (timeoutId) {
          clearTimeout(timeoutId);
        }
        sourceSub.unsubscribe();
      };
    });
}

3.2. Testing

With the operator implemented, let's write tests to ensure it behaves correctly. Create src/operators/throttleTime.test.ts.

As with delay and debounceTime, we must use Vitest's fake timers. This allows us to control the setTimeout behavior and test the time-based logic instantly. The pattern will be familiar to you from the last lesson.

// src/operators/throttleTime.test.ts
import { beforeEach, afterEach, describe, it, expect, vi } from 'vitest';
import { Observable } from '../observable';
import { throttleTime } from './throttleTime';

describe('throttleTime', () => {
  beforeEach(() => {
    vi.useFakeTimers();
  });

  afterEach(() => {
    vi.useRealTimers();
  });

  // Test cases go here
});

Now, add the following test cases inside the describe block. They cover the core functionality, edge cases, and cleanup.

1. Test 1: Should emit the first value immediately and ignore subsequent values within the duration.

   it('should emit the first value and ignore subsequent values for the duration', () => {
     const observer = { next: vi.fn() };
     const source$ = new Observable<number>(obs => {
       obs.next(1); // Emitted immediately at t=0
       setTimeout(() => obs.next(2), 50); // Ignored, t=50 < 200
       setTimeout(() => obs.next(3), 100); // Ignored, t=100 < 200
     });
   
     source$.pipe(throttleTime(200)).subscribe(observer);
   
     vi.advanceTimersByTime(150); // Advance time, but still within the throttle window
   
     expect(observer.next).toHaveBeenCalledTimes(1);
     expect(observer.next).toHaveBeenCalledWith(1);
   });
   

2. Test 2: Should emit a new value after the throttle duration has passed.

   it('should emit again after the throttle duration', () => {
     const observer = { next: vi.fn() };
     const source$ = new Observable<number>(obs => {
       obs.next(1); // Emitted at t=0
       setTimeout(() => obs.next(2), 250); // Emitted, t=250 > 200
     });
   
     source$.pipe(throttleTime(200)).subscribe(observer);
   
     vi.advanceTimersByTime(250); // Advance past the second emission
   
     expect(observer.next).toHaveBeenCalledTimes(2);
     expect(observer.next).toHaveBeenCalledWith(1);
     expect(observer.next).toHaveBeenCalledWith(2);
   });
   

3. Test 3: Should propagate errors immediately.

   it('should propagate errors immediately', () => {
     const error = new Error('fail');
     const observer = { error: vi.fn() };
     const source$ = new Observable<never>(obs => {
       obs.error(error);
     });
   
     source$.pipe(throttleTime(1000)).subscribe(observer);
   
     expect(observer.error).toHaveBeenCalledWith(error);
   });
   

4. Test 4: Should handle unsubscription correctly.
   This test ensures that if we unsubscribe, the timer is cleared and no more values are emitted.

   it('should not emit after unsubscription', () => {
     const observer = { next: vi.fn() };
     const source$ = new Observable<number>(obs => {
       obs.next(1); // Emitted
       setTimeout(() => obs.next(2), 300); // Should not be emitted
     });
   
     const subscription = source$.pipe(throttleTime(200)).subscribe(observer);
   
     // Unsubscribe before the throttle time is over
     subscription.unsubscribe();
     
     vi.runAllTimers(); // Run any pending timers
   
     expect(observer.next).toHaveBeenCalledTimes(1);
     expect(observer.next).toHaveBeenCalledWith(1);
   });
   

4. Advanced Configuration: Leading and Trailing

Our implementation handles the default "leading" throttle. However, the RxJS throttleTime operator is more configurable. It accepts a config object: { leading: boolean, trailing: boolean }.

• leading: true (our implementation): Emits the value at the beginning of the throttle window.
• trailing: true: Emits the most recent value at the end of the throttle window.

This diagram illustrates the difference beautifully.

Implementing a trailing throttle is a great extension exercise. It would require logic similar to debounceTime, where you store the last value and emit it when the timer fires. For our core library, the default leading behavior is the most common and a solid foundation.

Conclusion

Great job! You have now implemented throttleTime, adding another powerful rate-limiting operator to your reactive library. You've also deepened your understanding of how to test time-based asynchronous code with Vitest.

Key Takeaways:

• throttleTime enforces a cooldown period after an emission, ignoring subsequent values during that time.
• It differs from debounceTime, which waits for a period of inactivity before emitting.
• The implementation relies on a state flag (isThrottling) and a setTimeout to manage the cooldown window.
• Testing with fake timers (vi.useFakeTimers, vi.advanceTimersByTime) is essential for verifying the operator's behavior deterministically.

With this lesson, we conclude our work on timing operators in Module 5. In our next lesson, we will begin Module 6: Pipeable Operators: Error Handling & Completion by implementing catchError, a crucial operator for building resilient and fault-tolerant observable streams.