如何使用 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);

You can see it in action

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

Answer 3

我遇到了类似的问题并发现了这个 github 问题：https://github.com/Reactive-Extensions/RxJS/issues/648

更新到我的用例它会是这样的：

const throwingObservable = throwError(new Error('My Error'))
  .pipe(
    materialize(),
    delay(4000),
    dematerialize()
  );

throwingObservable.subscribe(console.log, console.error);

延迟 4 秒后抛出

如何使用 RxJS 延迟 throwError？

How to delay throwError with RxJS?

rxjs

rxjs6