javascript - RXJS Řetězit závislé observably sekvenčně, ale získávat každé vyzařování pro indikátor průběhu

Čelím problému a snažil jsem se najít řešení pomocí RxJs, ale zdá se, že žádné, které by tomu vyhovovalo, nenacházím...

• Mám 3 různé REST požadavky, které budou volány postupně, a každý z nich potřebuje odpověď předchozího jako argument
• Chci implementovat indikátor průběhu, který se zvyšuje, když požadavky uspějí

Tady je, co jsem si myslel:

• Budu používat pipes a concatMap(), abych se vyhnul vnořeným odběrům a přihlásil se k odběru každého požadavku, když je předchozí dokončen.

Zvažte tuto velmi zjednodušenou verzi. Předpokládejte, že každý of představuje celý úspěšný REST požadavek (chyby budou řešeny později) a že budu provádět nezobrazenou práci s parametrem n...

const request1 = of('success 1').pipe(
  delay(500),
  tap(n => console.log('received ' + n)),
);

const request2 = (n) => of('success 2').pipe(
  delay(1000),
  tap(n => console.log('received ' + n))
);

const request3 = (n) => of('success 3').pipe(
  delay(400),
  tap(n => console.log('received ' + n))
);

request1.pipe(
  concatMap(n => request2(n).pipe(
    concatMap(n => request3(n))
  ))
)

Nicméně, když se přihlásím k odběru posledního kusu kódu, dostanu pouze odpověď na poslední požadavek, což je očekávané, protože pipe se vyřeší na této odpovědi.

Takže s concatMap() mohu správně řetězit své závislé REST volání, ale nemohu sledovat postup.

I když bych mohl sledovat pokrok docela snadno pomocí vnořených odběrů, snažím se tomu však vyhnout a použít nejlepší postup.

Jak mohu propojit své závislé REST volání, a přitom být schopen dělat věci pokaždé, když volání uspěje?

Odpověď č. 1

Toto je obecné řešení, i když ne tak jednoduché. Ale umožňuje pozorovat pokrok, přičemž se stále vyhýbá operátoru share, který může při nesprávném použití zavést neočekávanou stavovost.

const chainRequests = (firstRequestFn, ...otherRequestFns) => (
  initialParams
) => {
  return otherRequestFns.reduce(
    (chain, nextRequestFn) =>
      chain.pipe(op.concatMap((response) => nextRequestFn(response))),
    firstRequestFn(initialParams)
  );
};

chainRequests přijímá proměnlivý počet funkcí a vrací funkci, která přijímá počáteční parametry a vrací observable, který concatMaps tyto funkce dohromady, jak je ukázáno ručně v otázce. Dělá to tak, že redukuje každou funkci do akumulační hodnoty, která je náhodou observable.

Pamatujte, RxJS nás vede ven z pekla zpětných volání, pokud známe cestu.

const chainRequestsWithProgress = (...requestFns) => (initialParams) =>  {
  const progress$ = new Rx.BehaviorSubject(0);
  const wrappedFns = requestFns.map((fn, i) => (...args) =>
    fn(...args).pipe(op.tap(() => progress$.next((i + 1) / requestFns.length)))
  );
  const chain$ = Rx.defer(() => {
    progress$.next(0);
    return chainRequests(...wrappedFns)(initialParams);
  });
  return [chain$, progress$];
};

chainRequestsWithProgress vrací dva observables - jeden, který nakonec vydá poslední odpověď, a druhý, který vydává hodnoty průběhu, když je první observable přihlášen. Děláme to tak, že vytvoříme BehaviorSubject, který slouží jako náš proud hodnot průběhu, a zabalíme každou naši funkci požadavku tak, aby vracela stejný observable, který by normálně vrátila, ale také ho propojíme s tap, aby mohl posílat novou hodnotu průběhu do BehaviorSubject.

Pokrok se vymaže při každém přihlášení k prvnímu pozorovatelnému.

Pokud jste chtěli vrátit jediný observable, který by produkoval stav průběhu i konečnou výslednou hodnotu, mohli jste nechat chainRequestsWithProgress místo toho vrátit:

chain$.pipe(
  op.startWith(null),
  op.combineLatest(progress$, (result, progress) => ({ result, progress }))
)

a budete mít observable, které vyzařuje objekt reprezentující průběh směrem k konečnému výsledku, a pak samotný výsledek. K zamyšlení - musí progress$ vyzařovat jen čísla?

Upozornění

To předpokládá, že požadavky pozorovatelů vydají přesně jednu hodnotu.

Odpověď č. 2

Nejjednodušším řešením by bylo mít proměnnou počítadla průběhu, která se při každé odpovědi aktualizuje klepnutím.

    let progressCounter = 0;

    request1.pipe(
      tap(_ => progressCounter = 0.33),
      concatMap(n => request2(n).pipe(
        tap(_ => progressCounter = 0.66),
        concatMap(n => request3(n)
          .pipe(tap(_ => progressCounter = 1)))
      ))
    );

Pokud chcete, aby byl samotný pokrok pozorovatelný, pak chcete sdílet observables požadavků, abyste nevytvářeli duplicitní požadavky, a pak je kombinovat, abyste získali pokrok.

Příklad toho, jak byste k tomu mohli přistoupit, lze nalézt na: https://www.learnrxjs.io/recipes/progressbar.html