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重新启动源可观察对象。

请尝试以下操作

retry的RxJS

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
);

retryWhen的RxJS

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