Transaction Guide

public interface TransactionStatus extends SavepointManager {
    boolean isNewTransaction();

    boolean hasSavepoint();

    void setRollbackOnly();

    boolean isRollbackOnly();

    void flush();

    boolean isCompleted();
}

<camel:camelContext trace="false" id="My.impl.CamelContext">
    <camel:route id="Queue2Backend">
        <camel:from uri="activemq:queue:TEST.amq.failover.inputQueue" /> 
        <camel:log message="Sending message to backend..."/>
        <camel:to uri="activemq:queue:TEST.amq.failover.backendQueue" />
    </camel:route>
</camel:camelContext>

<bean id="jmsConectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
    <property name="connectionFactory">
        <bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
           <property name="brokerURL" value="failover:(tcp://localhost:61616?wireFormat.
                  tightEncodingEnabled=false)?initialReconnectDelay=10000&useExponentialBackOff=false
                  &maxReconnectDelay=10000"/>
           <property name="userName" value="admin"/>
           <property name="password" value="admin"/>
       </bean>
    </property>
    <property name="maxConnections" value="1"/>
</bean>

<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
    <property name="connectionFactory" ref="jmsConectionFactory"/>
    <property name="lazyCreateTransactionManager" value="false"/>
    <property name="transacted" value="true"/>
    <property name="concurrentConsumers" value="1"/>
    <property name="cacheLevelName" value="CACHE_CONSUMER"/>
</bean>

<bean id="prefetchPolicy" class="org.apache.activemq.ActiveMQPrefetchPolicy">
    <property name="queuePrefetch" value="10"/>
</bean>

<bean id="redeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
    <property name="maximumRedeliveries" value="1"/>
</bean>

<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
    <property name="configuration" ref="jmsConfig"/>
</bean>

from("jmstx:queue:rawStockQuotes")
    .process(myFormatter)
    .to("jmstx:queue:formattedStockQuotes");

from("jmstx:queue:rawStockQuotes?disableReplyTo=true")
    .process(myFormatter)
    .to("jmstx:queue:formattedStockQuotes");

from("jmstx:queue:rawPayments")
    .process(inputReformatter)
    .to("jmstx:queue:formattedPayments")
    .process(outputReformatter);

Figure 3.2. Transactional JMS Route that Processes InOut Exchanges

from("jmstx:queue:rawPaymentsIn")
    .process(inputReformatter)
    .to("jmstx:queue:formattedPaymentsIn");

from("jmstx:queue:formattedPaymentsOut")
    .process(outputReformatter)
    .to("jmstx:queue:rawPaymentsOut");

Figure 3.3. Pair of Transactional JMS Routes that Support Request/Reply Semantics

from("jmstx:queue:inOutSource")
    .to(ExchangePattern.InOnly, "jmstx:queue:log")
    .process(myProcessor);

// Java
int origAmount = jdbc.queryForInt(
    "select amount from accounts where name = ?",
    new Object[]{name}
);

// Java
Map<String,Object> rowMap = jdbc.queryForMap(
    "select * from accounts where name = ?",
    new Object[]{name}
);

// Java
List<Map<String,Object> > rows = jdbc.queryForList(
    "select * from accounts"
);

// Java
List<Account> accountList = jdbc.query(
    "select * from accounts",
    new Object[]{},
    new RowMapper() {
        public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
            Account acc = new Account();
            acc.setName(rs.getString("name"));
            acc.setAmount(rs.getLong("amount"));
            return acc;
        }
    }
);

// Java
jdbc.update(
    "update accounts set amount = ? where name = ?",
    new Object[] {newAmount, name}
);

// Java
jdbc.execute("create table accounts (name varchar(50), amount int)");

<transaction>
  <transfer>
    <sender>Major Clanger</sender>
    <receiver>Tiny Clanger</receiver>
    <amount>90</amount>
  </transfer>
</transaction>

// Java
package com.fusesource.demo.tx.jdbc.java;

import javax.sql.DataSource;

import org.apache.log4j.Logger;
import org.springframework.jdbc.core.JdbcTemplate;

public class CreateTable {
    private static Logger log = Logger.getLogger(CreateTable.class);

    protected DataSource dataSource;
    protected JdbcTemplate jdbc;

    public DataSource getDataSource() {
        return dataSource;
    }

    public void setDataSource(DataSource dataSource) {
        this.dataSource = dataSource;
    }
    
