从流 onError 中恢复的惯用方法

免责声明：这是上一篇的延续2 个相关流的安全更新问题

是什么惯用语处理 RxJS（或任何其他 RX 实现）中允许流不终止的错误的方法？

相关代码是

function convert(unit, value) {
    var request = {};
    request[unit] = value;

    var conversion = $.ajax({
        method: 'POST',
        url: './convert.php',
        data: request,
        dataType: 'json'
    }).promise();

    return Rx.Observable.fromPromise(conversion).takeUntil(inInput.merge(cmInput));
}

var cmElement = document.getElementById('cm'),
    inElement = document.getElementById('in');

var cmInput = Rx.Observable.fromEvent(cmElement, 'input').map(targetValue),
    inInput = Rx.Observable.fromEvent(inElement, 'input').map(targetValue);

var inches = cmInput
    .flatMap(convert.bind(null, 'cm'))
    .startWith(0);

var centimeters = inInput
    .flatMap(convert.bind(null, 'in'))
    .startWith(0);

正如您所看到的，我们使用输入字段更改的流并将其传递给convert将其转换为另一个单位并将结果进一步传递的函数。

如果期间出现错误$.ajax()调用发生然后它向上传播并且整个inches or cetimeters流停止（实际上是预期的）。

但我该如何实施才能不这样做呢？

这样我就可以优雅地处理错误，例如显示错误消息并在新数据到达时重试？

我目前的想法是引入像 Haskell 那样的复合类型Data.Either并流式传输它而不是标量双精度数。

想法？

UPD: 是的，我读过在不停止序列的情况下处理反应式扩展中的异常但我还是希望有更好的方法。

你确实有两个选择：

1. 正如你所说，返回某种形式的Either这可能是结果，也可能是错误。

由于这是 JavaScript，因此您显然不需要正式类型，只需将 Error 实例与数字一起流式传输，您的订阅者可以在收到它们时通过检查接收到的值的运行时类型来区分它们。所以这就像添加一样简单.catch(function (e) { return Rx.Observable.of(e); }在你之后.fromPromise调用（或代替.promise(), use .then()使用错误过滤器来生成一个承诺，当出现错误时该承诺将具有您想要的任何值）。

1. 将错误发送到单独的流上。

基本上都有convert采用另一个参数，它是一个观察者，它应该用来发出错误：

function convert(errorObserver, unit, value) {
    ...
    return Rx.Observable
        .fromPromise(conversion)
        .catch(function (e) {
            errorObserver.onNext(e); // or whatever you want to emit here
            return Rx.Observable.empty(); // or possibly Rx.Observable.of(0) to reset?
        })
        ...
}

然后只需创建一个Subject对于您的错误流并将其作为第一个参数提供convert。或者如果您想拥有 2 个主题cm错误与in errors.

我个人倾向于使用第一种方法。