Updated perftests to include an Asynchronous ping sender

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

4 files changed, 118 insertions, 97 deletions

diff --git a/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java b/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java
index 33207790e7..34938a64bf 100644
--- a/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java
+++ b/java/perftests/src/main/java/org/apache/qpid/ping/AbstractPingProducer.java
@@ -265,6 +265,7 @@ public abstract class AbstractPingProducer implements Runnable, ExceptionListene
         _logger.trace("Batch time reached");

         if (_failAfterSend)

         {

+            _logger.trace("Batch size reached");

             if (_failOnce)

             {

                 _failAfterSend = false;

diff --git a/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java b/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java
index 35320d8b55..9a2e2c7e8c 100644
--- a/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java
+++ b/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java
@@ -20,6 +20,7 @@ package org.apache.qpid.ping;
 import javax.jms.JMSException;
 import javax.jms.MessageConsumer;
 import javax.jms.ObjectMessage;
+import javax.jms.MessageListener;
 
 import org.apache.log4j.Logger;
 
@@ -41,6 +42,7 @@ public class TestPingItself extends PingPongProducer
      * If queueCount is > 1 : This creats a client for tests with multiple queues. Creates as many consumer instances
      * as there are queues, each listening to a Queue. A producer is created which picks up a queue from
      * the list of queues to send message
+     *
      * @param brokerDetails
      * @param username
      * @param password
@@ -81,6 +83,25 @@ public class TestPingItself extends PingPongProducer
         }
     }
 
+
+    /**
+     * Sets the replyQueue to be the same as ping queue.
+     */
+    @Override
+    public void createConsumer(String selector) throws JMSException
+    {
+        // Create a message consumer to get the replies with and register this to be called back by it.
+        setReplyQueue(getPingQueue());
+        MessageConsumer consumer =
+                getConsumerSession().createConsumer(getReplyQueue(), PREFETCH, false, EXCLUSIVE, selector);
+        consumer.setMessageListener(this);
+    }
+
+    public void setMessageListener(MessageListener messageListener) throws JMSException
+    {
+        getConsumerSession().setMessageListener(messageListener);
+    }
+
     /**
      * Starts a ping-pong loop running from the command line.
      *
@@ -147,10 +168,10 @@ public class TestPingItself extends PingPongProducer
         }
 
         // Create a ping producer to handle the request/wait/reply cycle.
-        TestPingItself pingItself =
-            new TestPingItself(brokerDetails, "guest", "guest", virtualpath, queue, null, transacted, persistent,
-                               messageSize, verbose, afterCommit, beforeCommit, afterSend, beforeSend, failOnce, batchSize,
-                               queueCount, rate);
+        TestPingItself pingItself = new TestPingItself(brokerDetails, "guest", "guest", virtualpath, queue, null,
+                                                       transacted, persistent, messageSize, verbose,
+                                                       afterCommit, beforeCommit, afterSend, beforeSend, failOnce,
+                                                       batchSize, queueCount, rate);
 
         pingItself.getConnection().start();
 
@@ -194,16 +215,5 @@ public class TestPingItself extends PingPongProducer
         System.exit(0);
     }
 
-    /**
-     * Sets the replyQueue to be the same as ping queue.
-     */
-    @Override
-    public void createConsumer(String selector) throws JMSException
-    {
-        // Create a message consumer to get the replies with and register this to be called back by it.
-        setReplyQueue(getPingQueue());
-        MessageConsumer consumer =
-            getConsumerSession().createConsumer(getReplyQueue(), PREFETCH, false, EXCLUSIVE, selector);
-        consumer.setMessageListener(this);
-    }
+
 }
diff --git a/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java b/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
index 8d57174457..e68c988d08 100644
--- a/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
+++ b/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
@@ -140,7 +140,9 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
 

     protected Session _consumerSession;

 

-    /** Used to restrict the sending rate to a specified limit. */

+    /**

+     * Used to restrict the sending rate to a specified limit.

+     */

     private Throttle rateLimiter = null;

 

     /**

@@ -152,7 +154,7 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
     private PingPongProducer(String brokerDetails, String username, String password, String virtualpath, boolean transacted,

                              boolean persistent, int messageSize, boolean verbose, boolean afterCommit, boolean beforeCommit,

                              boolean afterSend, boolean beforeSend, boolean failOnce, int batchSize, int rate)

-                      throws Exception

+            throws Exception

     {

         // Create a connection to the broker.

         InetAddress address = InetAddress.getLocalHost();

@@ -238,6 +240,77 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
     }

 

     /**

+     * Creates the producer to send the pings on.  If the tests are with nultiple queues, then producer

+     * is created with null destination, so that any destination can be specified while sending

+     *

+     * @throws JMSException

+     */

