重试 observable 队列三次（从失败的索引开始）

Question

我得到了一组可观察值。我想订阅一次只提供一个的每个可观察源，并且在前一个完成之前不订阅下一个。如果有任何失败，我想从失败的 observable 重试（再试几次）。我找不到一个好的、聪明的方法来做到这一点。

我敢肯定有不止一种方法，但我什至还没有接近 'fluent' 在 rxjs 中，所以我有点迷路了 rn。我正在处理的项目正在使用 rxjs 6.4。

干杯！

Answer 1

您可以使用 RxJS 运算符 SwitchMap 和 retry。 SwitchMaps 将 observable 链接在一起，这样每个对象都等待最后一个发出一些东西。

obs1.pipe(
  retry(2)
  switchMap(res1 => obs2.pipe(retry(2)))
  switchMap(res2 => obs3.pipe(retry(2)))
).subscribe(res3 => console.log(res3))

编辑：

如果不知道可观察量的数量怎么办：

let arr = [obs1$, obs2$, obs3$]

of(arr).pipe(
  ...(arr.reduce((p, q) => [...p, switchMap(e), retry(2)]))
).subscribe(console.log)

Answer 2

您可以在 concatMap() 中使用 retry(X) 但这实际上取决于您想要的行为：

source$
  .pipe(
    concatMap(val => makeRequest(val).pipe(
      retry(3), // Will retry only this request 3 times.
    )),
  );

与concat:

concat(...sources.map(obs => obs.pipe(retry(3))))
  .subscribe(result => {});

Retry observable queue three times (from the index where it failed)