QPID-2920: Improve benchmark tests, allow saving received messages.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid-2920-active@1177832 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 17 insertions, 3 deletions

diff --git a/qpid/cpp/src/tests/qpid-cluster-benchmark b/qpid/cpp/src/tests/qpid-cluster-benchmark
index 7a7542d419..da6a835a8b 100755
--- a/qpid/cpp/src/tests/qpid-cluster-benchmark
+++ b/qpid/cpp/src/tests/qpid-cluster-benchmark
@@ -30,7 +30,7 @@ RECEIVERS="-r 3"
 BROKERS=			# Local broker
 CLIENT_HOSTS=			# No ssh, all clients are local
 
-while getopts "m:f:n:b:q:s:r:c:" opt; do
+while getopts "m:f:n:b:q:s:r:c:x" opt; do
     case $opt in
 	m) MESSAGES="-m $OPTARG";;
 	f) FLOW="--flow-control $OPTARG";;
@@ -40,13 +40,14 @@ while getopts "m:f:n:b:q:s:r:c:" opt; do
 	s) SENDERS="-s $OPTARG";;
 	r) RECEIVERS="-r $OPTARG";;
 	c) CLIENT_HOSTS="-c $OPTARG";;
+	x) SAVE_RECEIVED="--save-received";;
 	*) echo "Unknown option"; exit 1;;
     esac
 done
 BROKER=$(echo $BROKERS | sed s/,.*//)
 run_test() { echo $*; shift; "$@"; echo; echo; echo; }
 
-run_test "Multiple active brokers:" qpid-cpp-benchmark $REPEAT $BROKERS --summarize $QUEUES $SENDERS $RECEIVERS $MESSAGES $CLIENT_HOSTS
-run_test "Single active broker :" qpid-cpp-benchmark $REPEAT $BROKER --summarize $QUEUES $SENDERS $RECEIVERS $MESSAGES $CLIENT_HOSTS
+run_test "Multiple active brokers:" qpid-cpp-benchmark $REPEAT $BROKERS --summarize $QUEUES $SENDERS $RECEIVERS $MESSAGES $CLIENT_HOSTS $SAVE_RECEIVED
+run_test "Single active broker :" qpid-cpp-benchmark $REPEAT $BROKER --summarize $QUEUES $SENDERS $RECEIVERS $MESSAGES $CLIENT_HOSTS $SAVE_RECEIVED
 run_test "Latency under low load:" qpid-cpp-benchmark $REPEAT $BROKERS --connection-options "{tcp-nodelay:true}" $MESSAGES $FLOW $CLIENT_HOSTS
 
diff --git a/qpid/cpp/src/tests/qpid-cpp-benchmark b/qpid/cpp/src/tests/qpid-cpp-benchmark
index 1bc8770915..c951a02498 100755
--- a/qpid/cpp/src/tests/qpid-cpp-benchmark
+++ b/qpid/cpp/src/tests/qpid-cpp-benchmark
@@ -67,6 +67,8 @@ op.add_option("--flow-control", default=0, type="int", metavar="N",
               help="Flow control each sender to limit queue depth to 2*N. 0 means no flow control.")
 op.add_option("--durable", default=False, action="store_true",
               help="Use durable queues and messages")
+op.add_option("--save-received", default=False, action="store_true",
+              help="Save received message content to files <queuename>-receiver-<n>.msg")
 
 single_quote_re = re.compile("'")
 def posix_quote(string):
@@ -111,6 +113,8 @@ def start_receive(queue, index, opts, ready_queue, broker, host):
                "--ready-address", "%s;{create:always}"%ready_queue,
                "--report-header=no"
                ]
+    if opts.save_received:
+        command += ["--save-content=%s-receiver-%s.msg"%(queue,index)]
     command += opts.receive_arg
     if opts.connection_options:
         command += ["--connection-options",opts.connection_options]
diff --git a/qpid/cpp/src/tests/qpid-receive.cpp b/qpid/cpp/src/tests/qpid-receive.cpp
index 9783316449..a4e7653e8d 100644
--- a/qpid/cpp/src/tests/qpid-receive.cpp
+++ b/qpid/cpp/src/tests/qpid-receive.cpp
@@ -34,6 +34,7 @@
 #include "Statistics.h"
 
 #include <iostream>
+#include <fstream>
 #include <memory>
 
 using namespace qpid::messaging;
@@ -60,6 +61,7 @@ struct Options : public qpid::Options
     uint rollbackFrequency;
     bool printContent;
     bool printHeaders;
+    std::string saveContent;
     bool failoverUpdates;
     qpid::log::Options log;
     bool reportTotal;
@@ -105,6 +107,7 @@ struct Options : public qpid::Options
             ("rollback-frequency", qpid::optValue(rollbackFrequency, "N"), "rollback frequency (0 implies no transaction will be rolledback)")
             ("print-content", qpid::optValue(printContent, "yes|no"), "print out message content")
             ("print-headers", qpid::optValue(printHeaders, "yes|no"), "print out message headers")
+            ("save-content", qpid::optValue(saveContent, "FILE"), "save message content to FILE")
             ("failover-updates", qpid::optValue(failoverUpdates), "Listen for membership updates distributed via amq.failover")
             ("report-total", qpid::optValue(reportTotal), "Report total throughput and latency statistics")
             ("report-every", qpid::optValue(reportEvery,"N"), "Report throughput and latency statistics every N messages.")
@@ -196,6 +199,10 @@ int main(int argc, char ** argv)
 
             std::map<std::string,Sender> replyTo;
 
+            std::ofstream saveContent;
+            if (opts.saveContent.size()) 
+                saveContent.open(opts.saveContent.c_str());
+
             while (!done && receiver.fetch(msg, timeout)) {
                 reporter.message(msg);
                 if (!opts.ignoreDuplicates || !sequenceTracker.isDuplicate(msg)) {
@@ -217,6 +224,8 @@ int main(int argc, char ** argv)
                         }
                         if (opts.printContent)
                             std::cout << msg.getContent() << std::endl;//TODO: handle map or list messages
+                        if (opts.saveContent.size())
+                            saveContent << msg.getContent() << std::endl;
                         if (opts.messages && count >= opts.messages) done = true;
                     }
                 } else if (opts.checkRedelivered && !msg.getRedelivered()) {