QPID-388 : hand merged the changes done in perftesting branch

QPID-395 : hand merged the changes done in perftesting branch
QPID-375 : default queue config properties should now be under <queues> tag

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

3 files changed, 35 insertions, 34 deletions

diff --git a/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java b/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java
index 7fd215d410..0483a6f20a 100644
--- a/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java
+++ b/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java
@@ -70,7 +70,7 @@ public class PingAsyncTestPerf extends PingTestPerf implements TimingControllerA
 
     /** Holds test specifics by correlation id. This consists of the expected number of messages and the timing controler. */
     private Map<String, PerCorrelationId> perCorrelationIds =
-        Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
+            Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
 
     /** Holds the batched results listener, that does logging on batch boundaries. */
     private BatchedResultsListener batchedResultsListener = null;
@@ -86,12 +86,12 @@ public class PingAsyncTestPerf extends PingTestPerf implements TimingControllerA
 
         // Sets up the test parameters with defaults.
         testParameters.setPropertyIfNull(TEST_RESULTS_BATCH_SIZE_PROPNAME,
-                                         Integer.toString(DEFAULT_TEST_RESULTS_BATCH_SIZE));
+                                              Integer.toString(DEFAULT_TEST_RESULTS_BATCH_SIZE));
     }
 
     /**
      * Compile all the tests into a test suite.
-     * @return The test suite to run. Should only contain testAsyncPingOk method.
+     * @return The test suite to run. Should only contain testAsyncPingOk method. 
      */
     public static Test suite()
     {
@@ -129,7 +129,7 @@ public class PingAsyncTestPerf extends PingTestPerf implements TimingControllerA
      * all replies have been received or a time out occurs before exiting this method.
      *
      * @param numPings The number of pings to send.
-     * @throws Exception pass all errors out to the test harness
+     * @throws Exception pass all errors out to the test harness  
      */
     public void testAsyncPingOk(int numPings) throws Exception
     {
@@ -146,7 +146,8 @@ public class PingAsyncTestPerf extends PingTestPerf implements TimingControllerA
         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;
         _logger.debug("messageCorrelationId = " + messageCorrelationId);
 
         // Initialize the count and timing controller for the new correlation id.
@@ -154,33 +155,20 @@ public class PingAsyncTestPerf extends PingTestPerf implements TimingControllerA
         TimingController tc = getTimingController().getControllerForCurrentThread();
         perCorrelationId._tc = tc;
         perCorrelationId._expectedCount = pingClient.getExpectedNumPings(numPings);
-        perCorrelationIds.put(messageCorrelationId, 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 =
-            pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),
-                                      testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME),
-                                      testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME));
+        perCorrelationIds.put(perThreadSetup._correlationId, perCorrelationId);
 
         // 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(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);
     }
 
     /**
@@ -258,9 +246,9 @@ public class PingAsyncTestPerf extends PingTestPerf implements TimingControllerA
                 // Extract the correlation id from the message.
                 String correlationId = message.getJMSCorrelationID();
 
-                _logger.debug("public void onMessage(Message message, int remainingCount = " + remainingCount
-                              + "): called on batch boundary for message id: " + correlationId + " with thread id: "
-                              + Thread.currentThread().getId());
+                _logger.debug("public void onMessage(Message message, int remainingCount = " + remainingCount +
+                              "): called on batch boundary for message id: " + correlationId +
+                              " with thread id: " + Thread.currentThread().getId());
 
                 // Get the details for the correlation id and check that they are not null. They can become null
                 // if a test times out.
diff --git a/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java b/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
index 2057f09bc7..92acd18a38 100644
--- a/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
+++ b/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
@@ -66,7 +66,7 @@ public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
     ThreadLocal<PerThreadSetup> threadSetup = new ThreadLocal<PerThreadSetup>();

 

     /** Holds a property reader to extract the test parameters from. */

-    protected ParsedProperties testParameters = new TestContextProperties(System.getProperties());

+    protected ParsedProperties testParameters = new ParsedProperties(System.getProperties());

 

     public PingTestPerf(String name)

     {

@@ -107,6 +107,9 @@ 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));

+        testParameters.setSysPropertyIfNull(PingPongProducer.PAUSE_AFTER_BATCH_PROPNAME, 0l);

     }

 

     /**

@@ -141,11 +144,11 @@ public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
 

         // Generate a sample message. This message is already time stamped and has its reply-to destination set.

         ObjectMessage msg =

-            perThreadSetup._pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),

-                                                      testParameters.getPropertyAsInteger(

-                                                          PingPongProducer.MESSAGE_SIZE_PROPNAME),

-                                                      testParameters.getPropertyAsBoolean(

-                                                          PingPongProducer.PERSISTENT_MODE_PROPNAME));

+                perThreadSetup._pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),

+                                                          testParameters.getPropertyAsInteger(

+                                                                  PingPongProducer.MESSAGE_SIZE_PROPNAME),

+                                                          testParameters.getPropertyAsBoolean(

+                                                                  PingPongProducer.PERSISTENT_MODE_PROPNAME));

 

         // start the test

         long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));

@@ -190,10 +193,12 @@ 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);

+            int pausetime = testParameters.getPropertyAsInteger(PingPongProducer.PAUSE_AFTER_BATCH_PROPNAME);

 

             // Extract the test set up paramaeters.

             int destinationscount =

-                Integer.parseInt(testParameters.getProperty(PingPongProducer.PING_DESTINATION_COUNT_PROPNAME));

+                    Integer.parseInt(testParameters.getProperty(PingPongProducer.PING_DESTINATION_COUNT_PROPNAME));

 

             // This is synchronized because there is a race condition, which causes one connection to sleep if

             // all threads try to create connection concurrently.

@@ -203,7 +208,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, pausetime);

             }

             // Start the client connection

             perThreadSetup._pingClient.getConnection().start();

@@ -255,5 +261,6 @@ public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
          * Holds the test ping client.

          */

         protected PingClient _pingClient;

+        protected String _correlationId;

     }

 }

diff --git a/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java b/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
index 3a89b1044e..d2e0367bba 100644
--- a/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
+++ b/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
@@ -111,6 +111,9 @@ 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));

+        ParsedProperties.setSysPropertyIfNull(PingPongProducer.PAUSE_AFTER_BATCH_PROPNAME, 0l);

     }

 

     /**

@@ -188,6 +191,8 @@ 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);

+            long pause = testParameters.getPropertyAsInteger(PingPongProducer.PAUSE_AFTER_BATCH_PROPNAME);

 

             synchronized (this)

             {

@@ -205,7 +210,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, pause);

                 perThreadSetup._testPingProducer.getConnection().start();

             }