Home » General

Using Spring JMS

Written By: Pascal Alma on March 12, 2007 4 Comments

Last week I discovered a piece of code in our project that didn't use Spring yet. It was the piece of code that used JMS to access some JBoss queues. Since we did use Spring everywhere in our code where it was possible, I decided to rewrite this piece so it also uses Spring (and so it also profits of all the advantages that Spring offers).
The original code looked like this:

PLAIN TEXT
JAVA:
  1. package net.pascalalma.jms;
  2.  
  3. import javax.jms.JMSException;
  4. import javax.jms.MessageProducer;
  5. import javax.jms.ObjectMessage;
  6. import javax.jms.Queue;
  7. import javax.jms.QueueBrowser;
  8. import javax.jms.QueueConnection;
  9. import javax.jms.QueueConnectionFactory;
  10. import javax.jms.QueueReceiver;
  11. import javax.jms.QueueSession;
  12. import javax.naming.InitialContext;
  13. import javax.naming.NamingException;
  14.  
  15. import org.apache.commons.logging.Log;
  16. import org.apache.commons.logging.LogFactory;
  17.  
  18. public class QueueBase {
  19.  
  20.    private static final Log LOG = LogFactory.getLog(QueueBase.class);
  21.  
  22.    private static InitialContext context = null;
  23.    private static QueueConnectionFactory qcf = null;
  24.    private boolean transacted;
  25.    private int ackMode;
  26.    private String queueName;
  27.    private boolean init = false;
  28.    protected QueueConnection qc;
  29.    protected QueueSession qs;
  30.    protected Queue queue;
  31.    protected QueueReceiver qr;
  32.  
  33.    protected MessageProducer mp;
  34.  
  35.    protected void init() throws QueueException {
  36.       LOG.info("Queue init started");
  37.       if (context == null) {
  38.          
  39.          try {
  40.             LOG.info("QueueBase: creating InitialContext for queues (see jndi.properties)");
  41.             context = new InitialContext();
  42.             qcf = (QueueConnectionFactory) context.lookup("/ConnectionFactory");
  43.          } catch (Exception e) {
  44.             throw new QueueException(e);
  45.          }
  46.       }
  47.  
  48.       try {
  49.          queue = (Queue) context.lookup(queueName);
  50.          qc = qcf.createQueueConnection();
  51.          qs = qc.createQueueSession(transacted, ackMode);
  52.  
  53.          assert (queue != null) : "queue is null";
  54.          assert (qc != null) : "qc is null";
  55.          assert (qs != null) : "qs is null";
  56.  
  57.          init = true;
  58.       } catch (Exception e) {
  59.          throw new QueueException(e);
  60.       }
  61.    }
  62.  
  63.    public QueueBase() {
  64.       this.init = false;
  65.    }
  66.  
  67.    public QueueBase(String queueName) {
  68.       this.queueName = queueName;
  69.       this.init = false;
  70.    }
  71.  
  72.    /**
  73.     * Constructor
  74.     */
  75.    public QueueBase(String queueName, boolean transacted, int ackMode) throws QueueException {
  76.       this.queueName = queueName;
  77.       this.transacted = transacted;
  78.       this.ackMode = ackMode;
  79.       this.init = false;
  80.    }
  81.  
  82.    /**
  83.     * Sends a message to the queue..
  84.     *
  85.     * @param msg
  86.     * @throws QueueException
  87.     */
  88.    public void send(BulkMessage msg) throws QueueException {
  89.       try {
  90.          if ((qs != null) && (mp != null)) {
  91.             ObjectMessage om = qs.createObjectMessage(msg);
  92.             mp.send(om);
  93.          } else {
  94.             LOG.error("Queue is not running / mp==null ....");
  95.          }
  96.       } catch (JMSException e) {
  97.          throw new QueueException(e);
  98.       }
  99.    }
  100.  
  101.    /*
  102.     * Get / Receive a message from the queue The receive block at most
  103.     * <timeoutInMilisecs> miliseconds (0=block)
  104.     */
  105.    public ObjectMessage receive(long timeoutInMilisecs) throws QueueException {
  106.  
  107.       BulkMessage m = null;
  108.       ObjectMessage om = null;
  109.       try {
  110.          if (qr != null) {
  111.             om = (ObjectMessage) qr.receive(timeoutInMilisecs);
  112.             if (om != null) {
  113.                m = (BulkMessage) om.getObject();
  114.             }
  115.          } else {
  116.             throw new QueueException("Queue receiver is null");
  117.          }
  118.       } catch (JMSException e) {
  119.          throw new QueueException(e);
  120.       }
  121.       return om;
  122.    }
  123.  
  124.    public QueueBrowser getBrowser() throws JMSException, QueueException {
  125.       return qs.createBrowser(queue);
  126.    }
  127.  
  128.    /**
  129.     *
  130.     */
  131.    public void destroy() {
  132.       try {
  133.          if (qs != null) {
  134.             qs.close();
  135.             qs = null;
  136.          }
  137.          if (qc != null) {
  138.             qc.stop();
  139.             qc.close();
  140.             qc = null;
  141.          }
  142.       } catch (Exception e) {
  143.          LOG.error(e);
  144.       }
  145.    }
  146.  
  147.    public String getQueueName() {
  148.       return queueName;
  149.    }
  150.  
  151.    public void setQueueName(String queueName) {
  152.       this.queueName = queueName;
  153.    }
  154.  
  155.    public int getAckMode() {
  156.       return ackMode;
  157.    }
  158.  
  159.    public void setAckMode(int ackMode) {
  160.       this.ackMode = ackMode;
  161.    }
  162.  
  163.    public Queue getQueue() {
  164.       return queue;
  165.    }
  166.  
  167.    public void setQueue(Queue queue) {
  168.       this.queue = queue;
  169.    }
  170.  
  171.    public boolean isTransacted() {
  172.       return transacted;
  173.    }
  174.  
  175.    public void setTransacted(boolean transacted) {
  176.       this.transacted = transacted;
  177.    }
  178.  
  179. }

