takeUntil

Let’s start with a basic example where we’ll manually unsubscribe from two subscriptions. In this example we’ll subscribe to an Apollo watchQuery to get data from a GraphQL endpoint, and we’ll also create an interval observable that we subscribe to when an onStartInterval method gets called:

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

import { Observable } from 'rxjs/Observable';
import { Subscription } from 'rxjs/Subscription';
import 'rxjs/add/observable/interval';

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

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

  myQueryObs: Observable;

  constructor(private apollo: Apollo) {}

  ngOnInit() {
    this.myQueryObs = this.apollo.watchQuery({
      query: gql`
        query getAllPosts {
          allPosts {
            title
            description
            publishedAt
          }
        }
      `
    });

    this.myQuerySub = this.myQueryObs.subscribe(({data}) => {
      console.log(data);
    });
  }

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

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

    if (this.myIntervalSub) {
      this.myIntervalSub.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 our 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 we unsubscribe declaratively. You’ll notice that an added benefit is that we don’t need to keep references to our subscriptions anymore:

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

import { Observable } from 'rxjs/Observable';
import { Subject } from 'rxjs/Subject';
import 'rxjs/add/observable/interval';
import 'rxjs/add/operator/takeUntil';

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

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

  constructor(private apollo: Apollo) {}

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

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

  ngOnDestroy() {
    this.destroy$.next(true);
    // Now let's also unsubscribe from the subject itself:
    this.destroy$.unsubscribe();
  }
}
Note that Using an operator like takeUntil instead of manually unsubscribing will also complete the observable, triggering any completion event on the observable. Check your code to make sure this doesn’t create any unintended side effects. Thanks to @gerardsans for pointing this out. This is the same for other similar operators like take, takeWhile and first, which will all complete the observable.