Reactor how to repeat some step n times if condition is met

Question

Please help me with reactor I need to check one condition max n times and return the final result after all

I found that reactor has reactor-extra module

https://projectreactor.io/docs/extra/snapshot/api/reactor/retry/Repeat.html

It has construction Repeat.create(java.util.function.Predicate<? super RepeatContext<T>> predicate, long n) Repeat function that repeats n times, only if the predicate returns true.

It looks like right solution, but I can't understand where should be
tha action, which I want to repeat? I have Flux with many actions, but I want to repeat only one

Please make an example of code

Thank you

private int culculateNextResult(some params) {
          // some implementation  
 }



private Boolean compareResults(int prevRes, int nextRes) {
          // some implementation
 }

 public Flux<Boolean> run(some params, Flux<Integer> prevResults){

      return prevResults.map(elem -> compareResults(elem, culculateNextResult(some params)));

 // THIS LOGIC SHOULD BE REPEATED N times if compareResults(elem,       
 // culculateNextResult(some params))) == false, if true, we don't need 
// to repeat 
     }

I want to repeat compareResults(elem, culculateNextResult(some params))) until it's nit true. but n times maximum and return Flux as a result

Phil Clay · Accepted Answer · 2019-03-24T02:06:39.567

Flux.repeat and Mono.repeat will re-subscribe to the source, so every previous step of the source will be repeated with a new subscription.

Since calculateNextResult and compareResults are both synchronous operations in your example, you can use a simple for loop to repeat...

    public Flux<Boolean> run(some params, Flux<Integer> prevResults){
        return prevResults.map(elem -> {
            for (int i = 0; i < 5; i++) {
                if (compareResults(elem, calculateNextResult(some params))) {
                    return true;
                }
            }
            return false;
        });
    }

If calculateNextResult or compareResults were reactive methods returning Mono, then you could use flatMap instead of map, and use one of the Mono.repeat* methods.

For example, something like this:

    private Mono<Integer> calculateNextResult(some params) {
        // some implementation
    }

    private Mono<Boolean> compareResults(int prevRes, int nextRes) {
        // some implementation
    }
    public Flux<Boolean> run(some params, Flux<Integer> prevResults){

        return prevResults.flatMap(prevResult -> 

            calculateNextResult(some params)
                    .flatMap(nextResult -> compareResults(prevResult, nextResult))
                    .filter(comparisonResult -> comparisonResult)
                    .repeatWhenEmpty(Repeat.times(5))
                    .defaultIfEmpty(false));
    }

In this example, the repeatWhenEmpty will cause a new subscription to the Mono created within the flatMap, which will cause calculateNextResult to recalculate (assuming the Mono returned from calculateNextResult is setup to calculate a value on every subscription).

Reactor how to repeat some step n times if condition is met

1 Answers1

Linked