From 35384de3a73e96c0fef1f585923986d54dfaa493 Mon Sep 17 00:00:00 2001
From: Gary Tully <gtully@apache.org>
Date: Fri, 26 Aug 2011 17:15:46 +0000
Subject: [PATCH] second route is the way to use a second connection and let
 spring do the caching

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1162175 13f79535-47bb-0310-9956-ffa450edef68
---
 .../activemq/camel/TransactedConsumeTest.java | 23 +++++++++++++++----
 .../activemq/camel/transactedconsume.xml      | 20 ++++++++++++----
 2 files changed, 34 insertions(+), 9 deletions(-)

diff --git a/activemq-camel/src/test/java/org/apache/activemq/camel/TransactedConsumeTest.java b/activemq-camel/src/test/java/org/apache/activemq/camel/TransactedConsumeTest.java
index a325248807..a84f96add1 100644
--- a/activemq-camel/src/test/java/org/apache/activemq/camel/TransactedConsumeTest.java
+++ b/activemq-camel/src/test/java/org/apache/activemq/camel/TransactedConsumeTest.java
@@ -16,6 +16,8 @@
  */
 package org.apache.activemq.camel;
 
+import java.io.File;
+import java.util.concurrent.atomic.AtomicLong;
 import javax.jms.Connection;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
@@ -26,6 +28,7 @@ import org.apache.activemq.broker.region.policy.PolicyEntry;
 import org.apache.activemq.broker.region.policy.PolicyMap;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.command.ActiveMQTextMessage;
+import org.apache.activemq.store.amq.AMQPersistenceAdapter;
 import org.apache.activemq.util.Wait;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
@@ -40,7 +43,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
 public class TransactedConsumeTest extends CamelSpringTestSupport {
     private static final Logger LOG = LoggerFactory.getLogger(TransactedConsumeTest.class);
     BrokerService broker = null;
-    int messageCount = 1000;
+    int messageCount = 100000;
 
     @Test
     public void testConsume() throws Exception {
@@ -53,6 +56,8 @@ public class TransactedConsumeTest extends CamelSpringTestSupport {
                 return broker.getAdminView().getTotalDequeueCount() >= messageCount;
             }
         }, 20 * 60 * 1000));
+        long duration = System.currentTimeMillis() - firstConsumed.get();
+        LOG.info("Done message consumption in " + duration + "millis");
     }
 
     private void sendJMSMessageToKickOffRoute() throws Exception {
@@ -78,12 +83,14 @@ public class TransactedConsumeTest extends CamelSpringTestSupport {
 
         PolicyMap policyMap = new PolicyMap();
         PolicyEntry defaultPolicy = new PolicyEntry();
-       // defaultPolicy.setStrictOrderDispatch(false);
         policyMap.setDefaultEntry(defaultPolicy);
         brokerService.setDestinationPolicy(policyMap);
 
         brokerService.setAdvisorySupport(false);
         brokerService.setDataDirectory("target/data");
+        AMQPersistenceAdapter amq = new AMQPersistenceAdapter();
+        amq.setDirectory(new File("target/data"));
+        brokerService.setPersistenceAdapter(amq);
         brokerService.addConnector("tcp://localhost:61616");
         return brokerService;
     }
@@ -110,13 +117,21 @@ public class TransactedConsumeTest extends CamelSpringTestSupport {
         return new ClassPathXmlApplicationContext("org/apache/activemq/camel/transactedconsume.xml");
     }
 
+    static AtomicLong firstConsumed = new AtomicLong();
+    static AtomicLong consumed = new AtomicLong();
+
     static class ConnectionLog implements Processor  {
 
         @Override
         public void process(Exchange exchange) throws Exception {
+            if (consumed.getAndIncrement() == 0) {
+                firstConsumed.set(System.currentTimeMillis());
+            }
             ActiveMQTextMessage m = (ActiveMQTextMessage) ((JmsMessage)exchange.getIn()).getJmsMessage();
-            Thread.currentThread().sleep(10);
-            LOG.info("received on " + m.getConnection().toString());
+            //Thread.currentThread().sleep(500);
+            if (consumed.get() %500 == 0) {
+                LOG.info("received on " + m.getConnection().toString());
+            }
         }
     }
 
diff --git a/activemq-camel/src/test/resources/org/apache/activemq/camel/transactedconsume.xml b/activemq-camel/src/test/resources/org/apache/activemq/camel/transactedconsume.xml
index c315a1baf6..a6b1c4a423 100644
--- a/activemq-camel/src/test/resources/org/apache/activemq/camel/transactedconsume.xml
+++ b/activemq-camel/src/test/resources/org/apache/activemq/camel/transactedconsume.xml
@@ -28,12 +28,12 @@
     <context:annotation-config/>
 
     <bean id="vhfBatchListenerJMSConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
-        <property name="brokerURL" value="tcp://localhost:61616?jms.prefetchPolicy.all=1"/>
+        <property name="brokerURL" value="tcp://localhost:61616?jms.prefetchPolicy.all=1000"/>
     </bean>
 
     <bean id="vhfBatchListenerPooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
-        <property name="maxConnections" value="10"/>
-        <property name="maximumActive" value="10"/>
+        <property name="maxConnections" value="1"/>
+        <property name="maximumActive" value="1"/>
         <property name="connectionFactory" ref="vhfBatchListenerJMSConnectionFactory"/>
     </bean>
 
@@ -66,13 +66,23 @@
         <property name="configuration" ref="vhfBatchListenerJMSConfig"/>
     </bean>
 
+    <bean id="activemq2" class="org.apache.activemq.camel.component.ActiveMQComponent">
+        <property name="configuration" ref="vhfBatchListenerJMSConfig"/>
+    </bean>
+
     <camelContext xmlns="http://camel.apache.org/schema/spring">
         <route>
             <from uri="activemq:queue:scp_transacted"/>
             <!-- transacted /-->
-            <process ref="connectonLog"/>
+            <process ref="connectionLog"/>
+        </route>
+
+        <!-- marginally better through put with a second route/connection -->
+        <route>
+            <from uri="activemq2:queue:scp_transacted"/>
+            <process ref="connectionLog"/>
         </route>
     </camelContext>
 
-    <bean id="connectonLog" class="org.apache.activemq.camel.TransactedConsumeTest.ConnectionLog"/>
+    <bean id="connectionLog" class="org.apache.activemq.camel.TransactedConsumeTest.ConnectionLog"/>
 </beans>