在 RxJS 操作符中调用异步操作

Question

使用下面的代码，我每 500 毫秒得到一个值，我取其中的 10 个。

现在我想过滤掉无效值，但我想调用的函数是异步的（但发生得很快），我不知道如何在过滤器运算符中做到这一点。

this.subscription = this.input.getStream()
    .throttleTime(500)
    .filter(value => {
      // I want to filter out invalid values
      // but the function I want to call here is async
    })
    .take(10)  // and get 10 valid ones
    .subscribe(value => {
      this.values.push(value)
    })

注意：我希望过滤器在 时间限制之后发生。

Answer 1

一种方法是：

1. 使用flatMap调用异步函数，返回值和结果用于过滤；
2. 基于后者的过滤器；那么
3. 稍后，再次打开该值。

在你的情况下，可能是这样的：

this.subscription = this.input.getStream()
    .throttleTime(500)
    .flatMap(value => asyncFunction(value).map(include => ({ value, include }))
    .filter(({ include }) => include)
    .take(10)
    .subscribe(({ value }) => {
      this.values.push(value)
    })