Angular - Using Takeuntil RxJS Operator

In the last post , we have seen different types of observable. Observable is basically a container which produces asynchronous stream of data, and emit values over time. We have to subscribe to an observable in order to consume or receive data. But you have to be careful with observables as it may leads to memory leaks and affect the application performance. To avoid this issue, one approach is to keep the reference at the time of subscription and unsubscribe from the observable by using the same reference when you no longer need to receive observable's data stream.

Lets see the example code:

import { Component, OnInit, OnDestroy } from '@angular/core';
import { Subscription } from 'rxjs';
import { Service1 } from 'Service1';
import { Service2 } from 'Service2';

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

  constructor(private myservice1: Service1, private myservice2: Service2) {}

  ngOnInit() {
    
    this.mySubscription1 = this.myservice1.getData()
    .subscribe(({data}) => {
      console.log(data);
    });
	
    this.mySubscription2 = this.myservice2.getData()
    .subscribe(({data}) => {
      console.log(data);
    });
	
  }
 
  ngOnDestroy() {
    this.mySubscription1.unsubscribe();
    this.mySubscription2.unsubscribe();
  }
 
 }

In this code snippet, we have two services to consume Service1 and Service2. In ngOnInit handler, we maintained the subscription references for both the services in two different variables mySubscription1 and mySubscription2. Then in the ngOnDestroy handler, we are using the same reference variables to unsubscribe from the observable.

The above code works perfectly, but it would be cumbersome to maintain in long term when you have more number of subscriptions which makes you to keep the references for each subscription and then unsubscribe from each one in ngOnDestroy handler.

A better approach is to use takeUntil operator from RxJS library, it is used to automatically unsubscribe from an observable. takeUntil refelects the source Observable. It also monitors a second Observable (the notifier) that you provide. If the notifier emits a value, the output Observable stops reflecting the source Observable and completes itself.

Here is the same exmaple, this time unsubscribe using takeUntil operator.

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

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

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

  constructor(private myservice1: Service1, private myservice2: Service2) {}

  ngOnInit() {

    this.myservice1.getData()
    .pipe(takeUntil(this.destroy$))
    .subscribe(({data}) => {
      console.log(data);
    });
	
    this.myservice2.getData()
    .pipe(takeUntil(this.destroy$))
    .subscribe(({data}) => {
      console.log(data);
    });
  }

  ngOnDestroy() {
    this.destroy$.next(true);
    this.destroy$.unsubscribe();
    
    ////some people prefer to call complete() on destroy$ here, instead of unsubscribe()
    //this.destroy$.complete();
  }
} 

This code snippet also behaves the same as before, but its easier to manage when you have more number of subscriptions. In ngOnDestroy handler, we have called the unsubscribe() method, some people may prefer to call complete() method instead. But in the end it does not make much difference in this context, the purpose here is just to stop receiving more values from this subject.

References:

• Learn RxJS - takeUntil
• Observables vs Subjects vs Behavior Subjects