如何在 Rxjs 中创建条件轮询运算符答案

【问题标题】：How to create a conditonal polling operator in Rxjs如何在 Rxjs 中创建条件轮询运算符
【发布时间】：2020-02-12 23:05:37
【问题描述】：

我需要使用 Rxjs 创建一个轮询运算符，它不使用时间源（例如计时器或间隔）。这将用于轮询与返回值所需的时间完全不一致的第三方 api。如果返回的值为空，我只想再次轮询。

到目前为止我尝试过的是

this.someService.getStoredValue().pipe(
  repeatWhen(newPoll => newPoll.pipe( // delay each resubmission by two seconds
    delay(2000)
  )),
  pluck('Data), // only interested in the Data property of the returned object
  filter(isNotNull => isNotNull !== ''), 
  timeout(20000), // if gone 20 seconds without a successfull polling attempt, throw error
).subscribe({
 next: value => {
    // Only if polling is succesfull
   },
 error: err => {
      // I want to end up here if the api can not be reached, or the timeout occurs.
   },
 complete: () => console.log('completed')
})

我希望操作员的输出尝试轮询，例如 10 次。如果返回对象的 Data 属性具有非空值，则取消订阅轮询服务。如果尝试了 10 次，则取消订阅并发出错误。

但是如果轮询从未成功，订阅会无限期地继续下去。

【问题讨论】：

也许你想要this answer。它使用计时器，但我不知道你为什么要求不包括这个。
我肯定会为此推荐 rxjs-take-while-inclusive：npmjs.com/package/rxjs-take-while-inclusive
interval(1000).pipe( takeWhileInclusive(v => v < 5), ).subscribe(console.log);
@AndrewAllen 嗨，一定会去看看。我不想依赖计时器的原因是在第 3 方 api 需要超过一秒的时间来回复的情况下。当第 3 方仍在制定有效回复时，这将导致我的轮询函数再次轮询。
@Anders 在这种情况下，链接不太相关 - 它更多的是关于何时重试失败的时间。我会使用 mbojko 答案（可能与链接的 retryWhen(generalRetryStrategy()) 一起增加失败的延迟）

标签： angular rxjs

【解决方案1】：

我肯定会推荐 takeWhileInclusive 运算符，可以在这里下载：https://www.npmjs.com/package/rxjs-take-while-inclusive

interval(1000).pipe(
   takeWhileInclusive((data) => runCondition(data)))
   .subscribe(console.log)
);

只需将您的条件作为函数传递并添加您的间隔时间，它就会起作用。

【讨论】：

【解决方案2】：

这似乎是一个retry 案例，其中包含一个空答案应该被视为错误的问题。比如：

this.someService.getStoredValue().pipe(
    pluck('Data'),
    flatMap(value => !!value ? of(value) : throwError("Empty!")),
    retry(10),
)

【讨论】：

如果您想在每次重新提交之间添加延迟，您会怎么做？因此，在您从第 3 方获得空值后，您会将下一个轮询请求延迟 1 秒。我尝试在 throwError 运算符上添加delay(2000)，但没有效果。
看看这个帖子：stackoverflow.com/questions/44979131/…