Adding a volume test script(volumetestServiceRequestingClient.sh)

This script checks if all the messages were sent received back successfully.  ServiceRequestingClient.java and ServiceProvidingClient.java modified to add a message identifier to match the request and response message. log4j file modified to create a log file

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

3 files changed, 41 insertions, 39 deletions

diff --git a/java/perftests/src/main/java/org/apache/qpid/requestreply/ServiceProvidingClient.java b/java/perftests/src/main/java/org/apache/qpid/requestreply/ServiceProvidingClient.java
index 84cfc8934f..66950ddf3e 100644
--- a/java/perftests/src/main/java/org/apache/qpid/requestreply/ServiceProvidingClient.java
+++ b/java/perftests/src/main/java/org/apache/qpid/requestreply/ServiceProvidingClient.java
@@ -21,11 +21,11 @@
 package org.apache.qpid.requestreply;
 
 import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.jms.Session;
 import org.apache.qpid.jms.ConnectionListener;
-import org.apache.qpid.AMQException;
+import org.apache.qpid.jms.Session;
 import org.apache.qpid.url.URLSyntaxException;
 
 import javax.jms.*;
@@ -35,7 +35,7 @@ import java.net.UnknownHostException;
 public class ServiceProvidingClient
 {
     private static final Logger _logger = Logger.getLogger(ServiceProvidingClient.class);
-
+    private static final String MESSAGE_IDENTIFIER = "MessageIdentifier";
     private MessageProducer _destinationProducer;
 
     private Destination _responseDest;
@@ -102,9 +102,7 @@ public class ServiceProvidingClient
             public void onMessage(Message message)
             {
                 //_logger.info("Got message '" + message + "'");
-
                 TextMessage tm = (TextMessage) message;
-
                 try
                 {
                     Destination responseDest = tm.getJMSReplyTo();
@@ -147,6 +145,12 @@ public class ServiceProvidingClient
                         _logger.info("timeSent value is: " + timesent);
                         msg.setLongProperty("timeSent", timesent);
                     }
+                    // this identifier set in the serviceRequestingClient is used to match the response with the request
+                    if (tm.propertyExists(MESSAGE_IDENTIFIER))
+                    {
+                        msg.setIntProperty(MESSAGE_IDENTIFIER, tm.getIntProperty(MESSAGE_IDENTIFIER));
+                    }
+                    
                     _destinationProducer.send(msg);
 
 					if(_isTransactional)
@@ -182,7 +186,8 @@ public class ServiceProvidingClient
 
         if (args.length < 7)
         {
-            System.out.println("Usage: brokerDetails username password virtual-path serviceQueue <P[ersistent]|N[onPersistent]>  <T[ransacted]|N[onTransacted]> [selector]");
+            System.out.println("Usage: <brokerDetails> <username> <password> <virtual-path> <serviceQueue>" +
+                               " <P[ersistent]|N[onPersistent]>  <T[ransacted]|N[onTransacted]> [selector]");
             System.exit(1);
         }
         String clientId = null;
@@ -214,9 +219,6 @@ public class ServiceProvidingClient
         {
             _logger.error("Error: " + e, e);
         }
-
-
-
     }
 
 }
diff --git a/java/perftests/src/main/java/org/apache/qpid/requestreply/ServiceRequestingClient.java b/java/perftests/src/main/java/org/apache/qpid/requestreply/ServiceRequestingClient.java
index 7cbbbfe9b0..de49124035 100644
--- a/java/perftests/src/main/java/org/apache/qpid/requestreply/ServiceRequestingClient.java
+++ b/java/perftests/src/main/java/org/apache/qpid/requestreply/ServiceRequestingClient.java
@@ -22,13 +22,14 @@ package org.apache.qpid.requestreply;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
-import org.apache.qpid.url.URLSyntaxException;
 import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.message.TestMessageFactory;
 import org.apache.qpid.jms.MessageConsumer;
 import org.apache.qpid.jms.MessageProducer;
 import org.apache.qpid.jms.Session;
+import org.apache.qpid.url.URLSyntaxException;
 
 import javax.jms.*;
 import java.net.InetAddress;
@@ -46,8 +47,8 @@ public class ServiceRequestingClient implements ExceptionListener
 {
     private static final Logger _log = Logger.getLogger(ServiceRequestingClient.class);
 
-    private static final String MESSAGE_DATA_BYTES = "jfd ghljgl hjvhlj cvhvjf ldhfsj lhfdsjf hldsjfk hdslkfj hsdflk  ";
-
+    private static final String MESSAGE_IDENTIFIER = "MessageIdentifier";
+    private static int _messageIdentifier = 0;
     private String MESSAGE_DATA;
 
     private AMQConnection _connection;
@@ -68,24 +69,6 @@ public class ServiceRequestingClient implements ExceptionListener
 
     private Object _waiter;
 
-    private static String createMessagePayload(int size)
-    {
-        _log.info("Message size set to " + size + " bytes");
-        StringBuffer buf = new StringBuffer(size);
-        int count = 0;
-        while (count < size + MESSAGE_DATA_BYTES.length())
-        {
-            buf.append(MESSAGE_DATA_BYTES);
-            count += MESSAGE_DATA_BYTES.length();
-        }
-        if (count < size)
-        {
-            buf.append(MESSAGE_DATA_BYTES, 0, size - count);
-        }
-
-        return buf.toString();
-    }
-
     private class CallbackHandler implements MessageListener
     {
         private int _expectedMessageCount;
@@ -125,7 +108,11 @@ public class ServiceRequestingClient implements ExceptionListener
                         _log.info("Average latency now: " + _averageLatency);
                     }
                 }
