Debouncing and Auditing Emissions: debounceTime vs auditTime

More than two years have passed since my initial write-up on Rx.js operators, and the library has evolved considerably in that stretch. Notably, the .pipe method didn't even exist when that first piece went out.

As I've integrated Rx.js more deeply into the projects I've worked on since then, one observation has come up again and again:

The core challenge developers face with Rx.js isn't a shortage of known operators

It's remarkable how much cleaner code becomes simply by leveraging one operator or a combination of several. Obviously, memorizing all of them isn't realistic—there are over 100 in total—but grasping the key categories can significantly streamline coding, especially in Angular contexts.

Let's jump straight into the details.

debounceTime and the Autocomplete Scenario

If you've ever built an autocomplete input with Rx.js, you've likely encountered debounceTime. According to the official documentation, this operator:

Emits a value from the source Observable only after a particular time span has passed without another source emission.

Essentially, applying debounceTime(3000) will only trigger notifications once the source has been quiet—meaning it has settled—for at least 3000 milliseconds. This proves invaluable when you want to fetch data from a server only after the user has paused typing, preventing a barrage of futile requests mid-keystroke. Consider this illustration:

const inputEl = document.querySelector('input');
fromEvent(inputEl, 'input')
  .pipe(
    debounceTime(300),
    map((event) => event.target.value),
    switchMap((query) => from(fetch(`https://some-url?q=${query}`)))
  )
  .subscribe(console.log);

