rxjs 错误处理 > 在 catchError 源上停止发射

Question

我对 rxjscatchError运算符有点困惑。这是一个使用 Angular 的简单示例：

（现场演示在这里）

import { Component } from '@angular/core';
import { of, timer } from 'rxjs'
import { tap, catchError } from 'rxjs/operators'

@Component({
  selector: 'my-app',
  templateUrl: './app.component.html',
})
export class AppComponent {
  constructor() {
    const source$ =  timer(1000, 1000).pipe(tap(n => {
      if (n === 3) {
        throw new Error('n === 3')
      }
    }))

    this.value$ = source$.pipe(catchError(err => {
      return of(42)
    }))
  }

  value$
}

{{ value$ | async }}

source$异步管道订阅的observable 发出 0,1,2 然后错误。这个错误是在catchError操作符中捕获的，它默默地吞下错误并发出 42。我想我明白了。然而，发射然后停止（我期待 4,5,6,...）。有人可以解释为什么不会发生这种情况，以及是否有任何方法可以实现这种行为？

在实践中，这对我来说很重要，例如以下情况，每次路由参数发出时我们都会加载新数据。假设我们导航到一条引发错误的路线（例如 404、500，...），我不希望事件流停止发射，因为这样我就无法导航回任何其他页面。我只想优雅地处理错误并继续监听路由更改。

this.foo$ = this._route.params.pipe(
  map(params => params.id),
  switchMap(id => this._fooServerClient.getOneById(id)),
  catchError(err => {
    return of(undefined)
  })
)

Answer 1

如果在 Observable 管道中发生异常，则是 Observable 的设计，相应的 observable [抛出错误] 将处于错误状态并且它不能发出新的/进一步的值 [即它就像取消订阅]。

现在，为了保持外部 observable 的活动 [即继续发出新值] 然后处理内部 observable 中的错误 [即catchError在内部 observable 管道中使用运算符，如下所示：

this.foo$ = this._route.params
                .pipe(
                      map(params => params.id),
                      switchMap(id => {
                          return this._fooServerClient.getOneById(id)
                                     .pipe(
                                      catchError(err => {
                                        return of(undefined);
                                      })
                                     )
                        }),

                );

其catchError在内部观察到的管道将保持外观察到活[即不断发出新的值，即使内观察到抛出异常。