These days I’m working on re-engineering our existing messaging platform. One of our messaging platform’s main functionality is to connect with different external SMS Gateway providers and submit the sms messages. Existing platform was build on top of Jboss Messaging Queue /MDB. In the process of re-engineering, platform should use RabbitMQ as a new broker to store the messages before processing. RabbitMQ is based on wire protocol and Jboss Messaging framework is based on Java JMS API. Reason for switching is that messaging producers are build on django/python and it wants to submit request using JSON. Instead of writing a python JMS communication, I want utilize the  AMQP (Advanced Messaging Queue protocol ). Currently rabbitMQ is known for amqp implementation and of course, it is build on erlang and is robust. Lets look into details.

1. Using spring-amqp 1.1.2.RELESE to access rabbitMQ. Spring-amqp abstracts the AMQP provider, therefore we can change the broker easily. Using following maven dependency:

<dependency>
   <groupId>org.springframework.amqp</groupId>
   <artifactId>spring-amqp</artifactId>
   <version>1.1.2.RELEASE</version>
</dependency>
<dependency>
   <groupId>org.springframework.amqp</groupId>
   <artifactId>spring-rabbit</artifactId>
   <version>1.1.2.RELEASE</version>
   </dependency>
<dependency>
   <groupId>com.rabbitmq</groupId>
   <artifactId>amqp-client</artifactId>
   <version>2.8.7</version>
</dependency>

2. Spring-amqp web site documentation specifics version 1.1.1, but 1.1.2 includes a fix for the transaction support. We have to include spring-rabbit, it contains the rabbitMQ specific implementation and utilizes the com.rabbitmq library.

3.  Brief look into rabbitMQ. Producers uses the ‘Routing’ and connects to the ‘Exchange’ to submit the messages. ‘Queue’s are connected to ‘Exchange’ using ‘Binding’. Consumers connects to ‘Queue’ and receives the messages and process it. Consumers could receive the messages synchronously and/or asynchronously. We’ll be using async to receive the messages from the broker.

4. Using following spring configuration xml.

<?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:rabbit="http://www.springframework.org/schema/rabbit"
       xmlns:aop="http://www.springframework.org/schema/aop"
       xmlns:tx="http://www.springframework.org/schema/tx"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
                           http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
                           http://www.springframework.org/schema/rabbit
                           http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
                           http://www.springframework.org/schema/tx
                           http://www.springframework.org/schema/tx/spring-tx-3.0.xsd
                           http://www.springframework.org/schema/aop
                           http://www.springframework.org/schema/aop/spring-aop-3.0.xsd" ></code>

<bean id="rabbitConnectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
<constructor-arg index="0" value="localhost" />
<constructor-arg index="1" value="5672" />
<property name="username" value="guest" />
<property name="password" value="guest" />
<property name="channelCacheSize" value="10"/>
</bean>

<rabbit:queue id="requestQueue" durable="true" auto-delete="false" exclusive="false" name="RequestQueue"/>
<rabbit:queue id="smsGatewayQueue" durable="true" auto-delete="false" exclusive="false" name="SmsGatewayQueue"/>

<rabbit:direct-exchange name="messagingExchange" durable="true" auto-delete="false" id="PlatformExchange">
<rabbit:bindings>
<rabbit:binding queue="SmsGatewayQueue" key="SmsGatewayQueue"/>
<rabbit:binding queue="RequestQueue" key="RequestQueue"/>
</rabbit:bindings>
</rabbit:direct-exchange>

<bean id="requestTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate">
<property name="connectionFactory" ref="rabbitConnectionFactory"/>
<property name="exchange" value="PlatformExchange"/>
<property name="queue" value="RequestQueue"/>
<property name="routingKey" value="RequestQueue"/>
<property name="messageConverter">
<bean class="org.springframework.amqp.support.converter.JsonMessageConverter"></bean>
</property>
</bean>

