techblog 

This probably isn’t news to many but I only recently noticed that Oracle AQ can manage JMS queues. We don’t use a full blown J2EE application server so our options when using a JMS broker is to either have a standalone server/cluster (more work for SAs) or to embed the JMS server in to our middleware (more work for me). By using the database we add no new dependencies or software so this is a pretty neat solution. Another big advantage for us is that it looks like we would be able to create messages inside PL/SQL procedures but I may be getting ahead of myself here.

There isn’t much to it but it was a lot harder to get to the same point than with the other JMS brokers I have been testing. There is a lot of documentation to go through and it is all fairly general. I also couldn’t find a good example so I’ve created a quick howto…

First thing is to choose a user that is going to use the queue and set up the grants…

grant execute on sys.dbms_aqadm to mquser;
grant execute on sys.dbms_aq to mquser;
grant execute on sys.dbms_aqin to mquser;
grant execute on sys.dbms_aqjms to mquser;
exec dbms_aqadm.grant_system_privilege('ENQUEUE_ANY','mquser');
exec dbms_aqadm.grant_system_privilege('DEQUEUE_ANY','mquser');

And then as that user create the queue

exec dbms_aqadm.create_queue_table(queue_table=>'JMS_QUEUE_TABLE', queue_payload_type=>'sys.aq$_jms_text_message',multiple_consumers=>false);
exec dbms_aqadm.create_queue(queue_name=>'testmq', queue_table=>'JMS_QUEUE_TABLE');
exec dbms_aqadm.start_queue(queue_name=>'testmq');

This example sets up a point to point queue. I haven’t tried using a pub/sub model but at the very least

multiple_consumers=>true

would need to be set.

For the Java/Spring side you’ll need to download the Oracle Containers for Java (OC4J) pack and add aqapi.jar to your classpath.

For sending JMS messages from Spring you need to set up at least three beans. The interesting one is an instance of JmsTemplate which saves us from writing any boiler-plate JMS code. This requires a JMS ConnectionFactory and a default destination which are the two other beans. These are not so trivial to set up with Oracle AQ and at the moment I am creating these with extra factory-beans.

public class OracleAqConnectionFactoryInitialiser {

      public ConnectionFactory createConnectionFactory() throws Exception{
        return oracle.jms.AQjmsFactory.getQueueConnectionFactory("jdbc:oracle:thin:mquser/mqpass@myhost:1521:testtom", new Properties());
    }
}

(There is also a getQueueConnectionFactory(DataSource ds) method which is probably what you really want)

Finding the JMS destination has to be done via the connection so we have to create a Spring FactoryBean…

public class OracleAqDestinationFactoryBean implements FactoryBean {
    private QueueConnectionFactory connectionFactory;
    private String queueName;
    private String queueUser;    
 
    @Required
    public void setConnectionFactory(QueueConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }    
 
    @Required
    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }    
 
    @Required
    public void setQueueUser(String queueUser) {
        this.queueUser = queueUser;
    }    
 
    public Object getObject() throws Exception {
        // createQueueSession(true,0) - create a transactional session
        AQjmsSession session = (AQjmsSession) connectionFactory.createQueueConnection().createQueueSession(true, 0);
        return session.getQueue(queueUser, queueName);
    }    
 
    public Class getObjectType() {
        return javax.jms.Queue.class;
    }    
 
    public boolean isSingleton() {
		return false;
    }
}

Wire up these factory-beans and the ConnectionFactory and Queue…

<bean id="aqConnectionFactoryInitialiser" class="uk.nominet.mq.OracleAqConnectionFactoryInitialiser">
    <property name="dataSource" ref="oracleDs"/>
</bean>

<bean id="testmq" class="uk.nominet.mq.OracleAqDestinationFactoryBean">
    <property name="connectionFactory" ref="aqConnectionFactory"/>
    <property name="queueName" value="testmq"/>
    <property name="queueUser" value="mquser"/>
</bean>    

<bean id="aqConnectionFactory" factory-bean="aqConnectionFactoryInitialiser"
    factory-method="createConnectionFactory" >
</bean>

…and the JmsTemplate…

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="aqConnectionFactory"/>
    <property name="defaultDestination" ref="testmq"/>
    <property name="receiveTimeout" value="30000"/>
</bean>

Once these are configured you can send a message by calling jmsTemplate.send() from anywhere you wish…

        jmsTemplate.send(new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                TextMessage txtMsg = session.createTextMessage();
                txtMsg.setText("Hello World!");
                return txtMsg;
            }
        });

Configuring a listener is now simple

<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="aqConnectionFactory"/>
    <property name="destination" ref="testmq"/>
    <property name="messageListener" ref="messageListener1" />
    <property name="sessionTransacted" value="true"/>
</bean><bean id="messageListener1"
    class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
    <constructor-arg>
        <bean class="uk.nominet.contact.mq.TestMqMessageListener"/>
    </constructor-arg>
</bean>

Read the docs for details on using the MessageListenerAdapter.

The documentation for managing queues and using JMS can be found here and here