Added class to ping itself and a junit test for it.

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

3 files changed, 184 insertions, 53 deletions

diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java
new file mode 100644
index 0000000000..6bb4c08e6d
--- /dev/null
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java
@@ -0,0 +1,116 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.ping;
+
+import org.apache.qpid.requestreply.PingPongProducer;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.jms.Session;
+import org.apache.qpid.jms.MessageProducer;
+import org.apache.log4j.Logger;
+
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.ObjectMessage;
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.Queue;
+import java.net.InetAddress;
+
+/**
+ * This class is used to test sending and receiving messages to (pingQueue) and from a queue (replyQueue).
+ * The producer and consumer created by this test send and receive messages to and from the same Queue. ie.
+ * pingQueue and replyQueue are same.
+ * This class extends @see org.apache.qpid.requestreply.PingPongProducer which different ping and reply Queues
+ */
+public class TestPingItself extends PingPongProducer
+{
+    private static final Logger _logger = Logger.getLogger(TestPingItself.class);
+    
+    public TestPingItself(String brokerDetails, String username, String password, String virtualpath, String queueName,
+                          String selector, boolean transacted, boolean persistent,  int messageSize, boolean verbose)
+            throws Exception
+    {
+        super(brokerDetails, username, password, virtualpath, queueName, selector, transacted, persistent, messageSize, verbose);
+    }
+
+    @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);
+    }
+
+    
+
+    /**
+     * 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/>The command line takes from 2 to 4 arguments:
+     * <p/><table>
+     * <tr><td>brokerDetails <td> The broker connection string.
+     * <tr><td>virtualPath   <td> The virtual path.
+     * <tr><td>transacted    <td> A boolean flag, telling this client whether or not to use transactions.
+     * <tr><td>size          <td> The size of ping messages to use, in bytes.
+     * </table>
+     *
+     * @param args The command line arguments as defined above.
+     */
+    public static void main(String[] args) throws Exception
+    {
+        // Extract the command line.
+        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]");
+            System.exit(0);
+        }
+
+        String brokerDetails = args[0];
+        String virtualpath = args[1];
+        boolean verbose = (args.length >= 3) ? Boolean.parseBoolean(args[2]) : true;
+        boolean transacted = (args.length >= 4) ? Boolean.parseBoolean(args[3]) : false;
+        boolean persistent = (args.length >= 5) ? Boolean.parseBoolean(args[4]) : false;
+        int messageSize = (args.length >= 6) ? Integer.parseInt(args[5]) : DEFAULT_MESSAGE_SIZE;
+
+        String queue = "ping_"+ System.currentTimeMillis();
+        _logger.info("Queue:" + queue + ", Transacted:" + transacted + ", persistent:"+ persistent +
+                     ",MessageSize:" + messageSize + " bytes");
+
+        // 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);
+        pingItself.getConnection().start();
+
+        // Run a few priming pings to remove warm up time from test results.
+        pingItself.prime(PRIMING_LOOPS);
+        // Create a shutdown hook to terminate the ping-pong producer.
+        Runtime.getRuntime().addShutdownHook(pingItself.getShutdownHook());
+
+        // Ensure that the ping pong producer is registered to listen for exceptions on the connection too.
+        pingItself.getConnection().setExceptionListener(pingItself);
+
+        // Create the ping loop thread and run it until it is terminated by the shutdown hook or exception.
+        Thread pingThread = new Thread(pingItself);
+        pingThread.run();
+        pingThread.join();
+    }
+}
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 00b01f1025..031a5c5299 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
@@ -79,7 +79,7 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
     protected static final long TIMEOUT = 9000;

 

     /** Holds the name of the queue to send pings on. */

-    private static final String PING_QUEUE_NAME = "ping";

+    protected static final String PING_QUEUE_NAME = "ping";

 

     /** The batch size. */

     protected static final int BATCH_SIZE = 100;

@@ -91,7 +91,7 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
     protected static final boolean EXCLUSIVE = false;

 

     /** The number of priming loops to run. */

-    private static final int PRIMING_LOOPS = 3;

+    protected static final int PRIMING_LOOPS = 3;

 

     /** A source for providing sequential unique correlation ids. */

     private AtomicLong idGenerator = new AtomicLong(0L);

@@ -106,19 +106,19 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
     private Queue _pingQueue;

 

     /** Determines whether this producer sends persistent messages from the run method. */

-    private boolean _persistent;

+    protected boolean _persistent;

 

     /** Holds the message size to send, from the run method. */

-    private int _messageSize;

+    protected int _messageSize;

 

     /** Holds a map from message ids to latches on which threads wait for replies. */

     private Map<String, CountDownLatch> trafficLights = new HashMap<String, CountDownLatch>();

 

     /** Used to indicate that the ping loop should print out whenever it pings. */

