Spring Integration. Send message to HornetQ in JBoss 7.1.1

[Gary Russell]

That's a tricky one - are you saying that sometimes replyQueue contains replies, and sometimes just unsolicited messages?

Given that the gateway uses a selector looking for the JMSMessageID in the JMSCorrelationID, you could, for unsolicited messages, have the other side set JMSCorrelationID to, say "foo" and have a <int-jms:message-driven-channel-adapter ... selector="JMSCorrelationID='foo'" .../>.

[RobertVox1977]

Thanks a lot Gary!

That works.

My spring configuration file:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
       xmlns:jee="http://www.springframework.org/schema/jee" xmlns:jms="http://www.springframework.org/schema/jms"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/integration
       http://www.springframework.org/schema/integration/spring-integration.xsd
       http://www.springframework.org/schema/integration/jms
       http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.0.xsd
       http://www.springframework.org/schema/jee
       http://www.springframework.org/schema/jee/spring-jee.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms.xsd">

    <int:gateway id="gatewayAsynchService" service-interface="pl.firstdata.maqc.integration.GatewayAsynchService"
                 default-request-channel="requestChannel"/>

    <int:poller default="true" fixed-delay="250"/>

    <int:channel id="requestChannel">
        <int:queue capacity="10"/>
    </int:channel>

    <int-jms:outbound-gateway request-channel="requestChannel" request-destination="requestQueue"
                              connection-factory="connFactory"/>

    <int-jms:message-driven-channel-adapter id="mdChannel" connection-factory="connFactory" destination="replyQueue"
                                            selector="JMSCorrelationID='foo'" channel="replyChannel" />

    <int:outbound-channel-adapter id="replyChannel" ref="integrationAsynchService" method="logMessage"  />

    <bean id="integrationAsynchService" class="pl.firstdata.maqc.integration.IntegrationAsynchService"/>

    <jee:jndi-lookup id="requestQueue" expected-type="javax.jms.Destination" jndi-name="jms/queue/req"/>

    <jee:jndi-lookup id="replyQueue" expected-type="javax.jms.Destination" jndi-name="jms/queue/reply"/>

    <bean id="myTargetConnectionFactory" class="org.springframework.jndi.JndiObjectFactoryBean">
        <property name="jndiName" value="jms/RemoteConnectionFactory"/>
    </bean>

    <bean id="connFactory" class="org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapter">
        <property name="targetConnectionFactory" ref="myTargetConnectionFactory"/>
        <property name="username" value="robert"/>
        <property name="password" value="sys"/>
    </bean>

</beans>

and Test file:

@ContextConfiguration({ "/integration_asynch.xml"})
@RunWith(SpringJUnit4ClassRunner.class)
public class IntegrationAsynchTest {

    @Autowired
    private GatewayAsynchService gatewayAsynchService;

    @Autowired
    private IntegrationAsynchService integrationAsynchService;


    @Test
    public void integrAsynchTest() throws InterruptedException {
        gatewayAsynchService.process("test message");
        Thread.sleep(10000);
        List<String> messages = integrationAsynchService.getMessages();
        Assert.isTrue(messages.get(0).equals("Test message from MDB"));
        Assert.isTrue(!messages.isEmpty());
    }
}

A send message asynch, sleep main thread and reach to answers from queue.

MDB code:

@MessageDriven(name = "ReqMBEJB", activationConfig = {
        @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
        @ActivationConfigProperty(propertyName = "destination", propertyValue = "java:jboss/exported/jms/queue/req"),
        @ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge")})

public class RequestMBBean implements javax.jms.MessageListener {

    @Resource(mappedName = "java:/ConnectionFactory")
    private ConnectionFactory connectionFactory;

    @Resource(mappedName = "java:jboss/exported/jms/queue/reply")
    private Queue replyQueue;

    @Override
    public void onMessage(Message message) {
        System.out.println("-------------------" + message + "---------------------");
        QueueConnection con = null;
        QueueSession ses = null;
        QueueSender sender = null;
        try {
            QueueConnectionFactory qcf = (QueueConnectionFactory) connectionFactory;
            con = qcf.createQueueConnection();
            ses = con.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            sender = ses.createSender(replyQueue);
            TextMessage msg = ses.createTextMessage("Test message from MDB");
            //msg.setJMSCorrelationID(message.getJMSMessageID());
            msg.setJMSCorrelationID("foo");
            sender.send(msg);
        } catch (JMSException e) {
            throw new RuntimeException(e);
        } finally {
            JmsHelper.close(sender, ses, con);
        }
    }
}

and

public class IntegrationAsynchService {

    private List<String> messages = new ArrayList<String>();


    public void logMessage(String message) {
        messages.add(message);
    }

    public List<String> getMessages() {
        return Collections.unmodifiableList(messages);
    }
}

[RobertVox1977]

The example of my spring configuration file I've attached doesn't work.

When I send in a test a message twice:

gatewayAsynchService.process(claimDto1);
gatewayAsynchService.process(claimDto2);

then is sent only the first.

I must simply add dispatcher to my requestChannel to hand the request off to another thread - exactly as wrote Gary.

So, instead of:

 <int:channel id="requestChannel">
        <int:queue capacity="10"/>
    </int:channel>

I should have:

<int:channel id="requestChannel">
        <int:dispatcher task-executor="myTaskExecutor"/>
    </int:channel>

where myTaskExecutor could be class that extends, for instance: SimpleAsyncTaskExecutor.

I should have read more carefully, Gary, thanks