javascript - RXJS Kæd afhængige observables sekventielt, men få hver emission til en fremdriftslinje

Jeg står over for et problem, og jeg har forsøgt at finde en løsning ved hjælp af RxJs, men jeg kan ikke synes at finde en, der passer...

• Jeg har 3 forskellige REST-forespørgsler, som vil blive kaldt sekventielt, og hver af dem har brug for svaret fra den forrige som et argument
• Jeg vil implementere en fremdriftslinje, som øges, når forespørgslerne lykkes

Her er, hvad jeg tænkte:

• Jeg vil bruge pipes og concatMap() for at undgå indlejrede abonnementer og abonnere på hver anmodning, når den forrige er færdig.

Overvej denne meget forenklede version. Antag, at hver of repræsenterer en hel REST-forespørgsel, der lykkes (håndterer fejl senere), og at jeg vil udføre ikke-viste handlinger med n-parameteren...

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))
  ))
)

Men når jeg abonnerer på det sidste stykke kode, vil jeg kun få svaret fra den sidste forespørgsel, hvilket er forventet, da pipen løser til det.

Så med concatMap() kan jeg kæde mine afhængige REST-kald korrekt, men kan ikke følge fremskridtet.

Selvom jeg kunne følge fremskridtet ganske let med indlejrede abonnementer, prøver jeg dog hårdt at undgå dette og bruge den bedste praksis.

Hvordan kan jeg kæde mine afhængige REST-opkald sammen, men stadig være i stand til at gøre ting, hver gang et opkald lykkes?

Svar #1

Dette er en generaliseret løsning, omend ikke så enkel. Men det gør fremskridt observerbart, samtidig med at share-operatoren undgås, som kan introducere uventet tilstandsafhængighed, hvis den bruges forkert.

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

chainRequests tager et variabelt antal funktioner og returnerer en funktion, der accepterer startparametre og returnerer et observable, som concatMaps funktionerne sammen, som vist manuelt i spørgsmålet. Det gør dette ved at reducere hver funktion til en akkumuleringsværdi, som tilfældigvis er et observable.

Husk, RxJS leder os ud af callback-helvede, hvis vi kender vejen.

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 returnerer to observables - den, der til sidst udsender det sidste svar, og en, der udsender fremdriftsværdier, når den første observable abonneres på. Dette gør vi ved at oprette et BehaviorSubject til at fungere som vores strøm af fremdriftsværdier, og pakker hver af vores forespørgselsfunktioner ind, så de returnerer den samme observable, som de normalt ville, men vi sender den også gennem tap, så den kan udsende en ny fremdriftsværdi til BehaviorSubject.

Fremgangen nulstilles ved hvert abonnement på det første observable.

Hvis du ønskede at returnere et enkelt observerbart, der producerede fremdriftstilstanden samt den endelige resultatværdi, kunne du have chainRequestsWithProgress i stedet returnere:

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

og du vil have et observable, der udsender et objekt, der repræsenterer fremskridtet mod det endelige resultat, og derefter selve resultatet. Til eftertanke - skal progress$ kun udsende tal?

Advarsel

Dette antager, at forespørgselsobservationer udsender præcist én værdi.

Svar #2

Den simpleste løsning ville være at have en fremdriftsvariabel, der opdateres fra et tryk, når hvert svar kommer tilbage.

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

Hvis du vil have, at selve fremgangen skal være observerbar, så vil du dele forespørgselsobserables for ikke at lave dublerede forespørgsler, og derefter kombinere dem for at få fremgangen.

Et eksempel på, hvordan du måske ønsker at nærme dig det, kan findes på: https://www.learnrxjs.io/recipes/progressbar.html