QPID-388 (option to provide ack mode for tests)

changed config to include latest junit-toolkit snapshot (to get more results info in case of failure)
changed the generation of unique ping destination to enable parallel tests in different JVM

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/perftesting@511821 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 68 insertions, 35 deletions

diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingClient.java b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingClient.java
index a52b5a0c49..2a384c5a84 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingClient.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingClient.java
@@ -73,11 +73,12 @@ public class PingClient extends PingPongProducer
     public PingClient(String brokerDetails, String username, String password, String virtualpath, String destinationName,
                       String selector, boolean transacted, boolean persistent, int messageSize, boolean verbose,
                       boolean afterCommit, boolean beforeCommit, boolean afterSend, boolean beforeSend, boolean failOnce,
-                      int txBatchSize, int noOfDestinations, int rate, boolean pubsub, boolean unique) throws Exception
+                      int txBatchSize, int noOfDestinations, int rate, boolean pubsub, boolean unique,
+                      int ackMode) throws Exception
     {
         super(brokerDetails, username, password, virtualpath, destinationName, selector, transacted, persistent, messageSize,
               verbose, afterCommit, beforeCommit, afterSend, beforeSend, failOnce, txBatchSize, noOfDestinations, rate,
-              pubsub, unique);
+              pubsub, unique, ackMode);
 
         _pingClientCount++;
     }
@@ -104,5 +105,4 @@ public class PingClient extends PingPongProducer
             return _pingClientCount;
         }
     }
-
 }
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java b/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
index 6f444bd290..ce581a6582 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
@@ -144,6 +144,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
 
     public static final String UNIQUE_PROPNAME = "uniqueDests";
 
+    public static final String ACK_MODE_PROPNAME = "ackMode";
+
     /** Used to set up a default message size. */
     public static final int DEFAULT_MESSAGE_SIZE = 0;
 
@@ -218,6 +220,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
 
     public static final boolean DEFAULT_UNIQUE = true;
 
+    public static final int DEFAULT_ACK_MODE = Session.NO_ACKNOWLEDGE;
+
     /** Holds the name of the property to store nanosecond timestamps in ping messages with. */
     public static final String MESSAGE_TIMESTAMP_PROPNAME = "timestamp";
 
@@ -248,6 +252,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
     /** Determines whether this producer sends persistent messages. */
     protected boolean _persistent;
 
+    private int _ackMode = Session.NO_ACKNOWLEDGE;
+
     /** Determines what size of messages this producer sends. */
     protected int _messageSize;
 
@@ -347,7 +353,7 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
                             String destinationName, String selector, boolean transacted, boolean persistent, int messageSize,
                             boolean verbose, boolean afterCommit, boolean beforeCommit, boolean afterSend,
                             boolean beforeSend, boolean failOnce, int txBatchSize, int noOfDestinations, int rate,
-                            boolean pubsub, boolean unique) throws Exception
+                            boolean pubsub, boolean unique, int ackMode) throws Exception
     {
         _logger.debug("public PingPongProducer(String brokerDetails = " + brokerDetails + ", String username = " + username
                       + ", String password = " + password + ", String virtualpath = " + virtualpath
@@ -357,7 +363,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
                       + afterCommit + ", boolean beforeCommit = " + beforeCommit + ", boolean afterSend = " + afterSend
                       + ", boolean beforeSend = " + beforeSend + ", boolean failOnce = " + failOnce + ", int txBatchSize = "
                       + txBatchSize + ", int noOfDestinations = " + noOfDestinations + ", int rate = " + rate
-                      + ", boolean pubsub = " + pubsub + ", boolean unique = " + unique + "): called");
+                      + ", boolean pubsub = " + pubsub + ", boolean unique = " + unique
+                      + ", ackMode = " + ackMode + "): called");
 
         // Keep all the relevant options.
         _persistent = persistent;
@@ -371,6 +378,10 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
         _txBatchSize = txBatchSize;
         _isPubSub = pubsub;
         _isUnique = unique;
+        if (ackMode != 0)
+        {
+            _ackMode = ackMode;
+        }
 
         // Check that one or more destinations were specified.
         if (noOfDestinations < 1)
@@ -385,8 +396,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
         _connection = new AMQConnection(brokerDetails, username, password, clientID, virtualpath);
 
         // Create transactional or non-transactional sessions, based on the command line arguments.
