Весенняя интеграция - реализация запроса-ответа

Question

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

Я использую JMS для облегчения отправки и получения сообщений на основе XML. Наша подкрепляющая реализация - IBM Websphere MQ.

Я также использую Spring Boot (версия 1.3.6.RELEASE) и пытаюсь использовать подход с настройкой на основе чистой аннотации (если это возможно). Я искал в Интернете и вижу пример, но ничего, что я пока вижу, помогает мне понять, как все это сочетается. Документация Spring Integration отличная, но я все еще борется с тем, как все части подходят друг другу. Я заранее извиняюсь, если есть что-то, что я пропустил. Я рассматриваю публикацию здесь как последнюю альтернативу.

Вот что у меня есть для моей конфигурации:

package com.daluga.spring.integration.configuration 

import com.ibm.mq.jms.MQConnectionFactory; 
import com.ibm.mq.jms.MQQueue; 
import com.ibm.msg.client.wmq.WMQConstants; 
import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
import org.springframework.beans.factory.annotation.Autowired; 
import org.springframework.beans.factory.annotation.Qualifier; 
import org.springframework.beans.factory.annotation.Value; 
import org.springframework.context.annotation.Bean; 
import org.springframework.context.annotation.Configuration; 
import org.springframework.integration.annotation.InboundChannelAdapter; 
import org.springframework.integration.annotation.IntegrationComponentScan; 
import org.springframework.integration.annotation.Poller; 
import org.springframework.integration.channel.QueueChannel; 
import org.springframework.integration.config.EnableIntegration; 
import org.springframework.jms.annotation.EnableJms; 
import org.springframework.jms.connection.CachingConnectionFactory; 
import org.springframework.jms.core.JmsTemplate; 

import javax.jms.ConnectionFactory; 
import javax.jms.DeliveryMode; 
import javax.jms.Destination; 
import javax.jms.JMSException; 

//import com.ibm.msg.client.services.Trace; 

@Configuration 
public class MQConfiguration { 

    private static final Logger LOGGER = LoggerFactory.getLogger(MQConfiguration.class); 

    @Value("${host-name}") 
    private String hostName; 

    @Value("${port}") 
    private int port; 

    @Value("${channel}") 
    private String channel; 

    @Value("${time-to-live}") 
    private int timeToLive; 

    @Autowired 
    @Qualifier("MQConnectionFactory") 
    ConnectionFactory connectionFactory; 

    @Bean(name = "jmsTemplate") 
    public JmsTemplate provideJmsTemplate() { 
     JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); 
     jmsTemplate.setExplicitQosEnabled(true); 
     jmsTemplate.setTimeToLive(timeToLive); 
     jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);  
     return jmsTemplate; 
    } 

    @Bean(name = "MQConnectionFactory") 
    public ConnectionFactory connectionFactory() { 
     CachingConnectionFactory ccf = new CachingConnectionFactory(); 

     //Trace.setOn(); 

     try { 
      MQConnectionFactory mqcf = new MQConnectionFactory(); 
      mqcf.setHostName(hostName); 
      mqcf.setPort(port); 
      mqcf.setChannel(channel); 
      mqcf.setTransportType(WMQConstants.WMQ_CM_CLIENT); 
      ccf.setTargetConnectionFactory(mqcf); 
      ccf.setSessionCacheSize(2); 
     } catch (JMSException e) { 
      throw new RuntimeException(e); 
     } 

     return ccf; 
    } 

    @Bean(name = "requestQueue") 
    public Destination createRequestQueue() { 

     Destination queue = null; 

     try { 
      queue = new MQQueue("REQUEST.QUEUE"); 
     } catch (JMSException e) { 
      throw new RuntimeException(e); 
     } 

     return queue; 
    } 

    @Bean(name = "replyQueue") 
    public Destination createReplyQueue() { 

     Destination queue = null; 

     try { 
      queue = new MQQueue("REPLY.QUEUE"); 
     } catch (JMSException e) { 
      throw new RuntimeException(e); 
     } 

     return queue; 
    } 

    @Bean(name = "requestChannel") 
    public QueueChannel createRequestChannel() { 

     QueueChannel channel = new QueueChannel(); 

     return channel; 
    } 

    @Bean(name = "replyChannel") 
    public QueueChannel createReplyChannel() { 

     QueueChannel channel = new QueueChannel(); 

     return channel; 
    } 

} 

А вот мой сервис класс:

package com.daluga.spring.integration.service 

import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
import org.springframework.beans.factory.annotation.Autowired; 
import org.springframework.integration.annotation.ServiceActivator; 
import org.springframework.stereotype.Service; 


@Service 
public class MyRequestReplyService { 

    private static final Logger LOGGER = LoggerFactory.getLogger(MyRequestReplyService.class); 

    @ServiceActivator(inputChannel = "replyChannel") 
    public void sendAndReceive(String requestPayload) { 
     // How to get replyPayload 
    } 

} 

Таким образом, на данный момент, я не совсем уверен, как склеить все это вместе, чтобы сделать эту работу. Я не понимаю, как склеить мой запрос и ответить на очереди активатора службы, чтобы все это работало.

Служба, которую я вызываю (JMS/Webshere MQ based), использует типичный идентификатор сообщения и корреляции, чтобы я мог правильно связать запрос с соответствующим ответом.

Может ли кто-нибудь предоставить мне какие-либо рекомендации о том, как заставить это работать? Пожалуйста, дайте мне знать, какую дополнительную информацию я могу предоставить, чтобы это было ясно.

Заранее благодарим за вашу помощь!

Dan

Answer 1

Шлюзы предоставляют семантику запроса/ответа.

Вместо прямого использования JmsTemplate вы должны использовать Spring Integration's built-in JMS Support.

@Bean 
@ServiceActivator(inputChannel="requestChannel") 
public MessageHandler jmsOutGateway() { 
    JmsOutboundGateway outGateway = new JmsOutboundGateway(); 
    // set properties 
    outGateway.setOutputChannel(replyChannel()); 
    return outGateway; 
} 

Если вы хотите, чтобы свернуть свой собственный, изменить способ обслуживания активатором возвращать тип ответа и использовать один из шаблонов sendAndReceive() или convertSendAndReceive() методов.

sample app использует конфигурацию XML, но должен предоставить некоторые дополнительные указания.