Обработка ошибок – важная и часто непростая штука, особенно когда речь идёт об асинхронных операциях. В предыдущем постингея вскользь упомянул о том, какие возможности для этого предоставляет Rx, теперь же постараюсь рассказать об этих аспектах подробнее.

Для начала некий экспериментальный базис, который я буду использовать в своих примерах.
Базис не очень большой и состоит из одной строчки:

//раз существует Enumerable.Range, то почему бы не быть Observable.Range, подумали создатели Rx
var digits = Observable.Range(1, 10).Do(ThrowIf5);

Функция .Do(Action), которую я не показывал раньше, просто производит какое-то действие (побочный эффект) над каждым элементом и просто возвращает исходный Observable. В моём случае это функция, выкидывающая исключение при значении элемента равном 5, как и следует из её названия.

Самый простой способ обрабатывать исключения – это не обрабатывать их вообще. Этот способ мы рассматривать не будем.

Второй по простоте способ – это перехватить исключение “в конце”, на этапе получения результата. Как я уже показывал, в Rx это можно сделать в момент подписки:

digits.Subscribe(
    Console.WriteLine, 
    e=>Console.WriteLine(e.Message) //подписываемся на исключение, которое придёт в OnError
);

Такой способ означает, что клиент, подписывающийся на данный аспект, должен как-то отреагировать на наличие ошибки. Это правильно, так как, в реальной жизни, гарантировать “безошибочность” невозможно.
Кроме того, подписываясь на что-то клиент не знает (и не должен знать) того, как это “что-то” устроено и, следовательно, всегда должен предполагать возможность появления ошибки и пути её обработки.

Однако, далеко не все исключения имеет смысл доводить до клиента. Некоторые из них могут быть вполне успешно обработаны “по пути” к клиенту. Возможность такой обработки в Rx предоставляет метод .Catch(…), который можно читать как “В случае возникновения ошибки продолжить с…”:

//в случае любой ошибки перейти к другому observable
digits = digits.Catch(Observable.Range(-5, 6)); 

Или так:

//а теперь перехватываем только ArgumentException
digits = digits.Catch((ArgumentException e) => Observable.Range(-5, 6));

В качестве результата мы увидим (клиент получит) значения 1,2,3,4,-5,-4,-3,-2,-1,0.

Не смотря на простоту (и благодаря возможности композиции) .Catch(…) может играть очень значительную роль в построении алгоритмов.
Часто встречающаяся ситуация: “если не получилось – попробовать ещё раз” конечно может быть обработана клиентом с помощью метода .Subscribe(…). Но как только мы начинаем об этом думать, так сразу встают вопросы от “как подписаться на ошибку так, чтобы повтор был только однократный” и до “а почему, собственно, клиент должен каждый раз об этом заботиться”? Попробуйте представить себе такое решение.

А вот гораздо более удобный способ:

//при возникновении любой ошибки просто вернуть тот же observable ещё один раз (повторить)
digits = digits.Catch(digits);

Коментарий – и тот получился длиннее

Не без оснований мы можем сказать, что совершать повторную попытку сразу же при получении ошибки может быть и не лучший способ. Возможно, было бы разумнее подождать какое-то время и после этого попытаться повторить.
Опять же, этого очень легко добиться с помощью композиции.
Рализуем метод WaitFor(TimeSpan), позволяющий подождать указанный промежуток времени:

public static IObservable<T> WaitFor<T>(this IObservable<T> source, TimeSpan time)
{
    return Observable.Timer(time).SelectMany(source);
}

Одна строчка кода, однако, возможно, требует пояснений.
Когда я столкнулся с этой задачей (в моём случае это было не просто “подождать”, а “спросить у сервера что происходит”, я рассуждал так: Мне нужно дождаться результата из одного Observable и только потом перейти к другому. Использовать метод .Concat(…) для конкатенации результатов двух аспектов я не могу: во-первых, их результаты разных типов, а во-вторых, даже если я как-то трансформирую результат первого во второй (с помощью метода .Select(…) например, хотя как можно сделать манго из огурца?), то я нне хочу, чтобы этот трансформированный результат попал клиенту! Мне важен факт того, что ожидание завершилось, а результат – нет.
И тут я вспомнил недавнюю беседу о монадах и “вырожденных методах bind” И подумал: а ведь и верно. Мне ведь просто нужен какой-то метод с сигнатурой “M a –> (a –> M b) –> M b”!
То есть, мне нужен метод Bind!
Точнее даже мне нужен метод “M a –> (_ –> M b) –> M b” (подчёркивание означает, что значение меня не интересует и имя ему присваивать я не намерен). То есть, при любом значении я возвращаю M b – и всё. Т
о есть, долой функцию и получаем M a –> M b –> M b.
То есть, мне даже лучше иметь “вырожденный” метод Bind
Далее, что у нас является методом Bind в монаде LINQ? Да это же .SelectMany(…)!
Rx определяет .SelectMany(…) для IObservable. Более того, оказалось, что Rx определяет и “вырожденный” случай, тоже.

Вот так и получается с .SelectMany(…): Дождаться результата, игнорировать его и вернуть то, что передано в параметре.
Так же и WaitFor: создаётся Observable, который получит новое значение в OnNext через промежуток времени time, этот результат будет игнорирован и SelectMany просто вернёт тот аспект, который нам нужен после ожидания.
Надеюсь, понятно объяснил.

Ну а дальше дело техники:

//в случае ошибки повторить через три секунды
digits = digits.Catch(digits.WaitFor(TimeSpan.FromSeconds(3)));

Примерно таким же образом я решал задачу “если вдруг возникло исключение “session timeout”, то сходить на сервер, получить новую сессию, а потом повторить запрос”. Абсолютно то же самое, только запрос к веб-серверу из предыдущего постинга вместо таймера.

Ну а теперь, поняв всю прелесть композиции, можно сделать столько полезных методов, сколько нужно. Вся изюминка ведь как раз в том, что ничего из того, что мы уже написали, менять не приходится!
Например, можно реализовать метод, позволяющий повторить попытку не один, а N раз перед тем, как пропустить исключение к клиенту:

public static IObservable<T> TryMultipleTimes<T>(this IObservable<T> response, int maxTries)
{
    return response.TryMultipleTimes<T>(0, maxTries);
}

//метод приватный, так как мы не хотим, чтобы текущее значение счётчика было доступно извне
private static IObservable<T> TryMultipleTimes<T>(this IObservable<T> response, int i, int maxTries)
{
    if (++i < maxTries)
        return response.Catch(response
                                .WaitFor(TimeSpan.FromSeconds(.25))
                                            .TryMultipleTimes(i, maxTries));

    return response;
}  

Использовать, опять же, очень просто:

digits = digits.TryMultipleTimes(3);

Ну и в заключение хочу отметить ещё один метод: .Finally(Action).
Как видно из названия, этот “обработчик” срабатывает как в случае OnComplete, так и в случае OnError.
Использование его и вовсе тривиально:

digits = digits
    .TryMultipleTimes(3)
    .Finally(() => Console.WriteLine("We've done!"));

P.S. В этом постинге я ничего не сказал о методе Observable.Throw<TResult>(Exception ex). Но он есть. И он “производит” observable указанного типа, сразу вызывая в нём OnError и передавая в него указанное исключение. Ой, сказал