Red Hat Training

A Red Hat training course is available for Red Hat Fuse

10.4. Define a Transactional Route

Overview

This section describes how to create a transactional route and package it as an OSGi bundle. The route described here is based on the AccountService class (see Appendix A), implementing a transfer of funds from one account to another, where the account data is stored in an Apache Derby database instance.

Database schema

The database schema for the accounts consists of just two columns: the name column (identifying the account holder) and the amount column (specifying the amount of money left in the account). Formally, the schema is defined by the following SQL command: 
CREATE TABLE accounts (name VARCHAR(50), amount INT);

Sample incoming message

The following XML snippet demonstrates the format of a typical message that is processed by the route: 
<transaction>
  <transfer>
    <sender>Major Clanger</sender>
    <receiver>Tiny Clanger</receiver>
    <amount>90</amount>
  </transfer>
</transaction>
The message requests a transfer of money from one account to another. It specifies that 90 units should be subtracted from the Major Clanger account and 90 units should be added to the Tiny Clanger account.

The transactional route

The incoming messages are processed by the following transactional route: 
<route>
  <from uri="jmstx:queue:giro">
  <bean ref="accountService" method="credit">
  <bean ref="accountService" method="debit">
  <bean ref="accountService" method="dumpTable">
  <to uri="jmstx:queue:statusLog">
</route>
Money is transferred by calling the AccountService.credit and AccountService.debit bean methods (which access the Derby database). The AccountService.dumpTable method then dumps the complete contents of the database table into the current exchange and the route sends this to the statusLog queue.

Provoking a transaction rollback

The AccountService.debit method imposes a limit of 100 on the amount that can be withdrawn from any account and throws an exception if this limit is exceeded. This provides a simple means of provoking a transaction rollback, by sending a message containing a transfer request that exceeds 100.

Steps to define a transactional route

Perform the following steps to define a route that uses XA to coordinate global transactions across a JMS XA resource and an Apache Derby XA resource:
  1. Use the quickstart archetype to create a basic Maven project for the route bundle. Open a new command prompt, change directory to a convenient location, and enter the following command: 
    mvn archetype:generate
  -DarchetypeArtifactId=maven-archetype-quickstart
  -DgroupId=org.fusesource.example
  -DartifactId=tx-xa
    The preceding command creates a new Maven project for the org.fusesource.example/tx-xa artifact under the tx-xa directory.
  2. Change the project packaging type to bundle. Under the tx-xa directory, open the pom.xml file with a text editor and change the contents of the packaging element from jar to bundle, as shown in the following highlighted line: 
    <project ...>
  ...
  <groupId>org.fusesource.example</groupId>
  <artifactId>tx-xa</artifactId>
  <version>1.0-SNAPSHOT</version>
  <packaging>bundle</packaging>
  ...
</project>
  3. Add the bundle configuration to the POM. In the pom.xml file, add the following build element as a child of the project element: 
    <project ...>
  ...
  <build>
    <defaultGoal>install</defaultGoal>

    <plugins>

      <plugin>
        <groupId>org.apache.felix</groupId>
        <artifactId>maven-bundle-plugin</artifactId>
        <extensions>true</extensions>
        <configuration>
          <instructions>
            <Bundle-SymbolicName>${project.groupId}.${project.artifactId}</Bundle-SymbolicName>
            <Import-Package>
              org.springframework.core,
              org.apache.camel,
              org.apache.camel.component.jms,
              org.apache.activemq,
              org.apache.activemq.xbean,
              org.apache.activemq.pool,
              org.apache.xbean.spring,
              org.apache.commons.pool,
              *
            </Import-Package>
            <Private-Package>
              org.fusesource.example.*
            </Private-Package>
            <DynamicImport-Package>
              org.apache.activemq.*
            </DynamicImport-Package>
          </instructions>
        </configuration>
      </plugin>
      
    </plugins>
  </build>

</project>
  4. Customize the Maven compiler plug-in to enforce JDK 1.7 coding syntax. In the pom.xml file, add the following plugin element as a child of the plugins element, to configure the Maven compiler plug-in: 
    <project ...>
  ...
  <build>
    <defaultGoal>install</defaultGoal>

    <plugins>

      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <configuration>
          <source>1.7</source>
          <target>1.7</target>
        </configuration>
      </plugin>
      ...
    </plugins>
  </build>

