NO-JIRA: Fix qpid-cpp-benchmark to work with activemq broker.

The activemq broker appears not to respect the AMQP drain flag, so calling fetch() on
an empty queue hangs. Use get() with a timeout instead to drain the queues.

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

2 files changed, 19 insertions, 2 deletions

diff --git a/qpid/cpp/bindings/qpid/python/qpid_messaging.i b/qpid/cpp/bindings/qpid/python/qpid_messaging.i
index f31b734efa..513cc531ae 100644
--- a/qpid/cpp/bindings/qpid/python/qpid_messaging.i
+++ b/qpid/cpp/bindings/qpid/python/qpid_messaging.i
@@ -151,6 +151,7 @@ QPID_EXCEPTION(UnauthorizedAccess, SessionError)
 %rename(_next_receiver) qpid::messaging::Session::nextReceiver;
 
 %rename(_fetch) qpid::messaging::Receiver::fetch;
+%rename(_get) qpid::messaging::Receiver::get;
 %rename(unsettled) qpid::messaging::Receiver::getUnsettled;
 %rename(available) qpid::messaging::Receiver::getAvailable;
 
@@ -346,6 +347,16 @@ QPID_EXCEPTION(UnauthorizedAccess, SessionError)
                  # but C++ API uses milliseconds
                  return self._fetch(Duration(int(1000*timeout)))
     %}
+
+    %pythoncode %{
+         def get(self, timeout=None) :
+             if timeout is None :
+                 return self._get()
+             else :
+                 # Python API uses timeouts in seconds,
+                 # but C++ API uses milliseconds
+                 return self._get(Duration(int(1000*timeout)))
+    %}
 }
 
 %extend qpid::messaging::Sender {
diff --git a/qpid/cpp/src/tests/qpid-cpp-benchmark b/qpid/cpp/src/tests/qpid-cpp-benchmark
index 70a8a849ad..2d5ec711fe 100755
--- a/qpid/cpp/src/tests/qpid-cpp-benchmark
+++ b/qpid/cpp/src/tests/qpid-cpp-benchmark
@@ -78,6 +78,8 @@ op.add_option("--connection-options", type="str",
               help="Connection options for senders & receivers")
 op.add_option("--durable", default=False, action="store_true",
               help="Use durable queues and messages")
+op.add_option("-t", "--timeout", default=1.0, type="float", metavar="SECONDS",
+              help="Timeout for fetch operations (default %default)")
 op.add_option("--save-received", default=False, action="store_true",
               help="Save received message content to files <queuename>-receiver-<n>.msg")
 op.add_option("--verbose", default=False, action="store_true",
@@ -195,7 +197,11 @@ def drain(queue, session, opts):
     n = 0
     try:
         while True:
-            r.fetch(timeout=0)
+            # FIXME aconway 2014-11-21: activemq broker does not respect the drain flag
+            # so fetch on an empty queue will hang forever, use get with timeout instead.
+            # r.fetch(timeout=0)
+            m = qm.Message()
+            r.get(timeout=opts.timeout)
             n += 1
             if n % 500 == 0: r.session.acknowledge()
         r.session.acknowledge()
@@ -271,7 +277,7 @@ class ReadyReceiver:
         self.connection = connect(broker, opts)
         self.receiver = self.connection.session().receiver(queue)
         self.receiver.session.sync()
-        self.timeout=10
+        self.timeout=opts.timeout
 
     def wait(self, receivers):
         try: