javascript - RxJS: Batch forespørsler og del respons

La oss forestille oss at jeg har en funksjon fetchUser som tar som parameter userId og returnerer et observabel av bruker.

Siden jeg kaller denne metoden ofte, vil jeg batch ID-ene for å utføre én forespørsel med flere ID-er i stedet!

Her begynte mine problemer...

Jeg kan ikke finne en løsning for å gjøre det uten å dele en observable mellom de forskjellige kallene av fetchUser.

import { Subject, from } from "rxjs"
import { bufferTime, mergeMap, map, toArray, filter, take, share } from "rxjs/operators"

const functionThatSimulateAFetch = (userIds: string[]) => from(userIds).pipe(
    map((userId) => ({ id: userId, name: "George" })),
    toArray(),
)

const userToFetch$ = new Subject<string>()

const fetchedUser$ = userToFetch$.pipe(
    bufferTime(1000),
    mergeMap((userIds) => functionThatSimulateAFetch(userIds)),
    share(),
)

const fetchUser = (userId: string) => {
    const observable = fetchedUser$.pipe(
        map((users) => users.find((user) => user.id === userId)),
        filter((user) => !!user),
        take(1),
    )
    userToFetch$.next(userId)
    return observable
}

Men det er stygt og det har flere problemer:

  • Hvis jeg avslutter abonnementet på observablen som returneres av fetchUser før timeren til bufferTime er ferdig, hindrer det ikke henting av brukeren.
  • Hvis jeg avslutter abonnementet på alle observables som returneres av fetchUser før hentingen av batchen er ferdig, avbryter det ikke forespørselen.
  • Feilhåndtering er mer kompleks
  • osv.

Mer generelt: jeg vet ikke hvordan jeg skal løse problemene som krever deling av ressurser ved bruk av RxJS. Det er vanskelig å finne avanserte eksempler på RxJS.

Svar #1

Det du har er bra, men som med alt i RxJS, ligger djevelen i detaljene.

Problemer

  1. Å switchMape
        mergeMap((userIds) => functionThatSimulateAFetch(userIds)),

Dette er hvor du først tar feil. Ved å bruke et flettingskart her, gjør du det umulig å skille mellom «strømmen av forespørsler» og «strømmen som returneres av en enkelt forespørsel»:

  • Du gjør det nesten umulig å avslutte et individuelt forespørsel (for å kansellere den)
  • Du gjør det umulig å håndtere feil
  • Det faller fra hverandre hvis din indre observable sender ut mer enn én gang.

Snarere, det du ønsker er å sende ut individuelle BatchEvent-er, via en vanlig map (som produserer et observabelt av observabler), og switchMap/mergeMap disse etter filtreringen.

  1. Bivirkninger når man oppretter en observerbar og sender før abonnement
    userToFetch$.next(userId)
    return observable

Ikke gjør dette. Et observable i seg selv gjør egentlig ingenting. Det er en «blåkopi» for en sekvens av handlinger som skal skje når du abonnerer på det. Ved å gjøre dette, vil du bare opprette en batch-handling ved opprettelse av observable, men du er i trøbbel hvis du får flere eller forsinkede abonnementer.

Snarere ønsker du å lage en observable fra defer som sender til userToFetch$ ved hvert abonnement.

Selv da vil du ønske å abonnere på ditt observable før du sender til userToFetch: Hvis du ikke er abonnert, lytter ikke ditt observable til emnet, og hendelsen vil gå tapt. Du kan gjøre dette i et defer-lignende observable.

Løsning

Kort, og ikke veldig forskjellig fra koden din, men strukturer det slik.

const BUFFER_TIME = 1000;

type BatchEvent = { keys: Set<string>, values: Observable<Users> };

/** The incoming keys */
const keySubject = new Subject<string>();

const requests: Observable<{ keys: Set<string>, values: Observable<Users> }> =
  this.keySubject.asObservable().pipe(
    bufferTime(BUFFER_TIME),
    map(keys => this.fetchBatch(keys)),
    share(),
  );

