Observable.catch() 和 RxJs 中的异常处理

Question

我很难理解 RxJS 以及如何处理错误。我正在使用 Angular (4+) 并且他们切换到 RxJS 来处理简单的 HTTP 请求，我发现自己不得不非常努力地处理一些看似微不足道的事情。

这是我非常人为的代码：

import { Component, OnInit } from '@angular/core';

import 'rxjs/add/observable/from';
import 'rxjs/add/observable/of';
import 'rxjs/add/observable/empty';
import 'rxjs/add/observable/throw';

import 'rxjs/add/operator/catch';
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/mergeMap';
import 'rxjs/add/operator/switchMap';

import { Observable } from 'rxjs/Observable';
import { Subscription } from 'rxjs/Subscription';
import { BehaviorSubject } from 'rxjs/BehaviorSubject';

@Component({
    selector: 'app-root',
    templateUrl: './app.component.html',
    styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {
    title = 'app';
    index = 0;
    sub1: Subscription;
    fail = false;

    // the event generated by the user (click)
    eventA = new BehaviorSubject<number>(0);

    // another event, downstream from eventA
    eventB = this.eventA.mergeMap(x => this.fetchData(x));

    constructor() { }

    ngOnInit(): void {
        this.sub1 = this.eventB.catch(err => {
            console.log(`error was caught!`);
            return this.eventB;
        }).subscribe(x => {
            this.title = x.toString();
        });
    }

    doIt() {
        this.eventA.next(Date.now());
    }

    fetchData(input: number) {
        if (this.fail) {
            return Observable.throw(new Error(`I failed!`));
        }
        return Observable.of(input);
    }
}

和 html:

<input type="checkbox" [checked]="fail" (change)="fail = !fail" >Fail<br>
<button (click)="doIt()">doit()</button><br>
{{title}}

这是演示

可以看到，一旦失败，管道就不再执行了。我可以确认 eventA 和 eventB 仍然很好。但是，似乎 sub1 未订阅。

最后一部分是我不明白的地方。我明确地返回 eventB 以便它可以继续......或者我完全错了吗？

我的用例是这样的：

我有一个业务事件 (eventA) 强制在应用程序中刷新数据。
我不想强制应用程序中的所有组件监听该事件然后请求它们的新数据，我想利用 Rx 的强大功能来简单地让数据流通过应用程序。
因此，我的组件订阅了各种下游事件
我遇到的问题是，与任何数据调用一样，它可能会出错。我希望每个组件 自己处理 错误。
下游事件可能会为那些错误的数据调用产生错误，因此如何在不终止流的情况下让订阅者知道错误？

Answer 1

在触发器可观察对象上使用 .switchMap() 来获取每次点击的新提取，并在 switchMap 内捕获而不是在 eventB 之后捕获。

switchMap 每次都有效地给出一个新的内部订阅，所以如果旧订阅因为错误而关闭，下一次点击它会再次启动它。

唯一的潜在缺点是，如果在提取完成之前发生另一次点击，则第一次点击的提取会被丢弃。

eventB = this.eventA.switchMap(x => {
  return this.fetchData(x)
    .catch(err => {
      console.log(`error was caught!`, err);
      return Observable.empty();
    })
});

...

ngOnInit(): void {
  this.sub1 = this.eventB
    .subscribe(x => {
        this.title = x.toString();
      }
    );

这是我测试过的 StackBlitz。

Observable.catch() 和 RxJs 中的异常处理

Observable.catch() and exception handling in RxJs

exception-handling

rxjs

angular