-    private boolean _verbose = false;

-

-    private Session _consumerSession;

+    protected boolean _verbose = false;

 

+    protected Session _consumerSession;

+    

     /**

      * Creates a ping pong producer with the specified connection details and type.

      *

@@ -151,9 +151,6 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
         createProducer(queueName, persistent);

         createConsumer(selector);

 

-        // Run a few priming pings to remove warm up time from test results.

-        prime(PRIMING_LOOPS);

-

         _persistent = persistent;

         _messageSize = messageSize;

         _verbose = verbose;

@@ -168,7 +165,8 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
     public void createProducer(String queueName, boolean persistent) throws JMSException

     {

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

-        _pingQueue = new AMQQueue(queueName);

+        if (_pingQueue == null)

+            _pingQueue = new AMQQueue(queueName);

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

         _producer.setDisableMessageTimestamp(true);

         _producer.setDeliveryMode(persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);

@@ -199,6 +197,11 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
         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.

@@ -235,6 +238,8 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
                                              persistent, messageSize, verbose);

         _pingProducer.getConnection().start();

 

+        // Run a few priming pings to remove warm up time from test results.

+        _pingProducer.prime(PRIMING_LOOPS);

         // Create a shutdown hook to terminate the ping-pong producer.

         Runtime.getRuntime().addShutdownHook(_pingProducer.getShutdownHook());

 

@@ -445,6 +450,11 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable,
         return _replyQueue;

     }

 

+    protected void setReplyQueue(Queue queue)

+    {

+        _replyQueue = queue;

+    }

+

     /**

      * A connection listener that logs out any failover complete events. Could do more interesting things with this

      * at some point...

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 4b42283023..7cdfd29120 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
@@ -1,14 +1,16 @@
 package org.apache.qpid.ping;

 

-import java.net.InetAddress;

 import java.util.Properties;

 

 import javax.jms.*;

 

-import junit.framework.TestCase;

+import junit.framework.Test;

+import junit.framework.TestSuite;

+import junit.framework.Assert;

 

 import org.apache.log4j.Logger;

 import org.apache.log4j.NDC;

+import uk.co.thebadgerset.junit.extensions.AsymptoticTestCase;

 

 /**

  *

@@ -16,8 +18,7 @@ import org.apache.log4j.NDC;
  * simultaneously to simluate many clients/producers/connections.

  *

  * <p/>A single run of the test using the default JUnit test runner will result in the sending and timing of a single

- * full round trip ping. This test may be scaled up using a suitable JUnit test runner. See {@link TKTestRunner} or

- * {@link PPTestRunner} for more information on how to do this.

+ * full round trip ping. This test may be scaled up using a suitable JUnit test runner.

  *

  * <p/>The setup/teardown cycle establishes a connection to a broker and sets up a queue to send ping messages to and a

  * temporary queue for replies. This setup is only established once for all the test repeats/threads that may be run,

@@ -35,7 +36,7 @@ import org.apache.log4j.NDC;
  *

  * @author Rupert Smith

  */

-public class PingTestPerf extends TestCase //implements TimingControllerAware

+public class PingTestPerf extends AsymptoticTestCase //implements TimingControllerAware

 {

     private static Logger _logger = Logger.getLogger(PingTestPerf.class);

 

@@ -57,6 +58,9 @@ public class PingTestPerf extends TestCase //implements TimingControllerAware
     /** Holds the name of the property to get the test broker virtual path. */

     private static final String VIRTUAL_PATH_PROPNAME = "virtualPath";

 

+    /** Holds the waiting timeout for response messages */

+    private static final String TIMEOUT_PROPNAME = "timeout";

+

     /** Holds the size of message body to attach to the ping messages. */

     private static final int MESSAGE_SIZE_DEFAULT = 0;

 

@@ -76,7 +80,7 @@ public class PingTestPerf extends TestCase //implements TimingControllerAware
     private static final String VIRTUAL_PATH_DEFAULT = "/test";

 

     /** Sets a default ping timeout. */

-    private static final long TIMEOUT = 3000;

+    private static final long TIMEOUT_DEFAULT = 3000;

 

     // Sets up the test parameters with defaults.

     static

@@ -87,13 +91,11 @@ public class PingTestPerf extends TestCase //implements TimingControllerAware
         setSystemPropertyIfNull(TRANSACTED_PROPNAME, Boolean.toString(TRANSACTED_DEFAULT));

         setSystemPropertyIfNull(BROKER_PROPNAME, BROKER_DEFAULT);

         setSystemPropertyIfNull(VIRTUAL_PATH_PROPNAME, VIRTUAL_PATH_DEFAULT);

+        setSystemPropertyIfNull(TIMEOUT_PROPNAME, Long.toString(TIMEOUT_DEFAULT));

     }

 

