Throw an exception if a Reactor Flux doesn't complete in a set time

user_3650662 0
Viewed 513

I have a potentially long-running Flux that I'd like to stop after a certain duration has passed. I've found several methods of doing this, however what I'm struggling with is how to be able to tell that the Flux timed out rather than just completed naturally.

Sample (very simple) code:

Flux.range(0, 10000)
        .take(Duration.ofMillis(1))
        .doOnNext(System.out::println)
        .collectList()
        .block();

What I'd like is something like this:

Flux.range(0, 10000)
        .take(Duration.ofMillis(1))
        .doOnNext(System.out::println)
        .doOnError(t -> {
            if (t instanceof TimeoutException) {
                System.out.println("I timed out");
            }
        })
        .collectList()
        .block();

however take doesn't seem to notify on error; all I see is a terminate and complete signal which is what I get if I don't include the take in the Flux and just let it complete naturally.

I've looked briefly into the timeout operator which does throw an exception, however timeout looks like it'll only throw an exception if the Flux doesn't emit a single element within a certain time, rather than if the whole Flux doesn't complete in a certain time.

Does anyone have any tips or examples of how they've solved this?

Thanks in advance!

project-reactor java
2 Answers

UPDATED

You can use takeUntilOther operator to signal an error after a given time:

Duration timeout = Duration.ofMillis(500);
Flux.range(0, 10000)
    .delayElements(Duration.ofMillis(10))
    .doOnNext(System.out::println)
    .takeUntilOther(Mono.delay(timeout).then(Mono.error(new TimeoutException())))
    // .takeUntilOther(Mono.never().timeout(Duration.ofMillis(500))) // based on Michael's answer this is also an option
    .doOnError(t -> {
        if (t instanceof TimeoutException) {
            System.out.println("I timed out");
        }
    })
    .blockLast();

If Flux is converted into Mono using for example .collectList() or .then() operator, then you can simply apply the .timeout(...) operator after that and it will behave as you require:

Flux.range(0, 10000)
    .delayElements(Duration.ofMillis(10))
    .doOnNext(System.out::println)
    .collectList()
    .timeout(Duration.ofMillis(500))
    .doOnError(t -> {
        if (t instanceof TimeoutException) {
            System.out.println("I timed out");
        }
    })
    .block();
user_60511760

Instead of using take(), you can use .mergeWith(Flux.never().timeout(Duration.ofMillis(500))) to merge your flux with another that will always throw a timeout exception after a certain timeframe.

To take your example, you'd do something like:

Flux.range(0, 10000)
        .delayElements(Duration.ofMillis(10))
        .mergeWith(Flux.<Integer>never().timeout(Duration.ofMillis(500)))
        .doOnNext(System.out::println)
        .doOnError(t -> {
            if (t instanceof TimeoutException) {
                System.out.println("I timed out");
            }
        })
        .collectList()
        .block();

...which will give you something like:

(...snip)
24
25
26
27
28
29
30
31
I timed out
user_5514060
Related