Client API change: Centralize access to subscription status, better control of acquire/accept.

client/AckPolicy: removed, functionality moved to Subscription and SubscriptionSettings
client/SubscriptionSettings: struct aggregates flow control & accept-acquire parameters for subscribe.
client/Subscription: represents active subscription. Query settings, unacked messages, manual accept/acquire
client/SubscriptionManager: use AcceptMode, AcquireMode enums rather than confusing bools.

Issues addressed by the change:
 - old use of bool for acceptMode was inverted wrt AMQP enum values, bools are confusing.
 - old AckPolicy was broken - not possible to access the instance associated with an active subscription
 - old AckPolicy did not provide a way to do manual acquire, only accept.
 - setting values on SubscriptionManager to apply to subsequent subscriptions is awkward & error-prone, now can use SubscriptionSettings to control on each subscribe individually.
 - a subscription is a central concept in AMQP, it deserves to be a class. Subscription and SubscriptionSettings provides a single point for future expansion of interactions with a a Subscription.


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

1 files changed, 6 insertions, 5 deletions

diff --git a/qpid/cpp/src/tests/latencytest.cpp b/qpid/cpp/src/tests/latencytest.cpp
index 524870a0e8..a980a43322 100644
--- a/qpid/cpp/src/tests/latencytest.cpp
+++ b/qpid/cpp/src/tests/latencytest.cpp
@@ -204,14 +204,15 @@ Receiver::Receiver(const string& q, Stats& s) : Client(q), mgr(session), count(0
         std::cout << "Warning: found " << msgCount << " msgs on " << queue << ". Purging..." << std::endl;
         session.queuePurge(arg::queue=queue);
     }
+    SubscriptionSettings settings;
     if (opts.prefetch) {
-        mgr.setAckPolicy(AckPolicy(opts.ack ? opts.ack : (opts.prefetch / 2)));
-        mgr.setFlowControl(opts.prefetch, SubscriptionManager::UNLIMITED, true);
+        settings.autoAck = (opts.ack ? opts.ack : (opts.prefetch / 2));
+        settings.flowControl = FlowControl::messageWindow(opts.prefetch);
     } else {
-        mgr.setAcceptMode(1/*not-required*/);
-        mgr.setFlowControl(SubscriptionManager::UNLIMITED, SubscriptionManager::UNLIMITED, false);
+        settings.acceptMode = ACCEPT_MODE_NONE;
+        settings.flowControl = FlowControl::unlimited();
     }
-    mgr.subscribe(*this, queue);    
+    mgr.subscribe(*this, queue, settings);    
 }
 
 void Receiver::test()