javascript - RxJS Debounce 优先级

Question

我在想出这个流时遇到了问题。

我正在寻找类似 debounceTime 但具有优先级的东西。

因此，如果我有形状为 { type: 'a', priority: 2 } 的事件。这些事件需要几秒钟的时间去抖动，但不是发出最后一个事件，而是发出具有最高优先级的事件。

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


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

我尝试查看其他运算符，如 window 并过滤最后一个事件的结果，但这并不理想，因为 window 在我想要的固定节奏下工作像去抖动一样在第一个事件上启动的计时器。

Answer 1

您必须存储和更新具有最高优先级的项并映射到此highest 值，然后将其传递给debounceTime。

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

您可以创建自己的运算符，在 defer 的帮助下执行此操作。 defer 确保每个订阅者都有自己的 highest 变量，因为每个订阅者都将通过调用给定的工厂函数获得自己的新 Observable。

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

上面的代码是Typescript。如果您想要纯 Javascript，只需删除所有类型。

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