Get all current (active) subscriptions
Yeap. Just call .observers
property if you is using Subject
object of the rxjs
package.
Hope this helps.
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);
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
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