javascript - RxJS Debounce s prioritou

Mám problém vymyslieť tento stream.

To, čo hľadám, je niečo ako debounceTime, ale s prioritou.

Takže ak mám udalosti s tvarom { type: 'a', priority: 2 }. Tieto udalosti je potrebné spomaliť o niekoľko sekúnd, ale namiesto poslednej vyslanej udalosti sa vyšle udalosť s najvyššou prioritou.

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


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

Skúšal som pozerať sa aj na iné operátory ako window a filtrovať výsledok pre poslednú udalosť, ale nie je to ideálne, pretože window funguje na pevne stanovenej frekvencii, zatiaľ čo chcem, aby časovač začal pri prvej udalosti, ako to robí debouncing.

Odpoveď č. 1

Musíte uložiť a aktualizovať položku s najvyššou prioritou a namapovať ju na túto hodnotu highest, ktorú potom odovzdáte do debounceTime.

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

Môžete si vytvoriť vlastný operátor, ktorý to robí pomocou defer. defer zabezpečí, že každý odberateľ dostane svoju vlastnú premennú highest, pretože každý odberateľ dostane vlastný nový Observable vytvorený volaním zadanej továrenskej funkcie.

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

Kód vyššie je v Typescripte. Ak chcete obyčajný Javascript, jednoducho odstráňte všetky typy.

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

Odpoveď č. 2

Navrhnem nasledujúce riešenie, založené na použití scan, aby sa ponúklo doteraz najvyššie priorizované vyžarovanie na zváženie pre debounceTime(). Všimnite si, že scan potrebuje po každom úspešnom debounce znova zvážiť nové dáta, takže používam operátor window() na rozdelenie vyžarovaní, začínajúc nové pozorovateľné okno po každom vyžarovaní z debounceTime().

Tu je CodeSandbox

A tu je trochu zjednodušený kód z CodeSandboxu, ktorý ukazuje dôležité časti:

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