<bean id="smsGatewayTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate">
<property name="connectionFactory" ref="rabbitConnectionFactory"/>
<property name="exchange" value="PlatformExchange"/>
<property name="queue" value="SmsGatewayQueue"/>
<property name="routingKey" value="SmsGatewayQueue"/>
<property name="messageConverter">
<bean class="org.springframework.amqp.support.converter.JsonMessageConverter"></bean>
</property>
</bean>

<!-- Containers -->
<bean id="requestContainer" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
<constructor-arg ref= "rabbitConnectionFactory" />
<property name="concurrentConsumers" value="4" />
<property name="messageListener" ref="requestHandler" />
<property name="queues" ref="RequestQueue" />
<property name="channelTransacted" value="true" />
<property name="autoStartup" value="true" />
</bean>

<bean id="smsGatewayContainer" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
<constructor-arg ref= "rabbitConnectionFactory" />
<property name="concurrentConsumers" value="4" />
<property name="messageListener" ref="smsGatewayHandler" />
<property name="queues" ref="SmsGatewayQueue" />
<property name="channelTransacted" value="true" />
<property name="autoStartup" value="true" />
</bean>

<!-- message listeners -->
<bean id="requestHandler" class="com.senthadev.RequestHandler" />
<bean id="smsGatewayHandler" class="com.senthadev.SmsGatewayHandler" />

<!-- rabbit mq broker admin -->
<bean id="rabbitAdmin" class="org.springframework.amqp.rabbit.core.RabbitAdmin">
<constructor-arg ref="rabbitConnectionFactory" />
</bean>

<bean id="rabbitTxManager" class="org.springframework.amqp.rabbit.transaction.RabbitTransactionManager">
<property name="connectionFactory" ref="rabbitConnectionFactory"/>
</bean>

<tx:annotation-driven transaction-manager="rabbitTxManager"/>

</beans>

5.  Following Java code is a sample producer, it will use RabbitTemplate to submit the message.

package com.senthadev;

import java.util.HashMap;
import org.apache.log4j.Logger;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;

public class RequestDB {
    private static final Logger log = Logger.getLogger(MootaDB.class);

    public static void main(String a[]) throws Exception{
        ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:spring-context.xml");
        AmqpTemplate requestTemplate = (AmqpTemplate)ctx.getBean("RequestTemplate");

        while (true){
            RequestDto req = new RequestDto();
            //Converts the RequestDto to json using org.codehaus.jackson
            requestTemplate.convertAndSend(req);
            Thread.sleep(1000);
        }
    }
}

6. Following Java code is a sample async consumer, it will be used by SimpleMessageListenerContainer to processes the messages.

package com.senthadev;

import org.apache.log4j.Logger;
import org.codehaus.jackson.map.ObjectMapper;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;

public class RequestHandler implements MessageListener{

    private static final Logger log = Logger.getLogger(RequestHandler.class);

    @Override
    public void onMessage(Message message) {
        try{
        ObjectMapper convert = new ObjectMapper();
        RequestDto data = convert.readValue(message.getBody(), RequestDto.class);

        AmqpTemplate smsgwTemplate = (AmqpTemplate)SpringLoader.instance().getBean("smsGatewayTemplate");
        SmsMesgDto smsDto = factory.buildSms(data);
        smsgwTemplate.convertAndSend(smsDto);
        }catch (Exception e){
            log.error(e);
            throw new AmqpException(e);
        }
    }
}

7. According to the spring configuration, consumer container (requestContainer) will use 4 threads to process the messages from rabbitmq. When a message is received, it will call ‘RequestHandler’ to process it. Remember to write thread safety code.

8. Limitation that I’m facing now is that in JMS/Jboss its possible to set a delivery delay for the messages. Means, we can set a delivery time for the messages and the container will deliver the message to the consumer when the delivery delay is arrived. But in rabbitMQ its not possible to set a delivery delay. I have to find a technique to achieve delivery delay. Spring-amqp/Container provides methods to start and stop the container operations. Have to use them to achieve the delivery delay functionality. (Will be covering that soon.)

I have described the steps in high level. I will be adding a sample project in github soon.