Message Driven POGOs: When Groovy, Spring and OpenMQ collide

An article by Glen    3 Comments

When we last left this story, I was talking about integrating Grails and JMS so that groovyblogs can put all of its feed and thumbnail fetch requests on a JMS queue. Now it’s time to look at the second half of the story. How my standalone groovy feed/thumbnail fetcher can pick requests off the queue, and put a response back on a different queue.

First a little background. Spring makes writing message driven POJOs really tidy. Groovy makes it even tidier. To implement a POGO (Plain Old Groovy Object) that listens on given Queue, I just have to implement the MessageListener interface, and provide an onMessage routine. Spring will setup the necessary background thread poller that will invoke my bean when a new message arrives on the queue. It will also look after the connection pooling to my JMS server so I’m not creating/dissolving connections every time. Nice.

But if I want to put something on the queue? Then it’s time to use Spring’s JmsTemplate class which provides the necessary convertAndSend that I’ll need to put my Map object back on the queue with my thumbnail result. The rest is just Spring wiring. Let’s take a look at the code for my ThumbnailListener first:

package org.groovyblogs.queue

import javax.jms.MapMessage
import javax.jms.Message
import javax.jms.MessageListener
import org.apache.log4j.Logger

import org.springframework.jms.core.JmsTemplate

/**
 * Listener for Map messages containing a URL to fetch.
 *
 * @author Glen Smith
 */
class ThumbnailListener implements MessageListener {

    private static final Logger log = Logger.getLogger(ThumbnailListener.class)

    JmsTemplate jmsTemplate
    ImageGrabber imageGrabber

    public void onMessage(Message message) {

        MapMessage map = (MapMessage) message
        String url = map.getString("url")
        String blogEntryId = map.getString("id")

        println "Got Thumbnail Request for: ${url} for ${blogEntryId}"

        def thumbs = imageGrabber.getImages(url)

        sendReply( [ id : blogEntryId, url : url,
            bigImage: thumbs[0].encodeBase64().toString(),
            smallImage: thumbs[1].encodeBase64().toString() ] )

    }

    public void sendReply(Map map) {
        try {
          jmsTemplate.convertAndSend(map)
        } catch (Exception e) {
           log.error("Error sending reply to thumbnail queue", e)
        }

    }

}

Wow. The whole receive, gather, and send takes about 20ish lines. Of course, we’re going to need a few things injected, right? We’ll need our JmsTemplate injected, and it will need to be wired up to our JMS endpoint (in my case, an OpenMQ queue). We’ll also need to inject whatever class implements our ImageGrabber interface (but that’s for another day).

Finally, we need to tell Spring that our class is message listener, and we’ll need invoking if messages arrive on a specified queue. So far we’ve implementation agnostic, but it’s time to get some religion at the config level…

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

    <bean id="connectionFactory" class="com.sun.messaging.ConnectionFactory">
        <!-- wrap with a custom factory bean to call setProperty("imqAddressList", "yourhost:7676")
                with your host/port if not using default of localhost:7676 -->
    </bean>

    <bean id="jmsTemplateUrl" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="defaultDestination" ref="replyQueueUrl"/>
    </bean>

    <bean id="requestQueueThumbnail" class="com.sun.messaging.Queue">
        <constructor-arg value="thumbnailRequest"/>
    </bean>

    <bean id="replyQueueThumbnail" class="com.sun.messaging.Queue">
        <constructor-arg value="thumbnailResponse"/>
    </bean>

    <bean id="jmsTemplateThumbnail" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="defaultDestination" ref="replyQueueThumbnail"/>
    </bean>

    <bean id="swtImageGrabber" class="org.groovyblogs.queue.SWTImageGrabber"/>

    <bean id="myThumbnailListener" class="org.groovyblogs.queue.ThumbnailListener">
        <property name="jmsTemplate" ref="jmsTemplateThumbnail"/>
        <property name="imageGrabber" ref="swtImageGrabber"/>
    </bean>

    <bean id="jmsContainerThumbnail" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="requestQueueThumbnail"/>
        <property name="messageListener" ref="myThumbnailListener" />
    </bean>

</beans>

A few pieces of magic here. We construct the com.sun.messaging.Queue with a constructor argument that is the name of the queue. The connection factory uses localhost:7676 by default, and there are not convenience spring-friendly setters for changing it. You’ll need to wrap it in a custom FactoryBean if you need to use a different host/port (since you’ll need to make special a setProperty() call).

The work of listening on the queue and invoking our onMessage routine is handled by Spring’s DefaultMessageListenerContainer class. By default that class only creates a single listener on the queue. But if you need to scale out with more listeners, you can increase its concurrentConsumers property from the default of 1.

So there you have it. A couple of dozen lines of Groovy and a couple of dozen lines of XML and we’re in business with some serious Queue listening and responding. Add some GMaven to the mix (must blog about that at some stage), and you’ve got very few lines of code to handle you whole build process and jar bundling! I think my next step is to add some JMX instrumentation to my POGO so I can use JConsole to see some stats on recent messages processed…