[angular] Angular/RxJs When should I unsubscribe from `Subscription`

When should I store the Subscription instances and invoke unsubscribe() during the NgOnDestroy life cycle and when can I simply ignore them?

Saving all subscriptions introduces a lot of mess into component code.

HTTP Client Guide ignore subscriptions like this:

getHeroes() {
  this.heroService.getHeroes()
                  .subscribe(
                     heroes => this.heroes = heroes,
                     error =>  this.errorMessage = <any>error);
}

In the same time Route & Navigation Guide says that:

Eventually, we'll navigate somewhere else. The router will remove this component from the DOM and destroy it. We need to clean up after ourselves before that happens. Specifically, we must unsubscribe before Angular destroys the component. Failure to do so could create a memory leak.

We unsubscribe from our Observable in the ngOnDestroy method.

private sub: any;

ngOnInit() {
  this.sub = this.route.params.subscribe(params => {
     let id = +params['id']; // (+) converts string 'id' to a number
     this.service.getHero(id).then(hero => this.hero = hero);
   });
}

ngOnDestroy() {
  this.sub.unsubscribe();
}

In my case I am using a variation of the solution proposed by @seanwright:
https://github.com/NetanelBasal/ngx-take-until-destroy

It's a file used in the ngx-rocket / starter-kit project. You can access it here until-destroyed.ts

The component would look like so

/**
 * RxJS operator that unsubscribe from observables on destory.
 * Code forked from https://github.com/NetanelBasal/ngx-take-until-destroy
 *
 * IMPORTANT: Add the `untilDestroyed` operator as the last one to
 * prevent leaks with intermediate observables in the
 * operator chain.
 *
 * @param instance The parent Angular component or object instance.
 * @param destroyMethodName The method to hook on (default: 'ngOnDestroy').
 */
import { untilDestroyed } from '../../core/until-destroyed';

@Component({
  selector: 'app-example',
  templateUrl: './example.component.html'
})
export class ExampleComponent implements OnInit, OnDestroy {

  ngOnInit() {
    interval(1000)
        .pipe(untilDestroyed(this))
        .subscribe(val => console.log(val));

    // ...
  }


  // This method must be present, even if empty.
  ngOnDestroy() {
    // To protect you, an error will be thrown if it doesn't exist.
  }
}

A Subscription essentially just has an unsubscribe() function to release resources or cancel Observable executions. In Angular, we have to unsubscribe from the Observable when the component is being destroyed. Luckily, Angular has a ngOnDestroy hook that is called before a component is destroyed, this enables devs to provide the cleanup crew here to avoid hanging subscriptions, open portals, and what nots that may come in the future to bite us in the back

@Component({...})
export class AppComponent implements OnInit, OnDestroy {
    subscription: Subscription 
    ngOnInit () {
        var observable = Rx.Observable.interval(1000);
        this.subscription = observable.subscribe(x => console.log(x));
    }
    ngOnDestroy() {
        this.subscription.unsubscribe()
    }
}

We added ngOnDestroy to our AppCompoennt and called unsubscribe method on the this.subscription Observable

If there are multiple subscriptions:

@Component({...})
export class AppComponent implements OnInit, OnDestroy {
    subscription1$: Subscription
    subscription2$: Subscription 
    ngOnInit () {
        var observable1$ = Rx.Observable.interval(1000);
        var observable2$ = Rx.Observable.interval(400);
        this.subscription1$ = observable.subscribe(x => console.log("From interval 1000" x));
        this.subscription2$ = observable.subscribe(x => console.log("From interval 400" x));
    }
    ngOnDestroy() {
        this.subscription1$.unsubscribe()
        this.subscription2$.unsubscribe()
    }
}

It depends. If by calling someObservable.subscribe(), you start holding up some resource that must be manually freed-up when the lifecycle of your component is over, then you should call theSubscription.unsubscribe() to prevent memory leak.

Let's take a closer look at your examples:

getHero() returns the result of http.get(). If you look into the angular 2 source code, http.get() creates two event listeners:

_xhr.addEventListener('load', onLoad);
_xhr.addEventListener('error', onError);

and by calling unsubscribe(), you can cancel the request as well as the listeners:

