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