Get all current (active) subscriptions

Question

Is it possible to get all the "active" subscriptions without storing them manually?

I'd like to unsubscribe all of the "active" subscriptions and don't want to reference each of them in an array or a variable.

Answer 1

I depends on whether you're using a Subject or an Observable but there's probably no way to do this "automatically".

Observables

I don't think you can have such thing as "subscribed Observable" because you either store an Observable or Subscription:

const source = Observable.of(...)
  .map(...);

const subscription = source
  .subscribe();

Here source represents an Observable and subscription represents a single subscription.

Note that you can have a Subscription instance that stores multiple other subscriptions:

const subscriptions = new Subscription();

const sub1 = Observable...subscribe();
const sub2 = Observable...subscribe();
const sub3 = Observable...subscribe();

subscriptions.add(sub1).add(sub2).add(sub3);

// Then unsubscribe all of them with a single 
subscriptions.unsubscribe();

Subjects

If you're using Subjects they do have the unsubscribe method themselves, see https://github.com/ReactiveX/rxjs/blob/master/src/Subject.ts#L96.

However be aware that this makes the Subject "stopped", for more info see https://medium.com/@martin.sikora/rxjs-subjects-and-their-internal-state-7cfdee905156

Answer 2

Yeap. Just call .observers property if you is using Subject object of the rxjs package.

Hope this helps.

Answer 3

One of the options is to use takeUntil in combination with Subject to stop all (Observable) subscriptions.

terminator$: Subject<boolean> = new Subject();

Observable.timer(1000)
  .do(i => console.log(`timer 1: ${i}`))
  .takeUntil(terminator$)
  .subscribe();

Observable.timer(5000)
  .do(i => console.log(`timer 2: ${i}`))
  .takeUntil(terminator$)
  .subscribe();

const stopSubscriptions = () => terminator$.next(true);

Answer 4

I think the basic problem is that an Observable (with exception of Subject and derivatives) does not keep a reference to it's observers.

Without built-in references, you need to handle them externally in some form.

I think the best you could achieve is to create a reusable subscription 'override' to wrap the mechanism, although I doubt it's worth it.

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;
  }
}

Ref this question Angular/RxJs When should I unsubscribe from Subscription

Answer 5

You can just store one Subscription, and add to it.

private _subscriptions = new Subscription();   // yes you can do this!

Then add to it

_subscriptions.add(
    this.layoutManager.width$.subscribe((w) => {

        // any nested subscriptions can also use this mechanism
        _subscriptions.add(...);
    });

Then in ngOnDestroy() you just call _subscriptions.unsubscribe();.

I've been trying to make a 'clever' way to do this, but to be frank I've overengineered it quite a bit (including logging) and this is the simplest way I've come across.

Answer 6

You can inherit a Base component that implements the OnDestroy interface and keep an array to store all the Subscription and finally on destroy you can find the object with the instanceof Subscription and unsubscribe all. Let me write the code below.

1. Create a Base class that implements OnInit, ODestroy

export class BaseComponent implements OnInit, OnDestroy {

  subscriptions: Subscription[] = [];

  ngOnInit(): void {
  }

  ngOnDestroy(): void {
    console.log('OnDestory - Baselass');
    this.subscribers.forEach(sub => {
      sub.unsubscribe();
      console.log('Unsub');
    })
  }


   addObserver(subscription: Subscription) {
      this.subscribers.push(subscription);
   }
}

2. Create a component and inherit the Base class

export class SomeComponent extends BaseComponent implements OnInit, OnDestroy {

    constructor(private someService: SomeService) {
      super();
    }

    ngOnInit(): void {
      super.ngOnInit();
    }

    ngOnDestroy(): void {
      super.ngOnDestroy()
    }

    private initObservers() {
      super.addObserver(this.someService.someObservable$.subscribe(res => {

      }));
    }

    
}

That's it, You don't now need to worry about unsubscribing a Subscription manually, BaseComponent with taking care of that.

Important point:- Pass the subscription to superclass is necessary, which keeps the list and unsubscribe all Subscriptions on component destroy lifecycle event.

super.addObserver(this.someService.someObservable$.subscribe(res => {

}));

Thanks!!

Answer 7

The easiest way is to make use of ngx-auto-unsubscribe. With this decorator you shouldn't put the unsubscribe(), this is done automatically.

For use this you only need declare onDestroy, you don't need to put anything in this.

It is very easy to use, I encourage you to use it.

Npm installation

Official documentation