How to conditionally execute Reactor's Mono/Flux supplier operator?

2
votes

How to conditionally execute the then(Mono<T>) operator?

I have a method that returns Mono<Void>. It can also return an error signal. I want to use the then operator (or any other operator), only when the previous operation completes without an error signal.

Can someone help me to find the right supplier operator?

    public static void main(String[] args) {
        Mono.just("GOOD_SIGNAL")//It can also be a BAD_SIGNAL
                .flatMap(s -> firstMethod(s))
                .then(secondMethod())
                .subscribe()
        ;
    }

    private static Mono<String> secondMethod() {
        //This method call is only valid when the firstMethod is success
        return Mono.just("SOME_SIGNAL");
    }

    private static Mono<Void> firstMethod(String s) {
        if ("BAD_SIGNAL".equals(s)) {
            Mono.error(new Exception("Some Error occurred"));
        }

        return Mono
                .empty()//Just to illustrate that this function return Mono<Void>
                .then();
    }

-Thanks

2
javaspring-webfluxproject-reactor

2 Answers

1
votes

then propagates an error in its source, so it covers this aspect. From what I understand, the reason you cannot use flatMap instead of then is that the source is empty, due to firstMethod(). In that case, combine defer() and then() to achieve the laziness you seek:

Mono.just("GOOD_SIGNAL")//It can also be a BAD_SIGNAL
    .flatMap(s -> firstMethod(s))
    .then(Mono.defer(this::secondMethod))
    .subscribe();
1
votes

First of all, I want to underline that Reactor's Mono/Flux (will consider Mono next) have the following conditioning operators (at least what I know):

The second point is that Mono#then:

ignore element from this Mono and transform its completion signal into the emission and completion signal of a provided Mono. An error signal is replayed in the resulting Mono.

So it means that then is about to return the value (empty or provided) despite what was before it.

Considering all of that, your solution would look something like that:

public static void main(String[] args) {
        Mono.just("GOOD_SIGNAL")//It can also be a BAD_SIGNAL
                .flatMap(s -> firstMethod(s))
                .switchIfEmpty(secondMethod())
                .doOnError(...)//handle you error here
                .subscribe();
    }

private static Mono<String> secondMethod() {
     //This method call is only valid when the firstMethod is success
     return Mono.just("SOME_SIGNAL");
}

private static Mono<Void> firstMethod(String str) {
    return Mono.just(str)
               .filter(s -> "BAD_SIGNAL".equals(s))
               .map(s -> new Exception("Some Error occurred: "+s))
               .flatMap(Mono::error);
}