    public CreateTable(DataSource ds) {
        log.info("CreateTable constructor called");
        setDataSource(ds);
        setUpTable();
    }
    
    public void setUpTable() {
        log.info("About to set up table...");
        jdbc = new JdbcTemplate(dataSource);
        jdbc.execute("create table accounts (name varchar(50), amount int)");
        jdbc.update("insert into accounts (name,amount) values (?,?)",
                new Object[] {"Major Clanger", 2000}
        );
        jdbc.update("insert into accounts (name,amount) values (?,?)",
                new Object[] {"Tiny Clanger", 100}
        );
        log.info("Table created");
    }
}

<beans ...>
    <!-- datasource to the database -->
    <bean id="dataSource" class="org.springframework.jdbc.datasource.SimpleDriverDataSource">
        <property name="driverClass" value="org.hsqldb.jdbcDriver"/>
        <property name="url" value="jdbc:hsqldb:mem:camel"/>
        <property name="username" value="sa"/>
        <property name="password" value=""/>
    </bean>

    <!--  Bean to initialize table in the DB -->
    <bean id="createTable" class="com.fusesource.demo.tx.jdbc.java.CreateTable">
        <constructor-arg ref="dataSource" />
    </bean>
    ...
</beans>

package com.fusesource.demo.tx.jdbc.java;

import java.util.List;

import javax.sql.DataSource;


import org.apache.camel.Exchange;
import org.apache.camel.language.XPath;
import org.apache.log4j.Logger;
import org.springframework.jdbc.core.JdbcTemplate;


public class AccountService {
    private static Logger log = Logger.getLogger(AccountService.class);
    private JdbcTemplate jdbc;

    public AccountService() {
    }

    public void setDataSource(DataSource ds) {
        jdbc = new JdbcTemplate(ds);
    }
    ...
    // Service methods (see below)
    ...
}

<beans ...>
    <!-- Bean for account service -->
    <bean id="accountService" class="com.fusesource.demo.tx.jdbc.java.AccountService">
        <property name="dataSource" ref="dataSource"/>
    </bean>
    ...
</beans>

    public void credit(
            @XPath("/transaction/transfer/receiver/text()") String name, 1
            @XPath("/transaction/transfer/amount/text()") String amount
            )
    {
        log.info("credit() called with args name = " + name + " and amount = " + amount);
        int origAmount = jdbc.queryForInt( 2
                "select amount from accounts where name = ?",
                new Object[]{name}
        );
        int newAmount = origAmount + Integer.parseInt(amount);
        
        jdbc.update( 3
                "update accounts set amount = ? where name = ?",
                new Object[] {newAmount, name}
        );
    }

    public void debit(
            @XPath("/transaction/transfer/sender/text()") String name, 1
            @XPath("/transaction/transfer/amount/text()") String amount
            )
    {
        log.info("debit() called with args name = " + name + " and amount = " + amount);
        int iamount = Integer.parseInt(amount);
        if (iamount > 100) { 2
            throw new IllegalArgumentException("Debit limit is 100");
        }
        int origAmount = jdbc.queryForInt(
                "select amount from accounts where name = ?",
                new Object[]{name}
        );
        int newAmount = origAmount - Integer.parseInt(amount);
        if (newAmount < 0) { 3
            throw new IllegalArgumentException("Not enough in account");
        }
        
        jdbc.update(
                "update accounts set amount = ? where name = ?",
                new Object[] {newAmount, name}
        );
    }

    public void dumpTable(Exchange ex) {
        log.info("dump() called");
        List<?> dump = jdbc.queryForList("select * from accounts");
        ex.getIn().setBody(dump.toString());
    }

// Java
import org.apache.camel.spring.SpringRouteBuilder;

public class MyRouteBuilder extends SpringRouteBuilder {
    ...
    public void configure() {
        from("file:src/data?noop=true")
          .transacted()
          .beanRef("accountService","credit")
          .beanRef("accountService","debit");
    }
}

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" ... >

  <camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
      <from uri="file:src/data?noop=true"/>
      <transacted/>
      <bean ref="accountService" method="credit"/>
      <bean ref="accountService" method="debit"/>
    </route>
  </camelContext>

</beans>

// Java
import org.apache.camel.spring.SpringRouteBuilder;