_xhr.removeEventListener('load', onLoad);
_xhr.removeEventListener('error', onError);
_xhr.abort();

Note that _xhr is platform specific but I think it's safe to assume that it is an XMLHttpRequest() in your case.

Normally, this is enough evidence to warrant a manual unsubscribe() call. But according this WHATWG spec, the XMLHttpRequest() is subject to garbage collection once it is "done", even if there are event listeners attached to it. So I guess that's why angular 2 official guide omits unsubscribe() and lets GC clean up the listeners.

As for your second example, it depends on the implementation of params. As of today, the angular official guide no longer shows unsubscribing from params. I looked into src again and found that params is a just a BehaviorSubject. Since no event listeners or timers were used, and no global variables were created, it should be safe to omit unsubscribe().

The bottom line to your question is that always call unsubscribe() as a guard against memory leak, unless you are certain that the execution of the observable doesn't create global variables, add event listeners, set timers, or do anything else that results in memory leaks.

When in doubt, look into the implementation of that observable. If the observable has written some clean up logic into its unsubscribe(), which is usually the function that is returned by the constructor, then you have good reason to seriously consider calling unsubscribe().

Some of the best practices regarding observables unsubscriptions inside Angular components:

A quote from Routing & Navigation

When subscribing to an observable in a component, you almost always arrange to unsubscribe when the component is destroyed.

There are a few exceptional observables where this is not necessary. The ActivatedRoute observables are among the exceptions.

The ActivatedRoute and its observables are insulated from the Router itself. The Router destroys a routed component when it is no longer needed and the injected ActivatedRoute dies with it.

Feel free to unsubscribe anyway. It is harmless and never a bad practice.

And in responding to the following links:

• (1) Should I unsubscribe from Angular 2 Http Observables?
• (2) Is it necessary to unsubscribe from observables created by Http methods?
• (3) RxJS: Don’t Unsubscribe
• (4) The easiest way to unsubscribe from Observables in Angular
• (5) Documentation for RxJS Unsubscribing
• (6) Unsubscribing in a service is kind of pointless since there is no chance of memory leaks
• (7) Do we need to unsubscribe from observable that completes/errors-out?
• (8) A comment about the http observable

I collected some of the best practices regarding observables unsubscriptions inside Angular components to share with you:

• http observable unsubscription is conditional and we should consider the effects of the 'subscribe callback' being run after the component is destroyed on a case by case basis. We know that angular unsubscribes and cleans the http observable itself (1), (2). While this is true from the perspective of resources it only tells half the story. Let's say we're talking about directly calling http from within a component, and the http response took longer than needed so the user closed the component. The subscribe() handler will still be called even if the component is closed and destroyed. This can have unwanted side effects and in the worse scenarios leave the application state broken. It can also cause exceptions if the code in the callback tries to call something that has just been disposed of. However at the same time occasionally they are desired. Like, let's say you're creating an email client and you trigger a sound when the email is done sending - well you'd still want that to occur even if the component is closed (8).
• No need to unsubscribe from observables that complete or error. However, there is no harm in doing so(7).
• Use AsyncPipe as much as possible because it automatically unsubscribes from the observable on component destruction.
• Unsubscribe from the ActivatedRoute observables like route.params if they are subscribed inside a nested (Added inside tpl with the component selector) or dynamic component as they may be subscribed many times as long as the parent/host component exists. No need to unsubscribe from them in other scenarios as mentioned in the quote above from Routing & Navigation docs.
• Unsubscribe from global observables shared between components that are exposed through an Angular service for example as they may be subscribed multiple times as long as the component is initialized.
• No need to unsubscribe from internal observables of an application scoped service since this service never get's destroyed, unless your entire application get's destroyed, there is no real reason to unsubscribe from it and there is no chance of memory leaks. (6).
  
  Note: Regarding scoped services, i.e component providers, they are destroyed when the component is destroyed. In this case, if we subscribe to any observable inside this provider, we should consider unsubscribing from it using the OnDestroy lifecycle hook which will be called when the service is destroyed, according to the docs.
