Отказ от ответственности: это продолжение предыдущегоБезопасное обновление для 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 или жеcetimeters поток останавливается (это на самом деле ожидается).

Но как мне это реализовать, чтобы не делать этого?

Чтобы я мог корректно обработать ошибку, например, показать сообщение об ошибке и повторить попытку при получении новых данных?

Моя текущая идея состоит в том, чтобы ввести составной тип как у HaskellData.Either и поток вместо скалярных двойников.

Мысли?

UPDДа, я прочиталОбработка исключений в реактивных расширениях без остановки последовательности но я все еще надеюсь, что есть лучшие способы.