How do I retry an Flux invoker?

user_1281389 0
Viewed 127

In case of resilience i am trying to simulate a database breakdown.

  public void createParticipationCheckWorkflowsForThis(Integer numberOfAdvices) {
    participationAdviceSource.getParticipationAdvicesByAMaximumLimitOf(numberOfAdvices) // i want to retry this
        .subscribe(participationAdviceSender::sendAdviceToWorkflowEngine);
  }

My test scenario specifies that on the first database call, which is using Spring-R2DBC i return a flux error and on its second call a correct result.

when(participationAdviceSource.getParticipationAdvicesByAMaximumLimitOf(anyInt()))
    .thenReturn(Flux.error(new RuntimeException()))
    .thenReturn(Flux.just(ParticipationAdvice.builder().participationId(1L).build()));

My specific question is how can i retry the invoker/producer, because the retry mechanism of reactive retries the subscribe not the producer

project-reactor spring-webflux
1 Answers

You can use retry or retryWhen along with the Retry API:

 Flux.defer(() -> participationAdviceSource.getParticipationAdvicesByAMaximumLimitOf(numberOfAdvices))
    .retryWhen(Retry.backoff(retryMaxAttempts, Duration.ofMillis(retryMinBackoff)).maxBackoff(Duration.ofMillis(retryMaxBackoff)))

A Flux.defer wrapper is required because retry and retryWhen work by re-subscribing to the Flux (if the Flux throws an error and the retry conditions are fulfilled).

user_89092350
Related