- added confirm mode to perftest

- added acquire mode to perftest



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

3 files changed, 22 insertions, 5 deletions

diff --git a/cpp/src/qpid/client/SubscriptionManager.cpp b/cpp/src/qpid/client/SubscriptionManager.cpp
index 1282a1cf61..c482c824f7 100644
--- a/cpp/src/qpid/client/SubscriptionManager.cpp
+++ b/cpp/src/qpid/client/SubscriptionManager.cpp
@@ -35,14 +35,14 @@ namespace client {
 SubscriptionManager::SubscriptionManager(Session_0_10& s)
     : dispatcher(s), session(s),
       messages(UNLIMITED), bytes(UNLIMITED), window(true),
-      confirmMode(true)
+      confirmMode(true), acquireMode(true)
 {}
 
 void SubscriptionManager::subscribeInternal(
     const std::string& q, const std::string& dest)
 {
     session.messageSubscribe(arg::queue=q, arg::destination=dest,
-                             arg::confirmMode=confirmMode);
+                             arg::confirmMode=confirmMode, arg::acquireMode=acquireMode);
     setFlowControl(dest, messages, bytes, window);
 }
 
@@ -82,6 +82,8 @@ void SubscriptionManager::setFlowControl(
 
 void SubscriptionManager::setConfirmMode(bool c) { confirmMode=c; }
 
+void SubscriptionManager::setAcquireMode(bool a) { acquireMode=a; }
+
 void SubscriptionManager::setAckPolicy(const AckPolicy& a) { autoAck=a; }
 
 void SubscriptionManager::cancel(const std::string dest)
diff --git a/cpp/src/qpid/client/SubscriptionManager.h b/cpp/src/qpid/client/SubscriptionManager.h
index 5cff46f0f2..f163360f5d 100644
--- a/cpp/src/qpid/client/SubscriptionManager.h
+++ b/cpp/src/qpid/client/SubscriptionManager.h
@@ -46,6 +46,7 @@ class SubscriptionManager
     bool window;
     AckPolicy autoAck;
     bool confirmMode;
+    bool acquireMode;
 
 public:
     SubscriptionManager(Session_0_10& session);
@@ -107,6 +108,13 @@ public:
      */
     void setConfirmMode(bool confirm);
 
+    /** Set the acquire-mode for new subscriptions. Defaults to true.
+     *@param acquire: if false messages are not dequed, if true
+     * messages are dequed on acknowledgement or on transfer 
+	 * depending on confirmMode.
+     */
+    void setAcquireMode(bool acquire);
+
     /** Set the acknowledgement policy for new subscriptions.
      * Default is to acknowledge every message automatically.
      */
diff --git a/cpp/src/tests/perftest.cpp b/cpp/src/tests/perftest.cpp
index 019b1e1fce..c7e3940086 100644
--- a/cpp/src/tests/perftest.cpp
+++ b/cpp/src/tests/perftest.cpp
@@ -48,12 +48,14 @@ struct Opts : public TestOptions {
     std::string mode;
     size_t autoAck;
     bool summary;
+	bool confirmMode;
+	bool acquireMode;
     
     Opts() :
         listen(false), publish(false), purge(false),
         count(500000), size(64), consumers(1),
         mode("shared"), autoAck(100),
-        summary(false)
+        summary(false), confirmMode(false), acquireMode(true)
     {
         addOptions() 
             ("listen", optValue(listen), "Consume messages.")
@@ -65,7 +67,9 @@ struct Opts : public TestOptions {
             ("consumers", optValue(consumers, "N"), "Number of consumers.")
             ("mode", optValue(mode, "shared|fanout|topic"), "consume mode")
             ("auto-ack", optValue(autoAck, "N"), "ack every N messages.")
-            ("summary,s", optValue(summary), "summary output only");
+            ("summary,s", optValue(summary), "summary output only")
+            ("confirm-mode", optValue(confirmMode, "N"), "confirm mode")
+            ("acquire-mode", optValue(acquireMode, "Y"), "acquire mode");
     }
 };
 
@@ -175,7 +179,8 @@ void PublishThread::run() {
             char* data = const_cast<char*>(msg.getData().data());
             *reinterpret_cast<uint32_t*>(data) = i;
             session.messageTransfer(arg::destination=exchange(),
-                                    arg::content=msg);
+                                    arg::content=msg, arg::confirmMode=opts.confirmMode,
+									arg::acquireMode=opts.acquireMode);
             if (!opts.summary && (i%10000)==0){
                  cout << "." << flush;
                  session.execution().sendSyncRequest();
@@ -267,6 +272,8 @@ void ListenThread::run() {
 
         SubscriptionManager subs(session);
         LocalQueue consume(AckPolicy(opts.autoAck));
+		subs.setConfirmMode(opts.confirmMode);
+		subs.setAcquireMode(opts.acquireMode);
         subs.subscribe(consume, consumeQueue);
         int consumed=0;
         AbsTime start=now();