• Use an abstract technique to avoid any code mess that may be resulted from unsubscriptions. You can manage your subscriptions with takeUntil (3) or you can use this npm package mentioned at (4) The easiest way to unsubscribe from Observables in Angular.
• Always unsubscribe from FormGroup observables like form.valueChanges and form.statusChanges
• Always unsubscribe from observables of Renderer2 service like renderer2.listen
• Unsubscribe from every observable else as a memory-leak guard step until Angular Docs explicitly tells us which observables are unnecessary to be unsubscribed (Check issue: (5) Documentation for RxJS Unsubscribing (Open)).
• Bonus: Always use the Angular ways to bind events like HostListener as angular cares well about removing the event listeners if needed and prevents any potential memory leak due to event bindings.

A nice final tip: If you don't know if an observable is being automatically unsubscribed/completed or not, add a complete callback to subscribe(...) and check if it gets called when the component is destroyed.

I like the last two answers, but I experienced an issue if the the subclass referenced "this" in ngOnDestroy.

I modified it to be this, and it looks like it resolved that issue.

export abstract class BaseComponent implements OnDestroy {
    protected componentDestroyed$: Subject<boolean>;
    constructor() {
        this.componentDestroyed$ = new Subject<boolean>();
        let f = this.ngOnDestroy;
        this.ngOnDestroy = function()  {
            // without this I was getting an error if the subclass had
            // this.blah() in ngOnDestroy
            f.bind(this)();
            this.componentDestroyed$.next(true);
            this.componentDestroyed$.complete();
        };
    }
    /// placeholder of ngOnDestroy. no need to do super() call of extended class.
    ngOnDestroy() {}
}

I tried seangwright's solution (Edit 3)

That is not working for Observable that created by timer or interval.

However, i got it working by using another approach:

import { Component, OnDestroy, OnInit } from '@angular/core';
import 'rxjs/add/operator/takeUntil';
import { Subject } from 'rxjs/Subject';
import { Subscription } from 'rxjs/Subscription';
import 'rxjs/Rx';

import { MyThingService } from '../my-thing.service';

@Component({
   selector: 'my-thing',
   templateUrl: './my-thing.component.html'
})
export class MyThingComponent implements OnDestroy, OnInit {
   private subscriptions: Array<Subscription> = [];

  constructor(
     private myThingService: MyThingService,
   ) { }

  ngOnInit() {
    const newSubs = this.myThingService.getThings()
        .subscribe(things => console.log(things));
    this.subscriptions.push(newSubs);
  }

  ngOnDestroy() {
    for (const subs of this.subscriptions) {
      subs.unsubscribe();
   }
 }
}

For observables that complete directly after emitting the result like AsyncSubject or for example observables from http requests and such you don't need to unsubscribe. It doesn't hurt to to call unsubscribe() for those, but if the observable is closed the unsubscribe method will simply not do anything:

if (this.closed) {
  return;
}

When you have long-lived observables that emit several values over time (like for example a BehaviorSubject or a ReplaySubject) you need to unsubscribe to prevent memory leaks.

You can easily create an observable that completes directly after emitting a result from such long lived observables using a pipe operator. In some answers here the take(1) pipe is mentioned. But I prefer the first() pipe. The difference to take(1) is that it will:

deliver an EmptyError to the Observer's error callback if the Observable completes before any next notification was sent.

Another advantage of the first pipe is that you can pass a predicate that will help you to return the first value that satisfies certain criteria:

const predicate = (result: any) => { 
  // check value and return true if it is the result that satisfies your needs
  return true;
}
observable.pipe(first(predicate)).subscribe(observer);

First will complete directly after emitting the first value (or when passing a function argument the first value that satisfies your predicate) so there is no need to unsubscribe.

Sometimes you are not sure about whether you have a long lived observable or not. I am not saying it is good practice but you could then always add the first pipe just to make sure you won't need to manually unsubscribe. Adding an additional first pipe on an observable that will emit only one value doesn't hurt.

During development you can use the single pipe that will fail if source observable emits several events. This can help you to explore the type of observable and whether it is necessary to unsubscribe from it or not.

observable.pipe(single()).subscribe(observer);

The first and single seem very similar, both pipes can take an optional predicate but the differences are important and nicely summarized in this stackoverflow answer here:

First

