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

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

- Lukas Renggli

Answer 1

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

解决方案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

- Yanis-git

Answer 2

我发现了一种（在我看来）更简单的延迟抛出错误的方法：

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

- nsacerdote

---

很好 - 像那样使用throwError现在已被弃用，现在需要像这样 throwError(() => new Error('My Error')))。 - Joe Keene

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秒延迟后抛出异常。

- Vee6

---

哈，这是一个相当聪明的用法，利用了平时很少使用的一对运算符。 - Lazar Ljubenović