https://issues.apache.org/jira/browse/AMQ-3283 - transaction support for perf plugin

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1092753 13f79535-47bb-0310-9956-ffa450edef68

activemq-tooling/maven-activemq-perf-plugin/src/main/java/org/apache/activemq/tool/AbstractJmsClient.java

@@ -37,6 +37,8 @@ public abstract class AbstractJmsClient {
     protected int destCount = 1;
     protected int destIndex;
     protected String clientName = "";
+    
+    private int internalTxCounter = 0;
 
     public AbstractJmsClient(ConnectionFactory factory) {
         this.factory = factory;
@@ -159,4 +161,25 @@ public abstract class AbstractJmsClient {
         }
     }
 
+    /** 
+     * Helper method that checks if session is 
+     * transacted and whether to commit the tx based on commitAfterXMsgs 
+     * property. 
+     * 
+     * @return true if transaction was committed. 
+     * @throws JMSException in case the call to JMS Session.commit() fails.
+     */
+    public boolean commitTxIfNecessary() throws JMSException {
+    	
+    	internalTxCounter++;
+        if (getClient().isSessTransacted()) {
+        	if ((internalTxCounter % getClient().getCommitAfterXMsgs()) == 0) {
+        		LOG.debug("Committing transaction.");
+        		internalTxCounter = 0;
+        		getSession().commit();
+        		return true;
+        	}
+        }
+        return false;
+    }
 }

activemq-tooling/maven-activemq-perf-plugin/src/main/java/org/apache/activemq/tool/JmsConsumerClient.java

@@ -82,10 +82,14 @@ public class JmsConsumerClient extends AbstractJmsMeasurableClient {
 
             LOG.info("Starting to synchronously receive messages for " + duration + " ms...");
             long endTime = System.currentTimeMillis() + duration;
+
+            int counter = 0;
             while (System.currentTimeMillis() < endTime) {
                 getJmsConsumer().receive();
                 incThroughput();
+                counter++;
                 sleep();
+                commitTxIfNecessary();
             }
         } finally {
             if (client.isDurable() && client.isUnsubscribe()) {
@@ -112,6 +116,7 @@ public class JmsConsumerClient extends AbstractJmsMeasurableClient {
                 incThroughput();
                 recvCount++;
                 sleep();
+                commitTxIfNecessary();
             }
         } finally {
             if (client.isDurable() && client.isUnsubscribe()) {
@@ -132,6 +137,11 @@ public class JmsConsumerClient extends AbstractJmsMeasurableClient {
             public void onMessage(Message msg) {
                 incThroughput();
                 sleep();
+                try {
+                	commitTxIfNecessary();
+                } catch (JMSException ex) {
+                	LOG.error("Error committing transaction: " + ex.getMessage());
+                }
             }
         });
 
@@ -165,6 +175,12 @@ public class JmsConsumerClient extends AbstractJmsMeasurableClient {
                 recvCount.incrementAndGet();
                 synchronized (recvCount) {
                     recvCount.notify();
+                } 
+                
+                try {
+                	commitTxIfNecessary();
+                } catch (JMSException ex) {
+                	LOG.error("Error committing transaction: " + ex.getMessage());
                 }
             }
         });
@@ -244,6 +260,10 @@ public class JmsConsumerClient extends AbstractJmsMeasurableClient {
         client = (JmsConsumerProperties)clientProps;
     }
     
+    /**
+     * A way to throttle the consumer. Time to sleep is 
+     * configured via recvDelay property. 
+     */
     protected void sleep() {
         if (client.getRecvDelay() > 0) {
         	try {

activemq-tooling/maven-activemq-perf-plugin/src/main/java/org/apache/activemq/tool/JmsProducerClient.java

@@ -97,6 +97,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                             getJmsProducer().send(dest[j], getJmsTextMessage());
                             incThroughput();
                             sleep();
+                            commitTxIfNecessary();
                         }
                     }
                     // Send to only one actual destination
@@ -105,6 +106,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                         getJmsProducer().send(getJmsTextMessage());
                         incThroughput();
                         sleep();
+                        commitTxIfNecessary();
                     }
                 }
 
@@ -119,6 +121,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                             getJmsProducer().send(dest[j], createJmsTextMessage("Text Message [" + i + "]"));
                             incThroughput();
                             sleep();
+                            commitTxIfNecessary();
                         }
                     }
 
@@ -128,6 +131,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                         getJmsProducer().send(createJmsTextMessage("Text Message [" + i + "]"));
                         incThroughput();
                         sleep();
+                        commitTxIfNecessary();
                     }
                 }
             }
@@ -168,6 +172,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                             getJmsProducer().send(dest[j], getJmsTextMessage());
                             incThroughput();
                             sleep();
+                            commitTxIfNecessary();
                         }
                     }
                     // Send to only one actual destination
@@ -176,6 +181,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                         getJmsProducer().send(getJmsTextMessage());
                         incThroughput();
                         sleep();
+                        commitTxIfNecessary();
                     }
                 }
 
@@ -191,6 +197,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                             getJmsProducer().send(dest[j], createJmsTextMessage("Text Message [" + count++ + "]"));
                             incThroughput();
                             sleep();
+                            commitTxIfNecessary();
                         }
                     }
 
@@ -201,6 +208,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                         getJmsProducer().send(createJmsTextMessage("Text Message [" + count++ + "]"));
                         incThroughput();
                         sleep();
+                        commitTxIfNecessary();
                     }
                 }
             }

activemq-tooling/maven-activemq-perf-plugin/src/main/java/org/apache/activemq/tool/properties/JmsClientProperties.java

@@ -27,6 +27,9 @@ public class JmsClientProperties extends AbstractObjectProperties {
 
     protected String sessAckMode = SESSION_AUTO_ACKNOWLEDGE;
     protected boolean sessTransacted;
+    
+    // commit transaction after X msgs only. 
+    protected int commitAfterXMsgs = 1;
 
     protected String jmsProvider;
     protected String jmsVersion;
@@ -63,6 +66,14 @@ public class JmsClientProperties extends AbstractObjectProperties {
     public void setSessTransacted(boolean sessTransacted) {
         this.sessTransacted = sessTransacted;
     }
+    
+    public void setCommitAfterXMsgs(int commitAfterXMsg) {
+    	this.commitAfterXMsgs = commitAfterXMsg;
+    }
+    
+    public int getCommitAfterXMsgs() {
+    	return this.commitAfterXMsgs;
+    }
 
     public String getJmsProvider() {
         return jmsProvider;