javascript - RxJS Entprellung mit Priorität

Ich habe Schwierigkeiten, mir diesen Stream auszudenken.

Wonach ich suche, ist etwas wie debounceTime, aber mit Priorität.

Also, wenn ich Ereignisse mit der Form { type: 'a', priority: 2 } habe. Diese Ereignisse müssen um einige Sekunden gedämpft werden, aber anstelle des letzten Ereignisses wird das Ereignis mit der höchsten Priorität ausgegeben.

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


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

Ich habe versucht, andere Operatoren wie window zu betrachten und das Ergebnis nach dem letzten Ereignis zu filtern, aber es ist nicht ideal, weil window in einem festen Rhythmus arbeitet, während ich möchte, dass der Timer beim ersten Ereignis startet, wie es beim Debouncing der Fall ist.

Antwort #1

Sie müssen das Element mit der höchsten Priorität speichern und aktualisieren und auf diesen highest-Wert abbilden, den Sie dann an debounceTime übergeben.

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

Sie können Ihren eigenen Operator erstellen, der dies mit Hilfe von defer tut. defer stellt sicher, dass jeder Subscriber seine eigene highest-Variable erhält, da jeder Subscriber sein eigenes neues Observable bekommt, das durch Aufrufen der angegebenen Fabrikfunktion erstellt wird.

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

Der obige Code ist Typescript. Wenn Sie einfaches Javascript möchten, entfernen Sie einfach alle Typen.

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

Antwort #2

Ich werde die folgende Lösung anbieten, die darauf basiert, scan zu verwenden, um das bisher gegebene Emissionsergebnis mit der höchsten Priorität zur Betrachtung durch debounceTime() bereitzustellen. Beachten Sie, dass scan neue Daten nach jedem erfolgreichen Debounce erneut berücksichtigen muss, daher verwende ich den Operator window(), um die Emissionen aufzuteilen und ein neues Observable-Fenster nach jeder Emission durch debounceTime() zu starten.

Hier ist der CodeSandbox

Und hier ist etwas vereinfachter Code von CodeSandbox, der die wichtigen Teile zeigt:

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