QPID-420 (merged from trunk) And r518998:518999 and r520846:520850

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

3 files changed, 39 insertions, 8 deletions

diff --git a/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java b/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java
index c0f236b833..eeb4021f34 100644
--- a/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java
+++ b/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java
@@ -103,7 +103,7 @@ public class TestMessageFactory
     {
         StringBuffer buf = new StringBuffer(size);
         int count = 0;
-        while (count < size)
+        while (count <= (size - MESSAGE_DATA_BYTES.length()))
         {
             buf.append(MESSAGE_DATA_BYTES);
             count += MESSAGE_DATA_BYTES.length();
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 2a3aff4692..c4074806b5 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
@@ -497,11 +497,12 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
         boolean transacted = config.isTransacted();

         boolean persistent = config.usePersistentMessages();

         int messageSize = (config.getPayload() != 0) ? config.getPayload() : MESSAGE_SIZE_DEAFULT;

-        // int messageCount = config.getMessages();

+        int messageCount = config.getMessages();

         int destCount = (config.getDestinationsCount() != 0) ? config.getDestinationsCount() : DESTINATION_COUNT_DEFAULT;

         int batchSize = (config.getBatchSize() != 0) ? config.getBatchSize() : TX_BATCH_SIZE_DEFAULT;

         int rate = (config.getRate() != 0) ? config.getRate() : RATE_DEFAULT;

         boolean pubsub = config.isPubSub();

+        long timeout = (config.getTimeout() != 0) ? config.getTimeout() : TIMEOUT_DEFAULT;

 

         String destName = config.getDestination();

         if (destName == null)

@@ -561,10 +562,20 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
         // Ensure that the ping pong producer is registered to listen for exceptions on the connection too.

         pingProducer.getConnection().setExceptionListener(pingProducer);

 

-        // Create the ping loop thread and run it until it is terminated by the shutdown hook or exception.

-        Thread pingThread = new Thread(pingProducer);

-        pingThread.run();

-        pingThread.join();

+        // If messageount is 0, then continue sending

+        if (messageCount == 0)

+        {

+            // Create the ping loop thread and run it until it is terminated by the shutdown hook or exception.

+            Thread pingThread = new Thread(pingProducer);

+            pingThread.start();

+            pingThread.join();

+        }

+        else

+        {

+            // This feature is needed, when we want to send fix no of messages

+            pingProducer.pingLoop(messageCount, timeout);

+        }

+        pingProducer.close();

     }

 

     /**

@@ -963,7 +974,7 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
     /**

      * The ping loop implementation. This sends out pings waits for replies and inserts short pauses in between each.

      */

-    public void pingLoop()

+    public void pingLoop(int pingCount, long timeout)

     {

         try

         {

@@ -972,7 +983,7 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
             msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());

 

             // Send the message and wait for a reply.

-            pingAndWaitForReply(msg, TX_BATCH_SIZE_DEFAULT, TIMEOUT_DEFAULT);

+            pingAndWaitForReply(msg, pingCount, timeout);

         }

         catch (JMSException e)

         {

@@ -986,6 +997,11 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis
         }

     }

 

+    public void pingLoop()

+    {

+        pingLoop(TX_BATCH_SIZE_DEFAULT, TIMEOUT_DEFAULT);

+    }

+

     public Destination getReplyDestination()

     {

         return getReplyDestinations().get(0);

diff --git a/java/perftests/src/main/java/org/apache/qpid/topic/Config.java b/java/perftests/src/main/java/org/apache/qpid/topic/Config.java
index 60aa9f3930..342b28ca17 100644
--- a/java/perftests/src/main/java/org/apache/qpid/topic/Config.java
+++ b/java/perftests/src/main/java/org/apache/qpid/topic/Config.java
@@ -51,6 +51,7 @@ public class Config extends AbstractConfig implements ConnectorConfig
     private int batchSize;
     private int rate;
     private boolean ispubsub;
+    private long timeout;
 
     public Config()
     {
@@ -161,6 +162,16 @@ public class Config extends AbstractConfig implements ConnectorConfig
         this.delay = delay;
     }
 
+    public long getTimeout()
+    {
+        return timeout;
+    }
+
+    public void setTimeout(long time)
+    {
+        this.timeout = time;
+    }
+
     public String getClientId()
     {
         return clientId;
@@ -285,6 +296,10 @@ public class Config extends AbstractConfig implements ConnectorConfig
         {
             destinationName = value;
         }
+        else if("-timeout".equalsIgnoreCase(key))
+        {
+            setTimeout(parseLong("Bad timeout data", value));
+        }
         else
         {
             System.out.println("Ignoring unrecognised option: " + key);