处理“onNext”中为热、共享、可观察对象引发的错误的正确方法

Question

在 RxJS 版本 5 中，以下代码导致进程在两个订阅的 3 次迭代后终止：

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();

但是，我的理解是，在下一个处理程序中抛出的错误只会取消订阅该特定订阅，并且不会导致底层 observable 终止。我的预期输出是“一”订阅将取消订阅，但间隔将继续为“二”订阅产生结果。

这种行为给我带来了一些问题，我可能对一个底层的 hot observable 有多个订阅 - 但是在任何订阅中抛出一个异常会导致底层的 observable 完全终止。

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

有没有办法在不将我的每个订阅包装在 try/catch 中的情况下，在我的下一个处理程序中引发异常来简单地取消订阅该 ONE 订阅，而不终止底层的 observable？

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

通过在“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”处理程序会将错误重新冒泡？这对我来说似乎很奇怪。

Answer 1

不，不要使用该字段。如果您将其改回 true，您的订阅将开始吞下错误。

这是一个私有状态，我们用来知道订阅是同步通知（在与源 Observable 的订阅调用相同的块中）还是异步通知。如果在同步通知期间从订阅者的消息处理程序之一抛出错误，我们会推迟重新抛出它，直到我们退出 Observable 的订阅回调。[1]

如果您的处理程序抛出您希望转发到订阅的 onError 处理程序的错误，目前的指导是将它们移动到订阅上方的 do 块中。

不，我不同意这种行为。以下是一些上下文链接：

• https://github.com/zenparsing/es-observable/issues/47
• https://github.com/ReactiveX/RxJS/issues/1135

[1] 来源：这段代码是我写的。