rxjava: Posso usar retry (), mas com atraso?

Estou usando o rxjava em meu aplicativo Android para lidar com solicitações de rede de forma assíncrona. Agora, gostaria de repetir uma solicitação de rede com falha somente após um determinado período de tempo.

Existe alguma maneira de usar retry () em um Observable, mas tentar novamente apenas após um certo atraso?

Existe uma maneira de informar ao observável que está sendo tentado novamente (em vez de tentado pela primeira vez)?

Eu dei uma olhada em debounce () / throttleWithTimeout (), mas eles parecem estar fazendo algo diferente.

Editar:

Acho que encontrei uma maneira de fazer isso, mas estou interessado em confirmar se essa é a maneira correta de fazer isso ou em outras maneiras melhores.

O que estou fazendo é o seguinte: no método call () do meu Observable.OnSubscribe, antes de chamar o método Subscribers onError (), simplesmente deixo o Thread dormir pelo tempo desejado. Então, para tentar novamente a cada 1000 milissegundos, eu faço algo assim:

@Override
public void call(Subscriber<? super List<ProductNode>> subscriber) {
    try {
        Log.d(TAG, "trying to load all products with pid: " + pid);
        subscriber.onNext(productClient.getProductNodesForParentId(pid));
        subscriber.onCompleted();
    } catch (Exception e) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e1) {
            e.printStackTrace();
        }
        subscriber.onError(e);
    }
}

Como esse método está sendo executado em um thread de IO, ele não bloqueia a IU. O único problema que vejo é que até o primeiro erro é relatado com atraso, então o atraso existe mesmo se não houver nova tentativa (). Eu gostaria melhor se o atraso não fosse aplicado após um erro, mas antes de uma nova tentativa (mas não antes da primeira tentativa, obviamente).

Você pode usar o retryWhen()operador para adicionar lógica de repetição a qualquer observável.

A classe a seguir contém a lógica de repetição:

RxJava 2.x

public class RetryWithDelay implements Function<Observable<? extends Throwable>, Observable<?>> {
    private final int maxRetries;
    private final int retryDelayMillis;
    private int retryCount;

    public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
        this.maxRetries = maxRetries;
        this.retryDelayMillis = retryDelayMillis;
        this.retryCount = 0;
    }

    @Override
    public Observable<?> apply(final Observable<? extends Throwable> attempts) {
        return attempts
                .flatMap(new Function<Throwable, Observable<?>>() {
                    @Override
                    public Observable<?> apply(final Throwable throwable) {
                        if (++retryCount < maxRetries) {
                            // When this Observable calls onNext, the original
                            // Observable will be retried (i.e. re-subscribed).
                            return Observable.timer(retryDelayMillis,
                                    TimeUnit.MILLISECONDS);
                        }

                        // Max retries hit. Just pass the error along.
                        return Observable.error(throwable);
                    }
                });
    }
}

RxJava 1.x

public class RetryWithDelay implements
        Func1<Observable<? extends Throwable>, Observable<?>> {

    private final int maxRetries;
    private final int retryDelayMillis;
    private int retryCount;

    public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
        this.maxRetries = maxRetries;
        this.retryDelayMillis = retryDelayMillis;
        this.retryCount = 0;
    }

    @Override
    public Observable<?> call(Observable<? extends Throwable> attempts) {
        return attempts
                .flatMap(new Func1<Throwable, Observable<?>>() {
                    @Override
                    public Observable<?> call(Throwable throwable) {
                        if (++retryCount < maxRetries) {
                            // When this Observable calls onNext, the original
                            // Observable will be retried (i.e. re-subscribed).
                            return Observable.timer(retryDelayMillis,
                                    TimeUnit.MILLISECONDS);
                        }

                        // Max retries hit. Just pass the error along.
                        return Observable.error(throwable);
                    }
                });
    }
}

Uso:

// Add retry logic to existing observable.
// Retry max of 3 times with a delay of 2 seconds.
observable
    .retryWhen(new RetryWithDelay(3, 2000));

Inspirado pela resposta de Paul , e se você não está preocupado com os retryWhenproblemas declarados por Abhijit Sarkar , a maneira mais simples de atrasar a nova assinatura com rxJava2 incondicionalmente é:

source.retryWhen(throwables -> throwables.delay(1, TimeUnit.SECONDS))

Você pode querer ver mais exemplos e explicações sobre retryWhen e repeatWhen .

