RxJS 优先去抖动答案

【问题标题】：RxJS Debounce with priorityRxJS 优先去抖动
【发布时间】：2019-08-23 20:28:21
【问题描述】：

我无法创建此信息流。

我正在寻找类似 @987654321@ 但具有优先级的内容。

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

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


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

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

【问题讨论】：

标签： javascript rxjs

【解决方案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

【讨论】：

【解决方案2】：

我将提供以下解决方案，基于使用scan 提供迄今为止最高的给定优先级发射供debounceTime() 考虑。请注意，每次成功去抖动后，scan 都需要重新考虑新数据，因此我使用运算符 window() 来拆分发射，在 debounceTime() 每次发射后启动一个新的可观察窗口。

这里是CodeSandbox

下面是来自 CodeSandbox 的一些简化代码，显示了重要的部分：

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

【讨论】：

所有提交的解决方案都有效，但我觉得这个最优雅。我在想scan 将是保留最高优先级的好方法，你很好地展示了这一点。非常感谢大家！

【解决方案3】：

您可以结合 debounceTime 和 buffer 和 filter 操作符来实现您所需要的。我为它开发了这个小例子。

https://stackblitz.com/edit/typescript-lwzt4k

/*
Collect clicks that occur, after 250ms emit array of clicks
*/
clicks$.pipe(
  buffer(clicks$.pipe(debounceTime(1000))),
  // if array is greater than 1, double click occured
  map((clickArray) => {
    document.querySelector('#emittedObjects').innerHTML = (`<div>${JSON.stringify(clickArray)}</div>`); 
    const sortedArray = clickArray.sort((a, b) => {
      return a.priority < b.priority ? 1 : -1;
    });

    const output = sortedArray.length > 0 ? sortedArray[0] : null;
    document.querySelector('#mappedOutput').innerHTML = JSON.stringify(output);
    return output;
  })
)
.subscribe((obj) => {
  const str = obj ? JSON.stringify(obj) : 'NULL';
  document.querySelector('#throttledOutput').innerHTML = `<div>THROTTLED: ${str}</div>`;
});

【讨论】：