RxJS：如何延迟下一个发出的值？

Question

我有异步函数handleParamsChanges，它可能需要几秒钟才能解决。我在 observable 发出值时调用它：

this._activatedRoute.params
  .subscribe(params => {
    this.handleParamsChanges(params).then(() => {
      // new value can be processed now
    });
  });

如何修改我的代码，以便如果 observable 一个接一个地发出 2 个值，第一个值调用 handleParamsChanges，并且只有 在这个承诺解决之后，它被调用第二个值，等等。

编辑：

这是我想出的解决方案，但我猜有更好的方法来做到这一点：

const params$ = this._activatedRoute.params;
const canExecute$ = new BehaviorSubject(true);

combineLatest(params$, canExecute$)
  .pipe(
    filter(([_, canExecute]) => canExecute),
    map(([params]) => params),
    distinctUntilChanged()
  )
  .subscribe(async params => {
    canExecute$.next(false);
    try {
      await this.handleParamsChanges(params);
    } catch (e) {
      console.log(e);
    } finally {
      canExecute$.next(true);
    }
  })

我正在使用canExecute$ 来延迟新值的处理。

我需要在这里使用distinctUntilChanged 以避免创建无限循环。

Answer 1

您正在寻找的是concatMap。在再次订阅之前，它会等待先前的“内部可观察”完成。您还可以大大简化您的管道：

params$.pipe(
  concatMap(params => this.handleParamsChanges(params)),
).subscribe()