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, 6 insertions, 5 deletions

diff --git a/cpp/src/qpid/client/LocalQueue.cpp b/cpp/src/qpid/client/LocalQueue.cpp
index 99ab6f0133..229d3766ef 100644
--- a/cpp/src/qpid/client/LocalQueue.cpp
+++ b/cpp/src/qpid/client/LocalQueue.cpp
@@ -22,13 +22,15 @@
 #include "qpid/Exception.h"
 #include "qpid/framing/FrameSet.h"
 #include "qpid/framing/reply_exceptions.h"
+#include "HandlePrivate.h"
+#include "SubscriptionImpl.h"
 
 namespace qpid {
 namespace client {
 
 using namespace framing;
 
-LocalQueue::LocalQueue(AckPolicy a) : autoAck(a) {}
+LocalQueue::LocalQueue() {}
 LocalQueue::~LocalQueue() {}
 
 Message LocalQueue::pop() { return get(); }
@@ -48,7 +50,9 @@ bool LocalQueue::get(Message& result, sys::Duration timeout) {
     if (!ok) return false;
     if (content->isA<MessageTransferBody>()) {
         result = Message(*content);
-        autoAck.ack(result, session);
+        boost::intrusive_ptr<SubscriptionImpl> si = HandlePrivate<SubscriptionImpl>::get(subscription);
+        assert(si);
+        if (si) si->received(result);
         return true;
     }
     else
@@ -56,9 +60,6 @@ bool LocalQueue::get(Message& result, sys::Duration timeout) {
             QPID_MSG("Unexpected method: " << content->getMethod()));
 }
 
-void LocalQueue::setAckPolicy(AckPolicy a) { autoAck=a; }
-AckPolicy& LocalQueue::getAckPolicy() { return autoAck; }
-
 bool LocalQueue::empty() const
 { 
     if (!queue)