Настройка TcpInboundGateway для отправки нескольких ответов на одно и то же сообщение

Question

В настоящее время у меня есть TcpInboundGateway, который принимает сообщения, выполняет некоторую обработку сообщения и затем возвращает соответствующий ответ, все как TcpInboundGateway.

Однако мне любопытно, если это TcpInboundGateway можно настроить таким образом, чтобы он отправлял немедленный ответ на исходящий запрос, но продолжал обрабатывать запрос и также отправлять ответ на постобработку?

Думайте об этом немедленном ответе как о подтверждении отправителю, что сообщение было получено.

Возможное решение:

Изучив эту публикацию , я пришел к тому, что считаю жизнеспособным решением этой проблемы.

@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 и подтверждения с последующим ответом вам необходимо использовать PublishSubscribeChannel с Executor, введенным для выполнения асинхронной обработки.

Первый подписчик должен вернуть некотороеПодтвердите в заголовке replyChannel.Таким образом, ваш TcpInboundGateway выполнит запрос-ответ и возвратит подтверждение в подключенный сокет.

В то же время, как вы хотите, второй подписчик может выполнить желаемую логику и построить реальный ответ позже.Только то, что нам нужно использовать упоминание в документах Совместная работа адаптеров исходящих и входящих каналов (как вы уже заметили).Итак, поскольку 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. a PublishSubscribeChannel с executor для вашего TcpInboundGateway
2. Простой обработчик для ответа с подтверждением в качестве первого подписчика
3. Подпоток для обработки запроса
4. A TcpSendingMessageHandler для отправки ответа процесса в то же TCP-соединение.