-        _producerSession = (Session) getConnection().createSession(transacted, Session.AUTO_ACKNOWLEDGE);
-        _consumerSession = (Session) getConnection().createSession(transacted, Session.AUTO_ACKNOWLEDGE);
+        _producerSession = (Session) getConnection().createSession(transacted, _ackMode);
+        _consumerSession = (Session) getConnection().createSession(transacted, _ackMode);
 
         // Set up a throttle to control the send rate, if a rate > 0 is specified.
         if (rate > 0)
@@ -484,7 +495,7 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
         PingPongProducer pingProducer =
                 new PingPongProducer(brokerDetails, DEFAULT_USERNAME, DEFAULT_PASSWORD, virtualpath, destName, selector,
                                      transacted, persistent, messageSize, verbose, afterCommit, beforeCommit, afterSend,
-                                     beforeSend, failOnce, batchSize, destCount, rate, pubsub, false);
+                                     beforeSend, failOnce, batchSize, destCount, rate, pubsub, false, 0);
 
         pingProducer.getConnection().start();
 
@@ -574,31 +585,31 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
         {
             AMQDestination destination;
 
-            int id;
+            String id;
 
             // Generate an id, unique within this pinger or to the whole JVM depending on the unique flag.
             if (unique)
             {
                 _logger.debug("Creating unique destinations.");
-                id = _queueJVMSequenceID.incrementAndGet();
+                id = "_" + _queueJVMSequenceID.incrementAndGet() + "_" + _connection.getClientID();
             }
             else
             {
                 _logger.debug("Creating shared destinations.");
-                id = _queueSharedId.incrementAndGet();
+                id = "_" + _queueSharedId.incrementAndGet();
             }
 
             // Check if this is a pub/sub pinger, in which case create topics.
             if (_isPubSub)
             {
-                _logger.debug("Creating topics.");
                 destination = new AMQTopic(rootName + id);
+                _logger.debug("Creating topic " + destination);
             }
             // Otherwise this is a p2p pinger, in which case create queues.
             else
             {
-                _logger.debug("Creating queues.");
                 destination = new AMQQueue(rootName + id);
+                _logger.debug("Creating queue " + destination);
             }
 
             // Keep the destination.
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java b/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java
index 347031ff51..e0ce2b48b2 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java
@@ -145,42 +145,46 @@ public class PingAsyncTestPerf extends PingTestPerf implements TimingControllerA
         PerThreadSetup perThreadSetup = threadSetup.get();
         PingClient pingClient = perThreadSetup._pingClient;
 
+
         // Advance the correlation id of messages to send, to make it unique for this run.
-        String messageCorrelationId = Long.toString(corellationIdGenerator.incrementAndGet());
+        perThreadSetup._correlationId = Long.toString(corellationIdGenerator.incrementAndGet());
+        String messageCorrelationId = perThreadSetup._correlationId;//Long.toString(corellationIdGenerator.incrementAndGet());
         _logger.debug("messageCorrelationId = " + messageCorrelationId);
 
         // Initialize the count and timing controller for the new correlation id.
         PerCorrelationId perCorrelationId = new PerCorrelationId();
+
         TimingController tc = getTimingController().getControllerForCurrentThread();
         perCorrelationId._tc = tc;
         perCorrelationId._expectedCount = pingClient.getExpectedNumPings(numPings);
-        perCorrelationIds.put(messageCorrelationId, perCorrelationId);
-
+        perCorrelationIds.put(perThreadSetup._correlationId, perCorrelationId);
+        
         // Attach the chained message listener to the ping producer to listen asynchronously for the replies to these
         // messages.
         //pingClient.setChainedMessageListener(batchedResultsListener);
 
         // Generate a sample message of the specified size.
-        ObjectMessage msg =
+        perThreadSetup._message =
                 pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),
                                           testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME),
                                           testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME));
 
+
         // Send the requested number of messages, and wait until they have all been received.
         long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));
