Reactor-core: How to conditionally repeat/retry with backoff based on emitted elements and retry counter?

[asarkar]

Looking at the FluxRepeatWhenTest, there are tests for repeat with backoff, but I'm looking for a use case to stop resubscribing when a condition is met. Since repeatWhen/retryWhen get invoked on completion and error respectively, there's no way to check the values of the emitted elements and repeat/retry counter, and say, "I've tried maxRetries times, I give up", or "I've tried twice and it works, no need to subscribe anymore". This can be easily done using a loop in the generator but not when there is a need to introduce an exponential delay based on the repeat/retry count.
I hope the question makes sense.

[simonbasle]

You cannot react to the data (onNext) since retryWhen is triggered by onError. Note that for simpler use case like "I've tried maxRetries" you should definitely use retry(n).

However for your particular use case you don't seem to need the actual content, but rather the index of each retry?

In retryWhen, the companion Flux<Throwable> that you get will be notified of every error, including those which occur on a retry attempt, as a onNext(throwable) in the companion flux.

So what you can do is zip that Flux<Throwable> with a range to get the attempt's index, using a zip function to either emit the index (trigger a retry) or throw an error (complete cycle with onError).
Associated with a flatMap, you can then introduce a delay that depends on the retry index (backing off):

Flux<String> flux =
    source
    .retryWhen(companion -> companion
        .zipWith(Flux.range(1, 4), (error, index) -> { // index up to 4 retries
          if (index < 4) return index; //limit to 3 actual attempts
          else throw Exceptions.propagate(error); //terminate the source with the 4th `onError`
        })
        .flatMap(index -> Mono.delayMillis(index * 100)) // each retry has growing delay
    );

Alternative if you prefer is to do the "emit or error" choice from inside the flatMap using Mono.delay (emit) vs Mono.error (propagate error and terminate retry cycles).

I've added the above to the FAQ in the reference guide currently being written.

[asarkar]

@smaldini I do need to look at the element values. The workaround is instead of trying to do that in the retry logic, I create a custom generator using Mono.create and emit an empty success when the condition is met. This doesn't do what I wanted, which is to keep the value in the stream but at least I avoid retrying.
Perhaps I should rephrase the question as "retry when the emitted value is not x or when the max retries is not exceeded or the error matches certain conditions". The zipWith solves the max retries and error matching, but not when value is not x use case.
Does this make sense?

[simonbasle]

@asarkar that simply isn't in line with retry's semantics (react to the terminal event that is onError)... retryWhen may seem more general purpose, but it still adheres to these semantics by pushing these errors to you in the companion Flux<Throwable>.

Your usecase is still unclear. You seem to imply that you want to retry as soon as the value "x" stops matching a predicate, but still include that "x" value in the output Flux<T>?

Could you maybe rephrase again in terms of example inputs and outputs? Like 2 cases:

• a source sequence that don't trigger a retry, and the output
• a source sequence that triggers a retry on the element condition, and the output

It's still a bit unclear but I'm wondering if you could use

1. takeUntil(predicate) (forwards elements until one matches the predicate, including that last element in the output)
2. combined with .concatWith(Mono.error(myRetryError)) to trigger a retry after that.

The only way this would complete is by having only elements that don't match the take predicate