Idiomatic way to recover from stream onError

Question

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

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

Relevant code is

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

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.

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

But how would I implement it to not do so?

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

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

Thoughts?

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

Answer 1

You've really got two choices:

1. As you say, return some form of Either that can be either the result or an error.

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).

2. Send the errors out on a separate stream.

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?
        })
        ...
}

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.