Идиоматический способ восстановления из потока onError

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

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

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

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

Мысли?

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

Ответ 1

У вас действительно есть два варианта:

Как вы говорите, верните некоторую форму Either, которая может быть либо результатом, либо ошибкой.

Поскольку это JavaScript, вам, очевидно, не нужен формальный тип и он может просто передавать экземпляры ошибок вместе с цифрами, и ваш подписчик может рассказать им обособленно, когда он их получит, проверив тип времени выполнения полученного значения. Итак, это так же просто, как добавление .catch(function (e) { return Rx.Observable.of(e); } после вашего вызова .fromPromise (или вместо .promise(), используйте .then() с фильтром ошибок для получения обещания, которое будет иметь любое значение, которое вы хотите, когда есть ошибка).

Отправлять ошибки в отдельном потоке.

В принципе, 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. Или создайте 2 объекта, если вы хотите, чтобы ошибки cm были отделены от ошибок in.

Я лично предпочитаю использовать первый метод.

Ответ 2

Вы можете просто добавить catch() в цепочку fromPromise()

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