Will emit as soon as first item appears. Will complete right after that.

Single

Will fail if source observable emits several events.

Note I tried to be as accurate and complete as possible in my answer with references to the official documentation, but please comment if something important is missing...

--- Update Angular 9 and Rxjs 6 Solution

1. Using unsubscribe at ngDestroy lifecycle of Angular Component

class SampleComponent implements OnInit, OnDestroy {
  private subscriptions: Subscription;
  private sampleObservable$: Observable<any>;

  constructor () {}

  ngOnInit(){
    this.subscriptions = this.sampleObservable$.subscribe( ... );
  }

  ngOnDestroy() {
    this.subscriptions.unsubscribe();
  }
}

2. Using takeUntil in Rxjs

class SampleComponent implements OnInit, OnDestroy {
  private unsubscribe$: new Subject<void>;
  private sampleObservable$: Observable<any>;

  constructor () {}

  ngOnInit(){
    this.subscriptions = this.sampleObservable$
    .pipe(takeUntil(this.unsubscribe$))
    .subscribe( ... );
  }

  ngOnDestroy() {
    this.unsubscribe$.next();
    this.unsubscribe$.complete();
  }
}

3. for some action that you call at ngOnInit that just happen only one time when component init.

class SampleComponent implements OnInit {

  private sampleObservable$: Observable<any>;

  constructor () {}

  ngOnInit(){
    this.subscriptions = this.sampleObservable$
    .pipe(take(1))
    .subscribe( ... );
  }
}

We also have async pipe. But, this one use on the template (not in Angular component).

The official Edit #3 answer (and variations) works well, but the thing that gets me is the 'muddying' of the business logic around the observable subscription.

Here's another approach using wrappers.

Warining: experimental code

File subscribeAndGuard.ts is used to create a new Observable extension to wrap .subscribe() and within it to wrap ngOnDestroy().
Usage is the same as .subscribe(), except for an additional first parameter referencing the component.

import { Observable } from 'rxjs/Observable';
import { Subscription } from 'rxjs/Subscription';

const subscribeAndGuard = function(component, fnData, fnError = null, fnComplete = null) {

  // Define the subscription
  const sub: Subscription = this.subscribe(fnData, fnError, fnComplete);

  // Wrap component's onDestroy
  if (!component.ngOnDestroy) {
    throw new Error('To use subscribeAndGuard, the component must implement ngOnDestroy');
  }
  const saved_OnDestroy = component.ngOnDestroy;
  component.ngOnDestroy = () => {
    console.log('subscribeAndGuard.onDestroy');
    sub.unsubscribe();
    // Note: need to put original back in place
    // otherwise 'this' is undefined in component.ngOnDestroy
    component.ngOnDestroy = saved_OnDestroy;
    component.ngOnDestroy();

  };

  return sub;
};

// Create an Observable extension
Observable.prototype.subscribeAndGuard = subscribeAndGuard;

// Ref: https://www.typescriptlang.org/docs/handbook/declaration-merging.html
declare module 'rxjs/Observable' {
  interface Observable<T> {
    subscribeAndGuard: typeof subscribeAndGuard;
  }
}

Here is a component with two subscriptions, one with the wrapper and one without. The only caveat is it must implement OnDestroy (with empty body if desired), otherwise Angular does not know to call the wrapped version.

import { Component, OnInit, OnDestroy } from '@angular/core';
import { Observable } from 'rxjs/Observable';
import 'rxjs/Rx';
import './subscribeAndGuard';

@Component({
  selector: 'app-subscribing',
  template: '<h3>Subscribing component is active</h3>',
})
export class SubscribingComponent implements OnInit, OnDestroy {

  ngOnInit() {

    // This subscription will be terminated after onDestroy
    Observable.interval(1000)
      .subscribeAndGuard(this,
        (data) => { console.log('Guarded:', data); },
        (error) => { },
        (/*completed*/) => { }
      );

    // This subscription will continue after onDestroy
    Observable.interval(1000)
      .subscribe(
        (data) => { console.log('Unguarded:', data); },
        (error) => { },
        (/*completed*/) => { }
      );
  }

  ngOnDestroy() {
    console.log('SubscribingComponent.OnDestroy');
  }
}