</project>
  5. Add the JBoss Fuse Bill of Materials (BOM) as the parent POM. The JBoss Fuse BOM defines version properties (for example, camel-version, spring-version, and so on) for all of the JBoss Fuse components, which makes it easy to specify the correct versions for the Maven dependencies. Add the following parent element near the top of your POM and (if necessary) customize the version of the BOM: 
    <project ...>
  <modelVersion>4.0.0</modelVersion>

  <parent>
    <groupId>org.jboss.fuse.bom</groupId>
    <artifactId>jboss-fuse-parent</artifactId>
    <version>6.3.0.redhat-xxx</version>
  </parent>
  ...
</project>
  6. Add the required Maven dependencies to the POM and specify the derby-version property. In the pom.xml file, add the following elements as children of the project element: 
    <project ...>
  ...
  <name>Global transactions demo</name>
  <url>redhat.com</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <derby-version>10.10.1.1</derby-version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>

   <dependency>
      <groupId>org.apache.camel</groupId>
      <artifactId>camel-core</artifactId>
      <version>${camel-version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.camel</groupId>
      <artifactId>camel-blueprint</artifactId>
      <version>${camel-version}</version>
    </dependency>
    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-log4j12</artifactId>
      <version>${slf4j-version}</version>
    </dependency>
    <dependency>
      <groupId>log4j</groupId> 
      <artifactId>log4j</artifactId> 
      <version>${log4j-version}</version> 
    </dependency>
    
    <!-- Spring transaction dependencies -->
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-core</artifactId>
      <version>${spring-version}</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-context</artifactId>
      <version>${spring-version}</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-tx</artifactId>
      <version>${spring-version}</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-aop</artifactId>
      <version>${spring-version}</version>
    </dependency>
    
    <!-- Spring JDBC adapter -->
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-jdbc</artifactId>
      <version>${spring-version}</version>
    </dependency>

    <!-- Database dependencies -->
    <dependency>
      <groupId>org.apache.derby</groupId>
      <artifactId>derby</artifactId>
      <version>${derby-version}</version>
    </dependency>

    <!-- JMS/ActiveMQ artifacts -->
    <dependency> 
      <groupId>org.apache.camel</groupId> 
      <artifactId>camel-jms</artifactId> 
      <version>${camel-version}</version>
    </dependency> 
    <dependency> 
      <groupId>org.apache.activemq</groupId> 
      <artifactId>activemq-camel</artifactId> 
      <version>${activemq-version}</version> 
    </dependency>
    <!--  This is needed by the camel-jms component -->
    <dependency> 
      <groupId>org.apache.xbean</groupId> 
      <artifactId>xbean-spring</artifactId> 
      <version>${xbean-version}</version> 
    </dependency>
  </dependencies>
  ...
</project>
    Important
    Remember to customize the derby-version property to the version of Derby you are using.
  7. Define the AccountService class. Under the tx-xa project directory, create the following directory: 
    src/main/java/org/fusesource/example/tx/xa
    Create the file, AccountService.java, in this directory and add the contents of the listing from Example B.1, “The AccountService Class” to this file.
  8. Define the beans and resources needed by the route in a Blueprint XML file. Under the tx-xa project directory, create the following directory: 
    src/main/resources/OSGI-INF/blueprint
    Using a text editor, create the file, beans.xml, in this directory and add the following contents to the file: 
    <?xml version="1.0" encoding="UTF-8"?>
<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
        
    <!--
        JMS non-TX endpoint configuration
    -->
    <bean id="jms" class="org.apache.camel.component.jms.JmsComponent"> 
        <property name="configuration" ref="jmsConfig"> 
    </bean> 
    
    <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration"> 
        <property name="connectionFactory" ref="jmsPoolConnectionFactory"> 
    </bean>
    
    <!-- connection factory wrapper to support pooling -->
    <bean id="jmsPoolConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
        <property name="connectionFactory" ref="jmsConnectionFactory">
    </bean>
    
    <bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="vm:amq">
        <property name="userName" value="UserName">
        <property name="password" value="Password">
    </bean>


    <!--
        OSGi TM Service
    -->
    <!-- access through Spring's PlatformTransactionManager -->
    <reference id="osgiPlatformTransactionManager"
               interface="org.springframework.transaction.PlatformTransactionManager">
    <!-- access through javax TransactionManager -->
    <reference id="recoverableTxManager"
        interface="org.apache.geronimo.transaction.manager.RecoverableTransactionManager">
    
    <!--
        JMS TX endpoint configuration
    -->
    <bean id="jmstx" class="org.apache.activemq.camel.component.ActiveMQComponent"> 
        <property name="configuration" ref="jmsTxConfig"> 
    </bean> 
    
    <bean id="jmsTxConfig" class="org.apache.camel.component.jms.JmsConfiguration"> 
        <property name="connectionFactory" ref="jmsXaPoolConnectionFactory"> 
        <property name="transactionManager" ref="osgiPlatformTransactionManager"> 
        <property name="transacted" value="false">
        <property name="cacheLevelName" value="CACHE_CONNECTION">
    </bean> 
    
    <!-- connection factory wrapper to support auto-enlisting of XA resource -->
    <bean id="jmsXaPoolConnectionFactory" class="org.apache.activemq.pool.JcaPooledConnectionFactory">
        <!-- Defines the name of the broker XA resource for the Aries txn manager -->
        <property name="name" value="amq-broker">
        <property name="maxConnections" value="1">
        <property name="connectionFactory" ref="jmsXaConnectionFactory">
        <property name="transactionManager" ref="recoverableTxManager">
    </bean>
    
    <bean id="jmsXaConnectionFactory" class="org.apache.activemq.ActiveMQXAConnectionFactory">
        <property name="brokerURL" value="vm:amq">
        <property name="userName" value="UserName">
        <property name="password" value="Password">
        <property name="redeliveryPolicy">
            <bean class="org.apache.activemq.RedeliveryPolicy">
                <property name="maximumRedeliveries" value="0">
            </bean>
        </property>
    </bean>
    
    <!--
        ActiveMQ XA Resource Manager
    -->
    <bean id="resourceManager" class="org.apache.activemq.pool.ActiveMQResourceManager" init-method="recoverResource">
        <property name="transactionManager" ref="recoverableTxManager">
        <property name="connectionFactory" ref="jmsXaConnectionFactory">
        <property name="resourceName" value="activemq.default">
    </bean>
    
    <!--
        Import Derby data sources as OSGi services
    -->
    <reference id="derbyXADataSource"
               interface="javax.sql.DataSource"
               filter="(datasource.name=derbyXADB)">
    <reference id="derbyDataSource"
               interface="javax.sql.DataSource"
               filter="(datasource.name=derbyDB)">
    

    <!-- bean for account business logic -->
    <bean id="accountService" class="org.fusesource.example.tx.xa.AccountService">
        <property name="dataSource" ref="derbyXADataSource">
    </bean>

</blueprint>
    Important
    In the jmsConnectionFactory bean and in the jmsXaConnectionFactory bean, you must customize the JAAS user credentials, UserName and Password, that are used to log into the broker. You can use any JAAS user with the admin role (usually defined in the etc/users.properties file of your JBoss Fuse installation).
  9. Define the transactional route. In the src/main/resources/OSGI-INF/blueprint directory, create the new file, camelContext.xml, and add the following contents: 
    <?xml version="1.0" encoding="UTF-8"?>
<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xmlns:camel="http://camel.apache.org/schema/blueprint">

  <camelContext xmlns="http://camel.apache.org/schema/blueprint" trace="false">
    <!-- Transactional route -->
    <route>
      <from uri="jmstx:queue:giro">
      <bean ref="accountService" method="credit">
      <bean ref="accountService" method="debit">
      <bean ref="accountService" method="dumpTable">
      <to uri="jmstx:queue:statusLog">
    </route>
    
    <!-- Feeder route -->
    <route>
      <from uri="file:PathNameToMsgDir">
      <to uri="jms:queue:giro">
    </route>
  </camelContext>

</blueprint>
    Important
    Replace PathNameToMsgDir with the absolute path name of a temporary directory. When the application is running, you will use this directory as a convenient way of feeding XML messages into the route.
  10. To build the tx-xa bundle and install it in the local Maven repository, enter the following Maven command from the tx-xa directory: 
    mvn install