javascript - 从流 onError 中恢复的惯用方法

Question

免责声明:它是之前Safe update for 2 dependent streams的延续问题

在允许流不终止的 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 或 cetimeters 流停止(它实际上预计)。

但我要如何实现才能不这样做呢？

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

我目前的想法是引入一种复合类型，如 Haskell 的 Data.Either 并将其流式传输，而不是标量 double 。

想法？

UPD:是的，我读过Handling Exceptions in Reactive Extensions without stopping sequence但我仍然希望有更好的方法。

Answer 1

你真的有两个选择:

1. 如您所说，返回某种形式的 Either，它可以是结果也可以是错误。

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

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。如果您希望将 cm 错误与 in 错误分开，或者创建 2 个主题。

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