RXJS 重复查询直到满足条件？答案

【问题标题】：RXJS Repeat query until a condition is met?RXJS 重复查询直到满足条件？
【发布时间】：2018-08-01 18:56:08
【问题描述】：

我使用的 API 只返回有限数量的结果，比如 100 个。

我想重复查询，直到返回结果集为 < 100，这意味着我已经得到了最后一个结果。

所以它会是这样的：

查询
结果集是否小于限制？如果是这样，请再次执行并附加结果。
一旦结果集小于限制，就发出最终结果。

【问题讨论】：

标签： angular rxjs

【解决方案1】：

我们还可以使用repeatWhen 运算符（类似于retryWhen，但在完成而不是错误时起作用）：

this.queryData().pipe(
    repeatWhen(obs => obs),
    filter(data => this.checkLimit()),
    take(1)
).subscribe(result => console.log(result));

注意：take(1) 是停止 repeatWhen 循环所必需的。

如果我们需要重试之间的延迟，我们可以这样做：

repeatWhen(obs => obs.pipe(delay(1000)))

【讨论】：

这个方法很优雅。
嗨，是否可以为过滤器返回布尔可观察值。我有同样的情况，但我的条件返回 boolean observable，这个 checklimit 返回 observable。我们可以在这里用 observable 过滤吗
我认为可以这样做：``` this.queryData().pipe( repeatWhen(obs => obs), concatMap(data => this.checkLimitObs(data).pipe( map(test => ([test, data])))), filter(([test, data]) => test), take(1) ).subscribe(([test, data]) => console.log （数据））; ```
我认为你可以使用find(checkLimit)。和filter()和take(1)加在一起基本一样。

【解决方案2】：

 stop$: Subject = new Subject();

 query$.pipe(takeUntil(this.stop$)).subscribe( result => {
    if(result < limit) 
       this.stop$.next();
       this.stop$.complete();
     }
   else {
      process result, i.e append somewhere...
   }
 });

注意他是 RxJs 6 语法。为了便于阅读，您可以在方法中提取订阅逻辑。

在这种情况下，使用 takeWhile 可能更容易：

doQuery: boolean = true;

 query$.pipe(takeWhile(()=> this.doQuery).subscribe( result => {
    if(result < limit) 
       this.doQuery = false;
     }
   else {
      process result, i.e append somewhere...
   }
 });

【讨论】：

【解决方案3】：

您可以将 expand 运算符用于简单的“条件重复”行为。

仅作为示例，我更改了查询以返回一个数字，而不是结果集。以下不断查询，直到检索到的数字小于100

const { defer, empty } = rxjs;
const { expand, toArray} = rxjs.operators;

const query$ = defer(async () =>  Math.floor(Math.random()*1000));
   
query$
  .pipe(
    expand(result => result < 100 ? empty() : query$),
    toArray()
  )
  .subscribe(console.log);

&lt;script src="https://unpkg.com/rxjs/bundles/rxjs.umd.min.js"&gt;&lt;/script&gt;

【讨论】：

如何限制“重复”的次数？

【解决方案4】：

You can also use RXJS Interval with TakeWhile operator. Here is the sample code

In Component:

return Observable
  .interval(250)
  .flatMap(() => this.getQueryData())
  .takeWhile(data => this.checklimit(data))
  .subscribe(result => console.log(result);

getQueryData(){
   // HTTTP API call 
}

checklimit(){
  // return true/false
}

【讨论】：

不幸的是，interval() 方法没有考虑响应时间。我们每 250 毫秒重新发送一次 http 请求。但响应时间本身可能会超过 250 毫秒。完成上一个通话和开始新通话之间没有关联。