配置 TcpInboundGateway 以向同一消息发送多个响应

Question

我目前有一个 TcpInboundGateway 接收消息，对消息进行一些处理，然后返回适当的响应，所有这些都是 TcpInboundGateway 应该的。

但是，我很好奇这个TcpInboundGateway 是否可以配置为它会立即响应原始请求但继续处理请求并发送后处理响应？

将此立即响应视为对发送者收到消息的确认。

可能的解决方案：

在查看了这个post 之后，我想出了一个我认为可行的解决方案。

@Configuration
@EnableIntegration
public class Configuration {

    @Bean
    public AbstractServerConnectionFactory serverConnectionFactory() {
        return new TcpNetServerConnectionFactory(2002);
    }

    @Bean
    public TcpReceivingChannelAdapter inboundAdapter(AbstractServerConnectionFactory serverConnectionFactory) {
        TcpReceivingChannelAdapter inboundAdapter = new TcpReceivingChannelAdapter();
        inboundAdapter.setConnectionFactory(serverConnectionFactory);
        inboundAdapter.setOutputChannelName("sendAcknowledgement");
        return inboundAdapter;
    }

    @MessageEndpoint
    public class InboundMessageHandler {

        @Autowired
        private OutboundMessageGateway gateway;

        @ServiceActivator(inputChannel="sendAcknowledgement", outputChannel="doProcessing")
        public Message<String> initialAck(Message<String> message) {
            gateway.send("ACK", message.getHeaders().get(IpHeaders.CONNECTION_ID).toString());
            return message;
        }

        @ServiceActivator(inputChannel="doProcessing", outputChannel="sendResponse")
        public Message<String> mockDelay(Message<String> message) throws InterruptedException {
            return message;
        }
    }

    @MessagingGateway(defaultRequestChannel="sendResponse")
    public interface OutboundMessageGateway {
        void send(@Payload String message, @Header(IpHeaders.CONNECTION_ID) String connectionId);
    }

    @Bean
    @ServiceActivator(inputChannel="sendResponse")
    public TcpSendingMessageHandler outboundAdapter(AbstractServerConnectionFactory serverConnectionFactory) {
        TcpSendingMessageHandler outboundAdapter = new TcpSendingMessageHandler();
        outboundAdapter.setConnectionFactory(serverConnectionFactory);
        return outboundAdapter;
    }

}

Answer 1

对于带有TcpInboundGateway 并确认稍后回复的用例，您需要使用带有Executor 注入的PublishSubscribeChannel 来进行异步处理。

第一个订阅者应该在replyChannel 标头中返回一些确认。这样您的TcpInboundGateway 将执行请求-回复并将该确认返回到连接的套接字中。

在您希望的同时，第二个订阅者可以执行所需的逻辑并稍后构建真正的回复。只有我们需要使用文档Collaborating Outbound and Inbound Channel Adapters 中的提及（正如您已经注意到的那样）。因此，由于TcpInboundGateway 将IpHeaders.CONNECTION_ID 标头填充到请求消息中，它将在您的异步进程中可用，随后的TcpSendingMessageHandler 将知道将处理后的回复发送到哪里：

private void handleMessageAsServer(Message<?> message) {
    // We don't own the connection, we are asynchronously replying
    String connectionId = message.getHeaders().get(IpHeaders.CONNECTION_ID, String.class);
    TcpConnection connection = null;
    if (connectionId != null) {
        connection = this.connections.get(connectionId);
    }
    if (connection != null) {
        try {
            connection.send(message);
        }

所以，你需要的是这样的：

1. PublishSubscribeChannel 和 executor 用于您的 TcpInboundGateway
2. 作为第一个订阅者回复 ack 的简单处理程序
3. 处理请求的子流程
4. TcpSendingMessageHandler 将进程响应发送到同一 TCP 连接。