A demo plunker is here

An additional note: Re Edit 3 - The 'Official' Solution, this can be simplified by using takeWhile() instead of takeUntil() before subscriptions, and a simple boolean rather than another Observable in ngOnDestroy.

@Component({...})
export class SubscribingComponent implements OnInit, OnDestroy {

  iAmAlive = true;
  ngOnInit() {

    Observable.interval(1000)
      .takeWhile(() => { return this.iAmAlive; })
      .subscribe((data) => { console.log(data); });
  }

  ngOnDestroy() {
    this.iAmAlive = false;
  }
}

You usually need to unsubscribe when the components get destroyed, but Angular is going to handle it more and more as we go, for example in new minor version of Angular4, they have this section for routing unsubscribe:

Do you need to unsubscribe?

As described in the ActivatedRoute: the one-stop-shop for route information section of the Routing & Navigation page, the Router manages the observables it provides and localizes the subscriptions. The subscriptions are cleaned up when the component is destroyed, protecting against memory leaks, so you don't need to unsubscribe from the route paramMap Observable.

Also the example below is a good example from Angular to create a component and destroy it after, look at how component implements OnDestroy, if you need onInit, you also can implements it in your component, like implements OnInit, OnDestroy

import { Component, Input, OnDestroy } from '@angular/core';  
import { MissionService } from './mission.service';
import { Subscription }   from 'rxjs/Subscription';

@Component({
  selector: 'my-astronaut',
  template: `
    <p>
      {{astronaut}}: <strong>{{mission}}</strong>
      <button
        (click)="confirm()"
        [disabled]="!announced || confirmed">
        Confirm
      </button>
    </p>
  `
})

export class AstronautComponent implements OnDestroy {
  @Input() astronaut: string;
  mission = '<no mission announced>';
  confirmed = false;
  announced = false;
  subscription: Subscription;

  constructor(private missionService: MissionService) {
    this.subscription = missionService.missionAnnounced$.subscribe(
      mission => {
        this.mission = mission;
        this.announced = true;
        this.confirmed = false;
    });
  }

  confirm() {
    this.confirmed = true;
    this.missionService.confirmMission(this.astronaut);
  }

  ngOnDestroy() {
    // prevent memory leak when component destroyed
    this.subscription.unsubscribe();
  }
}

Angular 2 official documentation provides an explanation for when to unsubscribe and when it can be safely ignored. Have a look at this link:

https://angular.io/docs/ts/latest/cookbook/component-communication.html#!#bidirectional-service

Look for the paragraph with the heading Parent and children communicate via a service and then the blue box:

Notice that we capture the subscription and unsubscribe when the AstronautComponent is destroyed. This is a memory-leak guard step. There is no actual risk in this app because the lifetime of a AstronautComponent is the same as the lifetime of the app itself. That would not always be true in a more complex application.

We do not add this guard to the MissionControlComponent because, as the parent, it controls the lifetime of the MissionService.

I hope this helps you.

The SubSink package, an easy and consistent solution for unsubscribing

As nobody else has mentioned it, I want to recommend the Subsink package created by Ward Bell: https://github.com/wardbell/subsink#readme.

I have been using it on a project were we are several developers all using it. It helps a lot to have a consistent way that works in every situation.

Since seangwright's solution (Edit 3) appears to be very useful, I also found it a pain to pack this feature into base component, and hint other project teammates to remember to call super() on ngOnDestroy to activate this feature.

This answer provide a way to set free from super call, and make "componentDestroyed$" a core of base component.

class BaseClass {
    protected componentDestroyed$: Subject<void> = new Subject<void>();
    constructor() {

        /// wrap the ngOnDestroy to be an Observable. and set free from calling super() on ngOnDestroy.
        let _$ = this.ngOnDestroy;
        this.ngOnDestroy = () => {
            this.componentDestroyed$.next();
            this.componentDestroyed$.complete();
            _$();
        }
    }

    /// placeholder of ngOnDestroy. no need to do super() call of extended class.
    ngOnDestroy() {}
}

And then you can use this feature freely for example:

