RxJS Subject 有条件的 next() 更新，基于 current-new 值

Question

我有一个在主题中存储一些数据的 Angular 应用程序

dataSubject = new Subject<SomeDataType>();

我有两种不同的方式从服务器更新数据，看起来像

connection1.on('messageReceived', (newData: SomeDataType) => this.dataService.dataSubject.next(newData))

connection2.on('messageReceived', (newData: SomeDataType) => this.dataService.dataSubject.next(newData))

问题是最后到达的新数据不一定是最新的。 每个数据对象都包含一个我想要使用的时间戳。

有没有办法让 Subject 只接受具有 newData.timestamp > currentData.timestamp 的值？类似于条件 next()

Answer 1

听起来像是 scan 运算符的一个案例：

latestData$ = this.dataSubject.pipe(
  // scan will give you the last emitted value and the current value coming through
  // so you can compare them and select which to emit
  scan((last, current) => {
    if (last.timestamp > current.timestamp) {
      return last
    }
    return current;
  }),
  // optional operator to prevent emitting the same value twice 
  distinctUntilChanged()
);

latestData$ 只会在时间戳大于使用此设置的先前发出的值时发出值。