如何在RxJS流中条件插入数据、延迟后再发出原事件

你需要实现的是当流中出现{ data: 3 }时，先发出{ data: 'a' }，等待500ms后再发出原事件{ data: 3 }。可以通过concat操作符将两个Observable按顺序串联起来，修改后的代码如下：

from(Array(10).keys())
  .pipe(
    map(i => ({ data: i })),
    concatMap(e => {
      if (e.data === 3) {
        // 先发出{data:'a'}，延迟500ms后发出原事件e
        return concat(
          of({ data: 'a' }),
          of(e).pipe(delay(500))
        );
      }
      return of(e);
    }),
  )
  .subscribe(i => console.log(i));

逻辑说明： #

• concat操作符会按顺序订阅传入的Observable，只有前一个Observable完成后，才会订阅下一个。这里先订阅of({ data: 'a' })，立即发出数据；
• 接着订阅of(e).pipe(delay(500))，等待500ms后发出原事件{ data: 3 }；
• 其他非3的事件保持原有逻辑，直接发出。

内容的提问来源于stack exchange，提问作者phylax