rxjava :Puis-je utiliser retry() mais avec un délai ?

Question

J'utilise rxjava dans mon application Android pour gérer les requêtes réseau de manière asynchrone.Maintenant, je voudrais réessayer une requête réseau ayant échoué seulement après un certain temps.

Existe-t-il un moyen d'utiliser retry() sur un observable mais de réessayer seulement après un certain délai ?

Existe-t-il un moyen de faire savoir à l'Observable qu'il est en cours de nouvelle tentative (par opposition à la première tentative) ?

J'ai jeté un œil à debounce()/throttleWithTimeout() mais ils semblent faire quelque chose de différent.

Modifier:

Je pense avoir trouvé une façon de le faire, mais je serais intéressé soit par la confirmation que c'est la bonne façon de le faire, soit par d'autres, meilleures façons.

Ce que je fais c'est ceci :Dans la méthode call() de mon Observable.OnSubscribe, avant d'appeler la méthode Subscribers onError(), je laisse simplement le Thread dormir pendant la durée souhaitée.Donc, pour réessayer toutes les 1000 millisecondes, je fais quelque chose comme ceci :

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

Étant donné que cette méthode s’exécute sur un thread IO, elle ne bloque pas l’interface utilisateur.Le seul problème que je peux voir est que même la première erreur est signalée avec un délai, donc le délai est là même s'il n'y a pas de retry().J'aimerais mieux que le délai ne soit pas appliqué après une erreur mais à la place avant une nouvelle tentative (mais pas avant le premier essai, évidemment).

Answer 1

Vous pouvez utiliser le retryWhen() opérateur pour ajouter une logique de nouvelle tentative à n’importe quel observable.

La classe suivante contient la logique de nouvelle tentative :

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

Usage:

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

Answer 2

Inspiré par La réponse de Paul, et si cela ne vous concerne pas retryWhen problèmes évoqués par Abhijit Sarkar, le moyen le plus simple de retarder la réabonnement avec rxJava2 sans condition est :

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

Vous souhaiterez peut-être voir plus d'échantillons et d'explications sur réessayerQuand et répéterQuand.

Answer 3

Cet exemple fonctionne avec JXJAVA 2.2.2:

Réessayer sans délai:

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

Réessayer avec délai:

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."));

Notre source unique échoue si Someconnection.Send () échoue. Lorsque cela se produit, l'observable des défaillances à l'intérieur de RetryWhen émet l'erreur. Nous retardons cette émission de 300 ms et renvoyons-le à signaler une nouvelle tentative. Prendre (5) garantit que notre signalisation observable se terminera après avoir reçu cinq erreurs. RetryWhen voit la résiliation et ne réessaie pas après la cinquième échec.

Answer 4

Il s'agit d'une solution basée sur les extraits de Ben Christensen que j'ai vus, Exemple de réessayer quand, et RéessayerQuandTestsConditionnel (j'ai dû changer n.getThrowable() à n pour que ça marche).j'ai utilisé evant/gradle-retrolambda pour que la notation lambda fonctionne sur Android, mais vous n'êtes pas obligé d'utiliser des lambdas (bien que ce soit fortement recommandé).Pour le délai, j'ai implémenté un recul exponentiel, mais vous pouvez y brancher la logique de recul que vous souhaitez.Pour être complet, j'ai ajouté le subscribeOn et observeOn les opérateurs.j'utilise RéactifX/RxAndroid pour le 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);

Answer 5

au lieu d'utiliser myreQuestobservServable.Retry, j'utilise une fonction d'enveloppement réyobservable (myrequestobservatible, rétrycount, secondes) qui renvoient une nouvelle observable qui gère l'indirection du retard afin que je puisse faire

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

                }
            });

        }

    });

}

Answer 6

retryWhen est un opérateur compliqué, peut-être même bogué.Le document officiel et au moins une réponse ici utilisent range opérateur, qui échouera s’il n’y a aucune tentative à effectuer.Voir mon discussion avec David Karnok, membre de ReactiveX.

J'ai amélioré la réponse de Kjones en changeant flatMap à concatMap et en ajoutant un RetryDelayStrategy classe. flatMap ne préserve pas l'ordre d'émission pendant que concatMap le fait, ce qui est important pour les retards avec back-off.Le RetryDelayStrategy, comme son nom l'indique, permet à l'utilisateur de choisir parmi différents modes de génération de délais de nouvelle tentative, y compris l'attente.Le code est disponible sur mon GitHub compléter avec les cas de tests suivants :

1. Réussit à la 1ère tentative (aucune nouvelle tentative)
2. Échec après 1 nouvelle tentative
3. Tente de réessayer 3 fois mais réussit la 2ème et ne réessaye donc pas la 3ème fois
4. Réussi à la 3ème tentative

Voir setRandomJokes méthode.

Answer 7

Maintenant avec Rxjava Version 1.0+ Vous pouvez utiliser ZipWith pour réessayer avec Delay.

Ajout de modifications à KJONES Réponse.

modifié

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

Answer 8

Même réponse que de kjones mais mis à jour pour la dernière version Pour rxjava 2.x version: ('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);
            }
        });
    }
}

Utilisation:

// Ajoutez une nouvelle logique à l'observable existante. // réessaye max de 3 fois avec un délai de 2 secondes.

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

Answer 9

Vous pouvez ajouter un délai dans l'observable renvoyé dans l'opérateur 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);
}

Vous pouvez voir plus d'exemples ici. https://github.com/politrons/reactive

Answer 10

Basé sur kjones la réponse ici est la version Kotlin de la nouvelle tentative de RxJava 2.x avec un délai en tant qu'extension.Remplacer Observable pour créer la même extension pour 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)
            }
        }
    }
}

Ensuite, utilisez-le simplement sur observable observable.retryWithDelay(3, 1000)

Answer 11

Faites-le simplement comme ceci:

                  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) {

                                }
                            });

Answer 12

pour la version KOTLIN & 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
                    }
                })
    }
}

Answer 13

(kotlin) I Little Bit Amélioration du code avec des backtes exponentielles et une défense appliquée à l'émission d'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") }
}

Answer 14

Dans le cas où vous devez imprimer le décompte des tentatives, Vous pouvez utiliser l'exemple fourni dans la page Wiki de Rxjava https://github.com/REACTIVEX / RXJAVA / WIKI / TRAITEMENTS D'ERREUR - Opérateurs

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