It was doing a JNDI lookup of the queuefactory, created a new session and started and closed it.
After I rewrote it to use Spring it looked like:

PLAIN TEXT
JAVA:
  1. package nnet.pascalalma.jms;
  2.  
  3. import javax.jms.JMSException;
  4. import javax.jms.Message;
  5. import javax.jms.ObjectMessage;
  6. import javax.jms.Session;
  7.  
  8. import org.apache.commons.logging.Log;
  9. import org.apache.commons.logging.LogFactory;
  10. import org.springframework.jms.core.JmsTemplate;
  11. import org.springframework.jms.core.MessageCreator;
  12.  
  13. public class QueueBaseService {
  14.  
  15.    private static final Log LOG = LogFactory.getLog(QueueBaseService.class);
  16.  
  17.    private JmsTemplate jmsTemplate = null;
  18.  
  19.    public void setJmsTemplate(JmsTemplate jmsTemplate) {
  20.       this.jmsTemplate = jmsTemplate;
  21.    }
  22.  
  23.    public JmsTemplate getJmsTemplate() {
  24.       return jmsTemplate;
  25.    }
  26.  
  27.    public void sendMessage(final BulkMessage bm) {
  28.       LOG.debug("sendMessage(final BulkMessage bm)");
  29.  
  30.       jmsTemplate.send( new MessageCreator() {
  31.          public Message createMessage(Session session) throws JMSException {
  32.             ObjectMessage om = session.createObjectMessage();
  33.             om.setObject(bm);
  34.             return om;
  35.          }
  36.       });
  37.    }
  38.  
  39.    public ObjectMessage receiveMessage() {
  40.       return (ObjectMessage)jmsTemplate.receive();
  41.  
  42.    }
  43.  
  44.    public String getQueueName() {
  45.       return jmsTemplate.getDefaultDestinationName();
  46.    }
  47.  
  48.    public void setAcknowledgeMode(int mode){
  49.       jmsTemplate.setSessionAcknowledgeMode( mode);
  50.    }
  51.  
  52. }

And of course some configuration of the queues and JNDI/JMS:

