rxjs - 从不可预测的源 Observable 构建 "Heartbeat"Observable

Question

我有一个 Observable，source，它可能会在不可预测的时间发出项目。我正在尝试使用它来构建另一个每 500 毫秒可靠地发出其值的 Observable。

假设 source 在这些时间发出值:

• 100 毫秒 - 第一项
• 980 毫秒 - 第二项
• 1020 毫秒 - 第三项
• 1300 毫秒 - 第四项等

我想“平滑”这个流，以便我得到如下输出:

• 500 毫秒 - 第一项
• 1000 毫秒 - 第二项
• 1500 毫秒 - 第三项
• 2000 毫秒 - 第四项

一种天真的方法可能只是在源项的排放之间添加延迟。但是，这不会像我想要的那样创建均匀间隔的间隔。

我尝试了 .timer()、.interval() 和 .flatMap() 的各种组合，但没有任何希望, 然而。

Answer 1

我想你可以试试这个:

const src$ = merge(
  timer(100).pipe(mapTo(1)),
  timer(980).pipe(mapTo(2)),
  timer(1020).pipe(mapTo(3)),
  timer(1300).pipe(mapTo(4))
);

src$
  .pipe(
    bufferTime(500),
    mergeAll()
  )
  .subscribe(console.log);

bufferTime 用于创建一个计时器，该计时器将以恒定间隔发出，而不管发出的值如何。然后 mergeAll 用于分解由 bufferTime 产生的数组。

StackBlitz demo .