javascript - RxJS Debounce cu prioritate

Am probleme să vin cu acest flux.

Ceea ce caut este ceva de genul debounceTime, dar cu prioritate.

Deci, dacă am evenimente cu forma { type: 'a', priority: 2 }. Aceste evenimente trebuie să fie debounced pentru câteva secunde, dar în loc ca ultimul eveniment să fie emis, este emis evenimentul cu cea mai mare prioritate.

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


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

Am încercat să mă uit la alți operatori precum window și să filtrez prin rezultat pentru ultimul eveniment, dar nu este ideal deoarece window funcționează pe o cadentă fixă, în timp ce vreau ca temporizatorul să pornească la primul eveniment, așa cum face debouncing-ul.

Răspuns #1

Trebuie să stochezi și să actualizezi elementul cu cea mai mare prioritate și să îl mapezi la această valoare highest pe care o transmiți apoi către debounceTime.

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

Poți crea propriul tău operator care face asta cu ajutorul defer. defer se asigură că fiecare abonat primește propria sa variabilă highest, deoarece fiecare abonat va primi propriul său Observable nou creat apelând funcția de fabricație dată.

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

Codul de mai sus este Typescript. Dacă vrei Javascript simplu, doar elimină toate tipurile.

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

Răspunsul #2

Voi oferi următoarea soluție, bazată pe utilizarea scan pentru a oferi cea mai mare emisie de prioritate dată până acum pentru a fi luată în considerare de debounceTime(). Rețineți că scan trebuie să reanalizeze datele noi după fiecare debounce reușit, așa că folosesc operatorul window() pentru a împărți emisiile, începând o nouă fereastră observabilă după fiecare emisie prin debounceTime().

Iată CodeSandbox

Și aici este un cod simplificat din CodeSandbox care arată părțile importante:

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