Este exemplo funciona com jxjava 2.2.2:

Tentar novamente sem demora:

Single.just(somePaylodData)
   .map(data -> someConnection.send(data))
   .retry(5)
   .doOnSuccess(status -> log.info("Yay! {}", status);

Tentar novamente com atraso:

Single.just(somePaylodData)
   .map(data -> someConnection.send(data))
   .retryWhen((Flowable<Throwable> f) -> f.take(5).delay(300, TimeUnit.MILLISECONDS))
   .doOnSuccess(status -> log.info("Yay! {}", status)
   .doOnError((Throwable error) 
                -> log.error("I tried five times with a 300ms break" 
                             + " delay in between. But it was in vain."));

Nossa fonte falha se someConnection.send () falhar. Quando isso acontece, o observável de falhas dentro de retryWhen emite o erro. Atrasamos essa emissão em 300 ms e a enviamos de volta para sinalizar uma nova tentativa. take (5) garante que nossa sinalização observável terminará após recebermos cinco erros. retryWhen vê o encerramento e não tenta novamente após a quinta falha.

Esta é uma solução baseada nos trechos de Ben Christensen que vi, RetryWhen Example e RetryWhenTestsConditional (tive que mudar n.getThrowable()para npara que funcionasse). Usei evant / gradle-retrolambda para fazer a notação lambda funcionar no Android, mas você não precisa usar lambdas (embora seja altamente recomendado). Para o atraso, implementei o backoff exponencial, mas você pode inserir qualquer lógica de backoff que quiser. Para completar, adicionei os operadores subscribeOne observeOn. Estou usando ReactiveX / RxAndroid para o AndroidSchedulers.mainThread().

int ATTEMPT_COUNT = 10;

public class Tuple<X, Y> {
    public final X x;
    public final Y y;

    public Tuple(X x, Y y) {
        this.x = x;
        this.y = y;
    }
}


observable
    .subscribeOn(Schedulers.io())
    .retryWhen(
            attempts -> {
                return attempts.zipWith(Observable.range(1, ATTEMPT_COUNT + 1), (n, i) -> new Tuple<Throwable, Integer>(n, i))
                .flatMap(
                        ni -> {
                            if (ni.y > ATTEMPT_COUNT)
                                return Observable.error(ni.x);
                            return Observable.timer((long) Math.pow(2, ni.y), TimeUnit.SECONDS);
                        });
            })
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(subscriber);

em vez de usar MyRequestObservable.retry, uso uma função de invólucro retryObservable (MyRequestObservable, retrycount, segundos) que retorna um novo Observable que trata da indireção para o atraso para que eu possa fazer

retryObservable(restApi.getObservableStuff(), 3, 30)
    .subscribe(new Action1<BonusIndividualList>(){
        @Override
        public void call(BonusIndividualList arg0) 
        {
            //success!
        }
    }, 
    new Action1<Throwable>(){
        @Override
        public void call(Throwable arg0) { 
           // failed after the 3 retries !
        }}); 


// wrapper code
private static <T> Observable<T> retryObservable(
        final Observable<T> requestObservable, final int nbRetry,
        final long seconds) {

    return Observable.create(new Observable.OnSubscribe<T>() {

        @Override
        public void call(final Subscriber<? super T> subscriber) {
            requestObservable.subscribe(new Action1<T>() {

                @Override
                public void call(T arg0) {
                    subscriber.onNext(arg0);
                    subscriber.onCompleted();
                }
            },

            new Action1<Throwable>() {
                @Override
                public void call(Throwable error) {

                    if (nbRetry > 0) {
                        Observable.just(requestObservable)
                                .delay(seconds, TimeUnit.SECONDS)
                                .observeOn(mainThread())
                                .subscribe(new Action1<Observable<T>>(){
                                    @Override
                                    public void call(Observable<T> observable){
                                        retryObservable(observable,
                                                nbRetry - 1, seconds)
                                                .subscribe(subscriber);
                                    }
                                });
                    } else {
                        // still fail after retries
                        subscriber.onError(error);
                    }

                }
            });

        }

    });

}

retryWhené um operador complicado, talvez até com erros. O doc oficial e pelo menos uma resposta aqui usam rangeoperador, que irá falhar se não houver novas tentativas a serem feitas. Veja minha discussão com o membro do ReactiveX David Karnok.

Melhorei a resposta de kjones mudando flatMappara concatMape adicionando uma RetryDelayStrategyclasse. flatMapnão preserva a ordem de emissão enquanto o concatMapfaz, o que é importante para atrasos com retirada. O RetryDelayStrategy, como o nome indica, permite que o usuário escolha entre vários modos de geração de atrasos na nova tentativa, incluindo back-off. O código está disponível em meu GitHub completo com os seguintes casos de teste:

1. Sucesso na 1ª tentativa (sem novas tentativas)
2. Falha após 1 tentativa
3. Tenta repetir 3 vezes, mas consegue na 2ª, portanto, não tenta novamente na 3ª vez
4. Sucesso na 3ª tentativa

Veja o setRandomJokesmétodo.

Agora, com o RxJava versão 1.0+, você pode usar zipWith para obter uma nova tentativa com atraso.

Adicionando modificações à resposta do kjones .

Modificado

public class RetryWithDelay implements 
                            Func1<Observable<? extends Throwable>, Observable<?>> {

    private final int MAX_RETRIES;
    private final int DELAY_DURATION;
    private final int START_RETRY;

    /**
     * Provide number of retries and seconds to be delayed between retry.
     *
     * @param maxRetries             Number of retries.
     * @param delayDurationInSeconds Seconds to be delays in each retry.
     */
    public RetryWithDelay(int maxRetries, int delayDurationInSeconds) {
        MAX_RETRIES = maxRetries;
        DELAY_DURATION = delayDurationInSeconds;
        START_RETRY = 1;
    }

    @Override
    public Observable<?> call(Observable<? extends Throwable> observable) {
        return observable
                .delay(DELAY_DURATION, TimeUnit.SECONDS)
                .zipWith(Observable.range(START_RETRY, MAX_RETRIES), 
                         new Func2<Throwable, Integer, Integer>() {
                             @Override
                             public Integer call(Throwable throwable, Integer attempt) {
                                  return attempt;
                             }
                         });
    }
}

Mesma resposta de kjones, mas atualizado para a versão mais recente Para a versão RxJava 2.x : ('io.reactivex.rxjava2: rxjava: 2.1.3')

public class RetryWithDelay implements Function<Flowable<Throwable>, Publisher<?>> {

    private final int maxRetries;
    private final long retryDelayMillis;
    private int retryCount;

    public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
        this.maxRetries = maxRetries;
        this.retryDelayMillis = retryDelayMillis;
        this.retryCount = 0;
    }

    @Override
    public Publisher<?> apply(Flowable<Throwable> throwableFlowable) throws Exception {
        return throwableFlowable.flatMap(new Function<Throwable, Publisher<?>>() {
            @Override
            public Publisher<?> apply(Throwable throwable) throws Exception {
                if (++retryCount < maxRetries) {
                    // When this Observable calls onNext, the original
                    // Observable will be retried (i.e. re-subscribed).
                    return Flowable.timer(retryDelayMillis,
                            TimeUnit.MILLISECONDS);
                }

                // Max retries hit. Just pass the error along.
                return Flowable.error(throwable);
            }
        });
    }
}

Uso:

// Adicionar lógica de repetição ao observável existente. // Tente novamente no máximo 3 vezes com um atraso de 2 segundos.

observable
    .retryWhen(new RetryWithDelay(3, 2000));

Com base na resposta do kjones, aqui é a versão Kotlin do RxJava 2.x, tente novamente com um atraso como uma extensão. Substitua Observablepara criar a mesma extensão para Flowable.

fun <T> Observable<T>.retryWithDelay(maxRetries: Int, retryDelayMillis: Int): Observable<T> {
    var retryCount = 0

    return retryWhen { thObservable ->
        thObservable.flatMap { throwable ->
            if (++retryCount < maxRetries) {
                Observable.timer(retryDelayMillis.toLong(), TimeUnit.MILLISECONDS)
            } else {
                Observable.error(throwable)
            }
        }
    }
}

Em seguida, basta usá-lo em observáveis observable.retryWithDelay(3, 1000)

Você pode adicionar um atraso no Observable retornado no operador retryWhen

          /**
 * Here we can see how onErrorResumeNext works and emit an item in case that an error occur in the pipeline and an exception is propagated
 */
@Test
public void observableOnErrorResumeNext() {
    Subscription subscription = Observable.just(null)
                                          .map(Object::toString)
                                          .doOnError(failure -> System.out.println("Error:" + failure.getCause()))
                                          .retryWhen(errors -> errors.doOnNext(o -> count++)
                                                                     .flatMap(t -> count > 3 ? Observable.error(t) : Observable.just(null).delay(100, TimeUnit.MILLISECONDS)),
                                                     Schedulers.newThread())
                                          .onErrorResumeNext(t -> {
                                              System.out.println("Error after all retries:" + t.getCause());
                                              return Observable.just("I save the world for extinction!");
                                          })
                                          .subscribe(s -> System.out.println(s));
    new TestSubscriber((Observer) subscription).awaitTerminalEvent(500, TimeUnit.MILLISECONDS);
}

Você pode ver mais exemplos aqui. https://github.com/politrons/reactive

Simplesmente faça assim:

                  Observable.just("")
                            .delay(2, TimeUnit.SECONDS) //delay
                            .flatMap(new Func1<String, Observable<File>>() {
                                @Override
                                public Observable<File> call(String s) {
                                    L.from(TAG).d("postAvatar=");

                                    File file = PhotoPickUtil.getTempFile();
                                    if (file.length() <= 0) {
                                        throw new NullPointerException();
                                    }
                                    return Observable.just(file);
                                }
                            })
                            .retry(6)
                            .subscribe(new Action1<File>() {
                                @Override
                                public void call(File file) {
                                    postAvatar(file);
                                }
                            }, new Action1<Throwable>() {
                                @Override
                                public void call(Throwable throwable) {

                                }
                            });

Para a versão Kotlin e RxJava1

class RetryWithDelay(private val MAX_RETRIES: Int, private val DELAY_DURATION_IN_SECONDS: Long)
    : Function1<Observable<out Throwable>, Observable<*>> {

    private val START_RETRY: Int = 1

    override fun invoke(observable: Observable<out Throwable>): Observable<*> {
        return observable.delay(DELAY_DURATION_IN_SECONDS, TimeUnit.SECONDS)
            .zipWith(Observable.range(START_RETRY, MAX_RETRIES),
                object : Function2<Throwable, Int, Int> {
                    override fun invoke(throwable: Throwable, attempt: Int): Int {
                        return attempt
                    }
                })
    }
}

(Kotlin) Eu melhorei um pouco o código com backoff exponencial e defesa aplicada emitindo Observable.range ():

    fun testOnRetryWithDelayExponentialBackoff() {
    val interval = 1
    val maxCount = 3
    val ai = AtomicInteger(1);
    val source = Observable.create<Unit> { emitter ->
        val attempt = ai.getAndIncrement()
        println("Subscribe ${attempt}")
        if (attempt >= maxCount) {
            emitter.onNext(Unit)
            emitter.onComplete()
        }
        emitter.onError(RuntimeException("Test $attempt"))
    }

    // Below implementation of "retryWhen" function, remove all "println()" for real code.
    val sourceWithRetry: Observable<Unit> = source.retryWhen { throwableRx ->
        throwableRx.doOnNext({ println("Error: $it") })
                .zipWith(Observable.range(1, maxCount)
                        .concatMap { Observable.just(it).delay(0, TimeUnit.MILLISECONDS) },
                        BiFunction { t1: Throwable, t2: Int -> t1 to t2 }
                )
                .flatMap { pair ->
                    if (pair.second >= maxCount) {
                        Observable.error(pair.first)
                    } else {
                        val delay = interval * 2F.pow(pair.second)
                        println("retry delay: $delay")
                        Observable.timer(delay.toLong(), TimeUnit.SECONDS)
                    }
                }
    }

    //Code to print the result in terminal.
    sourceWithRetry
            .doOnComplete { println("Complete") }
            .doOnError({ println("Final Error: $it") })
            .blockingForEach { println("$it") }
}

no caso de precisar imprimir a contagem de novas tentativas, você pode usar o exemplo fornecido na página wiki de Rxjava https://github.com/ReactiveX/RxJava/wiki/Error-Handling-Operators

observable.retryWhen(errors ->
    // Count and increment the number of errors.
    errors.map(error -> 1).scan((i, j) -> i + j)  
       .doOnNext(errorCount -> System.out.println(" -> query errors #: " + errorCount))
       // Limit the maximum number of retries.
       .takeWhile(errorCount -> errorCount < retryCounts)   
       // Signal resubscribe event after some delay.
       .flatMapSingle(errorCount -> Single.timer(errorCount, TimeUnit.SECONDS));