typescript - Folosește condiționat switchmap sau exhaustmap în Angular

Îmi pare rău dacă această întrebare a fost deja răspunsă, dar am căutat și nu am putut să o găsesc singur.

Ceea ce vreau să fac este să apelez o funcție care returnează date din 1 din 2 locuri. Fie preia datele din cache, fie obține date noi de la API. Folosesc Angular 7 și am încercat să implementez un observable care folosește condiționat switchMap și exhaustMap. Am o înțelegere decentă a Observables, dar știu că încă există ceva ce nu înțeleg. Mă gândeam să folosesc exhaustMap ca metodă implicită care preia din cache. Dar dacă trebuie să obțin cele mai noi date de la API, vreau să folosesc switchMap pentru a ignora preluarea din cache și a obține date proaspete.

Am găsit un loc de joacă stackblitz și l-am bifurcat pentru a face niște teste. Iată încercarea mea https://stackblitz.com/edit/angular7-conditional-data

Și acesta este codul relevant.

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';
      }
    );
  }

ACTUALIZARE 2020-06-02 Am adăugat un serviciu care face un apel API pentru a se alinia mai bine cu configurația mea actuală la care se va aplica acest lucru. Am renunțat la exhaustMap și switchMap cu speranța de a rezolva aceeași problemă folosind ajutorul celor care au postat aici.

Acum cred că problema pe care o am este în această parte a codului, dar nu sunt sigur cum să o repar

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

  cacheSubject: Subject<any> = new Subject();
  cacheObservable: Observable<any> = 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));
  }

Problema pe care o am este că abonamentul cacheObservable (din getData) nu așteaptă ca valoarea cacheSubject (din apelarea next prin updateDataFromAPI) să răspundă cu datele API înainte ca cacheObservable să se încheie.

Răspuns #1

Asta nu va funcționa. Pentru că vrei să schimbi fluxul bazându-te pe emițerea this.observable.

Ideea lui exhaustMap este de a ignora toate emiterile lui this.observable până când fluxul copil al exhaustMap a fost finalizat.

În cazul tău aș sugera să folosești shareReplay(1) care ar continua să servească cache-ul până când apare o emitere de la 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.