javascript - RxJS Retraso con prioridad

Estoy teniendo problemas para idear esta transmisión.

Lo que estoy buscando es algo como debounceTime pero con prioridad.

Entonces, si tengo eventos con la forma { type: 'a', priority: 2 }. Estos eventos necesitan ser modulados por unos segundos, pero en lugar de que se emita el último evento, se emite el evento con la mayor prioridad.

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


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

He intentado mirar otros operadores como window y filtrar el resultado para el último evento, pero no es ideal porque window funciona con una cadencia fija, mientras que quiero que el temporizador comience en el primer evento, como lo hace el debounce.

Respuesta #1

Tienes que almacenar y actualizar el artículo con la mayor prioridad y mapear a este valor highest que luego pasas a debounceTime.

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

Puedes crear tu propio operador que haga esto con la ayuda de defer. defer asegura que cada suscriptor obtenga su propia variable highest, ya que cada suscriptor obtendrá su propio Observable nuevo creado al llamar a la función fábrica proporcionada.

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

El código de arriba es Typescript. Si quieres Javascript puro, solo elimina todos los tipos.

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

Respuesta #2

Ofreceré la siguiente solución, basada en usar scan para ofrecer la emisión de mayor prioridad dada hasta ahora para su consideración por debounceTime(). Nota que scan necesita reconsiderar los nuevos datos después de cada debounce exitoso, por lo que utilizo el operador window() para dividir las emisiones, iniciando una nueva ventana observable después de cada emisión por debounceTime().

Aquí está el CodeSandbox

Y aquí hay algo de código simplificado de CodeSandbox que muestra las partes importantes:

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