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/qpid@707808 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 9 insertions, 9 deletions

diff --git a/cpp/src/tests/txtest.cpp b/cpp/src/tests/txtest.cpp
index c285ff9fcc..9d253ddb7f 100644
--- a/cpp/src/tests/txtest.cpp
+++ b/cpp/src/tests/txtest.cpp
@@ -139,9 +139,10 @@ struct Transfer : public Client, public Runnable
             else session.txSelect();
             SubscriptionManager subs(session);
             
-            LocalQueue lq(AckPolicy(0));//manual acking
-            subs.setFlowControl(opts.msgsPerTx, SubscriptionManager::UNLIMITED, true);
-            subs.subscribe(lq, src);
+            LocalQueue lq;
+            SubscriptionSettings settings(FlowControl::messageWindow(opts.msgsPerTx));
+            settings.autoAck = 0; // Disabled
+            Subscription sub = subs.subscribe(lq, src, settings);
             
             for (uint t = 0; t < opts.txCount; t++) {
                 Message in;
@@ -157,7 +158,7 @@ struct Transfer : public Client, public Runnable
                     out.getDeliveryProperties().setDeliveryMode(in.getDeliveryProperties().getDeliveryMode());
                     session.messageTransfer(arg::content=out, arg::acceptMode=1);
                 }
-                lq.getAckPolicy().ackOutstanding(session);
+                sub.accept(sub.getUnaccepted());
                 if (opts.dtx) {
                     session.dtxEnd(arg::xid=xid);
                     session.dtxPrepare(arg::xid=xid);
@@ -230,8 +231,6 @@ struct Controller : public Client
     int check() 
     {
         SubscriptionManager subs(session);
-        subs.setFlowControl(SubscriptionManager::UNLIMITED, SubscriptionManager::UNLIMITED, false);
-        subs.setAcceptMode(1/*not-required*/);
 
         // Recover DTX transactions (if any)
         if (opts.dtx) {
@@ -262,9 +261,10 @@ struct Controller : public Client
         StringSet drained;
         //drain each queue and verify the correct set of messages are available
         for (StringSet::iterator i = queues.begin(); i != queues.end(); i++) {
-            //subscribe, allocate credit and flush
-            LocalQueue lq(AckPolicy(0));//manual acking
-            subs.subscribe(lq, *i, *i);
+            //subscribe, allocate credit and flushn
+            LocalQueue lq;
+            SubscriptionSettings settings(FlowControl::unlimited(), ACCEPT_MODE_NONE);
+            subs.subscribe(lq, *i, settings);
             session.messageFlush(arg::destination=*i);
             session.sync();