@Test public void onRetryWithMono() { Semaphore semaphore = new Semaphore(0); Retry<?> retry = RetryWithAsyncCallback.any() .retryOnce() .fixedBackoff(Duration.ofMillis(500)) .onRetryWithMono(context -> Mono.fromCallable(() -> { semaphore.release(); return 0; })); StepVerifier.withVirtualTime(() -> Flux.range(0, 2).concatWith(Mono.error(new SocketException())).retryWhen(retry)) .expectNext(0, 1) .then(semaphore::acquireUninterruptibly) .expectNoEvent(Duration.ofMillis(400)) .thenAwait(Duration.ofMillis(200)) .expectNext(0, 1) .verifyErrorMatches(e -> isRetryExhausted(e, SocketException.class)); StepVerifier.withVirtualTime(() -> Mono.error(new SocketException()).retryWhen(retry.noBackoff())) .then(semaphore::acquireUninterruptibly) .verifyErrorMatches(e -> isRetryExhausted(e, SocketException.class)); }
public RetryWithAsyncCallback<T> onRetryWithMono(Function<? super RetryContext<T>, Mono<?>> onRetryMono) { return new RetryWithAsyncCallback<>(retryPredicate, maxIterations, timeout, backoff, jitter, backoffScheduler, onRetry, onRetryMono, applicationContext); }
RetryWithAsyncCallback extends AbstractRetry<T, Throwable> implements Retry<T> { public RetryWithAsyncCallback<T> onRetryWithMono(Function<? super RetryContext<T>, Mono<?>> onRetryMono) { return new RetryWithAsyncCallback<>(retryPredicate, maxIterations, timeout, backoff, jitter, backoffScheduler, onRetry, onRetryMono, applicationContext); } }
RetryWithAsyncCallback extends AbstractRetry<T, Throwable> implements Retry<T> { public RetryWithAsyncCallback<T> onRetryWithMono(Function<? super RetryContext<T>, Mono<?>> onRetryMono) { return new RetryWithAsyncCallback<>(retryPredicate, maxIterations, timeout, backoff, jitter, backoffScheduler, onRetry, onRetryMono, applicationContext); } RetryWithAsyncCallback(Predicate<? super RetryContext<T>> retryPredicate,
                           long maxIterations,
                           Duration timeout,
                           Backoff backoff,
                           Jitter jitter,
                           Scheduler backoffScheduler,
                           final Consumer<? super RetryContext<T>> onRetry,
                           Function<? super RetryContext<T>, Mono<?>> onRetryMono,
                           T applicationContext); }
RetryWithAsyncCallback extends AbstractRetry<T, Throwable> implements Retry<T> { public RetryWithAsyncCallback<T> onRetryWithMono(Function<? super RetryContext<T>, Mono<?>> onRetryMono) { return new RetryWithAsyncCallback<>(retryPredicate, maxIterations, timeout, backoff, jitter, backoffScheduler, onRetry, onRetryMono, applicationContext); } RetryWithAsyncCallback(Predicate<? super RetryContext<T>> retryPredicate,
                           long maxIterations,
                           Duration timeout,
                           Backoff backoff,
                           Jitter jitter,
                           Scheduler backoffScheduler,
                           final Consumer<? super RetryContext<T>> onRetry,
                           Function<? super RetryContext<T>, Mono<?>> onRetryMono,
                           T applicationContext); static RetryWithAsyncCallback<T> any(); @SafeVarargs static RetryWithAsyncCallback<T> anyOf(Class<? extends Throwable>... retriableExceptions); @SafeVarargs static RetryWithAsyncCallback<T> allBut(final Class<? extends Throwable>... nonRetriableExceptions); static RetryWithAsyncCallback<T> onlyIf(Predicate<? super RetryContext<T>> predicate); static RetryWithAsyncCallback<T> create(Predicate<? super RetryContext<T>> retryPredicate); @Override RetryWithAsyncCallback<T> fixedBackoff(Duration backoffInterval); @Override RetryWithAsyncCallback<T> noBackoff(); @Override RetryWithAsyncCallback<T> exponentialBackoff(Duration firstBackoff, Duration maxBackoff); @Override RetryWithAsyncCallback<T> exponentialBackoffWithJitter(Duration firstBackoff, Duration maxBackoff); @Override RetryWithAsyncCallback<T> randomBackoff(Duration firstBackoff, Duration maxBackoff); @Override RetryWithAsyncCallback<T> withApplicationContext(T applicationContext); @Override RetryWithAsyncCallback<T> doOnRetry(Consumer<? super RetryContext<T>> onRetry); RetryWithAsyncCallback<T> onRetryWithMono(Function<? super RetryContext<T>, Mono<?>> onRetryMono); @Override RetryWithAsyncCallback<T> retryOnce(); @Override RetryWithAsyncCallback<T> retryMax(long maxIterations); @Override RetryWithAsyncCallback<T> timeout(Duration timeout); @Override RetryWithAsyncCallback<T> backoff(Backoff backoff); @Override RetryWithAsyncCallback<T> jitter(Jitter jitter); @Override RetryWithAsyncCallback<T> withBackoffScheduler(Scheduler scheduler); @Override Publisher<Long> apply(Flux<Throwable> errors); @Override String toString(); }
