Настройка канала ответа по умолчанию в Spring Integration MessagingGateway

Question

У меня есть приложение Spring Integration, которое использует комбинацию входящего / исходящего адаптера для отправки и получения TCP-сообщений. Я пытаюсь сопоставить сообщение, отправленное с ответом, полученным для этого запроса, но у меня возникают некоторые проблемы.

Моя текущая конфигурация выглядит следующим образом ...

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

@Bean
public TcpReceivingChannelAdapter inboundAdapter() {
    TcpReceivingChannelAdapter inboundAdapter = new TcpReceivingChannelAdapter();
    inboundAdapter.setConnectionFactory(serverConnectionFactory());
    inboundAdapter.setOutputChannel(receiveResponse());
    return inboundAdapter;
}

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

@Bean
public MessageChannel receiveResponse() {
    return new DirectChannel();
}

@MessagingGateway(defaultRequestChannel="sendRequest", defaultReplyChannel="receiveResponse") 
public interface RequestSender {
    public String sendRequest(@Payload String payload, @Header(IpHeaders.CONNECTION_ID) String connectionId);
}

Все настраивается, как и ожидалось, при загрузке, когда вызывается метод RequestSender.sendRequest(), сообщение отправляется, ответ отправляется обратно (у меня есть другое приложение, которое отвечает), но я сталкиваюсь с последующей ошибкой, когда ответ достигает моего inboundAdapter. Ошибка ...

org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.receiveResponse'.

Исходя из того, что я прочитал о MessagingGateway, replyChannel - это канал, для которого шлюз будет прослушивать ответ. Поэтому я думаю, что все, что получает входящий адаптер, в конечном итоге будет возвращено методом RequestSender.sendRequest(). Однако в моем случае кажется, что MessagingGateway не может подписаться на канал receiveResponse.

Есть ли способ реализовать MessagingGateway таким способом? Если так, что, по-видимому, мешает моей текущей конфигурации работать так, как я надеюсь?

Answer 1

Прежде всего, в вашем объяснении есть заблуждение. Вы говорите о receiveResponse, но ошибка показывает нам handleResponse, и в вашей конфигурации нет никого.

Пожалуйста, исправьте ваш вопрос так, чтобы он нас не смущал.

Еще одна проблема: какой смысл использовать комбинацию канальных адаптеров, если у вас все еще есть syn c шлюз поверх этого сетевого взаимодействия и вы все равно собираетесь блокировать вызывающий поток. Как насчет использования TcpOutboundGateway вместо этого? Кстати, начиная с последней версии 5.3, которая поддерживает асинхронный режим c, поэтому ваш @MessagingGateway может быть также асинхронным c: https://docs.spring.io/spring-integration/docs/5.3.0.BUILD-SNAPSHOT/reference/html/whats-new.html#x5 .3-tcp .

В любом случае проблема, с которой вы сталкиваетесь, - это передача идентификатора replyChannel по сети для сервера и обратно вместе с ответом.

Для этой задачи мы ввели функцию под названием header-channels-to-string: https://docs.spring.io/spring-integration/docs/5.3.0.M4/reference/html/message-transformation.html#header -channel-registry . Таким образом, объект TemporaryReplyChannel хранится в реестре, а его заголовок в сообщении запроса заменяется некоторым идентификатором строки. Это replyHeader должно быть отправлено по сети и должно возвращаться в заголовках ответного сообщения.

В вашем случае проблема намного больше, поскольку TCP / IP не поддерживает заголовки. Поэтому вам нужно придумать какой-нибудь механизм для переноса заголовков вместе с полезной нагрузкой в ​​пакете TCP. См. Некоторый подход в документах: https://docs.spring.io/spring-integration/docs/5.3.0.M4/reference/html/ip.html#ip -headers .

Но, несмотря на все это, я все еще думаю, что вам нужно вернуться к решению TcpOutboundGateway ...