Spring Integration DSL - 在流程中等待来自通道的输入

Question

我想知道在Spring Integration中是否可以在流程中包含一个外部通道。所以，我有一个HTTP入站网关流程，在触发后，它应该通过UDP端口与其他进程进行通信。我的最大担忧是如何在这个流程内从UDP端口接收消息。

@Bean
public IntegrationFlow httpInboundGatewayFlow() {
    return IntegrationFlows.from(Http.inboundGateway(httpInboundPath))
             .transform(niUdpRequestTransformer())
             /*sending a message to udp port*/
             .publishSubscribeChannel(runnable -> runnable
                  .subscribe(flow -> flow                      
                      .handle(udpOutboundChannel())))
            /*wait for input from udpResponse channel here (HOW TO?)*/
            /*process udpInboundFlow message*/
            .handle((payload, headers) -> successNetworkResponse())))
            .transform(new ObjectToJsonTransformer())
            .handle((payload, headers) -> payload)
            .get();
}

@Bean
public IntegrationFlow udpInboundFlow() {
    return IntegrationFlows.from(udpInboundChannel())
            .transform(niUdpResponseTransformer())
            .channel("udpResponse")
            .get();
}

使用udpInboundFlow应该实现为一种轮询器，它检查是否已到达正确的消息。

谢谢你的帮助。

- Stevan

Answer 1

你所说的是“相关性”问题。我认为你希望得到一些与UDP请求相关的回复。

因此，你确实需要以下内容：

.channel("udpResponse")
.aggregate(...)

您需要为请求消息确定一些“correlationKey”，并确保UDP的回复具有相同的关键字。聚合器应配置为“.releaseStrategy（group -> group.size() == 2）”，其中第一个消息将是请求消息，第二个消息实际上是来自外部udpResponse的结果。更多信息请参见参考手册。

- Artem Bilan