java - 如何有条件地执行 Reactor 的 Mono/Flux 供应商算子？

Question

如何有条件地执行then(Mono<T>)算子？

我有一个返回的方法Mono<Void>。它还可以返回错误信号。我想使用then运算符（或任何其他运算符），仅当前一个操作完成且没有错误信号时。

有人可以帮我找到合适的供应商运营商吗？

    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();
    }

-谢谢

标签： javaspring-webfluxproject-reactor

Answer 1

首先，我想强调一下 Reactor 的 Mono/Flux（接下来会考虑 Mono）有以下条件运算符（至少我知道的）：

• Mono#switchIfEmpty
• Mono#defaultIfEmpty
• 和其他一些供应商运营商的组合Mono#filter（例如Mono#flatMap）

第二点是Mono#then：

忽略此 Mono 中的元素并将其完成信号转换为提供的 Mono 的发射和完成信号。在生成的 Mono 中重放错误信号。

因此，这意味着无论then之前是什么，它都将返回值（空的或提供的）。

考虑到所有这些，您的解决方案将如下所示：

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);
}