Spring集成-重播相同的流

Question

我有一个SI DSL流，就像

@Bean
public IntegrationFlow processRequest() {
    return flow -> flow.channel(REQUEST_INPUT)
            .transform(...)
            .enrichHeaders(h -> h.<Storage>headerFunction(...)))
            .enrich(this::...)

            //.route(ifReplayIsNeeded(), routeToSameFlow())

            .enrich(this::...)
            .route(..., ...)
            .route(..., ...)
            .enrich(this::...)
            .handle(...)

            //.route(ifReplayWasNeeded(), routeBack())

            .route(..., ...)
            .enrich(this::...)
            .transform(...)
            .channel(REQUEST_OUTPUT);
}

因此，当满足一个条件时（请参见ifreplayineeded（）），则必须再次调用processRequest（）流。然而，并非必须执行整个流程，而是几乎在最后执行(-

routeToSameFlow（）看起来像（Storage用于存储请求/响应和流中使用的其他数据）

Consumer<RouterSpec<Boolean, MethodInvokingRouter>> routeToSameFlow() {
    return rs -> rs.resolutionRequired(false)
            .subFlowMapping(true, sf -> sf
                    // 1. storing the current Storage
                    .enrichHeaders(h -> h.<Storage>headerFunction("store", s -> s))
                    // 2. creating the req for the internal flow
                    .transform(internalRequestMapper, "mapFrom")
                    // 3. routing to the beginning of the flow
                    .route(Message.class, (m) -> REQUEST_INPUT)
                    // 4. defining the channel where the internal flow will return to
                    .channel("RETURN_CHANNEL")
            )
            .defaultOutputToParentFlow();
}

和routeBack（）

Consumer<RouterSpec<Boolean, MethodInvokingRouter>> routeBack() {
    return rs -> rs.resolutionRequired(false)
            .subFlowMapping(true, sf ->
                    sf.route(Message.class, (m) -> "RETURN_CHANNEL")
            )
            .defaultOutputToParentFlow();
}

我肯定错过了一些概念，因为我得到了以下错误：

原因：org.springframework.beans.factory.BeanCreationException：“当前组件”（org.springframework.integration.router.MethodInvokingRouter@60a0f09f）是单向的“MessageHandler”，不适合配置“outputChannel”。这是集成流程的结束。

你能帮我实现这样的逻辑吗？我应该把主流分成更小的集成流吗<我希望尽可能少地干扰主流，这就是为什么我只想在开始时添加分流路线，在结束时添加回流路线的原因。有可能做到吗？

谢谢你<问候，
V。

Answer 1

您可能会使所有这些路由器的逻辑过于复杂。考虑在您的主流中间只有简单的Channel（）定义，并作为您想要返回的子流的输出。JavaDSL中没有限制，只是在开头和结尾都有Channel（）定义：您可以在流定义的任何点添加Channel（）。例如：

 .handle(...)
 .channel("middleOfTheFlow")
 .route(ifReplayWasNeeded(), routeBack())

然后您的routeBack（）流定义最终可能只有. Channel（"middleOfTheFlow"），此处处理后的消息将被传递到主流的middleOfTheFlow通道。

换句话说，一个集成流（IntegrationFlow）只是将一些业务功能的endpoint保持在同一位置的逻辑组件，但这并不意味着流中的所有endpoint都非常严格地相互关联。您始终可以在中间有一个通道，并从其他地方向该通道发送消息，订阅该通道的endpoint将处理该消息。

查看文档中的更多信息：https://docs.spring.io/spring-integration/docs/current/reference/html/dsl.html#java-数字用户线