Spring-Integration-DSL:嵌套分散收集挂起

Question

这是一个损坏但可执行的示例代码:

@Bean
IntegrationFlow testFlow() {
    return IntegrationFlows
            .from(Http.inboundChannelAdapter("test")
                    .requestMapping(mapping -> mapping.methods(HttpMethod.GET))
                    .get())
            .scatterGather(
                    scatterer -> scatterer
                            .applySequence(true)
                            .recipientFlow(flow -> flow
                                    .scatterGather(
                                            scatterer1 -> scatterer1
                                                    .applySequence(true)
                                                    .recipientFlow(IntegrationFlowDefinition::bridge),
                                            gatherer -> gatherer.outputProcessor(MessageGroup::getOne))
                                    .log(INFO, m -> "THIS HAPPENS")),
                    gatherer -> gatherer.outputProcessor(MessageGroup::getOne))
            .log(INFO, m -> "THIS NEVER HAPPENS")
            .get();
}

预期输出是:

THIS HAPPENS
THIS NEVER HAPPENS

实际输出是:

THIS HAPPENS

我找到了this identical looking issue在 Github 上，但它声称已在 5.1.10 和 5.2.4 版本中修复。我正在运行 spring-boot-starter-integration 5.5.0 其中包括相同版本的 spring-integration-core 。

我该怎么做才能让这个嵌套的分散聚集工作？这是 DSL 还是我的代码的错误吗？

Answer 1

在仅使用一级分散收集遇到类似问题后，我意识到是日志消息阻止了输出返回到父流。将 .log() 替换为 .logAndReply() 或 .log().bridge()，一切都会再次正常工作。

像这样:

@Bean
IntegrationFlow testFlow() {
    return IntegrationFlows
            .from(Http.inboundChannelAdapter("test")
                    .requestMapping(mapping -> mapping.methods(HttpMethod.GET))
                    .get())
            .scatterGather(
                    scatterer -> scatterer
                            .applySequence(true)
                            .recipientFlow(flow -> flow
                                    .scatterGather(
                                            scatterer1 -> scatterer1
                                                    .applySequence(true)
                                                    .recipientFlow(IntegrationFlowDefinition::bridge),
                                            gatherer -> gatherer.outputProcessor(MessageGroup::getOne))
                                    .logAndReply(INFO, m -> "THIS HAPPENS")), // this fixes the problem
                    gatherer -> gatherer.outputProcessor(MessageGroup::getOne))
            .log(INFO, m -> "THIS NEVER HAPPENS")
            .get();
}