JMS Request-Response with Spring-Integration

A colleague of mine asked me “why would Spring need to write a framework to send / receive JMS messages”.

So here is why:

In short, avoiding frameworks usually leads to breaking the DRY principle. 

In order to send messages you need to

1. Fetch the connection factory from JNDI.
2. Create a JMS connection.
3. Create a JMS session.
4. Fetch / create the JMS destination.
5. Create and configure a Producer.
6. Create your JMS message.
7. Remember to start() the connection.
8. Remember to close / cache the JMS resources, and handle errors properly.

In order to receive messages you need to

1. Fetch the connection factory from JNDI.
2. Create a JMS connection.
3. Create a JMS session.
4. Fetch / create the JMS destination.
5. Create and configure a consumer - either add a MessageListener, or receive() messages in a loop.
6. Remember to close / cache the JMS resources, and handle errors properly.
   

If you’d also like to get a response you need to

1. Set the message reply to channel. This can be a temp destination, or an existing one. If this is an existing destination, you will also need a message selector and use the JMSCorrelationID.
2. Create a consumer for the reply message and receive() the message.
3. On the receiver side you need to create a response message, set the JMSCorrelationID, and send back to the JMSReplyTo destination that was specified on the incoming message.
   

Writing all this code is naturally prone to errors. Now throw in handling of connections errors (reconnects) as well, and what you get is a whole lot of messy code :P

When using Spring JMS integration framework you can avoid most of the boiler plate code.  When using spring-integration on top of that you can actually decouple your (Java) code from JMS and Spring all together. That is you can avoid having any compile time dependency on JMS or Spring!

The client code:

  <integration:channel id="publishChannel"/>
  <integration:channel id="jmsReplyToChannel"/>
 
  <integration:gateway service-interface="MyPublisher" id="publisher" default-request-channel="publishChannel" default-reply-channel="jmsReplyToChannel"/>

  <jms:outbound-gateway id="jmsOutGateway"
                       request-destination="topic"
                       request-channel="publishChannel"
                       reply-channel="jmsReplyToChannel"
                       connection-factory="jmsConnectionFactory"/>

The publisher is merely an interface :D

public interface MyPublisher {
    public long publish(String msg);
}

The server side code:

<jms:inbound-gateway id="jmsInGateway"
                      request-destination="topic"
                      request-channel="serviceInboundChannel"
                      connection-factory="jmsConnectionFactory"/>
 
  <integration:service-activator input-channel="serviceInboundChannel" ref="myService" method="print"/>

  <integration:channel id="serviceInboundChannel"/>

  <bean id="myService" class="MyServiceImpl"/>

The called service implementation can be anything you like. I only print the message to stdout and return the reception time – this can be later used to test that we are actually performing a synchronised request response style messaging, which is not the case when no JMS response is involved.

public class MyServiceImpl implements MyService {
    public long print(String msg) {
        long now = System.currentTimeMillis();
        System.out.format("%d print: %s\n", now, msg);
        return now;
    }
}

We also need some boring JMS Spring beans definitions to get this working. The code below assumes you have a running ActiveMQ server, but you may replace the provider url to use an embeded ActiveMQ broker (see the commented-out line).

<bean id="jmsJndiTemplate" class="org.springframework.jndi.JndiTemplate">
    <constructor-arg>
      <props>
        <prop key="java.naming.factory.initial">org.apache.activemq.jndi.ActiveMQInitialContextFactory</prop>
        <prop key="java.naming.provider.url">tcp://localhost:61616</prop>
        <!-- prop key="java.naming.provider.url">vm://localhost?broker.persistent=false</prop -->
      </props>
    </constructor-arg>
  </bean>

  <bean id="jmsConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
    <constructor-arg>
      <bean class="org.springframework.jndi.JndiObjectFactoryBean">
        <property name="jndiName" value="ConnectionFactory"/>
            <property name="jndiTemplate" ref="jmsJndiTemplate"/>
          </bean>
    </constructor-arg>
    <property name="reconnectOnException" value="true"/>
    <property name="cacheProducers" value="true"/>
  </bean>

  <bean id="topic" class="org.springframework.jndi.JndiObjectFactoryBean">
        <property name="jndiName" value="dynamicTopics/eran-test"/>
        <property name="jndiTemplate" ref="jmsJndiTemplate"/> 
  </bean>

The diagram below depicts our creation:

Notes:

• I used a topic here, but it would have been more appropriate to use a queue in most cases…
• Spring-Integration has it’s own cost. Although the code could be slightly improved performance wise, the internal message format, and the temp queue / channels creation impact performance. So don’t say I didn’t warn you… a plain async message passing style application should probably be written with plain Spring JMSteamplate, and message containers.
• The code above as is causes Spring to create a lot of temporary resources like queues, consumers, producers, channels. I left it this way for brevity, but most of the temp resources creation can be avoided, thus reducing the load on the message broker.