如何使用 RxJS 延迟 throwError？

Question

正如预期的那样，以下代码在 5 秒后发出 42：

const valueObservable = of(42).pipe(delay(5000));
valueObservable.subscribe((value) => console.log(value));

但是，订阅时会立即抛出版本错误：

const throwingObservable = throwError(new Error('My Error')).pipe(delay(5000));
throwingObservable.subscribe((value) => console.log(value), (error) => console.error(error));

为什么会发生这种情况？如何延迟抛出错误？

Answer 1

Rxjs 错误是异常，它立即停止流并让您捕获它以对意外情况做出反应。我想你没有办法操纵 throwError 流，除了catchError

解决方案 1：在抛出错误之前操作流。

const throwingObservable = throwError(new Error('My Error'));
timer(5000).pipe(mergeMap(e => throwingObservable))
  .subscribe((value) => console.log(value), (error) => console.error(error));

解决方案2：捕获错误，延迟流，然后再次调度

throwingObservable.pipe(
  // We catch the error, we delay by adding timer stream, then we mergeMap to the error.
  catchError(e => timer(1000).pipe(mergeMap(t => throwError(e)))
)).subscribe(console.log, console.error);

你可以看到它在行动

Answer 2

我找到了一种（IMO）更简单的方法来延迟抛出错误：

const throwingObservable = of(42).pipe(
    delay(5000),
    switchMap(() => throwError(() => new Error('My Error')))
);
throwingObservable.subscribe(
    value => console.log(value),
    error => console.error(error)
);