PLAIN TEXT
XML:
  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <!DOCTYPE beans SYSTEM "../dtd/spring-beans.dtd">
  3. <beans>
  4.  
  5.   <!--  JNDI properties -->
  6.   <bean id="jndiTemplate" class="org.springframework.jndi.JndiTemplate">
  7.     <property name="environment">
  8.       <props>
  9.         <prop key="java.naming.factory.initial">
  10.            org.jnp.interfaces.NamingContextFactory
  11.         </prop>
  12.         <prop key="java.naming.provider.url">
  13.            jnp://localhost:1099
  14.         </prop>
  15.         <prop key="java.naming.factory.url.pkgs">
  16.            org.jboss.naming:org.jnp.interfaces
  17.         </prop>
  18.       </props>
  19.     </property>
  20.   </bean>
  21.  
  22.   <!--  JNDI Connection Factory -->
  23.   <bean id="jmsConnectionFactory" class="org.springframework.jndi.JndiObjectFactoryBean">
  24.     <property name="jndiTemplate">
  25.       <ref bean="jndiTemplate"/>
  26.     </property>
  27.     <property name="jndiName">
  28.       <value>ConnectionFactory</value>
  29.     </property>
  30.   </bean>
  31.  
  32.    <!-- The default queue, for incoming bulk messages-->
  33.   <bean id="destination"  class="org.springframework.jndi.JndiObjectFactoryBean">
  34.     <property name="jndiTemplate">
  35.       <ref bean="jndiTemplate"/>
  36.     </property>
  37.     <property name="jndiName">
  38.       <value>${queue.in}</value>
  39.     </property>
  40.   </bean>
  41.  
  42.   <!-- Queue Access Object, for the default queue -->
  43.   <bean id="jmsTemplateQueue" class="org.springframework.jms.core.JmsTemplate102">
  44.     <property name="connectionFactory">
  45.       <ref bean="jmsConnectionFactory"/>
  46.     </property>
  47.     <property name="defaultDestination">
  48.       <ref bean="destination"/>
  49.     </property>
  50.     <property name="receiveTimeout">
  51.        <value>60000</value>
  52.     </property>
  53.     <!--  this value can be overwritten by setting this parameter at business level-->
  54.     <property name="sessionAcknowledgeModeName">
  55.        <value>AUTO_ACKNOWLEDGE</value>
  56.     </property>
  57.     <property name="sessionTransacted">
  58.       <value>false</value>
  59.     </property>
  60.   </bean>
  61.  
  62.   <!-- Queue service for the default queue -->
  63.   <bean id="queueService" class="net.pascalalma.jms.QueueBaseService">
  64.     <property name="jmsTemplate">
  65.       <ref bean="jmsTemplateQueue" />
  66.     </property>
  67.   </bean>
  68. </beans>

So not only does it take less code to achieve the same thing, it is also much less error prone. At least, that was what I would expect, but when I got back in the office one day later, started my machine and ran some unit tests (after synchronizing my code) I was running into an error:

PLAIN TEXT
CODE:
  1. Caused by: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'jmsConnectionFactory' defined in class path resource [jmsContext.xml]: Initialization of bean failed; nested exception is java.lang.NoSuchFieldError: doPruning
  2.     at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:355)
  3.     at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:226)
  4.     at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:147)
  5.     at org.springframework.beans.factory.support.BeanDefinitionValueResolver.resolveReference(BeanDefinitionValueResolver.java:176)
  6.     ... 33 more
  7. Caused by: java.lang.NoSuchFieldError: doPruning
  8.     at org.jboss.aop.AOPClassPool.<clinit>(AOPClassPool.java:49)
  9.     at java.lang.Class.forName0(Native Method)
  10.     at java.lang.Class.forName(Unknown Source)
  11.     at org.jboss.aop.instrument.InstrumentorFactory.class$(InstrumentorFactory.java:44)
  12.     at org.jboss.aop.instrument.InstrumentorFactory.<clinit>(InstrumentorFactory.java:44)
  13.     at org.jboss.aop.AspectManager$2.run(AspectManager.java:316)
  14.     at java.security.AccessController.doPrivileged(Native Method)
  15.     at org.jboss.aop.AspectManager.instance(AspectManager.java:261)
  16.     at org.jboss.aop.AspectManager.instance(AspectManager.java:254)
  17.     at org.jboss.jms.client.delegate.ClientConnectionFactoryDelegate.<clinit>(ClientConnectionFactoryDelegate.java)
  18.     at sun.misc.Unsafe.ensureClassInitialized(Native Method)
  19.     at sun.reflect.UnsafeFieldAccessorFactory.newFieldAccessor(Unknown Source)
  20.     at sun.reflect.ReflectionFactory.newFieldAccessor(Unknown Source)
  21.     at java.lang.reflect.Field.acquireFieldAccessor(Unknown Source)
  22.     at java.lang.reflect.Field.getFieldAccessor(Unknown Source)
  23.     at java.lang.reflect.Field.getLong(Unknown Source)
  24.     at java.io.ObjectStreamClass.getDeclaredSUID(Unknown Source)
  25.     at java.io.ObjectStreamClass.access$600(Unknown Source)
  26.     at java.io.ObjectStreamClass$2.run(Unknown Source)
  27.     at java.security.AccessController.doPrivileged(Native Method)
  28.     at java.io.ObjectStreamClass.<init>(Unknown Source)
  29.     at java.io.ObjectStreamClass.lookup(Unknown Source)
  30.     at java.io.ObjectStreamClass.initNonProxy(Unknown Source)
  31.     at java.io.ObjectInputStream.readNonProxyDesc(Unknown Source)
  32.     at java.io.ObjectInputStream.readClassDesc(Unknown Source)
  33.     at java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
  34.     at java.io.ObjectInputStream.readObject0(Unknown Source