结合最新的。收到错误后继续观察

Question

我有几个 Observable 与 combineLatest 结合在一个 Observable 中。我还有一个内部 switchMap observable，在实际示例中向远程服务器发出 http 请求。

现在我想知道，如果 switchMap returns 错误，组合的 observable 如何继续工作？

我创建了简化示例 here

//emit every 2.5 seconds
const first = interval(2500);
//emit every 2 seconds
const second = interval(2000);
//emit every 1.5 seconds
const third = interval(1500);
//emit every 1 second
const fourth = interval(1000);

let count = 0;

//emit outputs from one observable
const example = combineLatest(
  first.pipe(mapTo("FIRST!")),
  second.pipe(mapTo("SECOND!")),
  third.pipe(mapTo("THIRD")),
  fourth.pipe(mapTo("FOURTH"))
)
  .pipe(
    switchMap(data => {
      console.log(data);
      count++;
      // here lets asume in some cases http request getting error
      return count < 5 ? of("switchMap") : throwError("This is an error!");
    }),
    catchError(err => of(err))
  )
  .subscribe(val => console.log(val));

输出

["FIRST!", "SECOND!", "THIRD", "FOURTH"]
switchMap
["FIRST!", "SECOND!", "THIRD", "FOURTH"]
switchMap
["FIRST!", "SECOND!", "THIRD", "FOURTH"]
switchMap
["FIRST!", "SECOND!", "THIRD", "FOURTH"]
switchMap
["FIRST!", "SECOND!", "THIRD", "FOURTH"]
This is an error!

所以在收到错误 combineLatest 之后，observable 的工作停止了。在我的真实示例中，我有 4 个过滤器，更改过滤器后我发出 http 请求。

Answer 1

来自 combineLatest 的流本身将在发生错误时结束。
您可以通过将 catchError 添加到 switchMap.

中返回的 Observable 来阻止它

这样一来，主流就不会改变，会继续直播。

const first  = interval(2500);
const second = interval(2000);
const third  = interval(1500);
const fourth = interval(1000);

let count = 0;

combineLatest(
  first.pipe(mapTo("FIRST!")),
  second.pipe(mapTo("SECOND!")),
  third.pipe(mapTo("THIRD")),
  fourth.pipe(mapTo("FOURTH"))
).pipe(
  switchMap(data => {
    count++;
    const obs$ = count < 5
      ? of("switchMap")
      : throwError("This is an error!");

    return obs$.pipe(
      catchError(err => of(err))
    );
  })
).subscribe(val => console.log(val));

结合最新的。收到错误后继续观察

combineLatest. continue observables after getting error

javascript

observable

rxjs

combinelatest