@Component({
    selector: 'my-thing',
    templateUrl: './my-thing.component.html'
})
export class MyThingComponent extends BaseClass implements OnInit, OnDestroy {
    constructor(
        private myThingService: MyThingService,
    ) { super(); }

    ngOnInit() {
        this.myThingService.getThings()
            .takeUntil(this.componentDestroyed$)
            .subscribe(things => console.log(things));
    }

    /// optional. not a requirement to implement OnDestroy
    ngOnDestroy() {
        console.log('everything works as intended with or without super call');
    }

}

Another short addition to the above mentioned situations is:

• Always unsubscribe, when new values in the subscribed stream is no more required or don't matter, it will result in way less number of triggers and increase in performance in a few cases. Cases such as components where the subscribed data/event no more exists or a new subscription to an all new stream is required (refresh, etc.) is a good example for unsubscription.

Based on : Using Class inheritance to hook to Angular 2 component lifecycle

Another generic approach:

export abstract class UnsubscribeOnDestroy implements OnDestroy {_x000D_
  protected d$: Subject<any>;_x000D_
_x000D_
  constructor() {_x000D_
    this.d$ = new Subject<void>();_x000D_
_x000D_
    const f = this.ngOnDestroy;_x000D_
    this.ngOnDestroy = () => {_x000D_
      f();_x000D_
      this.d$.next();_x000D_
      this.d$.complete();_x000D_
    };_x000D_
  }_x000D_
_x000D_
  public ngOnDestroy() {_x000D_
    // no-op_x000D_
  }_x000D_
_x000D_
}

And use :

@Component({_x000D_
    selector: 'my-comp',_x000D_
    template: ``_x000D_
})_x000D_
export class RsvpFormSaveComponent extends UnsubscribeOnDestroy implements OnInit {_x000D_
_x000D_
    constructor() {_x000D_
        super();_x000D_
    }_x000D_
_x000D_
    ngOnInit(): void {_x000D_
      Observable.of('bla')_x000D_
      .takeUntil(this.d$)_x000D_
      .subscribe(val => console.log(val));_x000D_
    }_x000D_
}

The Subscription class has an interesting feature:

Represents a disposable resource, such as the execution of an Observable. A Subscription has one important method, unsubscribe, that takes no argument and just disposes the resource held by the subscription.
Additionally, subscriptions may be grouped together through the add() method, which will attach a child Subscription to the current Subscription. When a Subscription is unsubscribed, all its children (and its grandchildren) will be unsubscribed as well.

You can create an aggregate Subscription object that groups all your subscriptions. You do this by creating an empty Subscription and adding subscriptions to it using its add() method. When your component is destroyed, you only need to unsubscribe the aggregate subscription.

@Component({ ... })
export class SmartComponent implements OnInit, OnDestroy {
  private subscriptions = new Subscription();

  constructor(private heroService: HeroService) {
  }

  ngOnInit() {
    this.subscriptions.add(this.heroService.getHeroes().subscribe(heroes => this.heroes = heroes));
    this.subscriptions.add(/* another subscription */);
    this.subscriptions.add(/* and another subscription */);
    this.subscriptions.add(/* and so on */);
  }

  ngOnDestroy() {
    this.subscriptions.unsubscribe();
  }
}

You don't need to have bunch of subscriptions and unsubscribe manually. Use Subject and takeUntil combo to handle subscriptions like a boss:

import { Subject } from "rxjs"
import { takeUntil } from "rxjs/operators"

@Component({
  moduleId: __moduleName,
  selector: "my-view",
  templateUrl: "../views/view-route.view.html"
})
export class ViewRouteComponent implements OnInit, OnDestroy {
  componentDestroyed$: Subject<boolean> = new Subject()

  constructor(private titleService: TitleService) {}

  ngOnInit() {
    this.titleService.emitter1$
      .pipe(takeUntil(this.componentDestroyed$))
      .subscribe((data: any) => { /* ... do something 1 */ })

    this.titleService.emitter2$
      .pipe(takeUntil(this.componentDestroyed$))
      .subscribe((data: any) => { /* ... do something 2 */ })

    //...

    this.titleService.emitterN$
      .pipe(takeUntil(this.componentDestroyed$))
      .subscribe((data: any) => { /* ... do something N */ })
  }