/** Returns a single User from an ID. Batches the request */
function get(userId: string): Observable<User> {
  console.log("Creating observable for:", userId);
  // The money observable. See "defer":
  // triggers a new subject event on subscription
  const observable = new Observable<BatchEvent>(observer => {
    this.requests.subscribe(observer);
    // Emit *after* the subscription
    this.keySubject.next(userId);
  });
  return observable.pipe(
    first(v => v.keys.has(userId)),
    // There is only 1 item, so any *Map will do here
    switchMap(v => v.values),
    map(v => v[userId]),
  );
}

function fetchBatch(args: string[]): BatchEvent {
  const keys = new Set(args); // Do not batch duplicates
  const values = this.userService.get(Array.from(keys)).pipe(
    share(),
  );
  return { keys, values };
}

Dette gjør nøyaktig det du ba om, inkludert:

  • Feil blir spredt til mottakerne av batch-kallet, men ingen andre
  • Hvis alle avslutter abonnementet på en batch, blir observasjonen kansellert
  • Hvis alle melder seg av et batch før forespørselen til og med blir sendt, blir den aldri sendt
  • Observable oppfører seg som HttpClient: abonnering på observable utløser en ny (samlet) forespørsel om data. Kallere står fritt til å bruke shareReplay eller hva de vil. Så ingen overraskelser der.

Her er en fungerende stackblitz Angular-demo: https://stackblitz.com/edit/angular-rxjs-batch-request

Legg spesielt merke til oppførselen når du "bytter" visningen: Du vil merke at det å abonnere på eksisterende observables på nytt vil utløse nye batch-forespørsler, og at disse forespørslene vil bli avbrutt (eller rett og slett ikke utløst) hvis du bytter raskt nok.

Brukstilfelle

I vårt prosjekt bruker vi dette for Angular-tabeller, der hver rad trenger å hente ekstra data individuelt for å kunne vises. Dette lar oss:

  • grupper alle forespørslene for en «enkelt side», uten å måtte ha noen spesiell kunnskap om paginering
  • Potensielt hente flere sider samtidig hvis brukeren blar raskt
  • gjenbruk eksisterende resultater selv om sidestørrelsen endres

Begrensninger

Jeg ville ikke lagt til chunking eller rate-limiting i dette. Fordi den kilde-observerbare er en enkel bufferTime, støter du på problemer:

  • «Chunking» vil skje før duplikatsjekk. Så hvis du har 100 forespørsler for en enkelt userId, vil du ende opp med å sende flere forespørsler med bare 1 element.
  • Hvis du begrenser hastigheten, vil du ikke kunne inspisere køen din. Så du kan ende opp med en veldig lang kø som inneholder flere like forespørsler.

Dette er imidlertid et pessimistisk synspunkt. Å fikse det ville bety å gå helt inn med en tilstandsfylt kø-/batch-mekanisme, som er en størrelsesorden mer kompleks.

Svar #2

Jeg tror @Biggy har rett.

Slik forstår jeg problemet og hva du ønsker å oppnå

  1. Det finnes forskjellige steder i appen din hvor du ønsker å hente brukere
  2. Du vil ikke sende en fetch-forespørsel hele tiden, heller vil du bufre dem og sende dem med jevne mellomrom, la oss si hvert sekund
  3. Du vil avbryte en bestemt buffer og unngå at det for det 1-sekunders intervallet sendes en forespørsel om å hente en gruppe brukere
  4. Samtidig, hvis noen, la oss kalle det Kode på posisjon X har bedt om en bruker, og bare noen millisekunder senere noen andre, altså Kode på posisjon Y, kansellerer hele batchen med forespørsler, da må Kode på posisjon X få en slags svar, la oss si null.
  5. I tillegg kan det hende du vil kunne be om å hente en bruker og deretter ombestemme deg, hvis det er innenfor intervallet for buffertid, og unngå at denne brukeren blir hentet (jeg er langt fra sikker på om dette virkelig er noe du ønsker, men det ser ut til å på en eller annen måte å komme frem fra spørsmålet ditt)

Hvis dette er sant, må du sannsynligvis ha en slags kømekanisme, som Buggy foreslo.

Da kan det være mange implementeringer av en slik mekanisme.

Relaterte artikler