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, 29 deletions

diff --git a/cpp/src/qpid/client/Dispatcher.cpp b/cpp/src/qpid/client/Dispatcher.cpp
index c26dba188d..0b7618eb4c 100644
--- a/cpp/src/qpid/client/Dispatcher.cpp
+++ b/cpp/src/qpid/client/Dispatcher.cpp
@@ -19,6 +19,7 @@
  *
  */
 #include "Dispatcher.h"
+#include "SubscriptionImpl.h"
 
 #include "qpid/framing/FrameSet.h"
 #include "qpid/framing/MessageTransferBody.h"
@@ -37,18 +38,6 @@ using qpid::sys::Thread;
 namespace qpid {
 namespace client {
 
-Subscriber::Subscriber(const Session& s, MessageListener* l, AckPolicy a)
-    : session(s), listener(l), autoAck(a) {}
-
-void Subscriber::received(Message& msg)
-{
-
-    if (listener) {
-        listener->received(msg);
-        autoAck.ack(msg, session);
-    }
-}
-
 Dispatcher::Dispatcher(const Session& s, const std::string& q)
     : session(s), 
       running(false), 
@@ -78,7 +67,7 @@ void Dispatcher::run()
             FrameSet::shared_ptr content = queue->pop();
             if (content->isA<MessageTransferBody>()) {
                 Message msg(*content);
-                Subscriber::shared_ptr listener = find(msg.getDestination());
+                boost::intrusive_ptr<SubscriptionImpl> listener = find(msg.getDestination());
                 if (!listener) {
                     QPID_LOG(error, "No listener found for destination " << msg.getDestination());
                 } else {
@@ -121,7 +110,7 @@ void Dispatcher::setAutoStop(bool b)
     autoStop = b;
 }
 
-Subscriber::shared_ptr Dispatcher::find(const std::string& name)
+boost::intrusive_ptr<SubscriptionImpl> Dispatcher::find(const std::string& name)
 {
     ScopedLock<Mutex> l(lock);
     Listeners::iterator i = listeners.find(name);
@@ -131,24 +120,12 @@ Subscriber::shared_ptr Dispatcher::find(const std::string& name)
     return i->second;
 }
 
-void Dispatcher::listen(
-    MessageListener* listener, AckPolicy autoAck
-)
-{
-    ScopedLock<Mutex> l(lock);
-    defaultListener = Subscriber::shared_ptr(
-        new Subscriber(session, listener, autoAck));
-}
-
-void Dispatcher::listen(const std::string& destination, MessageListener* listener, AckPolicy autoAck)
-{
+void Dispatcher::listen(const boost::intrusive_ptr<SubscriptionImpl>& subscription) {
     ScopedLock<Mutex> l(lock);
-    listeners[destination] = Subscriber::shared_ptr(
-        new Subscriber(session, listener, autoAck));
+    listeners[subscription->getName()] = subscription;
 }
 
-void Dispatcher::cancel(const std::string& destination)
-{
+void Dispatcher::cancel(const std::string& destination) {
     ScopedLock<Mutex> l(lock);
     listeners.erase(destination);
     if (autoStop && listeners.empty())