-				if(_isTransactional)
+                if (m.propertyExists(MESSAGE_IDENTIFIER))
+                {
+                    _log.info("Received Message Identifier: " + m.getIntProperty(MESSAGE_IDENTIFIER));
+                }
+                if(_isTransactional)
                 {
                     _session.commit();
                 }
@@ -186,7 +173,7 @@ public class ServiceRequestingClient implements ExceptionListener
         _log.info("Delivery Mode: " + deliveryMode + "\t isTransactional: " + _isTransactional);
 
         _messageCount = messageCount;
-        MESSAGE_DATA = createMessagePayload(messageDataLength);
+        MESSAGE_DATA = TestMessageFactory.createMessagePayload(messageDataLength);
         try
         {
             createConnection(brokerHosts, clientID, username, password, vpath);
@@ -210,7 +197,7 @@ public class ServiceRequestingClient implements ExceptionListener
             //Send first message, then wait a bit to allow the provider to get initialised
             TextMessage first = _session.createTextMessage(MESSAGE_DATA);
             first.setJMSReplyTo(_tempDestination);
-            _producer.send(first);
+            send(first);
 			if(_isTransactional)
             {
                 _producerSession.commit();
@@ -234,6 +221,13 @@ public class ServiceRequestingClient implements ExceptionListener
         }
     }
 
+    private void send(TextMessage msg) throws JMSException
+    {
+        msg.setIntProperty(MESSAGE_IDENTIFIER, ++_messageIdentifier);
+        _producer.send(msg);
+        _log.info("Sent Message Identifier: " + _messageIdentifier);
+    }
+
     /**
      * Run the test and notify an object upon receipt of all responses.
      *
@@ -253,7 +247,7 @@ public class ServiceRequestingClient implements ExceptionListener
                 long timeNow = System.currentTimeMillis();
                 msg.setLongProperty("timeSent", timeNow);
             }
-            _producer.send(msg);
+            send(msg);
             if(_isTransactional)
             {
                 _producerSession.commit();
@@ -283,8 +277,9 @@ public class ServiceRequestingClient implements ExceptionListener
     {
         if (args.length < 9)
         {
-            System.err.println(
-                    "Usage: ServiceRequestingClient <brokerDetails - semicolon separated host:port list> <username> <password> <vpath> <command queue name> <P[ersistent]|N[onPersistent]>  <T[ransacted]|N[onTransacted]> <number of messages> <message size>");
+            System.err.println("Usage: ServiceRequestingClient <brokerDetails - semicolon separated host:port list>" +
+                               " <username> <password> <vpath> <command queue name> <P[ersistent]|N[onPersistent]>" +
+                               " <T[ransacted]|N[onTransacted]> <number of messages> <message size>");
             System.exit(1);
         }
         try
@@ -305,7 +300,6 @@ public class ServiceRequestingClient implements ExceptionListener
                     waiter.wait();
                 }
             }
-
         }
         catch (UnknownHostException e)
         {
diff --git a/java/perftests/src/main/java/perftests.log4j b/java/perftests/src/main/java/perftests.log4j
index 6c74baedf6..e15cb9b285 100644
--- a/java/perftests/src/main/java/perftests.log4j
+++ b/java/perftests/src/main/java/perftests.log4j
@@ -22,10 +22,16 @@ log4j.rootLogger=${root.logging.level}
 log4j.logger.org.apache.qpid=${amqj.logging.level}, console

 log4j.additivity.org.apache.qpid=false

 

-log4j.logger.org.apache.qpid.requestreply=${amqj.test.logging.level}

+log4j.logger.org.apache.qpid.requestreply=${amqj.test.logging.level}, fileApp

 log4j.logger.org.apache.qpid.pingpong=${amqj.test.logging.level}

 

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

 log4j.appender.console.Threshold=all

 log4j.appender.console.layout=org.apache.log4j.PatternLayout

 log4j.appender.console.layout.ConversionPattern=%t %d %p [%c{4}] %m%n

+

+log4j.appender.fileApp=org.apache.log4j.FileAppender

+log4j.appender.fileApp.file=${log.dir}/perftests.volumetest.log

+log4j.appender.fileApp.Threshold=info

+log4j.appender.fileApp.append=false

+log4j.appender.fileApp.layout=org.apache.log4j.PatternLayout