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

Question

我正在通过 :

中的方法对我的 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）以某种方式映射可观察对象，以便我可以识别它连续发出 'timeout' 两次的情况？

Answer 1

您可以使用 scan 运算符来执行您想要的操作：

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

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

Detect when an observable has emitted the same value twice in a row

timeout

websocket

node.js

observable

rxjs