RxJS Operators: takeUntil

RxJS is a JavaScript library that enables the creation of asynchronous and event-based programs. The main type is the Observable and a suite of functions and operators are provided to make it easier to work with the data coming through. This series will detail those operators (and a few stand-alone functions) and provide examples of their use.

In this post, we’re going to cover the takeUntil operator.

What does it do?

The takeUntil operator gives us a way to cancel a subscription to an observable. We add the operator to our observable pipeline and provide it with a subject. Whenever that subject emits a value the subscription will be cancelled.

Without using this we would need to track all the subscriptions individually. When it came time to cancel them we would have to run through them all. This can be error-prone since it’s easy to miss one or incorrectly track a reference. Using takeUntil means we just need to add the method and make a single call to cancel all the subscriptions.

Example

We’ll use interval as our observable:

import { Subject, interval, takeUntil } from 'rxjs';

const unsubscribe$ = new Subject<void>();

interval(1000)
    .pipe(takeUntil(unsubscribe$))
    .subscribe(x => {
        console.log(x);
    });

setTimeout(() => {
    unsubscribe$.next();
}, 10000);

A few things are going on here. We’ll break it down and see what’s happening bit by bit.

const unsubscribe$ = new Subject();

We create a new subject to pass to takeUntil. This is what we’ll be triggering later.

interval(1000)
    .pipe(takeUntil(unsubscribe$))
    .subscribe(x => {
        console.log(x);
    });

Subscribe to the observable and include takeUntil in the pipeline. This will handle every value until it has been unsubscribed.

setTimeout(() => {
    unsubscribe$.next(undefined);
}, 10000);

Use setTimeout to wait 10 seconds and send a value through the unsubscribe$ subject. This will cancel the subscription of the above observable. If we run this we’re going to see values emitted for 10 seconds and then stop, since we’re emitting an incremented number every second we should see:

0
1
2
3
4
5
6
7
8

Since we’re using interval the subject will complete and the program will end at this point. If we were using a typical long-lived observable the values would continue being sent out, we’re just not subscribing to them anymore.

This is the operator I use most when working in Angular. Since Angular is a single-page application framework, observables will never unsubscribe unless explicitly told to do so. This means I can set up subscriptions in components, configure them with takeUntil, and make a .next() call in the destroy handling.

That’s just one use case. There are other scenarios where this can be useful. The point is takeUntil makes it easy to control the lifetime of our subscriptions.

The source code for this example is available on GitHub: