rxjs + angular：错误“杀死”Web服务调用中的错误后的observable

Question

所以，我已经开始使用 rxjs，并且我有一个关于在从 Web 服务调用中收到错误后保持我的 observable 活动的最佳方法的问题。

在显示代码之前，这是我当前的场景：一个角度组件，它必须加载一个初始列表，该列表是分页的，也可以通过更改组合上的项目来过滤。为了用 rxjs 解决这个问题，我考虑过合并两个 observable：一个处理 select 的 change 事件，另一个用于加载更多项目。这是我正在使用的代码：

const filtro$ = this.estadoPedido.valueChanges.pipe(
    distinctUntilChanged(),
    tap(_ => {
        this._paginaAtual = 0;
        this.existemMais = true;
    }),
    startWith(this.estadoPedido.value),
    map(estado => new DadosPesquisa(this._paginaAtual,
        this._elemsPagina,
        estado,
        false))
);

Whenever the select changes, I end up resetting the global page counter (tap operator) and since I want to have an initial load, I'm also using the startWith operator.最后，我将当前状态转换为具有加载值所需的所有值的对象。

我还有一个主题，每当单击加载更多项目按钮时都会使用它：

dataRefresh$ = new Subject<DadosPesquisa>();

这两个 observable 被合并了，这样我就可以有一个单一的路径来调用我的 web 服务：

this.pedidosCarregados$ = merge(filtro$, this.dataRefresh$).pipe(
    tap(() => this.emChamadaRemota = true),
    switchMap(info => forkJoin(
        of(info),
        this._servicoPedidos.obtemPedidos(this._idInstancia,
                                info.paginaAtual,
                                info.elemsPagina,
                                info.estado)
    )),
    shareReplay(),
    tap(([info, pedidos]) => this.existemMais = pedidos.length === info.elemsPagina),
    scan((todosPedidos, info) => !info[0].addPedidosToExisting ?
        info[1] :
        todosPedidos.concat(info[1]), []),
    tap(() => this.emChamadaRemota = false),
    catchError(erro => {
        this.emChamadaRemota = false;
        this.trataErro(erro);
        this.existemMais = false;
        return of([]);
    })
);

只是快速回顾一下我在这里尝试做的事情...tap 用于设置和清理控制等待微调器 (emChamadaRemota) 的字段以及控制是否应该加载更多按钮显示（existemMais）。我在switchMap 中使用forkJoin，因为我需要访问有关跨管道的当前搜索的信息。 scan 存在是因为加载更多项目应该将项目添加到上一个加载的页面。

现在，我还使用了一个拦截器，它负责设置正确的标头并使用重试策略处理典型错误（401、503 等）。这是intercept 方法的代码：

intercept(req: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
    const headers = this.obtemHeaders();
    const requestClonado = req.clone({ headers });
    return next.handle(requestClonado).pipe(
                    retryWhen(this.retryStrategy()),
                    catchError(err => {
                        console.error(err);
                        let msgErro: string;
                        if(err instanceof HttpErrorResponse && this._servicoAutenticacao.trataErroFimSessao(err)) {
                            msgErro = "A sua sessão terminou. Vai ser redirecionado para a página de login" ;
                        }
                        else if(err.status === 503 ) {
                            msgErro = "O servidor não devolveu uma resposta válida (503).";
                        }
                        else {
                            msgErro = err.error && err.error.message ? err.error.message : "Ocorreu um erro no servidor.";
                        }
                        if(err.status !== 503) {
                            this._logger.adicionaInfoExcecao(msgErro).subscribe();
                        }
                        return throwError(msgErro);
                    }
                ));
} 

现在，问题是：如果我在 Web 服务调用中遇到错误，一切正常，但我的 observable 将被“杀死”......这是有道理的，因为操作员应该捕获错误并“取消订阅”流（至少，这是我从我读过的一些文章中理解的）。

我读过一些文章，说解决方案是创建一个从不抛出的内部可观察对象，并包装从 Web 服务调用返回的可观察对象。这是要走的路吗？如果是这样，我可以在拦截器级别进行吗？或者，如果出现错误，我应该简单地重建我的可观察链（但不使用 startWith 运算符自动启动它）？

Answer 1

好吧，经过一些测试，我设法使其工作的唯一方法（不放弃我拥有的重试/捕获错误传播）是在引发异常时重建管道。所以，我将创建代码移到了一个方法中：

private setupPipeline(runFirstTime = true) {
  const filtro$ = this.estadoPedido.valueChanges.pipe(
                   distinctUntilChanged(),
                   tap(_ => {
                         this._paginaAtual = 0;
                         this.existemMais = true;
                   }),
                   runFirstTime ? startWith(this.estadoPedido.value) : tap(),
                   map(estado => new DadosPesquisa(this._paginaAtual,
                                                  this._elemsPagina,
                                                  estado,
                                                  false))
                   );
  this.pedidosCarregados$ = merge(filtro$, this.dataRefresh$).pipe( 
       //same as before...
      catchError(erro => {
         this.emChamadaRemota = false;
         this.trataErro(erro);
         this.existemMais = false;
         setTimeout(() => this.setupRxjs(false), 100); // reset pipeline
         return of([]);
    })
}

从 init 方法和 catchError 操作符内部调用该方法。我确信有更好的方法，但是重新创建管道让我几乎可以按原样重用代码......