  ngOnDestroy() {
    this.componentDestroyed$.next(true)
    this.componentDestroyed$.complete()
  }
}

Alternative approach, which was proposed by @acumartini in comments, uses takeWhile instead of takeUntil. You may prefer it, but mind that this way your Observable execution will not be cancelled on ngDestroy of your component (e.g. when you make time consuming calculations or wait for data from server). Method, which is based on takeUntil, doesn't have this drawback and leads to immediate cancellation of request. Thanks to @AlexChe for detailed explanation in comments.

So here is the code:

@Component({
  moduleId: __moduleName,
  selector: "my-view",
  templateUrl: "../views/view-route.view.html"
})
export class ViewRouteComponent implements OnInit, OnDestroy {
  alive: boolean = true

  constructor(private titleService: TitleService) {}

  ngOnInit() {
    this.titleService.emitter1$
      .pipe(takeWhile(() => this.alive))
      .subscribe((data: any) => { /* ... do something 1 */ })

    this.titleService.emitter2$
      .pipe(takeWhile(() => this.alive))
      .subscribe((data: any) => { /* ... do something 2 */ })

    // ...

    this.titleService.emitterN$
      .pipe(takeWhile(() => this.alive))
      .subscribe((data: any) => { /* ... do something N */ })
  }

  ngOnDestroy() {
    this.alive = false
  }
}

For handling subscription I use a "Unsubscriber" class.

Here is the Unsubscriber Class.

export class Unsubscriber implements OnDestroy {
  private subscriptions: Subscription[] = [];

  addSubscription(subscription: Subscription | Subscription[]) {
    if (Array.isArray(subscription)) {
      this.subscriptions.push(...subscription);
    } else {
      this.subscriptions.push(subscription);
    }
  }

  unsubscribe() {
    this.subscriptions
      .filter(subscription => subscription)
      .forEach(subscription => {
        subscription.unsubscribe();
      });
  }

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

And You can use this class in any component / Service / Effect etc.

Example:

class SampleComponent extends Unsubscriber {
    constructor () {
        super();
    }

    this.addSubscription(subscription);
}

Following the answer by @seangwright, I've written an abstract class that handles "infinite" observables' subscriptions in components:

import { OnDestroy } from '@angular/core';
import { Subscription } from 'rxjs/Subscription';
import { Subject } from 'rxjs/Subject';
import { Observable } from 'rxjs/Observable';
import { PartialObserver } from 'rxjs/Observer';

export abstract class InfiniteSubscriberComponent implements OnDestroy {
  private onDestroySource: Subject<any> = new Subject();

  constructor() {}

  subscribe(observable: Observable<any>): Subscription;

  subscribe(
    observable: Observable<any>,
    observer: PartialObserver<any>
  ): Subscription;

  subscribe(
    observable: Observable<any>,
    next?: (value: any) => void,
    error?: (error: any) => void,
    complete?: () => void
  ): Subscription;

  subscribe(observable: Observable<any>, ...subscribeArgs): Subscription {
    return observable
      .takeUntil(this.onDestroySource)
      .subscribe(...subscribeArgs);
  }

  ngOnDestroy() {
    this.onDestroySource.next();
    this.onDestroySource.complete();
  }
}

To use it, just extend it in your angular component and call the subscribe() method as follows:

this.subscribe(someObservable, data => doSomething());

It also accepts the error and complete callbacks as usual, an observer object, or not callbacks at all. Remember to call super.ngOnDestroy() if you are also implementing that method in the child component.

Find here an additional reference by Ben Lesh: RxJS: Don’t Unsubscribe.

You can use latest Subscription class to unsubscribe for the Observable with not so messy code.

We can do this with normal variable but it will be override the last subscription on every new subscribe so avoid that, and this approach is very much useful when you are dealing with more number of Obseravables, and type of Obeservables like BehavoiurSubject and Subject

Subscription

Represents a disposable resource, such as the execution of an Observable. A Subscription has one important method, unsubscribe, that takes no argument and just disposes the resource held by the subscription.

you can use this in two ways,

• you can directly push the subscription to Subscription Array

   subscriptions:Subscription[] = [];
  
