如何选择性地跳过 Flux 上的几个处理步骤

Question

我有从套接字接收到的动态热数据流。我需要检查条件，如果值匹配，则跳转到带有新消息的步骤 3。

    final Flux<Msg> msgs = Flux.generate(receiver);

    final Flux<Msg> processed = msgs
        .map(this::checkCondition)  //step1
        .map(remote::doLongRunning) //optional step2
        .map(this::processFurther)  //step3
 ...

    public Msg checkCondition(Msg msg) {
        if(doCheck(msg)){
            //is there a way to jump to step3 here ?
            return new OtherMsg(msg, "someAdditionalData")) 
        } else{
            return msg
        }
    }

我能想到的唯一替代方案 - 拆分 Flux 并将其组装回去，有没有更清洁的方法？

   final Flux<Msg> msgs = Flux.generate(receiver);

        final Flux<OtherMsg> checked = msgs
            .filter(this::doCheck) //step1
            .map(msg -> new OtherMsg(msg, "someAdditionalData"));

        final Flux<OtherMsg> unchecked = msgs
            .filter(msg -> !doCheck(msg)) //step1
            .map(remote::doLongRunning);  //optional step2

        Flux.merge(checked, unchecked)
            .map(this::processFurther)  //step3

Answer 1

您不能跳过一个步骤，但您可以使用flatMap()和三元运算符来代替条件分支形式：

final Flux<Msg> processed = msgs
        .flatMap(msg -> doCheck(msg)
            ? Mono.just(new OtherMsg(msg, "someAdditionalData")).map(remote::doLongRunning)
            : Mono.just(msg))
        .map(this::processFurther);

这样您就可以调用任何其他方法来操作三元表达式第一部分中的值，如果 doCheck() returns false，第二部分将确保它被绕过。 processFurther() 将在 flatMap() 调用之后执行，因此无论如何都会执行。

如何选择性地跳过 Flux 上的几个处理步骤

How to optionally skip several processing steps on Flux

java

optional

flux

project-reactor