RetryHandlerBuilder { public Func1<Observable<? extends Throwable>, Observable<?>> buildExponentialBackoff() { Preconditions.checkState(retryCount > 0, "Retry count not defined"); Preconditions.checkState(retryDelayMs > 0, "Retry delay not defined"); return failedAttempts -> failedAttempts .doOnNext(error -> onErrorHook.call(error)) .zipWith(Observable.range(0, retryCount + 1), RetryItem::new) .flatMap(retryItem -> { if (retryWhenCondition != null && !retryWhenCondition.get()) { String errorMessage = String.format( "Retry condition not met for %s. Last error: %s. Returning an error to the caller", title, retryItem.cause.getMessage() ); return Observable.error(new IOException(errorMessage, retryItem.cause)); } if (retryOnThrowableCondition != null && !retryOnThrowableCondition.test(retryItem.cause)) { String errorMessage = String.format( "Retry condition for the last error not met for %s. Last error: %s. Returning an error to the caller", title, retryItem.cause.getMessage() ); return Observable.error(new IOException(errorMessage, retryItem.cause)); } if (retryItem.retry == retryCount) { String errorMessage = String.format( "Retry limit reached for %s. Last error: %s. Returning an error to the caller", title, retryItem.cause.getMessage() ); return Observable.error(new IOException(errorMessage, retryItem.cause)); } long expDelay = buildDelay(retryItem.retry); if (retryItem.cause instanceof TimeoutException) { logger.info("Delaying timed-out {} retry by {}[ms]", title, expDelay); } else { logger.info("Delaying failed {} retry by {}[ms]: {}", title, expDelay, ExceptionExt.toMessageChain(retryItem.cause)); logger.debug("Exception", retryItem.cause); } return Observable.timer(expDelay, TimeUnit.MILLISECONDS, scheduler); }); } private  RetryHandlerBuilder(); RetryHandlerBuilder withScheduler(Scheduler scheduler); RetryHandlerBuilder withReactorScheduler(reactor.core.scheduler.Scheduler scheduler); RetryHandlerBuilder withRetryCount(int retryCount); RetryHandlerBuilder withUnlimitedRetries(); RetryHandlerBuilder withRetryDelay(long retryDelay, TimeUnit timeUnit); RetryHandlerBuilder withMaxDelay(long maxDelay, TimeUnit timeUnit); RetryHandlerBuilder withDelay(long initial, long max, TimeUnit timeUnit); RetryHandlerBuilder withTitle(String title); RetryHandlerBuilder withOnErrorHook(Consumer<Throwable> hook); RetryHandlerBuilder withRetryWhen(Supplier<Boolean> retryWhenCondition); RetryHandlerBuilder withRetryOnThrowable(Predicate<Throwable> retryOnThrowable); RetryHandlerBuilder but(); Func1<Observable<? extends Throwable>, Observable<?>> buildExponentialBackoff(); Function<Flux<Throwable>, Publisher<?>> buildReactorExponentialBackoff(); static RetryHandlerBuilder retryHandler();  }
@Test public void testRetryOnError() throws Exception { Func1<Observable<? extends Throwable>, Observable<?>> retryFun = builder.buildExponentialBackoff(); observableOf("A", new IOException("Error1"), "B", new IOException("Error2"), "C") .retryWhen(retryFun, testScheduler) .subscribe(testSubscriber); testScheduler.triggerActions(); assertThat(testSubscriber.getLatestItem()).isEqualTo("A"); testScheduler.advanceTimeBy(RETRY_DELAY_SEC, TimeUnit.SECONDS); assertThat(testSubscriber.getLatestItem()).isEqualTo("B"); testScheduler.advanceTimeBy(2 * RETRY_DELAY_SEC, TimeUnit.SECONDS); assertThat(testSubscriber.getLatestItem()).isEqualTo("C"); assertThat(testSubscriber.isUnsubscribed()); }