javascript - rxjs Observable: handle unsubscribe of all subscribe - Stack Overflow

admin管理员组

文章数量:1289836

I have a method that return a Observable

subFoo(id): Observable<number> {
    return new Observable<number>(observer => {
        setTimeout(() => {
            observer.next(id);
        }, 1000);
    });
}

now i subscribe it three time, and after 5 second unsubscribe all:

const sub1 = subFoo(1).subscribe(result => console.log(result));
const sub2 = subFoo(2).subscribe(result => console.log(result));
const sub3 = subFoo(3).subscribe(result => console.log(result));

setTimeout(() => {
  sub1.unsubscribe();
  sub2.unsubscribe();
  sub3.unsubscribe();
}, 5000);

i can handle the plete unsubscrite of all listners?

eg. (in pseudo code):

subFoo(id): Observable<number> {
    return new Observable<number>(observer => {

        // something like this
        observer.onAllListenerAreUnsubscribed(() => {
           console.log('All Listener Are Unsubscribed!');
        });

        setTimeout(() => {
            observer.next(id);
        }, 1000);
    });
}

Live demo:

I have a method that return a Observable

subFoo(id): Observable<number> {
    return new Observable<number>(observer => {
        setTimeout(() => {
            observer.next(id);
        }, 1000);
    });
}

now i subscribe it three time, and after 5 second unsubscribe all:

const sub1 = subFoo(1).subscribe(result => console.log(result));
const sub2 = subFoo(2).subscribe(result => console.log(result));
const sub3 = subFoo(3).subscribe(result => console.log(result));

setTimeout(() => {
  sub1.unsubscribe();
  sub2.unsubscribe();
  sub3.unsubscribe();
}, 5000);

i can handle the plete unsubscrite of all listners?

eg. (in pseudo code):

subFoo(id): Observable<number> {
    return new Observable<number>(observer => {

        // something like this
        observer.onAllListenerAreUnsubscribed(() => {
           console.log('All Listener Are Unsubscribed!');
        });

        setTimeout(() => {
            observer.next(id);
        }, 1000);
    });
}

Live demo: https://stackblitz./edit/angular-ayl12r

• javascript
• rxjs

Share Improve this question Follow asked Oct 26, 2018 at 13:11 ar099968ar099968 7,5771515 gold badges7373 silver badges138138 bronze badges

Add a ment  | 

3 Answers 3

Sorted by: Reset to default 6

EDIT July 2022: The same functionality can be achieved since RxJS 7.0 with tap({ subscribe: () => ... }).

An Observable can't know about subscriptions to its chains. If you want to be able to tell how many times someone subscribed you can count it yourself:

let subscriptions = 0;

subFoo(id): Observable<number> {
  return new Observable<number>(observer => {
    subscriptions++;
    ...
    return (() => {
      if (--subscriptions === 0) {
        // do whatever...
      }
      ...
    })
  })
})

You can also collect all subscriptions on the "observer side" into a single Subscription and then add a custom handler when you unsubscribe:

const subs = new Subscription();
subs.add(subFoo(1).subscribe(...));
subs.add(subFoo(2).subscribe(...));
subs.add(subFoo(3).subscribe(...));
subs.add(() => {
  // do whatever...
});

subs.unsubscribe(); // Will unsubscribe all subscriptions and then call your custom method.

you can unsubscribe all listeners in one single line so no need to handle that event

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

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

By pleting all your observables at once, you are sure you will not get any data leakage. You can create a subject that will emit once the observables should stop emitting and use the takeUntil() operator on your observables, like so:

const pleteSubscription: Subject<void> = new Subject();

const sub1 = subFoo(1)
  .pipe(takeUntil(pleteSubscription))
  .subscribe(result => console.log(result));
const sub2 = subFoo(2)
  .pipe(takeUntil(pleteSubscription))
  .subscribe(result => console.log(result));
const sub3 = subFoo(3)
  .pipe(takeUntil(pleteSubscription))
  .subscribe(result => console.log(result));

setTimeout(() => {
  pleteSubscription.next();
  pleteSubscription.plete();
}, 5000);

本文标签： javascriptrxjs Observable handle unsubscribe of all subscribeStack Overflow