public class MyRouteBuilder extends SpringRouteBuilder {
    ...
    public void configure() {
        from("file:src/data?noop=true")
          .beanRef("accountService","credit")
          .transacted()  // <-- WARNING: Transaction started in the wrong place!
          .beanRef("accountService","debit");
    }
}

// Java
import org.apache.camel.spring.SpringRouteBuilder;

public class MyRouteBuilder extends SpringRouteBuilder {
    ...
    public void configure() {
        from("file:src/data?noop=true")
          .transacted()
          .threads(3)  // WARNING: Subthreads are not in transaction scope!
          .beanRef("accountService","credit")
          .beanRef("accountService","debit");
    }
}

// Java
import org.apache.camel.spring.SpringRouteBuilder;

public class MyRouteBuilder extends SpringRouteBuilder {
    ...
    public void configure() {
        from("file:src/data?noop=true")
            .transacted()
            .beanRef("accountService","credit")
            .choice().when(xpath("/transaction/transfer[amount > 100]"))
                .to("direct:txbig")
            .otherwise()
                .to("direct:txsmall");

        from("direct:txbig")
            .beanRef("accountService","debit")
            .beanRef("accountService","dumpTable")
            .to("file:target/messages/big");

        from("direct:txsmall")
            .beanRef("accountService","debit")
            .beanRef("accountService","dumpTable")
            .to("file:target/messages/small");
    }
}

from("file:src/data?noop=true")
    .transacted()
    .to("jmstx:queue:credits")
    .to("jmstx:queue:debits");

from("jmstx:queue:giro")
    .to("jmstx:queue:credits")
    .to("jmstx:queue:debits");

<beans ...>
  <bean id="jmstx" 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="jmsConnectionFactory"/> 
      <property name="transactionManager" ref="jmsTransactionManager"/> 
      <property name="transacted" value="true"/> 
  </bean>
  ...
</beans>

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" ... >

  <camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
      <from uri="jmstx:queue:giro"/>
      <to uri="jmstx:queue:credits"/>
      <to uri="jmstx:queue:debits"/>
    </route>
  </camelContext>

</beans>

from("jmstx:queue:giro")
    .transacted()
    .to("jmstx:queue:credits")
    .to("jmstx:queue:debits");

from("file:src/data?noop=true")
    .transacted("PROPAGATION_REQUIRES_NEW")
    .beanRef("accountService","credit")
    .beanRef("accountService","debit")
    .to("file:target/messages");

<beans ...>
  <bean id="PROPAGATION_MANDATORY "class="org.apache.camel.spring.spi.SpringTransactionPolicy">
    <property name="transactionManager" ref="txManager"/>
    <property name="propagationBehaviorName" value="PROPAGATION_MANDATORY"/>
  </bean>
  ...
</beans>

from("file:src/data?noop=true")
    .transacted()
    .beanRef("accountService","credit")
    .transacted("PROPAGATION_NEVER")
    .beanRef("accountService","debit");

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" ... >

  <camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
      <from uri="file:src/data?noop=true"/>
      <transacted/>
      <bean ref="accountService" method="credit"/>
      <transacted ref="PROPAGATION_NEVER"/>
      <bean ref="accountService" method="debit"/>
    </route>
  </camelContext>

</beans>

org.apache.aries.transaction.manager
org.apache.aries.transaction.wrappers
org.apache.aries.transaction.blueprint

JBossFuse:karaf@root> features:install transaction

EsbInstallDir/etc/org.apache.aries.transaction.cfg

aries.transaction.timeout=600
aries.transaction.howl.logFileDir=${karaf.data}/txlog/

<?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"
            xsi:schemaLocation="http://www.osgi.org/xmlns/blueprint/v1.0.0"
            default-activation="lazy">
  
    <!--
        OSGi TM Service
    -->
    <!-- access through Spring's PlatformTransactionManager -->
    <reference id="osgiPlatformTransactionManager"
               interface="org.springframework.transaction.PlatformTransactionManager"/>
    <!-- access through PlatformTransactionManager -->
    <reference id="osgiJtaTransactionManager"
               interface="javax.transaction.TransactionManager"/>

</blueprint>

<dependency>
   <groupId>org.apache.activemq</groupId>
   <artifactId>activemq-jms-pool</artifactId>
   <version>x.x.x</version>  <!-- Use same version as your activemq core  -->
<dependency>

