Я пытаюсь написать простой поток сообщений с использованием DSL API Spring Integration v4, который будет выглядеть следующим образом:
-> in.ch -> Processing -> JmsGatewayOut -> JMS_OUT_QUEUE
Gateway
<- out.ch <- Processing <- JmsGatewayIn <- JMS_IN_QUEUE
Поскольку запрос / ответ является асинхронным, когда я вставляю сообщение через начальный шлюз, сообщение проходит весь путь до JMS_OUT_QUEUE. Помимо этого потока сообщений ответное сообщение помещается обратно в JMS_IN_QUEUE, которое затем принимается JmsGatewayIn. На этом этапе сообщение обрабатывается и помещается в out.ch (я знаю, что ответ попадает в out.ch, потому что у меня есть перехватчик регистратора, который регистрирует помещаемое туда сообщение), но шлюз никогда не получает ответ.
Вместо ответа система за пределами этого потока сообщений, которая приняла сообщение из JMS_OUT_QUEUE и поместила ответ в JMS_IN_QUEUE, получает javax.jms.MessageFormatException: MQJMS1061: Unable to deserialize object на собственном JmsOutboundgateway (я думаю, что она не может десериализовать объект ответа jms, просматривая журналы ).
У меня явно что-то неправильно настроено, но я точно не знаю, что. Кто-нибудь знает, что мне не хватает?
Работа с spring-integration-core-4.0.3.RELEASE, spring-integration-jms-4.0.3.RELEASE, spring-integration-java-dsl-1.0.0.M2, spring-jms-4.0.6.RELEASE.
Мой шлюз настроен следующим образом:
@MessagingGateway
public interface WsGateway {
@Gateway(requestChannel = "in.ch", replyChannel = "out.ch",
replyTimeout = 45000)
AResponse process(ARequest request);
}
Мой поток интеграции настроен следующим образом:
@Configuration
@EnableIntegration
@IntegrationComponentScan
@ComponentScan
public class IntegrationConfig {
@Bean(name = "in.ch")
public DirectChannel inCh() {
return new DirectChannel();
}
@Bean(name = "out.ch")
public DirectChannel outCh() {
return new DirectChannel();
}
@Autowired
private MQQueueConnectionFactory mqConnectionFactory;
@Bean
public IntegrationFlow requestFlow() {
return IntegrationFlows.from("in.ch")
.handle("processor", "processARequest")
.handle(Jms.outboundGateway(mqConnectionFactory)
.requestDestination("JMS_OUT_QUEUE")
.correlationKey("JMSCorrelationID")
.get();
}
@Bean
public IntegrationFlow responseFlow() {
return IntegrationFlows.from(Jms.inboundGateway(mqConnectionFactory)
.destination("JMS_IN_QUEUE"))
.handle("processor", "processAResponse")
.channel("out.ch")
.get();
}
}
Спасибо за любую помощь в этом, личка.
