RxJS (Interval) Observable 等待最后一个 observable 完成

Question

我想要一个等待最后一次执行的可观察间隔。 这是我的尝试。

不等待的简单事情。

interval(1000)
  .subscribe(async x => {
    await new Promise(resolve => setTimeout(resolve, Math.floor(Math.random() * 10000) + 1000));
    console.log('Got counter', x);
  });

结果：4, 1, 2, 6, 9, 7, 6, 3, ...

下一次尝试，但有点糟糕。

let alreadyRunning = false;
interval(1000)
  .pipe(skipWhile(() => alreadyRunning))
  .subscribe(async x => {
    alreadyRunning = true;
    await new Promise(resolve => setTimeout(resolve, Math.floor(Math.random() * 10000) + 1000));
    console.log('Got counter', x, alreadyRunning);
    alreadyRunning = false;
  });

skipWhile 只等待在第一件事为真之前。

现在我尝试了同样不起作用的 switchMap。

interval(1000)
  .pipe(switchMap(() => from(new Promise(resolve => setTimeout(resolve, Math.floor(Math.random() * 10000) + 1000)))))
  .subscribe(async x => {
    console.log('Got counter', x);
  });

也不行：

interval(1000)
  .pipe(switchMap(x => from(async () => {
    await new Promise(resolve => setTimeout(resolve, Math.floor(Math.random() * 10000) + 1000));
    console.log('Got counter', x);
    return x;
  })))
  .subscribe(async x => {
    console.log('X', x);
  });

是否有解决方案来实现这一点？等待最后一个 observable 完成？ 订阅后没有机会这样做。 那么我以前怎么能做到这一点。

//编辑1： 我想要什么？

我有一个在其中执行 HTTP 请求的间隔。 因此，当 HTTP 请求等待几秒钟时，将执行下一个间隔，以便多次执行请求。

我想避免的。

MergeMap 也不起作用。

interval(1000)
  .pipe(mergeMap(x => from(new Promise(resolve => setTimeout(() => resolve(x), Math.floor(Math.random() * 10000) + 1000)))))
  .subscribe(async x => {
    console.log('Got counter', x);
  });

Answer 1

我不确定您在问什么，但如果用户 AJT82 是正确的（即，您想在上一个请求尚未完成的情况下从间隔中删除任何排放）

这是我将如何实现的：

interval(1000).pipe(
  tap(_ => console.log('interval hits every second!')),
  exhaustMap(_ => {
    console.log("request has started!")
    return timer(3000).pipe( // simulation that takes 3 seconds to complete
      tap(_ => console.log("request has finished!"))
    )
  })
).subscribe({
  next: v => console.log("Value emitted: ", v),
  error: e => console.log("Error emitted: ", e),
  complete: () => console.log("Complete emitted")
});

关于 async/await 的旁白

Observable 是 Promise 的严格超集。这意味着，如果你可以用 Promise 做某事，那么你也可以用 observable 做某事。

与 Promises 不同的是，Observables 没有像 async/await 这样的特殊语法糖。

Observables 与 Promises 的互操作性相当好。大多数操作员通过在后台为您转换 Promise 来处理 Promise 以及 Observables。

如果您的项目正在使用一个并且您希望包含一个使用另一个的库，这将很有帮助。

另一方面，如果你的代码已经在使用 observables 机制，那么永远不会有理由使用 Promise。除非您真的有充分的理由，否则应该避免这种代码异味。