How To Use the takeUntil RxJS Operator to Manage Subscriptions Declaratively

Introduction

Angular handles unsubscribing from observable subscriptions like those returned from the HTTP service or when using the async pipe. However, for other situations, it can quickly become difficult to manage all subscriptions and ensure to unsubscribe from those that are long-lived. A policy of unsubscribing from most subscriptions will also have its own problems.

In this article, you will be presented with an example Angular application that relies upon manually subscribing and unsubscribing. Then, you will compare it to an example Angular application that uses the takeUntil operator to declaratively manage subscriptions.

Prerequisites

If you would like to follow along with this article, you will need:

• Some familiarity with the RxJS library, in particular, Observable and Subscription will be beneficial.
• Some familiarity with Apollo and GraphQL will be helpful but is not required.

This tutorial was verified with Node v15.3.0, npm v6.14.9, @angular/core v11.0.4, rxjs v6.6.3, apollo-angular v2.1.0, graph-tag v2.11.0. This article was edited to reflect changes in migrating from earlier versions of @angular/core and rxjs.

Unsubscribing Manually

Let’s start with an example where you will manually unsubscribe from two subscriptions.

In this example, the code is subscribing to an Apollo watchQuery to get data from a GraphQL endpoint.

The code is also creating an interval observable that you subscribe to when an onStartInterval method gets called.

import { Component, OnInit, OnDestroy } from '@angular/core';

import { Subscription, interval } from 'rxjs';

import { Apollo } from 'apollo-angular';
import gql from 'graphql-tag';

@Component({ ... })
export class AppComponent implements OnInit, OnDestroy {
  myQuerySubscription: Subscription;
  myIntervalSubscription: Subscription;

  constructor(private apollo: Apollo) {}

  ngOnInit() {
    this.myQuerySubscription = this.apollo.watchQuery<any>({
      query: gql`
        query getAllPosts {
          allPosts {
            title
            description
            publishedAt
          }
        }
      `
    })
    .valueChanges
    .subscribe(({data}) => {
      console.log(data);
    });
  }

  onStartInterval() {
    this.myIntervalSubscription = interval(250).subscribe(value => {
      console.log('Current value:', value);
    });
  }

  ngOnDestroy() {
    this.myQuerySubscription.unsubscribe();

    if (this.myIntervalSubscription) {
      this.myIntervalSubscription.unsubscribe();
    }
  }
}

Now imagine that your component has many similar subscriptions, it can quickly become quite a process to ensure everything gets unsubscribed when the component is destroyed.

Unsubscribing Declaratively with takeUntil

The solution is to compose the subscriptions with the takeUntil operator and use a subject that emits a truthy value in the ngOnDestroy lifecycle hook.

The following snippet does the exact same thing, but this time the code will unsubscribe declaratively. You will notice that an added benefit is that you no longer need to keep references to our subscriptions anymore.

import { Component, OnInit, OnDestroy } from '@angular/core';

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

import { Apollo } from 'apollo-angular';
import gql from 'graphql-tag';

@Component({ ... })
export class AppComponent implements OnInit, OnDestroy {
  destroy$: Subject<boolean> = new Subject<boolean>();

  constructor(private apollo: Apollo) {}

  ngOnInit() {
    this.apollo.watchQuery<any>({
      query: gql`
        query getAllPosts {
          allPosts {
            title
            description
            publishedAt
          }
        }
      `
    })
    .valueChanges
    .pipe(takeUntil(this.destroy$))
    .subscribe(({data}) => {
      console.log(data);
    });
  }

  onStartInterval() {
    interval(250)
    .pipe(takeUntil(this.destroy$))
    .subscribe(value => {
      console.log('Current value:', value);
    });
  }

  ngOnDestroy() {
    this.destroy$.next(true);
    this.destroy$.unsubscribe();
  }
}

Notice how using an operator like takeUntil instead of manually unsubscribing will also complete the observable, triggering any completion event on the observable.

Be sure to check your code to make sure this does not create any unintended side effects.

Conclusion

In this article, you learned about using takeUntil to declaratively unsubscribe. Unsubscribing from unnecessary subscriptions contributes towards preventing memory leaks. Declaratively unsubscribing allows you to not require references to subscriptions.

There are other similar RxJS operators - like take, takeWhile, and first - which will all complete the observable.

If you’d like to learn more about Angular, check out our Angular topic page for exercises and programming projects.