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, 116 insertions, 0 deletions

diff --git a/cpp/src/qpid/client/SubscriptionImpl.cpp b/cpp/src/qpid/client/SubscriptionImpl.cpp
new file mode 100644
index 0000000000..3363dda11f
--- /dev/null
+++ b/cpp/src/qpid/client/SubscriptionImpl.cpp
@@ -0,0 +1,116 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "SubscriptionImpl.h"
+#include "SubscriptionManager.h"
+#include "SubscriptionSettings.h"
+
+namespace qpid {
+namespace client {
+
+using sys::Mutex;
+
+SubscriptionImpl::SubscriptionImpl(SubscriptionManager& m, const std::string& q, const SubscriptionSettings& s, const std::string& n, MessageListener* l)
+    : manager(m), name(n), queue(q), settings(s), listener(l)
+{
+    async(manager.getSession()).messageSubscribe( 
+        arg::queue=queue,
+        arg::destination=name,
+        arg::acceptMode=settings.acceptMode,
+        arg::acquireMode=settings.acquireMode);
+    setFlowControl(settings.flowControl);
+}
+
+std::string SubscriptionImpl::getName() const { return name; }
+
+std::string SubscriptionImpl::getQueue() const { return queue; }
+
+const SubscriptionSettings& SubscriptionImpl::getSettings() const {
+    Mutex::ScopedLock l(lock);
+    return settings;
+}
+
+void SubscriptionImpl::setFlowControl(const FlowControl& f) {
+    Mutex::ScopedLock l(lock);
+    AsyncSession s=manager.getSession();
+    if (&settings.flowControl != &f) settings.flowControl = f;
+    s.messageSetFlowMode(name, f.window); 
+    s.messageFlow(name, CREDIT_UNIT_MESSAGE, f.messages); 
+    s.messageFlow(name, CREDIT_UNIT_BYTE, f.bytes);
+    s.sync();
+}
+
+void SubscriptionImpl::setAutoAck(size_t n) {
+    Mutex::ScopedLock l(lock);
+    settings.autoAck = n;
+}
+
+SequenceSet SubscriptionImpl::getUnacquired() const { Mutex::ScopedLock l(lock); return unacquired; }
+SequenceSet SubscriptionImpl::getUnaccepted() const { Mutex::ScopedLock l(lock); return unaccepted; }
+
+void SubscriptionImpl::acquire(const SequenceSet& messageIds) {
+    Mutex::ScopedLock l(lock);
+    manager.getSession().messageAcquire(messageIds);
+    unacquired.remove(messageIds);
+    if (settings.acceptMode == ACCEPT_MODE_EXPLICIT)
+        unaccepted.add(messageIds);
+}
+
+void SubscriptionImpl::accept(const SequenceSet& messageIds) {
+    Mutex::ScopedLock l(lock);
+    manager.getSession().messageAccept(messageIds);
+    unaccepted.remove(messageIds);
+}
+
+Session SubscriptionImpl::getSession() const { return manager.getSession(); }
+
+SubscriptionManager&SubscriptionImpl:: getSubscriptionManager() const { return manager; }
+
+void SubscriptionImpl::cancel() { manager.cancel(name); }
+
+void SubscriptionImpl::received(Message& m) {
+    Mutex::ScopedLock l(lock);
+    manager.getSession().markCompleted(m.getId(), false, false);        
+    if (m.getMethod().getAcquireMode() == ACQUIRE_MODE_NOT_ACQUIRED) 
+        unacquired.add(m.getId());
+    else if (m.getMethod().getAcceptMode() == ACCEPT_MODE_EXPLICIT)
+        unaccepted.add(m.getId());
+
+    if (listener) {
+        Mutex::ScopedUnlock u(lock);
+        listener->received(m);
+    }
+
+    if (settings.autoAck) {
+        if (unacquired.size() + unaccepted.size() >= settings.autoAck) {
+            if (unacquired.size()) {
+                async(manager.getSession()).messageAcquire(unacquired);
+                unaccepted.add(unacquired);
+                unaccepted.clear();
+            }
+            async(manager.getSession()).messageAccept(unaccepted);
+            unaccepted.clear();
+        }
+    }
+}
+
+}} // namespace qpid::client
+