forketyfork

Java(TM) SE Runtime Environment, 1.7.0_51-b13
Java HotSpot(TM) 64-Bit Server VM, 24.51-b03
Mac OS X, 10.9.2, x86_64

Burning up to figure out the exact CPU count....... done!

Running with 1 threads and [-client]: 
       granularity_currentTime:  993598,116 +-    812,330 ns
          granularity_nanotime:    1007,995 +-      2,673 ns
           latency_currentTime:      42,144 +-     15,816 ns

forketyfork

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jta-atomikos</artifactId>
</dependency>

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-spring-boot-starter</artifactId>
</dependency>

forketyfork

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-sql</artifactId>
</dependency>

<dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-camel</artifactId>
</dependency>

forketyfork

<dependencyManagement>
    <dependencies>

        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-spring-boot-dependencies</artifactId>
            <version>2.24.3</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>

forketyfork

spring:
  activemq:
    # we'll use an external ActiveMQ broker
    broker-url: tcp://localhost:61616
  datasource:
    # we'll use an external PostgreSQL database
    url: jdbc:postgresql://localhost:5432/xatest
    xa:
      # this is needed to make sure the PostgreSQL data source is XA-aware
      data-source-class-name: org.postgresql.xa.PGXADataSource

forketyfork

drop table if exists message;

create table message (
  id serial primary key,
  contents varchar(256) not null
);

forketyfork

@Configuration
public class ActiveMQConfig {

    @Bean("activemq")
    public ActiveMQComponent activeMq(
            ConnectionFactory connectionFactory,
            JtaTransactionManager jtaTransactionManager) {
        ActiveMQComponent component = new ActiveMQComponent();
        component.setAcknowledgementMode(JmsProperties.AcknowledgeMode.CLIENT.getMode());
        component.setCacheLevelName("CACHE_CONSUMER");

forketyfork

@Configuration
public class TransactionConfig {

    @Bean("policyPropagationRequired")
    public SpringTransactionPolicy transactionPolicyPropagationRequired(
            @Autowired JtaTransactionManager transactionManager) {
        SpringTransactionPolicy policy = new SpringTransactionPolicy(transactionManager);
        policy.setPropagationBehaviorName("PROPAGATION_REQUIRED");
        return policy;
    }

forketyfork

@Component
public class XaRoute extends RouteBuilder {

    @Override
    public void configure() {
        from("activemq:TestQueue")
                .transacted("policyPropagationRequired")
                .to("sql:insert into message(contents) values(:#${body})")
                .log("Delay...")
                .delay(10000)

forketyfork

<dependency>
  <groupId>org.junit.jupiter</groupId>
  <artifactId>junit-jupiter-api</artifactId>
  <version>5.6.2</version>
  <scope>test</scope>
</dependency>

<dependency>
  <groupId>org.junit.jupiter</groupId>
  <artifactId>junit-jupiter-params</artifactId>