即使捕获到错误，Observable 也会停止触发

Question

我的项目遇到了一个非常奇怪的行为，我有一个简单的 Angular 服务，代码如下：

seatClick$ = new Subject<Seat>();

以及触发可观察对象的服务方法：

  handleSeatClick(seat: Seat) {
    this.seatClick$.next(seat);
  }

可观察的逻辑很简单：

this.seatClick$.pipe(
    exhaustMap((seat: Seat) => {
       this.someFunctionThatThrowsException(); // this function throws ref exception
      return of(null);
    })
    , catchError(err => {
        console.log('error handled');
        return of(null);
    })
)
.subscribe(() => {
    console.log('ok');
  },
  (error1 => {
    console.log('oops');
  })
);

这真的很奇怪，当调用 "someFunctionThatThrowsException" 时它抛出一些 ReferenceError 异常，然后用 catchError 捕获这个异常并触发 next() 事件。

但是，从这一刻起，seatClick observable 停止响应，就好像它已经完成了，在服务上调用 handleSeatClick 将不再响应。

我在这里错过了什么？

Answer 1

这是正确的行为，您需要 repeat 操作员在这里重新订阅。

this.seatClick$.pipe(
    exhaustMap((seat: Seat) => {
       this.someFunctionThatThrowsException();
       return of(null);
    })

    // in case of an error the stream has been completed.
    , catchError(err => {
        console.log('error handled');
        return of(null);
    })

    // now we need to resubscribe again
    , repeat() // <- here
)
.subscribe(() => {
    console.log('ok');
  },
  (error1 => {
    console.log('oops');
  })
);

此外，如果您知道某些事情可能会失败，您可以将其专用于内部流并在那里使用 catchError，那么您就不需要 repeat。

this.seatClick$.pipe(
  // or exhaustMap, or mergeMap or any other stream changer.
  switchMap(seal => of(seal).pipe(
    exhaustMap((seat: Seat) => {
       this.someFunctionThatThrowsException();
       return of(null);
    })
    , catchError(err => {
        console.log('error handled');
        return of(null);
    })
  )),
  // now switchMap always succeeds with null despite an error happened inside
  // therefore we don't need `repeat` outside and our main stream
  // will be completed only when this.seatClick$ completes
  // or throws an error.
)
.subscribe(() => {
    console.log('ok');
  },
  (error1 => {
    console.log('oops');
  })
);

Answer 2

使用 repeat() 运算符的一个很好的替代方法是将错误处理嵌套在内部管道中。完全没问题 - 这个可观察对象无论如何都应该终止。

this.seatClick$.pipe(
  exhaustMap((seat: Seat) => {
    // We moved the error handling down one level.
    this.someFunctionThatThrowsException().pipe(
      catchError(err => {
        console.log('error handled');
        return of(null);
      }),
    );
    return of(null);
  }),
).subscribe());

即使捕获到错误，Observable 也会停止触发

Observable stops firing even when catching the error

observable

rxjs

typescript

angular

angular-observable