从流onError恢复的惯用方式 [英] Idiomatic way to recover from stream onError

问题描述

免责声明:它是先前 2个相关流的安全更新问题的延续

Disclaimer: it is the continuation for the previous Safe update for 2 dependent streams question

用于处理RxJS(或任何其他RX实现)中的错误而导致流不终止的 idiomatic 方法是什么?

What is the idiomatic way to handle errors in RxJS (or any other RX implementation) that allows the stream to not terminate?

相关代码是

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 函数传递，该函数将其转换为另一个单位并进一步传递结果.

So as you can see we use the stream of input field changes and pass it through the convert function that converts it into another unit and passes the result further.

如果在 $.ajax()调用期间发生错误，则错误会传播并整个 inches 或 cetimeters 流停止(实际上是预期的.)

If the error during $.ajax() call occurs then it's propagated up and the whole inches or cetimeters stream stops (it actually is expected).

但是我将如何实现呢?

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

So that I could handle error gracefully, like show error message and try again when new data arrives?

我目前的想法是引入一种类似Haskell的 Data.Ether 的复合类型，并对其进行流式处理，而不是标量双精度.

My current idea is to introduce a composite type like Haskell's Data.Either and stream it instead of scalar doubles.

有想法吗?

UPD :是的，我已经阅读了在不停止序列的情况下处理Reactive Extensions中的异常但我仍然希望有更好的方法.

UPD: Yes, I've read Handling Exceptions in Reactive Extensions without stopping sequence but I still hope there are better ways.

推荐答案

您确实有两种选择:

1. 正如您所说，返回某种形式的 E 既可以是结果也可以是错误.

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

Since this is JavaScript, you obviously do not need a formal type and can just stream Error instances along with numbers and your subscriber can tell them apart when it receives them by checking the runtime type of the value received. So this is as simple as adding .catch(function (e) { return Rx.Observable.of(e); } after your .fromPromise call (or instead of .promise(), use .then() with an error filter to produce a promise that will have whatever value you want when there is an error).

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

基本上让 convert 带有另一个参数，这是它应该用来发出错误的观察者:

Basically have convert take another parameter, which is an observer it should use to emit errors:

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个主题.

Then just create a Subject for your error stream and supply it as the first argument to convert. Or create 2 subjects if you want to have the cm errors separate from the in errors.

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

I personally tend to use the first method.

这篇关于从流onError恢复的惯用方式的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！