javascript - RxJS zpoždění (Debounce) s prioritou

Mám problém vymyslet tento proud.

Co hledám, je něco jako debounceTime, ale s prioritou.

Takže pokud mám události ve tvaru { type: 'a', priority: 2 }. Tyto události je potřeba spouštět s debounce několika sekund, ale místo toho, aby byla vyvolána poslední událost, je vyvolána událost s nejvyšší prioritou.

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


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

Zkoušel jsem se podívat na jiné operátory jako window a filtrovat výsledky pro poslední událost, ale není to ideální, protože window pracuje na pevně stanoveném rytmu, zatímco já chci, aby časovač začal při první události, stejně jako debouncing.

Odpověď č. 1

Musíte uložit a aktualizovat položku s nejvyšší prioritou a mapovat na tuto hodnotu highest, kterou pak předá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 vytvořit vlastní operátor, který to dělá, s pomocí defer. defer zajišťuje, že každý odběratel dostane svou vlastní proměnnou highest, protože každý odběratel získá vlastní nový Observable vytvořený voláním dané tovární funkce.

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

Výše uvedený kód je v Typescriptu. Pokud chcete čistý Javascript, jednoduše odstraňte všechny typy.

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

Odpověď č. 2

Nabídnu následující řešení, založené na použití scan k nabídnutí dosud nejvyšší přiřazené prioritní emise k posouzení debounceTime(). Všimněte si, že scan musí po každém úspěšném debounce znovu zvážit nová data, takže používám operátor window() k rozdělení emisí, přičemž nové okno observable začíná po každé emisi debounceTime().

Tady je CodeSandbox

A zde je nějaký zjednodušený kód z CodeSandbox ukazující důležité části:

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