rxjs：错误后保持无限流活着

Question

我想知道我是否可以在我的无限主流中合并映射的每个外部流的管道中改为 catchError，在我的主流末尾使用一个 catchError。此 catchError 返回对主流本身的引用以进行救援。可能会导致内存泄漏或任何其他问题？

这是示例代码：

import { Observable, fromEvent } from "rxjs";
import { catchError, mergeMap } from "rxjs/operators";

function foreignStream() {
  return new Observable(observer => {
    observer.next(0);
    observer.next(1);
    observer.next(2);
    observer.error("error");
  });
}

const stream$ = fromEvent(document, "click").pipe(
  mergeMap(foreignStream),
  catchError(x => {
    console.log(x);
    return stream$;
  })
);
stream$.subscribe(
  console.log,
  x => console.log("err" + x),
  () => console.log("complete")
);

Answer 1

虽然该解决方案看起来确实很有趣，但它实际上会导致在每个错误上都有新的订阅，而旧的订阅一直处于打开状态。您可以改用retry 或retryWhen 来重新启动可观察源，以防出现错误。

试试下面的

RxJS retry

const stream$ = fromEvent(document, "click").pipe(
  mergeMap(foreignStream),
  mergeMap(foreignStream2),
  mergeMap(foreignStream3),
  retry() // <-- retry immediately infinite times
);

const stream$ = fromEvent(document, "click").pipe(
  mergeMap(foreignStream),
  mergeMap(foreignStream2),
  mergeMap(foreignStream3),
  retry(5) // <-- retry immediately 5 times on errors and complete
);

RxJS retryWhen

const stream$ = fromEvent(document, "click").pipe(
  mergeMap(foreignStream),
  mergeMap(foreignStream2),
  mergeMap(foreignStream3),
  retryWhen(error => error.pipe(delay(5000))) // <-- retry after 5 seconds infinite times
);