JMS and Oracle Streams AQ can support a variety of payloads. These payloads are stored in the database and need to be converted to a Java representation in order for our programs to manipulate them. The following table outlines what payloads are supported and the corresponding classes that will work with these payloads and that will be able to convert them to and from a Java representation.

When you use the JmsTemplate together with the Oracle AQ JMS support you can use the aq-jms-connection-factory entry to provide a connection factory to the JmsTemplate.

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:orcl="http://www.springframework.org/schema/data/orcl" 
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
       http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context-3.0.xsd
       http://www.springframework.org/schema/data/orcl
       http://www.springframework.org/schema/data/orcl/spring-data-orcl-1.0.xsd"> 

 
    <orcl:aq-jms-connection-factory id="connectionFactory" 
            data-source="dataSource"/> 

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="sessionTransacted" value="true"/>
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>

    <bean id="transactionManager"
          class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
        <property name="dataSource" ref="dataSource"/>
    </bean>

    <bean id="dataSource" ... />

</beans>

The configuration for a Message-Driven POJO with a MessageListenerContainer is very similar. You use the same type of connection factory configuration. This is passed in to the listener container configuration. Here is an example using the JMS namespace support.

<?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:tx="http://www.springframework.org/schema/tx"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:orcl="http://www.springframework.org/schema/data/orcl" 
       xmlns:jms="http://www.springframework.org/schema/jms"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
       http://www.springframework.org/schema/tx
       http://www.springframework.org/schema/tx/spring-tx-3.0.xsd
       http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context-3.0.xsd
       http://www.springframework.org/schema/data/orcl 
       http://www.springframework.org/schema/data/orcl/spring-data-orcl-1.0.xsd
       http://www.springframework.org/schema/jms
       http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">

    <context:annotation-config/>
    
    <tx:annotation-driven/>
    
    <bean id="messageDelegate" class="spring.test.MessageDelegate"/>
    
    <jms:listener-container connection-factory="connectionFactory" 
            transaction-manager="transactionManager">
        <jms:listener destination="jmsadmin.jms_text_queue" 
                ref="messageDelegate" method="handleMessage"/>
    </jms:listener-container> 
    
    <orcl:aq-jms-connection-factory id="connectionFactory" 
            data-source="dataSource"/> 
    
    <bean id="transactionManager"
          class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
        <property name="dataSource" ref="dataSource"/>
    </bean>

    <bean id="dataSource" ... />

</beans>

See the next section for how to configure the transaction support and use a the same local transaction as the JDBC or ORM data access.

The configurations in the previous section will take advantage of the transaction synchronization provided by Spring, but there will be two transactions. One transaction for the data access and one for the JMS messaging. They will be synchronized, so if the data access transaction commits then the messaging transaction will also commit while if the data access transaction roll back then the messaging transaction will also roll back.

There is always a chance that the commit for the messaging transaction could fail after the data access transaction has committed successfully. This is of course a problem that you would have to account for in your code by checking for duplicate delivery of a message.

A better solution is to configure both data access and the messaging to share a transaction. Most often this is done using JTA, and that works, but has some impact on performance. For JTA you need to use distributed transactions and XA capable resources designed for two-phase commits. This comes at an extra cost that we try to avoid if possible.

Another option is to have the data access and the messaging share a local data access transaction. This is possible since the Oracle AQ implementation consists of a set of tables and stored procedures running in the database accessed through a standard JDBC connection. If you use the same database for data access and messaging with AQ, then you can configure the connection factory to share the database connection and the local transaction. You configure this connection and transaction sharing by setting the attribute use-local-data-source-transaction to true.

    <orcl:aq-jms-connection-factory id="connectionFactory"
        use-local-data-source-transaction="true" 
        data-source="dataSource"/>

Configuring the connection factory to share a local data source transaction with the data access code has some implications for JMS connection and session caching. You can still configure a MessageListenerContainer to cache the JMS connection since each JMS session will be created as it's needed inside a data source transaction. However, if you cache the JMS session, then the database connection for it is established when the container starts up and it will not be possible to have this cached JMS session participite in the local data source transaction.

In many application server environments the JDBC connection is wrapped in an implementation specific class that delegates to the underlying native JDBC connection. Oracle's AQ connection factory needs the native Oracle connection and will throw an "oracle.jms.AQjmsException: JMS-112: Connection is invalid" exception if the connection is wrapped by a foreign class. To solve this problem you can specify a NativeJdbcExtractor that can be used to unwrap the connection. Spring provides a number of implementations to match the application server environment. Here is an example for specifying a NativeJdbcExtractor.

    <orcl:aq-jms-connection-factory id="connectionFactory"
        use-local-data-source-transaction="true"
        native-jdbc-extractor="dbcpNativeJdbcExtractor" 
        data-source="dataSource" />

    <bean id="dbcpNativeJdbcExtractor" 
        class="org.springframework.jdbc.support.nativejdbc.CommonsDbcpNativeJdbcExtractor"/>

    <bean id="dbcpDataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
        <property name="driverClassName" value="${jdbc.driverClassName}" />
        <property name="url" value="${jdbc.url}" />
        <property name="username" value="${jdbc.username}" />
        <property name="password" value="${jdbc.password}" />
    </bean>