-        int numReplies = pingClient.pingAndWaitForReply(msg, numPings, timeout, messageCorrelationId);
+        int numReplies = pingClient.pingAndWaitForReply(perThreadSetup._message, numPings, timeout, perThreadSetup._correlationId);
 
         // Check that all the replies were received and log a fail if they were not.
         if (numReplies < perCorrelationId._expectedCount)
         {
-            tc.completeTest(false, numPings - perCorrelationId._expectedCount);
+            perCorrelationId._tc.completeTest(false, numPings - perCorrelationId._expectedCount);
         }
 
         // Remove the chained message listener from the ping producer.
         //pingClient.removeChainedMessageListener();
 
         // Remove the expected count and timing controller for the message correlation id, to ensure they are cleaned up.
-        perCorrelationIds.remove(messageCorrelationId);
+        perCorrelationIds.remove(perThreadSetup._correlationId);
     }
 
     /**
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java b/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
index 6d024a189d..ecab3876ee 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
@@ -106,6 +106,8 @@ public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
                                          PingPongProducer.DEFAULT_FAIL_BEFORE_SEND);
         testParameters.setPropertyIfNull(PingPongProducer.FAIL_ONCE_PROPNAME, PingPongProducer.DEFAULT_FAIL_ONCE);
         testParameters.setPropertyIfNull(PingPongProducer.UNIQUE_PROPNAME, PingPongProducer.DEFAULT_UNIQUE);
+        testParameters.setSysPropertyIfNull(PingPongProducer.ACK_MODE_PROPNAME,
+                                              Integer.toString(PingPongProducer.DEFAULT_ACK_MODE));
     }
 
     /**
@@ -189,6 +191,7 @@ public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
             int batchSize = testParameters.getPropertyAsInteger(PingPongProducer.COMMIT_BATCH_SIZE_PROPNAME);
             Boolean failOnce = testParameters.getPropertyAsBoolean(PingPongProducer.FAIL_ONCE_PROPNAME);
             boolean unique = testParameters.getPropertyAsBoolean(PingPongProducer.UNIQUE_PROPNAME);
+            int ackMode = testParameters.getPropertyAsInteger(PingPongProducer.ACK_MODE_PROPNAME);
 
             // Extract the test set up paramaeters.
             int destinationscount =
@@ -202,7 +205,8 @@ public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
                 perThreadSetup._pingClient = new PingClient(brokerDetails, username, password, virtualPath, destinationName,
                                                             selector, transacted, persistent, messageSize, verbose,
                                                             failAfterCommit, failBeforeCommit, failAfterSend, failBeforeSend,
-                                                            failOnce, batchSize, destinationscount, rate, pubsub, unique);
+                                                            failOnce, batchSize, destinationscount, rate, pubsub,
+                                                            unique, ackMode);
             }
             // Start the client connection
             perThreadSetup._pingClient.getConnection().start();
@@ -252,5 +256,7 @@ public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
          * Holds the test ping client.
          */
         protected PingClient _pingClient;
+        protected Message _message;
+        protected String _correlationId;
     }
 }
diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java b/qpid/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
index 8515038d0f..4121811b8e 100644
--- a/qpid/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
+++ b/qpid/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
@@ -111,6 +111,8 @@ public class PingPongTestPerf extends AsymptoticTestCase
                                               PingPongProducer.DEFAULT_FAIL_BEFORE_SEND);
         ParsedProperties.setSysPropertyIfNull(PingPongProducer.FAIL_ONCE_PROPNAME, PingPongProducer.DEFAULT_FAIL_ONCE);
         ParsedProperties.setSysPropertyIfNull(PingPongProducer.UNIQUE_PROPNAME, Boolean.toString(PingPongProducer.DEFAULT_UNIQUE));