-    /** Holds the test ping-pong producer. */

-    private TestPingProducer _testPingProducer;

-

     /** Holds the test ping client. */

-    private TestPingClient _testPingClient;

+    private TestPingItself _pingItselfClient;

 

     // Set up a property reader to extract the test parameters from. Once ContextualProperties is available in

     // the project dependencies, use it to get property overrides for configurable tests and to notify the test runner

@@ -114,26 +116,22 @@ public class PingTestPerf extends TestCase //implements TimingControllerAware
         }

     }

 

-    public void testPingOk() throws Exception

+    public void testPingOk(int numPings) throws Exception

     {

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

-        ObjectMessage msg =

-            _testPingProducer.getTestMessage(null, Integer.parseInt(testParameters.getProperty(MESSAGE_SIZE_PROPNAME)),

+        ObjectMessage msg = _pingItselfClient.getTestMessage(null,

+                                             Integer.parseInt(testParameters.getProperty(MESSAGE_SIZE_PROPNAME)),

                                              Boolean.parseBoolean(testParameters.getProperty(PERSISTENT_MODE_PROPNAME)));

 

-        // Use the test timing controller to reset the test timer now and obtain the current time.

-        // This can be used to remove the message creation time from the test.

-        //TestTimingController timingUtils = getTimingController();

-        //long startTime = timingUtils.restart();

-

-        // Send the message.

-        _testPingProducer.ping(msg);

-

+        // start the test

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

+        int numReplies = _pingItselfClient.pingAndWaitForReply(msg, numPings, timeout);

+        _logger.info("replies = " + numReplies);

         // Fail the test if the timeout was exceeded.

-        /*if (reply == null)

+        if (numReplies != numPings)

         {

-            Assert.fail("The ping timed out for message id: " + msg.getJMSMessageID());

-        }*/

+            Assert.fail("The ping timed out. Messages Sent = " + numReplies + ", MessagesReceived = " + numPings);

+        }

     }

 

     protected void setUp() throws Exception

@@ -142,7 +140,7 @@ public class PingTestPerf extends TestCase //implements TimingControllerAware
         NDC.push(getName());

 

         // Ensure that the connection, session and ping queue are established, if they have not already been.

-        if (_testPingProducer == null)

+        if (_pingItselfClient == null)

         {

             // Extract the test set up paramaeters.

             String brokerDetails = testParameters.getProperty(BROKER_PROPNAME);

@@ -156,17 +154,12 @@ public class PingTestPerf extends TestCase //implements TimingControllerAware
             boolean verbose = false;

             int messageSize = Integer.parseInt(testParameters.getProperty(MESSAGE_SIZE_PROPNAME));

 

-            // Establish a bounce back client on the ping queue to bounce back the pings.

-            _testPingClient = new TestPingClient(brokerDetails, username, password, queueName, virtualpath, transacted,

-                                                 selector, verbose);

-

-            // Establish a ping-pong client on the ping queue to send the pings with.

-            _testPingProducer = new TestPingProducer(brokerDetails, username, password, virtualpath, queueName, transacted,

-                                                     persistent, messageSize, verbose);

+            // Establish a client to ping a Queue and listen the reply back from same Queue

+            _pingItselfClient = new TestPingItself(brokerDetails, username, password, virtualpath, queueName,

+                                                   selector, transacted, persistent, messageSize, verbose);

 

-            // Start the connections for client and producer running.

-            _testPingClient.getConnection().start();

-            _testPingProducer.getConnection().start();

+            // Start the client connection

+            _pingItselfClient.getConnection().start();

         }

     }

 

@@ -174,19 +167,31 @@ public class PingTestPerf extends TestCase //implements TimingControllerAware
     {

         try

         {

-            if ((_testPingClient != null) && (_testPingClient.getConnection() != null))

+            /*

+            if ((_pingItselfClient != null) && (_pingItselfClient.getConnection() != null))

             {

-                _testPingClient.getConnection().close();

-            }

-

-            if ((_testPingProducer != null) && (_testPingProducer.getConnection() != null))

-            {

-                _testPingProducer.getConnection().close();

+                _pingItselfClient.getConnection().close();

             }

+            */

         }

         finally

         {

             NDC.pop();

         }

     }

+

+    /**

+     * Compile all the tests into a test suite.

+     */

+    public static Test suite()

+    {      

+        // Build a new test suite

+        TestSuite suite = new TestSuite("Ping Performance Tests");

+

+        // Run performance tests in read committed mode.

+        suite.addTest(new PingTestPerf("testPingOk"));

+

+        return suite;

+        //return new junit.framework.TestSuite(PingTestPerf.class);

+    }

 }