   ngOnInit(): void {
  
     this.subscription.push(this.dataService.getMessageTracker().subscribe((param: any) => {
              //...  
     }));
  
     this.subscription.push(this.dataService.getFileTracker().subscribe((param: any) => {
          //...
      }));
   }
  
   ngOnDestroy(){
      // prevent memory leak when component destroyed
      this.subscriptions.forEach(s => s.unsubscribe());
    }
  

• using add() of Subscription

  subscriptions = new Subscription();
  
  this.subscriptions.add(subscribeOne);
  this.subscriptions.add(subscribeTwo);
  
  ngOnDestroy() {
    this.subscriptions.unsubscribe();
  }
  

A Subscription can hold child subscriptions and safely unsubscribe them all. This method handles possible errors (e.g. if any child subscriptions are null).

Hope this helps.. :)

in SPA application at ngOnDestroy function (angular lifeCycle) For each subscribe you need to unsubscribe it. advantage => to prevent the state from becoming too heavy.

for example: in component1 :

import {UserService} from './user.service';

private user = {name: 'test', id: 1}

constructor(public userService: UserService) {
    this.userService.onUserChange.next(this.user);
}

in service:

import {BehaviorSubject} from 'rxjs/BehaviorSubject';

public onUserChange: BehaviorSubject<any> = new BehaviorSubject({});

in component2:

import {Subscription} from 'rxjs/Subscription';
import {UserService} from './user.service';

private onUserChange: Subscription;

constructor(public userService: UserService) {
    this.onUserChange = this.userService.onUserChange.subscribe(user => {
        console.log(user);
    });
}

public ngOnDestroy(): void {
    // note: Here you have to be sure to unsubscribe to the subscribe item!
    this.onUserChange.unsubscribe();
}

In case unsubscribe is needed the following operator for observable pipe method can be used

import { Observable, Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';
import { OnDestroy } from '@angular/core';

export const takeUntilDestroyed = (componentInstance: OnDestroy) => <T>(observable: Observable<T>) => {
  const subjectPropertyName = '__takeUntilDestroySubject__';
  const originalOnDestroy = componentInstance.ngOnDestroy;
  const componentSubject = componentInstance[subjectPropertyName] as Subject<any> || new Subject();

  componentInstance.ngOnDestroy = (...args) => {
    originalOnDestroy.apply(componentInstance, args);
    componentSubject.next(true);
    componentSubject.complete();
  };

  return observable.pipe(takeUntil<T>(componentSubject));
};

it can be used like this:

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

@Component({ template: '<div></div>' })
export class SomeComponent implements OnInit, OnDestroy {

  ngOnInit(): void {
    const observable = Observable.create(observer => {
      observer.next('Hello');
    });

    observable
      .pipe(takeUntilDestroyed(this))
      .subscribe(val => console.log(val));
  }

  ngOnDestroy(): void {
  }
}

The operator wraps ngOnDestroy method of component.

Important: the operator should be the last one in observable pipe.

A lot of great answers here...

Let me add another alternative:

import { interval    } from "rxjs";
import { takeUntil   } from "rxjs/operators";
import { Component   } from "@angular/core";
import { Destroyable } from "@bespunky/angular-zen/core";

@Component({
    selector: 'app-no-leak-demo',
    template: ' Destroyable component rendered. Unload me and watch me cleanup...'
})
export class NoLeakComponent extends Destroyable
{
    constructor()
    {
        super();

        this.subscribeToInterval();
    }

    private subscribeToInterval(): void
    {
        const value    = interval(1000);
        const observer = {
            next    : value => console.log(` Destroyable: ${value}`),
            complete: ()    => console.log(' Observable completed.')
        };

        // ==== Comment one and uncomment the other to see the difference ====
        
        // Subscribe using the inherited subscribe method
         this.subscribe(value, observer);

        // ... or pipe-in the inherited destroyed subject
        //value.pipe(takeUntil(this.destroyed)).subscribe(observer);
    }
}

Live Example

What's happening here

The component/service extends Destroyable (which comes from a library called @bespunky/angular-zen).

The class can now simply use this.subscribe() or takeUntil(this.destroyed) without any additional boilerplate code.

To install the library use:
> npm install @bespunky/angular-zen