+        ParsedProperties.setSysPropertyIfNull(PingPongProducer.ACK_MODE_PROPNAME,
+                                              Integer.toString(PingPongProducer.DEFAULT_ACK_MODE));
     }
 
     /**
@@ -188,6 +190,7 @@ public class PingPongTestPerf extends AsymptoticTestCase
             int batchSize = testParameters.getPropertyAsInteger(PingPongProducer.COMMIT_BATCH_SIZE_PROPNAME);
             Boolean failOnce = testParameters.getPropertyAsBoolean(PingPongProducer.FAIL_ONCE_PROPNAME);
             boolean unique = testParameters.getPropertyAsBoolean(PingPongProducer.UNIQUE_PROPNAME);
+            int ackMode = testParameters.getPropertyAsInteger(PingPongProducer.ACK_MODE_PROPNAME);
 
             synchronized (this)
             {
@@ -205,7 +208,8 @@ public class PingPongTestPerf extends AsymptoticTestCase
                                                                         destinationName, selector, transacted, persistent,
                                                                         messageSize, verbose, failAfterCommit,
                                                                         failBeforeCommit, failAfterSend, failBeforeSend,
-                                                                        failOnce, batchSize, 0, rate, pubsub, unique);
+                                                                        failOnce, batchSize, 0, rate, pubsub,
+                                                                        unique, ackMode);
                 perThreadSetup._testPingProducer.getConnection().start();
             }
 
diff --git a/qpid/java/pom.xml b/qpid/java/pom.xml
index f5c4a2d309..4506555769 100644
--- a/qpid/java/pom.xml
+++ b/qpid/java/pom.xml
@@ -47,7 +47,7 @@
             <url>scp://people.apache.org/www/people.apache.org/repo/m2-snapshot-repository</url>
         </snapshotRepository>
 
-	<!-- Qpid has a Wiki site, maven generated site not used. This is just so that it can be created locally for viewing the reports. -->
+    <!-- Qpid has a Wiki site, maven generated site not used. This is just so that it can be created locally for viewing the reports. -->
         <site>
           <id>Qpid_Site</id>
           <name>Qpid Site</name>
@@ -97,17 +97,17 @@
         <java.source.version>1.5</java.source.version>
         <compile.flags>-Xlint:fallthrough,finally</compile.flags>
 
-	<!--
+    <!--
           This should always point to a default minimal log4j configuration that all developers are happy with as a useable default. To use your own
           log4j preferences set up an alternative in your settings.xml and avoid corrupting the default with private preferences.
-	  -->
-	<!--<log4j.configuration>file:/${topDirectoryLocation}/etc/log4j.xml</log4j.configuration>-->
+      -->
+    <!--<log4j.configuration>file:/${topDirectoryLocation}/etc/log4j.xml</log4j.configuration>-->
         <amqj.logging.level>warn</amqj.logging.level> <!-- This is referenced in the default log4j.xml -->
 
         <!--Versions for various plugins and features -->
         <antrun.version>1.2-SNAPSHOT</antrun.version>
         <!--<assembly.version>2.2-SNAPSHOT</assembly.version>-->
-	<assembly.version>2.1</assembly.version>
+    <assembly.version>2.1</assembly.version>
         <cobertura.version>2.0</cobertura.version>
         <compiler.version>2.0.1</compiler.version>
         <dependency.plugin.version>1.0</dependency.plugin.version>
@@ -288,12 +288,12 @@
                         <argLine>-ea</argLine>
                         <systemproperties>
                             <property>
-			        <name>amqj.logging.level</name>
+                    <name>amqj.logging.level</name>
                                 <value>${amqj.logging.level}</value>
-				<!--
-				<name>log4j.configuration</name>
-				<value>${log4j.configuration}</value>
-				-->
+                <!--
+                <name>log4j.configuration</name>
+                <value>${log4j.configuration}</value>
+                -->
                             </property>
                         </systemproperties>
                     </configuration>
@@ -351,8 +351,8 @@
                 <plugin>
                     <groupId>uk.co.thebadgerset</groupId>
                     <artifactId>junit-toolkit-maven-plugin</artifactId>
-                    <version>0.5-SNAPSHOT</version>
-		</plugin>
+                    <version>0.5</version>
+        </plugin>
 
             </plugins>
         </pluginManagement>
@@ -457,7 +457,7 @@
             <dependency>
                 <groupId>uk.co.thebadgerset</groupId>
                 <artifactId>junit-toolkit</artifactId>
-                <version>0.5-SNAPSHOT</version>
+                <version>0.6-SNAPSHOT</version>
                 <scope>compile</scope>
             </dependency>
 
@@ -545,6 +545,14 @@
           </snapshots>
         </repository>
         -->
+        <repository>
+              <id>sourceforge.snapshots</id>
+              <name>Sourceforge SNAPSHOT Repository</name>
+              <url>http://junit-toolkit.svn.sourceforge.net/svnroot/junit-toolkit/snapshots</url>
+              <snapshots>
+                <enabled>true</enabled>
+              </snapshots>
+        </repository>
     </repositories>
 
     <pluginRepositories>