+    public void createProducer() throws JMSException

+    {

+        if (getQueueCount() > 1)

+        {

+            // create producer with initial destination as null for test with multiple queues

+            // In this case, a different destination will be used while sending the message

+            _producer = (MessageProducer) getProducerSession().createProducer(null);

+        }

+        else

+        {

+            // Create a queue and producer to send the pings on.

+            _producer = (MessageProducer) getProducerSession().createProducer(_pingQueue);

+

+        }

+        _producer.setDisableMessageTimestamp(true);

+        _producer.setDeliveryMode(_persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);

+    }

+

+    /**

+     * Creates the temporary queue to listen to the responses

+     *

+     * @param selector

+     * @throws JMSException

+     */

+    public void createConsumer(String selector) throws JMSException

+    {

+        // Create a temporary queue to get the pongs on.

+        _replyQueue = _consumerSession.createTemporaryQueue();

+

+        // Create a message consumer to get the replies with and register this to be called back by it.

+        MessageConsumer consumer = _consumerSession.createConsumer(_replyQueue, PREFETCH, NO_LOCAL, EXCLUSIVE, selector);

+        consumer.setMessageListener(this);

+    }

+

+    /**

+     * Creates consumer instances for each queue. This is used when test is being done with multiple queues.

+     *

+     * @param selector

+     * @throws JMSException

+     */

+    public void createConsumers(String selector) throws JMSException

+    {

+        for (int i = 0; i < getQueueCount(); i++)

+        {

+            MessageConsumer consumer = getConsumerSession().createConsumer(getQueue(i), PREFETCH, false, EXCLUSIVE, selector);

+            consumer.setMessageListener(this);

+        }

+    }

+

+

+    protected Session getConsumerSession()

+    {

+        return _consumerSession;

+    }

+

+    public Queue getPingQueue()

+    {

+        return _pingQueue;

+    }

+

+    protected void setPingQueue(Queue queue)

+    {

+        _pingQueue = queue;

+    }

+

+    /**

      * Starts a ping-pong loop running from the command line. The bounce back client {@link org.apache.qpid.requestreply.PingPongBouncer} also needs

      * to be started to bounce the pings back again.

      * <p/>

@@ -257,8 +330,8 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
         if (args.length < 2)

         {

             System.err.println(

-                "Usage: TestPingPublisher <brokerDetails> <virtual path> [verbose (true/false)] "

-                + "[transacted (true/false)] [persistent (true/false)] [message size in bytes] [batchsize] [rate]");

+                    "Usage: TestPingPublisher <brokerDetails> <virtual path> [verbose (true/false)] "

+                    + "[transacted (true/false)] [persistent (true/false)] [message size in bytes] [batchsize] [rate]");

             System.exit(0);

         }

 

@@ -310,10 +383,10 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
         }

 

         // Create a ping producer to handle the request/wait/reply cycle.

-        PingPongProducer pingProducer =

-            new PingPongProducer(brokerDetails, "guest", "guest", virtualpath, PING_QUEUE_NAME, null, transacted, persistent,

-                                 messageSize, verbose, afterCommit, beforeCommit, afterSend, beforeSend, failOnce, batchSize,

-                                 0, rate);

+        PingPongProducer pingProducer = new PingPongProducer(brokerDetails, "guest", "guest", virtualpath, PING_QUEUE_NAME, null, transacted,

+                                                             persistent, messageSize, verbose,

+                                                             afterCommit, beforeCommit, afterSend, beforeSend, failOnce,

+                                                             batchSize, 0, rate);

 

         pingProducer.getConnection().start();

 

@@ -332,66 +405,11 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
     }

 

     /**

-     * Creates the producer to send the pings on.  If the tests are with nultiple queues, then producer

-     * is created with null destination, so that any destination can be specified while sending

-     *

-     * @throws JMSException

-     */

-    public void createProducer() throws JMSException

