В сети очень много русско- и англоязычных статей по Rx операторам retryWhen и repeatWhen.
Несмотря на это, очень часто встречаю нежелание их использовать (ввиду сложного синтаксиса и непонятных диаграмм).
Приведу несколько как можно с их помощью эффективно перезапускать участки цепи и делегировать обработку перезапусков при ошибках и завершениях потока.
В примерах будет Java код с лямбдами (Retrolamda), но переписать его на Kotlin или чистую Java не составит труда.
Императивный способ перезапуска цепи
Предположим, мы используем Retrofit и загрузку начинаем в методе load(). Repository.getSomething() возвращает Single<Something>().
@NonNull
private Subscription loadingSubscription = Subscriptions.unsubscribed();
private void load() {
subscription.unsubscribe();
subscription = repository
.getSomething()
.subscribe(result -> {}, err -> {});
}
private void update() {
load();
}
Из какого-нибудь листенера обновлений (e.g. PullToRefreshView) мы вызываем метод update(), который, в свою очередь, вызовет метод load(), где с нуля будет создана подписка.
Я предпочитаю ко вниманию вариант использования более реактивного, на мой взгляд, способа с вышеупомянутым оператором repeatWhen().
Реактивный способ перезапуска цепи — repeatWhen
Создадим объект PublishSubject updateSubject, и передадим в оператор лямбду
repeatHandler -> repeatHandler.flatMap(nothing -> updateSubject.asObservable())
@NonNull
private final PublishSubject<Void> updateSubject = PublishSubject.create();
private void load() {
repository
.getSomething()
.repeatWhen(repeatHandler ->
repeatHandler.flatMap(nothing -> updateSubject.asObservable()))
.subscribe(result -> {}, err -> {});
}
Теперь для обновления загруженных данных нужно заэмитить null в updateSubject.
private void update() {
updateSubject.onNext(null);
}
Нужно помнить, что работает такой реактивный способ только с Single, который вызывает onComplete() сразу после эмита единственного элемента (будет работать и с Observable, но только после завершения потока).
Реактивный способ обработки ошибок retryWhen
Подобным образом можно обрабатывать и ошибки. Предположим, у пользователя пропала сеть, что приведет к ошибке и вызову onError() внутри Single, который возвращается методом getUser().
В этот момент можно показать пользователю диалог с текстом «Проверьте соединение», а по нажатию кнопки OK вызвать метод retry().
@NonNull
private final PublishSubject<Void> retrySubject = PublishSubject.create();
private void load() {
repository
.getSomething()
.doOnError(err -> showConnectionDialog())
.retryWhen(retryHandler -> retryHandler.flatMap(nothing -> retrySubject.asObservable()))
.subscribe(result -> {}, err -> {});
}
private void retry() {
retrySubject.onNext(null);
}
По вызову retrySubject.onNext(null) вся цепочка выше retryWhen() переподпишется к источнику getUser(), и повторит запрос.
При таком подходе важно помнить, что doOnError() должен находиться выше в цепочке, чем retryWhen(), поскольку последний «поглощает» ошибки до эмита repeatHandler'а.
В данном конкретном случае выигрыша по производительности не будет, а кода стало даже чуть больше, но эти примеры помогут начать мыслить реактивными паттернами.
В следующем, бессовестно притянутом за уши, примере, в методе load() мы объединяем два источника оператором combineLatest.
Первый источник — repository.getSomething() загружает что-то из сети, второй, localStorage.fetchSomethingReallyHuge(), загружает что-то тяжелое из локального хранилища.
public void load() {
Observable.combineLatest(repository.getSomething(),
localStorage.fetchSomethingReallyHuge(),
(something, hugeObject) -> new Stuff(something, hugeObject))
.subscribe(stuff -> {}, err -> {});
}
При обработке ошибки императивным способом, вызывая load() на каждую ошибку, мы будем заново подписываться на оба источника, что, в данном примере, абсолютно ненужно. При сетевой ошибке, второй источник успешно заэмитит данные, ошибка произойдет только в первом. В этом случае императивный способ будет еще и медленней.
Посмотрим, как будет выглядеть реактивный способ.
public void load() {
Observable.combineLatest(
repository.getSomething()
.retryWhen(retryHandler ->
retryHandler.flatMap(
err -> retrySubject.asObservable())),
localStorage.fetchSomethingReallyHuge()
.retryWhen(retryHandler ->
retryHandler.flatMap(
nothing -> retrySubject.asObservable())),
(something, hugeObject) -> new Stuff(something, hugeObject))
.subscribe(stuff -> {}, err -> {});
}
Прелесть такого подхода в том, что лямбда, переданная в оператор retryWhen() исполняется только после ошибки внутри источника, соответственно, если «ошибется» только один из источников, то и переподписка произойдет только на него, а оставшаяся цепочка ниже будет ожидать переисполнения.
А если ошибка произойдет внутри обоих источников, то один и тот же retryHandler сработает в двух местах.
Делегирование обработки ошибок
Следующим шагом можно делегировать обработку повторов некоему RetryManager. Перед этим еще можно немного подготовиться к переезду на Rx2 и убрать из наших потоков null объекты, которые запрещены в Rx2. Для этого можно создать класс:
public class RetryEvent {
}
Без ничего. Позже туда можно будет добавлять разные флаги, но это другая история. Интерфейс RetryManager может выглядеть как-то так:
interface RetryManager {
Observable<RetryEvent> observeRetries(@NonNull Throwable error);
}
Реализация может проверять ошибки, показывать диалоги, снэкбар, устанавливать бесшумный таймаут — всё, что душе угодно. И слушать коллбэки от всех этих UI компонентов, чтобы в последствии заэмитить RetryEvent в наш retryHandler.
Предыдущий пример с использованием такого RetryManager будет выглядеть вот так:
//pass this through constructor, DI or use singleton (but please don't)
private final RetryManager retryManager;
public void load() {
Observable.combineLatest(
repository.getSomething()
.retryWhen(retryHandler ->
retryHandler.flatMap(
err -> retryManager.observeRetries())),
localStorage.fetchSomethingReallyHuge()
.retryWhen(retryHandler ->
retryHandler.flatMap(
nothing -> retryManager.observeRetries())),
(something, hugeObject) -> new Stuff(something, hugeObject))
.subscribe(stuff -> {}, err -> {});
}
Таким нехитрым образом обработка повторов при ошибках делегирована сторонней сущности, которую можно передавать как зависимость.
Надеюсь, эти примеры окажутся кому-то полезны и соблазнят попробовать repeatWhen() и retryWhen() в своих проектах.
Автор: yatsinar
