检测 observable 何时连续两次发出相同的值

Question

我使用来自this answer的方法对我的websocket连接使用保持活动机制：

Observable.timer(0, 5000)
  .map(i => 'ping')
  .concatMap(val => {
    return Observable.race(
      Observable.of('timeout').delay(3000),
      sendMockPing()
    );
  })

如果发生超时，我需要完全重置 websocket 连接（因为这可能意味着连接断开），但有时单个超时可能只是随机发生（我猜是由于服务器实现不佳？）

我的订阅逻辑目前是这样实现的

).subscribe(result => {
  if (result === 'timeout') {
    // Reconnect to server
  }
}

有什么方法（最好使用 RxJs）以某种方式映射 observable，以便我可以识别它连续两次发出 'timeout' 的情况？

Answer 1

你可以使用scan 操作符来做你想做的事：

source.pipe(
  scan((previous, next) => {
    if ((previous === 'timeout') && (next === 'timeout')) {
      throw new Error('Two consecutive timeouts occurred.');
    }
    return next;
  }, undefined);
);