How to unsubscribe from nested observable

Question

Using the ngOnDestroy() method on our Angular component we cancel http requests if they are still pending when leaving a page. On some pages we use a custom generic cache helper to prevent reloading data which are already loaded.

import { HttpClient } from "@angular/common/http";
import { Injectable } from "@angular/core";
import { AsyncSubject } from "rxjs";

@Injectable()
export class HttpCacheHelper {
    public cache = new Map<string, AsyncSubject<any>>();

    public constructor(private readonly http: HttpClient) {        

    }

    public get<T>(url: string): AsyncSubject<T> {
        if (!this.cache.has(url)) {
            const subject = new AsyncSubject<T>();
            this.cache.set(url, subject);
            this.http.get(url)
                .subscribe((data:any) => {
                    subject.next(data as T);
                    subject.complete();
                });
        }
        return this.cache.get(url);
    }
}

If I unsubscribe from the AsyncSubject my http call is (of course) not cancelled. How to accomplish that?

Can't you just use the share operator: http.get().pipe(share())? — Marcel Hoekstra, Dec 04 '19 at 09:19
You should consider picking an accepted answer as to help anyone else that might see this post — Jayme, Oct 12 '21 at 12:18

Ivan Tarskich · Answer 1 · 2019-12-04T09:11:26.450

1

I this case u dont have to unsubscribe from the http source ( see this question). Genarally you could use the take operator or flatmap for nested subscribtions.

edited Dec 04 '19 at 09:11

answered Dec 04 '19 at 09:02

Ivan Tarskich

1,332
1
12
19

As far as I understand it unsubscribes on completion of the request. In my case I want to unsubscribe on leaving a component. – Johan Versteeg Dec 04 '19 at 09:11
why dont u put this.http.get(url) in the cache ? – Ivan Tarskich Dec 04 '19 at 09:19
The second time I subscribe on the http.get the call is repeated. – Johan Versteeg Dec 04 '19 at 09:24
as @Marcel-Hoekstra metioned u can use the share operator – Ivan Tarskich Dec 04 '19 at 11:16
Does the share operator work with already completed calls? In this case I use it for caching and the cache can be used minutes after the original call is made. – Johan Versteeg Dec 05 '19 at 13:30

score 0 · Answer 2 · answered Dec 04 '19 at 09:01

Get the subscription Object and unsubscribe from it.

import { HttpClient } from "@angular/common/http";
import { Injectable } from "@angular/core";
import { AsyncSubject, Subscription } from "rxjs";

@Injectable()
export class HttpCacheHelper {
    public cache = new Map<string, AsyncSubject<any>>();
    private mySubscription: Subscription;
    public constructor(private readonly http: HttpClient) {        

    }

    public get<T>(url: string): AsyncSubject<T> {
        if (!this.cache.has(url)) {
            const subject = new AsyncSubject<T>();
            this.cache.set(url, subject);
    this.mySubscription = this.http.get(url)
                .subscribe((data:any) => {
                    subject.next(data as T);
                    subject.complete();
                });
        }
        return this.cache.get(url);
    }

// Use below function when required
ngOnDestroy(): void {
    this.mySubscription.unsubscribe();
  }

}

score 0 · Answer 3 · answered Dec 04 '19 at 09:08

0

You'll need to unsubscribe from the source:

public get<T>(url: string): AsyncSubject<T> {
    if (!this.cache.has(url)) {
        const subject = new AsyncSubject<T>();

        const subscription = this.http.get(url)
            .subscribe((data:any) => {
                subject.next(data as T);
                subject.complete();
            });

         this.cache.set(url, {
           subscribe(observer) { subject.subscribe(observer) },
           unsubscribe(observer) { subscription.unsubscribe() }
         });
    }

    return this.cache.get(url);
}

answered Dec 04 '19 at 09:08

undefined

6,366
12
46
90

This looks promising! Which type can I use for the cache map? new Map(); – Johan Versteeg Dec 04 '19 at 09:17
I think it can be Observable – undefined Dec 04 '19 at 09:23

score 0 · Answer 4 · answered Dec 04 '19 at 12:26

You could try unsubscribing this way as well

Create a subject we want to use to unsubscribe

private onDestroy$ = new Subject();

Add this before any .subscribe() that you need to unsubscribe from

.pipe(takeUntil(this.onDestroy$))

example

this.anyService.getData()
  .pipe(takeUntil(this.onDestroy$))
  .subscribe((data: any) => {
    // Do work ...
  });

Then use ngOnDestroy() like this

ngOnDestroy() {
  this.onDestroy$.next();
  this.onDestroy$.complete(); // Also unsubscribe from the subject itself
}

When we ngOnDestroy() runs and we complete the subject any subscriptions using the takeUntill will also automatically unsubscribe

score 0 · Answer 5 · answered Dec 05 '19 at 09:29

I would just go with a publishReplay and refCount:

@Injectable()
export class HttpCacheHelper {
  public cache = new Map<string, Observable<yourType>>();

  public constructor(private readonly http: HttpClient) {}

  public get<T>(url: string): Observable<T> {
    if (!this.cache.has(url)) {
      const obs = this.http.get(url).pipe(
         publishReplay(1),
         refCount()
      )

      this.cache.set(url, obs); 
    }

    return this.cache.get(url);
  }

  // Use below function when required
  ngOnDestroy(): void {
    this.mySubscription.unsubscribe();
  }
}

How to unsubscribe from nested observable

5 Answers5