Conditionally use switchmap or exhaustmap angular

Question

Sorry if this question has been answered but I've searched and couldn't find it myself.

What I want to do is call a function that returns data from 1 of 2 places. Either gets cached data or gets new data from API. I am using angular 7 and have been trying to implement an observable that conditionally uses switchMap and exhaustMap. I have a decent understanding of Observables, but I know there's still something I don't understand. I was thinking of using exhaustMap as the default method which pulls from cache. But if I need to get the newest data from the API, I want to use switchMap to ignore retrieving from cache and get fresh data.

I found a stackblitz playground and forked it to do some testing. Here's my attempt https://stackblitz.com/edit/angular7-conditional-data

And this is the relevant code.

this.observable.pipe(
      map((val: string) => {
        if (/^true/.test(val)) {
          this.mergeVal = 'switch ' + val;
          return switchMap(val => {
            // console.log("SwitchMap val: ", val);
            this.switchVal = val;
            return of(val);
          })
        } else {
          this.mergeVal = 'exhaust ' + val;
          return exhaustMap(val => {
            // console.log("resolve val: ", val);
            return new Promise(resolve => {
              setTimeout(() => {
                this.exhaustVal = val;
                resolve(val);
              }, 3000);
            });
          })
        }
      })
    ).subscribe(
      data => {
      //  this.observable; 
      this.subscribeVal = 'data: ' + data;
      },
      err => {
        this.subscribeVal = 'error';
      },
      () => {
        this.subscribeVal = 'complete';
      }
    );
  }

UPDATE 2020-06-02 I added a service that makes an api call to better align with my current setup that this is going to be applied to. moved away exhaustMap and switchMap with hopes of solving the same problem using the help from people who have posted here

Now I think the problem I have is in this part of the code but I'm not sure how to fix it

  dummyAPI = 'https://swapi.dev/api/';

  cacheSubject: Subject = new Subject();
  cacheObservable: Observable = this.cacheSubject.asObservable().pipe(
    startWith(this.http.get(this.dummyAPI).pipe(
      map(val => {
        console.log('startWith: ', val);
        return this.now('dataService', true);
      })
    )),
    concatMap(val => val),
    shareReplay({refCount: true})
  );

  constructor(private http: HttpClient) {}

  now(fromId, val = false) {
    return `${val} ${moment().format('hh:mm:ss')} COMP ${fromId}`;
  }

  updateDataFromAPI(fromId, getFromAPI) {
    const val = this.now(fromId, getFromAPI);
    if (getFromAPI) {
      this.cacheSubject.next(
        this.http.get(this.dummyAPI).pipe(
          tap(val => {
            console.log("tap dummyAPI before delay: ", val);
          }),
          delay(5000),
          tap(val => {
            console.log("tap dummyAPI after delay: ", val);
          }),
        )
      );
    } else {
      this.cacheSubject.next(of(val));
    }
  }

  getData(fromId, getFromAPI?) {
    if (getFromAPI) {
      this.updateDataFromAPI(fromId, getFromAPI);
    }
    return this.cacheObservable.pipe(take(1));
  }

The problem I'm having is that the cacheObservable subscription (from getData) isn't waiting for the cacheSubject value (from calling next through updateDataFromAPI) to respond with the api data before cacheObservable completes.

user_13112018 · Answer

This won't work. Because you want to change the stream based on the emit of this.observable.

The idea of exhaustMap is to skip all emits of this.observable until the child stream of exhaustMap has been finished.

In your case I would suggest to use shareReplay(1) that would keep serving cache until an emit comes from this.observable.

const cachedStream$ = this.observable.pipe(
  startWith(null), // to trigger the first load.
  switchMap(() => this.http.get('http://my-data.url')),
  shareReplay(1),
);

cachedStream$.subscribe(cached => this.cached1 = cached);
cachedStream$.subscribe(cached => this.cached2 = cached);
cachedStream$.subscribe(cached => this.cached3 = cached);

this.observable.next(); // refresh the value from the backend.

Conditionally use switchmap or exhaustmap angular

1 Answers