<beans ...>

    <!--
        OSGi TM Service
    -->
    <!-- access through Spring's PlatformTransactionManager -->
    <reference id="osgiPlatformTransactionManager"
               interface="org.springframework.transaction.PlatformTransactionManager">
    <!-- access through JTA TransactionManager -->
    <reference id="osgiJtaTransactionManager"
               interface="javax.transaction.TransactionManager">
    <reference id="recoverableTxManager"
        interface="org.apache.geronimo.transaction.manager.RecoverableTransactionManager">

</beans>

tcp://brokerhost:61616?jms.rmIdFromConnectionId=true

// Java
ActiveMQXAConnectionFactory cf = new ActiveMQXAConnectionFactory( ... );
cf.setRmIdFromConnectionId(true);

<!--
    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">
    <property name="rmIdFromConnectionId" value="true">
</bean>

// Java
import org.apache.activemq.ActiveMQXAConnectionFactory

import javax.jms.XAConnection;
import javax.jms.XASession;
import javax.jms.XATopicConnection;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
...
ActiveMQXAConnectionFactory cf = new ActiveMQXAConnectionFactory("tcp://brokerhost:61616?jms.rmIdFromConnectionId=true");
... // Configure other connection factory options (not shown)

XAConnection connection1 = (XAConnection)cf.createConnection();
XASession session1 = connection1.createXASession();
XAResource resource1 = session1.getXAResource();

XAConnection connection2 = (XAConnection)cf.createConnection();
XASession session2 = connection2.createXASession();
XAResource resource2 = session2.getXAResource();
...
// Send messages using 'connection1' AND connection2' in this thread
...
// Commit transaction => transaction manager sends xa.end() to BOTH XAResource objects

JBossFuse:karaf@root> osgi:ls 229

Derby XA data source (229) provides:
------------------------------------
datasource.name = derbyXADB
objectClass = javax.sql.XADataSource
osgi.jndi.service.name = jdbc/derbyXADB
osgi.service.blueprint.compname = derbyXADataSource
service.id = 423
----
aries.xa.aware = true
aries.xa.name = derbyDS
datasource.name = derbyXADB
objectClass = javax.sql.DataSource
osgi.jndi.service.name = jdbc/derbyXADB
osgi.service.blueprint.compname = derbyXADataSource
service.id = 424
----
...

// Java
import org.apache.camel.builder.RouteBuilder;

public class MyRouteBuilder extends RouteBuilder {
    ...
    public void configure() {
        from("jmstx:queue:giro")
          .beanRef("accountService","credit")
          .beanRef("accountService","debit");
    }
}

// Java
import javax.transaction.UserTransaction;
...
UserTransaction ut =
    (UserTransaction) camelContext.getRegistry().lookup(UserTransaction.class.getName());

<blueprint  xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
            xmlns:tx="http://aries.apache.org/xmlns/transactions/v1.1.0">

    <bean id="accountFoo" class="org.fusesource.example.Account">
        <tx:transaction method="*" value="Required"/>
        <property name="accountName" value="Foo"/>
    </bean>

    <bean id="accountBar" class="org.fusesource.example.Account">
        <tx:transaction method="*" value="Required"/>
        <property name="accountName" value="Bar"/>
    </bean>

</blueprint>

<blueprint  xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
            xmlns:tx="http://aries.apache.org/xmlns/transactions/v1.1.0">

    <tx:transaction bean="account*" value="Required"/>

    <bean id="accountFoo" class="org.fusesource.example.Account">
        <property name="accountName" value="Foo"/>
    </bean>

    <bean id="accountBar" class="org.fusesource.example.Account">
        <property name="accountName" value="Bar"/>
    </bean>

</blueprint>

http://db.apache.org/derby/derby_downloads.html

REM Set Apache Derby environment on Windows
SET DERBY_HOME=DerbyInstallDir

SET PATH=%DERBY_HOME%\bin;%PATH%

# Set Apache Derby environment on *NIX
DERBY_HOME=DerbyInstallDir

export PATH=$DERBY_HOME/bin:$PATH

derby.system.home=DerbySystemDirectory

CREATE TABLE accounts (name VARCHAR(50), amount INT);

<transaction>
  <transfer>
    <sender>Major Clanger</sender>
    <receiver>Tiny Clanger</receiver>
    <amount>90</amount>
  </transfer>
</transaction>

<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>