how to unsubscribe a RXJS subscription inside the subscribe method?

Question

I have some javascript:

this.mySubscription = someObservable.subscribe((obs: any) => {
   this.mySubscription.unsubscribe();
   this.mySubscription = undefined;
}

on execution, the console logs the error ERROR TypeError: Cannot read property 'unsubscribe' of undefined. I wonder why I can not unsubscribe inside the subscribe lambda function. Is there a correct way to do so? I have read a bit about using dummy-subjects and completing them or using takeUntil/takeWhile and other pipe operators workArounds.

What is a correct way/workaround to unsubscribe a subscription inside the subscription's subscribe-function?

I am currently using a dummy subscription like so:

mySubscription: BehaviorSubject<any> = new BehaviorSubject<any>(undefined);


// when I do the subscription:
dummySubscription: BehaviorSubject<any> = new BehaviourSubject<any>(this.mySubscription.getValue());
this.mySubscription = someObservable.subscribe((obs: any) => {
    // any work...
    dummySubscription.next(obs);
    dummySubscription.complete();
    dummySubscription = undefined;
}, error => {
    dummySubscription.error(error);
});

dummySubscription.subscribe((obs: any) => {
    // here the actual work to do when mySubscription  emits a value, before it should have been unsubscribed upon
}, err => {
    // if errors need be
});

Answer 1

Inspired by another answer here and especially this article, https://medium.com/@benlesh/rxjs-dont-unsubscribe-6753ed4fda87, I'd like to suggest takeUntil() with following example:

...
    let stop$: Subject<any> = new Subject<any>(); // This is the one which will stop the observable ( unsubscribe a like mechanism )

    obs$
      .pipe(
        takeUntil(stop$)
      )
      .subscribe(res => {
        if ( res.something === true ) {
          // This next to lines will cause the subscribe to stop
          stop$.next();
          stop$.complete();
        }

      });
...

And I'd like to quote sentence RxJS: Don’t Unsubscribe from those article title mentioned above :).

Answer 2

If you make your stream asynchronous, what you're doing should work. For example, this will not work:

const sub = from([1,2,3,4,5,6,7,8,9,10]).subscribe(val => {
  console.log(val);
  if(val > 5) sub.unsubscribe();
});

but this will work:

const sub2 = from([1,2,3,4,5,6,7,8,9,10]).pipe(
  delay(0)
).subscribe(val => {
  console.log(val);
  if(val > 5) sub2.unsubscribe();
});

---

Because the JS event loop is fairly predictable (blocks of code are always run to completion), If any part of your stream is asynchronous, then you can be sure that your subscription will be defined before your lambda callback is invoked.

Should you do this?

Probably not. If your code relies on the internal (otherwise hidden) machinations of your language/compiler/interpreter/etc, you've created brittle code and/or code that is hard to maintain. The next developer looking at my code is going to be confused as to why there's a delay(0) - that looks like it shouldn't do anything.

Notice that in subscribe(), your lambda has access to its closure as well as the current stream variable. The takeWhile() operator has access to the same closure and the same stream variables.

from([1,2,3,4,5,6,7,8,9,10]).pipe(
  takeWhile(val => {
    // add custom logic
    return val <= 5;
  })
).subscribe(val => {
  console.log(val);
});

takeWhile() can to anything that sub = subscribe(... sub.unsubscibe() ... ), and has the added benefit of not requiring you to manage a subscription object and being easier to read/maintain.