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

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

• javascript
• reactive-programming
• rxjs
• reactive-extensions-js

Share Improve this question Follow edited May 23, 2017 at 12:09 CommunityBot 111 silver badge asked Dec 24, 2014 at 1:59 zerkmszerkms 255k7373 gold badges447447 silver badges547547 bronze badges 1

• What is targetValue? – Brandon Commented Dec 30, 2014 at 14:24

Add a ment  | 

2 Answers 2

Sorted by: Reset to default 6

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.

You can simply add a catch() to the fromPromise() chain

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

function handleError() {
    //Do whatever you want to handle this exception then return empty.
    return Rx.Observable.Empty();
}