-    {

-        if (getQueueCount() > 1)

-        {

-            // create producer with initial destination as null for test with multiple queues

-            // In this case, a different destination will be used while sending the message

-            _producer = (MessageProducer) getProducerSession().createProducer(null);

-        }

-        else

-        {

-            // Create a queue and producer to send the pings on.

-            _producer = (MessageProducer) getProducerSession().createProducer(_pingQueue);

-

-        }

-

-        _producer.setDisableMessageTimestamp(true);

-        _producer.setDeliveryMode(_persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);

-    }

-

-    /**

-     * Creates the temporary queue to listen to the responses

-     *

-     * @param selector

-     * @throws JMSException

-     */

-    public void createConsumer(String selector) throws JMSException

-    {

-        // Create a temporary queue to get the pongs on.

-        _replyQueue = _consumerSession.createTemporaryQueue();

-

-        // Create a message consumer to get the replies with and register this to be called back by it.

-        MessageConsumer consumer = _consumerSession.createConsumer(_replyQueue, PREFETCH, NO_LOCAL, EXCLUSIVE, selector);

-        consumer.setMessageListener(this);

-    }

-

-    /**

      * Creates consumer instances for each queue. This is used when test is being done with multiple queues.

      *

      * @param selector

      * @throws JMSException

      */

-    public void createConsumers(String selector) throws JMSException

-    {

-        for (int i = 0; i < getQueueCount(); i++)

-        {

-            MessageConsumer consumer =

-                getConsumerSession().createConsumer(getQueue(i), PREFETCH, false, EXCLUSIVE, selector);

-            consumer.setMessageListener(this);

-        }

-    }

-

-    public Queue getPingQueue()

-    {

-        return _pingQueue;

-    }

 

     /**

      * Primes the test loop by sending a few messages, then introduces a short wait. This allows the bounce back client

@@ -415,7 +433,8 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
                 Thread.sleep(100);

             }

             catch (InterruptedException ignore)

-            { }

+            {

+            }

         }

     }

 

@@ -476,10 +495,8 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
      * @param message  The message to send.

      * @param numPings The number of ping messages to send.

      * @param timeout  The timeout in milliseconds.

-     *

      * @return The number of replies received. This may be less than the number sent if the timeout terminated the

      *         wait for all prematurely.

-     *

      * @throws JMSException All underlying JMSExceptions are allowed to fall through.

      */

     public int pingAndWaitForReply(Message message, int numPings, long timeout) throws JMSException, InterruptedException

@@ -570,18 +587,18 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
      * @return The reply, or null if no reply arrives before the timeout.

      * @throws JMSException All underlying JMSExceptions are allowed to fall through.

      */

-    /*public void pingNoWaitForReply(Message message, int numPings) throws JMSException, InterruptedException

+    public void pingNoWaitForReply(Message message, int numPings) throws JMSException, InterruptedException

     {

         for (int i = 0; i < numPings; i++)

         {

             sendMessage(message);

-

+            

             if (_verbose)

             {

                 _logger.info(timestampFormatter.format(new Date()) + ": Pinged at.");

             }

         }

-    }*/

+    }

 

     /**

      * The ping loop implementation. This send out pings of the configured size, persistence and transactionality, and

@@ -618,15 +635,6 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
         return _replyQueue;

     }

 

-    protected Session getConsumerSession()

-    {

-        return _consumerSession;

-    }

-

-    protected void setPingQueue(Queue queue)

-    {

-        _pingQueue = queue;

-    }

 

     protected void setReplyQueue(Queue queue)

     {

@@ -666,10 +674,12 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
     public static class FailoverNotifier implements ConnectionListener

     {

         public void bytesSent(long count)

-        { }

+        {

+        }

 

         public void bytesReceived(long count)

-        { }

+        {

+        }

 

         public boolean preFailover(boolean redirect)

         {

diff --git a/java/perftests/src/main/java/perftests.log4j b/java/perftests/src/main/java/perftests.log4j
index 7666459a79..a6e1bd5452 100644
--- a/java/perftests/src/main/java/perftests.log4j
+++ b/java/perftests/src/main/java/perftests.log4j
@@ -28,7 +28,7 @@ log4j.logger.org.apache.qpid.topic=${amqj.test.logging.level}
 

 

 log4j.logger.uk.co.thebadgerset.junit.extensions=info, console

-log4j.additivity.uk.co.thebadgerset.junit.extensions=fals

+log4j.additivity.uk.co.thebadgerset.junit.extensions=false

 log4j.logger.uk.co.thebadgerset.junit.extensions=info

 

 log4j.appender.console=org.apache.log4j.ConsoleAppender