How do you implement Polling Logic in Project Reactor?

user_906497 0
Viewed 2145

I have a method that sends a request to get the status of a job and returns the status, it looks like this:

Mono<JobStatus> getJobStatus() {...}

The JobStatus object has a method JobStatus.isDone() which returns whether the pending job is done or not.

Is there a way for me to keep subscribing to the mono until JobStatus.isDone() is true? i.e. something like getJobStatus().keepSubscribingUntil(status -> status.isDone())

project-reactor spring
2 Answers

One option would be for the getJobStatus() Mono to only emit when the job is done, which might not necessarily be easy depending on how that Mono is currently implemented.

For polling, assuming the Mono polls each time you subscribe to it you could use repeatWhen paired with takeUntil:

getJobStatus()
    .repeatWhen(completed -> completed.delayElements(Duration.ofMillis(pollDelay))) //(1)
    .takeUntil(JobStatus::isDone) //(2)
    .last() //(3)

(1) repeatedly re-subscribe to the source Mono (this produces a Flux<JobStatus>)

(2) cancel the above repeat cycle as soon as the returned status is marked as done

(3) switch back to a Mono<JobStatus> which emits the last iteration's status (so the one that is marked as done)

user_11134860

Other option would be using repeatWhenEmpty when you might want to poll till you get either success or failure from the publisher. When there are scenarios you don't want to wait forever to get the response, rather want to have a timeout, you could use the same method where you have own logic or libraries' method to timeout the operation.

AtomicInteger c = new AtomicInteger();
Mono<String> source = Mono.defer(() -> c.getAndIncrement() < 3 
? Mono.empty() 
: Mono.just("7"));
return source.repeatWhenEmpty(4, Flux::cache);

In the above example 4 specifies, maximum number of times you want to retry.

source.repeatWhenEmpty(exponentialBackOff(Duration.ofMillis(1), 
                Duration.ofMillis(15),
                Duration.ofSeconds(1)));

https://github.com/cloudfoundry/cf-java-client has the exponentialBackOff utils mehtod.

user_39624700
Related