javascript - RxJS Debounce med prioritet

Jeg har problemer med å komme opp med denne strømmen.

Det jeg leter etter er noe som debounceTime, men med prioritet.

Så hvis jeg har hendelser med formen { type: 'a', priority: 2 }. Disse hendelsene må debounses i noen sekunder, men i stedet for at den siste hendelsen blir sendt ut, blir hendelsen med høyest prioritet sendt ut.

input stream:
------(a|1)--(b|3)---(c|2)-----------------------(a|1)-----------------


output stream:
-----------------------------------(b|3)---------------------(a|1)-----

Jeg har prøvd å se på andre operatorer som window og filtrere gjennom resultatet for den siste hendelsen, men det er ikke ideelt fordi window fungerer på en fast rytme, mens jeg vil at timeren skal starte på den første hendelsen, slik som debouncing gjør.

Svar #1

Du må lagre og oppdatere elementet med høyest prioritet og koble det til denne highest-verdien som du deretter sender til debounceTime.

let highest = null;
source$.pipe(
  map(v => highest = highest && highest.priority > v.priority ? highest : v),
  debounceTime(2000),
  tap(() => highest = null)
);

Du kan lage din egen operator som gjør dette med hjelp av defer. defer sørger for at hver abonnent får sin egen highest-variabel, ettersom hver abonnent vil få sin egen nye Observable opprettet ved å kalle den gitte fabrikkfunksjonen.

function debounceTimeHighest<T>(dueTime: number, getHighest: (curr: T, high: T) => T): MonoTypeOperatorFunction<T> {
  return (source: Observable<T>) => defer(() => {
    let highest: T = null;
    return source.pipe(
      map(item => highest = highest ? getHighest(item, highest) : item),
      debounceTime(dueTime),
      tap(() => highest = null)
    );
  });
}

// Usage
source$.pipe(
  debounceTimeHighest(2000, (v1, v2) => v1.priority >= v2.priority ? v1 : v2)
)

Koden ovenfor er Typescript. Hvis du vil ha vanlig Javascript, fjern bare alle typene.

https://stackblitz.com/edit/rxjs-hitqxk

Svar #2

Jeg vil tilby følgende løsning, basert på bruk av scan for å tilby den høyest prioriterte utsendelsen så langt for vurdering av debounceTime(). Merk at scan må vurdere ny data på nytt etter hver vellykkede debounce, så jeg bruker operatoren window() for å dele opp utsendelsene, og starter et nytt observable-vindu etter hver utsendelse av debounceTime().

Her er CodeSandbox

Og her er litt forenklet kode fra CodeSandbox som viser de viktige delene:

const resetScan$ = new Subject();

source$.pipe(
  window(resetScan$),
  mergeMap(win$ => win$.pipe(
    scan((acc, cur) => acc.priority >= cur.priority ? acc : cur )
  )),
  debounceTime(debounceDelay),
  tap(() => resetScan$.next())
);