RetryWithAsyncCallback extends AbstractRetry<T, Throwable> implements Retry<T> { public RetryWithAsyncCallback<T> onRetryWithMono(Function<? super RetryContext<T>, Mono<?>> onRetryMono) { return new RetryWithAsyncCallback<>(retryPredicate, maxIterations, timeout, backoff, jitter, backoffScheduler, onRetry, onRetryMono, applicationContext); } RetryWithAsyncCallback(Predicate<? super RetryContext<T>> retryPredicate,
                           long maxIterations,
                           Duration timeout,
                           Backoff backoff,
                           Jitter jitter,
                           Scheduler backoffScheduler,
                           final Consumer<? super RetryContext<T>> onRetry,
                           Function<? super RetryContext<T>, Mono<?>> onRetryMono,
                           T applicationContext); static RetryWithAsyncCallback<T> any(); @SafeVarargs static RetryWithAsyncCallback<T> anyOf(Class<? extends Throwable>... retriableExceptions); @SafeVarargs static RetryWithAsyncCallback<T> allBut(final Class<? extends Throwable>... nonRetriableExceptions); static RetryWithAsyncCallback<T> onlyIf(Predicate<? super RetryContext<T>> predicate); static RetryWithAsyncCallback<T> create(Predicate<? super RetryContext<T>> retryPredicate); @Override RetryWithAsyncCallback<T> fixedBackoff(Duration backoffInterval); @Override RetryWithAsyncCallback<T> noBackoff(); @Override RetryWithAsyncCallback<T> exponentialBackoff(Duration firstBackoff, Duration maxBackoff); @Override RetryWithAsyncCallback<T> exponentialBackoffWithJitter(Duration firstBackoff, Duration maxBackoff); @Override RetryWithAsyncCallback<T> randomBackoff(Duration firstBackoff, Duration maxBackoff); @Override RetryWithAsyncCallback<T> withApplicationContext(T applicationContext); @Override RetryWithAsyncCallback<T> doOnRetry(Consumer<? super RetryContext<T>> onRetry); RetryWithAsyncCallback<T> onRetryWithMono(Function<? super RetryContext<T>, Mono<?>> onRetryMono); @Override RetryWithAsyncCallback<T> retryOnce(); @Override RetryWithAsyncCallback<T> retryMax(long maxIterations); @Override RetryWithAsyncCallback<T> timeout(Duration timeout); @Override RetryWithAsyncCallback<T> backoff(Backoff backoff); @Override RetryWithAsyncCallback<T> jitter(Jitter jitter); @Override RetryWithAsyncCallback<T> withBackoffScheduler(Scheduler scheduler); @Override Publisher<Long> apply(Flux<Throwable> errors); @Override String toString();  }