RXJS observable unsubscribe inner observables

Question

After this.route.params is destroyed and this.sub is unsubscribed, "interval run" keeps logging. This was against my intuition that with one subscription, after an unsubscribe, everything should stop happening. What's happening here & what's the best way to deal with it? Should I add a takeUntil, or should I use something other than concatMap? (i'm polling for new photos)

 constructor(private authService:AuthService,
            private route:ActivatedRoute,
            private router:Router,
            private photosApi:PhotosApi) {
}

ngOnInit() {
    this.sub = this.route.params
        .switchMap(()=>{
            console.log('switchmap Ruun')
            return Observable.interval(2000).startWith(0).concatMap(()=>{
                console.log('interval run')
                return this.photosApi.apiPhotosGet(this.authService.getAuth(), 0, 40)
            })
        })
        .subscribe((data:PagedResultPatientPhotoModel) => {
                this.processPhotos(data)
            }).add(()=>console.log('unsubscribe happened'))

edit: I tried rewriting to make it cleaner, the following with concatMap only does one call, i'm not sure why it stops working after the first call:

this.intervalSub = Observable.interval(3000).startWith(0).concatMap(()=>{
        console.log('getting photos interval', this.route.params)
        return this.route.params.switchMap(()=>{
            console.log('getting photos')
            return this.photosApi.apiPhotosGet(this.authService.getAuth(), 0, 40)
        })
    }).subscribe((data:PagedResultPatientPhotoModel) => {
                this.processPhotos(data)
            }).add(()=>console.log('unsubscribe happened'))

and the following with switchMap works as long as the api calls can run faster than 3000, otherwise they cancel each other:

this.intervalSub = Observable.interval(3000).startWith(0).switchMap(()=>{
        console.log('getting photos interval', this.route.params)
        return this.route.params.switchMap(()=>{
            console.log('getting photos')
            return this.photosApi.apiPhotosGet(this.authService.getAuth(), 0, 40)
        })
    }).subscribe((data:PagedResultPatientPhotoModel) => {
                this.processPhotos(data)
            }).add(()=>console.log('unsubscribe happened'))

i'm curious as to why switchMap (and mergemap) worked each time but concatMap only worked once, seems like a bug in RXJS

I've decided to go with mergeMap since i feel like it's unlikely for the order the requests come back in to be wrong very often, and if so, it will correct itself after a few seconds.

stackoverflow.com/questions/35765322/… - looks like switchmap can only cancel requests in the same underlying stream etc - doesn't propagate inward — robert king

cartant cartant · Accepted Answer · 2017-07-12T23:56:31

The problem is not switchMap, which behaves in the manner you would expect. In the following snippet, the inner interval observable is unsubscribed when the subscription to the composed observable is unsubscribed:

const source = new Rx.Subject();

const subscription = source
  .switchMap(() => {
    console.log('> switchmap');
    return Rx.Observable.interval(500);
  })
  .subscribe((value) => console.log(value));

source.next(1);
setTimeout(() => source.next(2), 2000);

setTimeout(() => {
  console.log('> unsubscribe');
  subscription.unsubscribe();
}, 5000);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

Your problem lies with the calling of the subscription's add method. That creates an additional subscription and it is that subscription that's assigned to sub.

The initial subscription - to which you have added the additional subscription - is not assigned to a variable and is never unsubscribed. And it's that subscription that sees the interval continue to run.

The following snippet shows the problem:

const source = new Rx.Subject();

const subscription1 = source
  .switchMap(() => {
    console.log('> switchmap');
    return Rx.Observable.interval(500);
  })
  .subscribe((value) => console.log(value));

const subscription2 = subscription1.add(() => console.log('> unsubscribed'));

source.next(1);
setTimeout(() => source.next(2), 2000);

setTimeout(() => {
  console.log('> unsubscribe 2');
  subscription2.unsubscribe();
}, 5000);
setTimeout(() => {
  console.log('> unsubscribe 1');
  subscription1.unsubscribe();
}, 6000);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

Note that calling unsubscribe on subscription1 will see subscription2 unsubscribed, too - but not vice versa.

RXJS observable unsubscribe inner observables

1 Answers