Экспоненциальный откат в RxJava

В вашем коде есть две ошибки:

1. Чтобы повторить наблюдаемую последовательность, она должна быть конечной. Т.е. вместо interval вам лучше использовать что-то вроде just или fromCallable, как я сделал в примере ниже.
2. Из внутренней функции repeatWhen вам нужно вернуть новый отсроченный наблюдаемый источник, поэтому вместо observable.delay() вы должны вернуть Observable.timer().

Рабочий код:

public void testRepeat() throws InterruptedException {
    logger.info("test start");

    int DEFAULT_DELAY = 100; // ms
    int ADDITIONAL_DELAY = 100; // ms
    AtomicInteger generator = new AtomicInteger(0);
    AtomicBoolean connectionAlive = new AtomicBoolean(true); // initially alive

    Disposable subscription = Observable.fromCallable(generator::incrementAndGet)
            .repeatWhen(counts -> {
                AtomicInteger retryCounter = new AtomicInteger(0);
                return counts.flatMap(c -> {
                    int retry = 0;
                    if (connectionAlive.get()) {
                        retryCounter.set(0); // reset counter
                    } else {
                        retry = retryCounter.incrementAndGet();
                    }
                    int additionalDelay = ADDITIONAL_DELAY * (int) Math.pow(retry, 2);
                    logger.info("retry={}, additionalDelay={}ms", retry, additionalDelay);
                    return Observable.timer(DEFAULT_DELAY + additionalDelay, TimeUnit.MILLISECONDS);
                });
            })
            .subscribe(v -> logger.info("got {}", v));

    Thread.sleep(220);
    logger.info("connection dropped");
    connectionAlive.set(false);
    Thread.sleep(2000);
    logger.info("connection is back alive");
    connectionAlive.set(true);
    Thread.sleep(2000);
    subscription.dispose();
    logger.info("test complete");
}

См. Подробную статью о repeatWhen здесь.

Я всегда считал retryWhen несколько низкоуровневым, поэтому для экспоненциального отката я использую конструктор (например, Abhijit), который протестирован и доступен для RxJava 1.x по адресу rxjava-extras. Я бы предложил использовать версию с ограничением, чтобы экспоненциальное увеличение задержки не выходило за пределы максимального значения, которое вы определяете.

Вот как вы его используете:

observable.retryWhen(
    RetryWhen.exponentialBackoff(
        delay, maxDelay, TimeUNIT.SECONDS)
    .build());

Я не согласен с тем, что retryWhen содержит ошибки, но если вы обнаружите ошибку, сообщите об этом в RxJava. Ошибки исправляются быстро!

Вам понадобится rxjava-extras 0.8.0.6 или более поздняя версия, которая находится на Maven Central:

<dependency>
    <groupId>com.github.davidmoten</groupId>
    <artifactId>rxjava-extras</artifactId>
    <version>0.8.0.6</version>
</dependency>

Сообщите мне, нужна ли вам версия RxJava 2.x. Те же функции доступны в rxjava2-extras начиная с версии 0.1.4.

Вы можете использовать оператор retryWhen для настройки задержки при отсутствии соединения. Как периодически выдавать элементы - отдельная тема (поищите операторы interval или timer). Если не можете разобраться, задайте отдельный вопрос.

В моем Github есть обширный пример, но здесь я расскажу вам суть.

RetryWithDelay retryWithDelay = RetryWithDelay.builder()
    .retryDelayStrategy(RetryDelayStrategy.RETRY_COUNT)
    .build()

Single.fromCallable(() -> {
    ...
}).retryWhen(retryWithDelay)
.subscribe(j -> {
    ...
})

RetryWithDelay определяется следующим образом. Я использовал RxJava 2.x, поэтому, если вы используете 1.x, подпись должна быть Func1<Observable<? extends Throwable>, Observable<Object>>.

public class RetryWithDelay implements
        Function<Flowable<? extends Throwable>, Publisher<Object>> {
    ...
}

RetryWithDelay класс.

RetryStrategy перечисление.

Это позволяет мне настраивать различные виды тайм-аутов, постоянные, линейные, экспоненциальные, на основе RetryDelayStrategy. Для вашего варианта использования вы должны выбрать CONSTANT_DELAY_TIMES_RETRY_COUNT стратегию задержки и вызвать retryDelaySeconds(2) при создании RetryWithDelay.

retryWhen - сложный, возможно, даже глючный оператор. В большинстве примеров, которые вы найдете в Интернете, используется оператор range, который завершится ошибкой, если не будет предпринято никаких повторных попыток. Подробнее см. Мой ответ здесь.