javascript - RxJS: Batcha förfrågningar och dela svar

Låt oss föreställa oss att jag har en funktion fetchUser som tar userId som parameter och returnerar en observerbar av användare.

Eftersom jag ofta anropar denna metod vill jag batcha id:n för att utföra en förfrågan med flera id:n istället!

Här började mina bekymmer...

Jag kan inte hitta en lösning för att göra det utan att dela ett observerbart mellan de olika anropen 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 är fult och det har flera problem:

  • Om jag avregistrerar mig från observabeln som returneras av fetchUser innan timern för bufferTime är klar, hindrar det inte hämtningen av användaren.
  • Om jag avregistrerar mig från alla observables som returneras av fetchUser innan hämtningen av batchen är klar, avbryts inte förfrågan.
  • Felh hantering är mer komplicerad
  • osv

Mer allmänt: jag vet inte hur man löser problem som kräver resursdelning med RxJS. Det är svårt att hitta avancerade exempel på RxJS.

Svar #1

Det du har är bra, men som med allt i RxJS, ligger djävulen i detaljerna.

Problem

  1. Att switchMapa
        mergeMap((userIds) => functionThatSimulateAFetch(userIds)),

Här går det först fel. Genom att använda en sammanslagningskarta här gör du det omöjligt att skilja på "ström av förfrågningar" och "ström som returneras av en enda förfrågan":

  • Du gör det nästan omöjligt att avbryta en enskild begäran (att avsluta den)
  • Du gör det omöjligt att hantera fel
  • Det faller isär om din inre observable sänder mer än en gång.

Snarare vill du sända individuella BatchEvent-händelser, via en vanlig map (som producerar en observerbar av observerbara), och switchMap/mergeMap dessa efter filtreringen.

  1. Biverkningar vid skapande av en observerbar & utsändning innan prenumeration
    userToFetch$.next(userId)
    return observable

Gör inte detta. Ett observable i sig gör egentligen ingenting. Det är en "ritning" för en sekvens av åtgärder som ska hända när du prenumererar på det. Genom att göra detta kommer du bara att skapa en batch-åtgärd vid skapandet av observable, men du är körd om du får flera eller förseningar i prenumerationerna.

Snarare vill du skapa ett observerbart från defer som skickar till userToFetch$ vid varje prenumeration.

Även då vill du prenumerera på ditt observerbara innan du skickar till userToFetch: Om du inte är prenumererad lyssnar inte ditt observerbara på ämnet, och händelsen kommer att gå förlorad. Du kan göra detta i ett defer-liknande observerbart.

Lösning

Kort, och inte särskilt annorlunda från din kod, men strukturera det så här.

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

Detta gör precis vad du bad om, inklusive:

  • Fel vidarebefordras till mottagarna av batchanropet, men ingen annan
  • Om alla avregistrerar sig från en batch avbryts observabeln
  • Om alla avregistrerar sig från en batch innan förfrågan ens skickas, skickas den aldrig
  • Observable:n beter sig som HttpClient: prenumeration på observable:n utlöser en ny (samlad) förfrågan om data. Användare kan dock fritt använda shareReplay eller vad de vill. Så inga överraskningar där.

Här är en fungerande stackblitz Angular-demo: https://stackblitz.com/edit/angular-rxjs-batch-request

Speciellt, lägg märke till beteendet när du "växlar" visningen: Du kommer att märka att att återprenumerera på befintliga observables kommer att utlösa nya batchförfrågningar, och att dessa förfrågningar kommer att avbrytas (eller helt enkelt inte utlösas) om du växlar snabbt nog.

Användningsfall

I vårt projekt använder vi detta för Angular-tabeller, där varje rad behöver hämta ytterligare data individuellt för att renderas. Detta gör att vi kan:

  • gruppera alla förfrågningar för en "enskild sida", utan att behöva någon särskild kunskap om paginering
  • Kan potentiellt hämta flera sidor samtidigt om användaren bläddrar snabbt
  • återanvänd befintliga resultat även om sidstorleken ändras

Begränsningar

Jag skulle inte lägga till chunking eller rate limiting i detta. Eftersom den ursprungliga observabla är en dum bufferTime stöter du på problem:

  • “Chunking” kommer att ske innan duplikatkontrollen. Så om du har 100 förfrågningar för ett enda userId, kommer du att sluta med att skicka flera förfrågningar med endast 1 element
  • Om du begränsar takten kommer du inte att kunna inspektera din kö. Så du kan sluta med en mycket lång kö som innehåller flera samma förfrågningar.

Detta är dock en pessimistisk synpunkt. Att åtgärda det skulle innebära att gå fullt ut med en tillståndsbaserad kö-/batch-mekanism, vilket är en storleksordning mer komplext.

Svar #2

Jag tror att @Biggy har rätt.

Det här är sättet jag förstår problemet på och vad du vill uppnå

  1. Det finns olika platser i din app där du vill hämta användare
  2. Du vill inte skicka en fetch-förfrågan hela tiden, utan du vill buffra dem och skicka dem med ett visst tidsintervall, låt oss säga 1 sekund
  3. Du vill avbryta en viss buffert och undvika att en förfrågan om att hämta en batch användare skickas under det där 1-sekundersintervallet
  4. Samtidigt, om någon, låt oss kalla det Code at Position X har begärt en användare och bara några millisekunder senare någon annan, dvs Code at Position Y avbryter hela batchen av förfrågningar, då måste Code at Position X få någon form av svar, låt oss säga null.
  5. Mer, du kanske vill kunna be om att hämta en användare och sedan ändra dig, om det är inom intervallet för bufferttiden, och undvika att denna användare hämtas (jag är långt ifrån säker på att detta verkligen är något du vill, men det verkar på något sätt framträda från din fråga)

Om detta är sant, då måste du förmodligen ha någon typ av köhanteringsmekanism, som Buggy föreslog.

Då kan det finnas många implementeringar av en sådan mekanism.

Relaterade artiklar