Without the debounceTime operator, typing "Hello" quickly would generate five requests to [https://some-url?q=](https://some-url/?q=,),, four of which wouldn't serve any real purpose. By incorporating debounceTime, the server interaction is deferred until the user has definitively finished their input.

So that covers one side. What exactly is the function of auditTime, and in what situations does it shine?

As described in the documentation, auditTime:

Ignores source values for duration milliseconds, then emits the most recent value from the source Observable, then repeats this process.

At first glance, this might seem akin to debounceTime; however, their behavior diverges significantly. Even though both operators filter emissions based on time intervals, the mechanism differs: debounceTime waits for a quiet period *after every* emission, only letting the last one through if no new value arrives in that window; auditTime, on the other hand, doesn't react to each emission. Instead, it checks in at regular intervals and, if any source value appeared since the last check, it allows the most recent one to pass. To visualize, imagine an Observable stream of notifications from your Facebook friends. If you apply auditTime(3000), and Anthony sends a message followed by Julia within that 3-second window, only Julia's message will surface at the next check—Anthony's will be permanently dropped.

Where might you actually employ this? Picture a highly frequent source, like a WebSocket broadcasting real-time stock exchange data. Each incoming rate change triggers a repaint of a chart, which is computationally heavy. Since these rates fluctuate rapidly and minor shifts under half a second are imperceptible to the user, you can use auditTime to throttle the repaints to every 500 milliseconds:

observableFromSocket$.pipe(auditTime(500)).subscribe(repaintChart);

Aggregation Strategies: scan vs reduce

Both of these operators serve to aggregate emitted values—for instance, summing a series of numbers. Their key divergence lies in their output timing. The reduce operator sends a final, single emission only upon the source Observable's completion. This makes it ideal for tasks like computing an average age from user data:

of(23, 25, 24, 25, 25, 25)
  .pipe(
    map((age) => [age, 1]),
    reduce(([accAge, accCount], [age, count]) => [
      accAge + age,
      accCount + count,
    ]),
    map(([sum, count]) => sum / count)
  )
  .subscribe(console.log);

Here, we convert each user's age into a tuple containing the age itself and a count of one. These tuples are then accumulated, summing the ages and incrementing the count, all while maintaining the tuple structure. Once the observable completes, a final transformation divides the total sum by the count to yield the average.

Contrastingly, scan produces an emission for *every* intermediate aggregated value. A classic use case is a donation tracker on a webpage. Suppose an Observable reports the amount of each individual donation, but you need to display a running total to the user. With scan, you can accumulate the incoming amounts and update the DOM each time a new donation comes through:

donations$.pipe(scan((acc, next) => acc + next)).subscribe(updateAmount);

Preventing Duplicates: distinct, distinctUntilChanged, and distinctUntilKeyChanged

The purpose of the distinct operator is self-explanatory: it filters out any emission that has occurred previously. Here's a simple demonstration:

of(1, 2, 3, 2, 4, 4).pipe(distinct()).subscribe(console.log);

This code will output 1, 2, 3, 4, filtering out the repeated instances of 2 and 4.

What's a practical application? Primarily, it prevents repetitive actions on the same asset. For instance, consider a Subject that emits a product object each time a user initiates a deletion. A user might click the button more than once before the first API call completes, so you'd want to ensure only the initial product ID is processed. The distinct operator can be configured with a selector function to differentiate emissions based on a property, like the product's ID, rather than the object reference itself:

interface Product {
  id: number;
  title: string;
}

deleteProduct$ // a Subject emitting Products
  .pipe(
    distinct(product => product.id),
  )
  .subscribe(product => ProductService.delete(product));

Now, what if you need to allow non-adjacent duplicates but disallow consecutive identical values? Picture a quiz game where players choose from categories like math, physics, history, geography, and literature. The game should force the player to pick a different field if they choose the same one twice *in succession*. In this case, distinctUntilChanged is the right tool:

fieldsOfStudy$
  .pipe(distinctUntilChanged())
  .subscribe((field) => presentQuestion(field));

This permits duplicates, but strictly prohibits back-to-back identical choices.

Naturally, similar to distinct, you can supply a custom comparison function to distinctUntilChanged for nested properties. However, Rx.js offers a shorthand for this common pattern with distinctUntilKeyChanged. It works like so:

of<Product>(
  {id: 1, title: 'Chair'},
  {id: 1, title: 'Chair'},
  {id: 2, title: 'Table'},
  {id: 1, title: 'Chair'},
).pipe(
  distinctUntilKeyChanged('id'),
  map(product => product.title),
).subscribe(console.log);

This code will output “Chair, Table, Chair”.

Controlling Streams: takeWhile vs takeUntil

Sometimes you need manual control over when an Observable stops emitting. There are three primary scenarios for manual termination:

  1. Take a specific number of emissions
  2. Take emissions as long as a certain condition is true (e.g., reading numbers until one exceeds 5)
  3. Take emissions until an external trigger fires

Rx.js offers operators with names that are quite descriptive of these use cases. I'll skip the straightforward take operator and focus on the other two. Consider this basic example:

of(1, 2, 3, 4, 5)
  .pipe(takeWhile((n) => n < 4))
  .subscribe(console.log);

This will only log "1, 2, 3". You might initially assume this is similar to the filter operator, but it's crucial to note that takeWhile fully unsubscribes from the source after the condition becomes false. A filter merely prevents certain emissions from passing while staying subscribed, allowing future values that meet the predicate to come through. Once takeWhile's condition is not met, it's over permanently.

But how do you gracefully stop a stream based on an external factor? For example, you might want to log a value every second, but only for a total of 10 seconds. How do you halt the process after that timeframe?

You could technically accomplish this using takeWhile:

let stopped = false;

setTimeout(() => (stopped = true), 10000);

interval(1000)
  .pipe(takeWhile(() => !stopped))
  .subscribe(console.log);

While this works, it isn't very elegant. Fortunately, Rx.js includes a dedicated operator for this scenario. takeUntil takes another Observable as its argument and will let the source stream pass values until that passed-in Observable emits. This is achieved by supplying an Observable that fires after 10 seconds (e.g., using the timer operator). Here's the improved code:

interval(1000)
  .pipe(takeUntil(timer(10000)))
  .subscribe(console.log);

This particular pattern is a common practice in Angular, especially for managing unsubscriptions in components. I delve further into this strategy in my article Harnessing the Power of Mixins in Angular.

For more details, the reference section has entries for takeWhile and takeUntil.

Wrapping Up

As previously noted, Rx.js comes with a vast collection of operators, each unlocking intriguing capabilities. Future articles will continue exploring this topic.