处理“ onNext”中热的，共享的，可观察到的错误的正确方法

debugcn 发表于 Dev

用户名

在RxJS版本5中，以下代码导致两个订阅都经过三轮迭代后终止了该过程：

var Rx = require("rxjs");

const published$ = Rx.Observable.interval(1000).publish();

published$.subscribe(index => {
    console.log(`One: ${index}`);

    if (index == 3) throw new Error("ded.");
});

published$.forEach(index => {
    console.log(`Two: ${index}`);
});

published$.connect();

但是，我的理解是，在下一个处理程序中引发的错误只会取消订阅该特定的订阅，而不会导致基础的可观察项终止。我的预期输出是“一个”订阅将取消订阅，但是该间隔将继续为“两个”订阅产生结果。

此行为引起我一些问题，其中我可能对基础的可观察对象有多个订阅，但是对这些订阅中的任何一个抛出的异常都会导致基础的可观察对象完全终止。

当我在使用热模块重载进行开发时，这尤其令人烦恼，因为任何订阅中的任何编程错误都使我不得不刷新整个页面以重新启动可观察的序列。

有没有一种方法可以在不将我的每个订阅都包装在try / catch中的情况下，在下一个处理程序中引发异常以简单地取消订阅该ONE订阅，而不终止基础的可观察对象？

- - - - - - 编辑 - - - - - -

通过在“ subscribe”返回的订阅对象上将syncErrorThrowable设置为true，我已经找到了所需的行为。似乎唯一在代码库中将此设置为true的时间是通过“ do”运算符。

我应该利用这个领域吗？我觉得使用它很脏，但是另一方面，我感到奇怪的是，“ do”运算符具有与“ next”订阅处理程序不同的错误处理语义。

这是受此标志影响的主要代码块：https : //github.com/ReactiveX/RxJS/blob/master/src%2FSubscriber.ts#L132

如果将其设置为false，则将调用此方法：https : //github.com/ReactiveX/RxJS/blob/master/src%2FSubscriber.ts#L179

如果将其设置为true，则改为使用此方法：https : //github.com/ReactiveX/RxJS/blob/master/src%2FSubscriber.ts#L188

不同之处在于，第一种方法将异常重新引发回调用堆栈，而第二种方法则将错误传播给后续订阅。

为什么do运算符会向前传播错误，而“ next”处理程序会使错误回退呢？这对我来说似乎很奇怪。