Reactor based polling is not terminating

Intention is to poll a certain number of times until I receive a particular response. Here is the piece of code I have compiled:

Mono.defer(() -> webClient.getResponse())
// repeat GET call after every 500 ms
.repeatWhen(repeat -> repeat.delayElements(Duration.ofMillis(500)))
.repeat(4) // max repeat attempts
.takeUntil(response -> response.equals("1"))
//.retryWhen(Retry.max(3))
.log()
.subscribe(response -> {
System.out.println("response received: " + response);
});

But, there's a problem. It is not stopping at all. I was expecting it to run 4 times and exit.

12:54:54.527 [parallel-7] INFO  reactor.Flux.TakeUntil.1 - onNext(2)
response received: 2
12:54:55.034 [parallel-8] INFO  reactor.Flux.TakeUntil.1 - onNext(2)
response received: 2
12:54:55.540 [parallel-9] INFO  reactor.Flux.TakeUntil.1 - onNext(2)
response received: 2
12:54:56.045 [parallel-10] INFO  reactor.Flux.TakeUntil.1 - onNext(2)
response received: 2
12:54:56.551 [parallel-1] INFO  reactor.Flux.TakeUntil.1 - onNext(2)
response received: 2
12:54:57.057 [parallel-2] INFO  reactor.Flux.TakeUntil.1 - onNext(2)
response received: 2
12:54:57.563 [parallel-3] INFO  reactor.Flux.TakeUntil.1 - onNext(2)
response received: 2
.......

repeat(int n) runs the flux n number of times. But in your case, you've used repeatWhen before that, and it produces an infinite flux. That's why your flux does not ends anymore.

If you want to retry subscription until you receive the right value, you can use retryWhen function. It takes a retry configuration as input, to limit the number of retries, specifies delays between retrys, etc. In your case, you can:

1. Fail the stream if you do not receive the right value
2. Append a retryWhen operation to retry subscription.

Let's illustrate it with an example:

import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

import static java.lang.System.nanoTime;

public class RetryWhen {

    /**
     * Emit a decreased value on each subscription
     * @param initial First value to emit
     */
    static Mono<Integer> countdown(int initial) {
        var current = new AtomicInteger(initial);
        var t0 = nanoTime();
        return Mono.fromCallable(() -> current.getAndDecrement())
                .doOnNext(i -> System.out.printf(
                        "Emit %d at %d ms%n",
                        i, (int)((nanoTime() - t0) * 1e-6)
                ));
    }

    public static void main(String[] args) {
        var source = countdown(4);
        // raise an error if value is different than one.
        var oneOrError = source.filter(i -> i == 1)
                .switchIfEmpty(Mono.error(new RuntimeException("bad value")));

        // Retry 4 times maximum. Each retry is spaced by ~ 500 ms
        var retried = oneOrError.retryWhen(Retry.fixedDelay(4, Duration.ofMillis(500)));

        var result = retried.block();
        System.out.println("Result: "+result);
    }
}

The above program produces:

Emit 4 at 173 ms
Emit 3 at 696 ms
Emit 2 at 1198 ms
Emit 1 at 1699 ms
Result: 1

Notes :

• in your case, repeatWhen produces an infinite stream, because the function you passed to it never returns an empty publisher. It causes continuous subscription to your original mono (which returns "2").
• Also, as repeatWhen repeat values from the source/caller, your .takeUntil(response -> response.equals("1")) statement will never be matched (it will always receive "2").