For some use cases the default plain ConnectionFactory does not work and you need to explicitly use a QueueConnectionFactory or a TopicConnectionFactory. To support this requirement it is possible to specify this using the connection-factory-type attribute. The default is CONNECTION but you can specify QUEUE_CONNECTION or TOPIC_CONNECTION instead. Here is an example for specifying the connection factory type.

    <orcl:aq-jms-connection-factory id="connectionFactory"
        use-local-data-source-transaction="true"
        connection-factory-type="QUEUE_CONNECTION" 
        data-source="dataSource" />

    <bean id="dbcpDataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
        <property name="driverClassName" value="${jdbc.driverClassName}" />
        <property name="url" value="${jdbc.url}" />
        <property name="username" value="${jdbc.username}" />
        <property name="password" value="${jdbc.password}" />
    </bean>

When you use a SYS.XMLType as payload there a few additional configuration settings are needed.

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/> 
        <property name="messageConverter">
            <bean id="messageConverter"> 
                    class="org.springframework.data.jdbc.jms.support.converter.oracle.XmlMessageConverter">
                <constructor-arg>
                    <bean class="org.springframework.data.jdbc.support.oracle.StringXmlTypeHandler"/> 
                </constructor-arg>
            </bean>
        </property>
    </bean>

        String xmlval = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" +
                "<product id=\"10\">\n" +
                " <description>Foo</description>\n" +
                " <price>2.05</price>\n" +
                "</product>";

        jmsTemplate.convertAndSend("jmsadmin.jms_xml_queue", xmlval);

    <bean id="messageDelegate" class="org.springframework.data.jdbc.test.xml.MessageDelegate" />

    <jms:listener-container connection-factory="connectionFactory" 
        transaction-manager="transactionManager"
        message-converter="messageConverter" 
        container-class="org.springframework.data.jdbc.jms.listener.oracle.XmlMessageListenerContainer"> 
        <jms:listener destination="jmsadmin.jms_xml_queue" 
            ref="messageDelegate" method="handleMessage">
        </jms:listener>
    </jms:listener-container>

    <bean id="messageConverter" 
            class="org.springframework.data.jdbc.jms.support.converter.oracle.XmlMessageConverter">
        <constructor-arg>
            <bean class="org.springframework.data.jdbc.support.oracle.DocumentXmlTypeHandler"/> 
        </constructor-arg>
    </bean>

public class MessageDelegate {

    @Autowired
    private DomainService domainService;

    public void handleMessage(Document xmlDoc) throws MessageConversionException, JMSException {
        domainService.processXmlMessage(xmlDoc);
    }

}

When you use a custom ADT as payload there are certain configuration settings that are needed. When creating the queue and its queue table you specify the custom type as the "queue_payload_type". This custom type is defined using a regular "CREATE TYPE" statement. In the code example that follow we have defined a PRODUCT type:

create or replace TYPE PRODUCT_TYPE AS OBJECT
(
  id INTEGER,
  description VARCHAR(50),
  price DECIMAL(12,2)
);

public interface DatumMapper {

    public Datum toDatum(Object object, Connection conn) throws SQLException;

    public Object fromDatum(Datum datum) throws SQLException;

}

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/> 
        <property name="messageConverter">
            <bean id="messageConverter" 
                    class="org.springframework.data.jdbc.jms.support.converter.oracle.MappingAdtMessageConverter">
                <constructor-arg>
                    <bean class="org.springframework.data.jdbc.jms.support.oracle.StructDatumMapper"> 
                        <constructor-arg index="0" value="JMSADMIN.PRODUCT_TYPE"/>
                        <constructor-arg index="1" value="org.springframework.data.jdbc.test.domain.Product"/>
                    </bean>
                </constructor-arg>
            </bean>
        </property>
    </bean>

        Product product = new Product();
        product.setId(22L);
        product.setDescription("Foo");
        product.setPrice(new BigDecimal("42.95"));

        jms.convertAndSend("jmsadmin.jms_product_queue", product);

    <bean id="messageDelegate" class="org.springframework.data.jdbc.test.adt.MessageDelegate" />

    <jms:listener-container connection-factory="connectionFactory" 
        transaction-manager="transactionManager"
        message-converter="messageConverter" 
        container-class="org.springframework.data.jdbc.jms.listener.oracle.AdtMessageListenerContainer"> 
        <jms:listener destination="jmsadmin.jms_product_queue" 
            ref="messageDelegate" method="handleMessage">
        </jms:listener>
    </jms:listener-container>

    <bean id="messageConverter" 
            class="org.springframework.data.jdbc.jms.support.converter.oracle.MappingAdtMessageConverter">
        <constructor-arg>
            <bean class="org.springframework.data.jdbc.jms.support.oracle.StructDatumMapper"> 
                <constructor-arg index="0" value="JMSADMIN.PRODUCT_TYPE"/>
                <constructor-arg index="1" value="org.springframework.data.jdbc.test.domain.Product"/>
            </bean>
        </constructor-arg>
    </bean>

public class MessageDelegate {

    @Autowired
    private DomainService domainService;

    public void handleMessage(Product product) throws MessageConversionException, JMSException {
        domainService.saveProduct(product);
    }

}