Added ability to release messages through the Subscription class (+test)

Added another mode for managing completion (+test)
Fixed regression where bytes credit was not reallocated in windowing mode after an accept/release
Fixed regression where subscribe request is issued before listener is registered with dispatcher



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

1 files changed, 28 insertions, 4 deletions

diff --git a/cpp/src/qpid/client/SubscriptionImpl.cpp b/cpp/src/qpid/client/SubscriptionImpl.cpp
index 684cca031a..0ccf5674fd 100644
--- a/cpp/src/qpid/client/SubscriptionImpl.cpp
+++ b/cpp/src/qpid/client/SubscriptionImpl.cpp
@@ -31,6 +31,9 @@ using framing::MessageAcquireResult;
 
 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)
+{}
+
+void SubscriptionImpl::subscribe()
 {
     async(manager.getSession()).messageSubscribe( 
         arg::queue=queue,
@@ -79,11 +82,25 @@ void SubscriptionImpl::accept(const SequenceSet& messageIds) {
     Mutex::ScopedLock l(lock);
     manager.getSession().messageAccept(messageIds);
     unaccepted.remove(messageIds);
-    if (settings.autoComplete) {
+    switch (settings.completionMode) {
+      case COMPLETE_ON_ACCEPT:
+        manager.getSession().markCompleted(messageIds, true);
+        break;
+      case COMPLETE_ON_DELIVERY:
         manager.getSession().sendCompletion();
+        break;
+      default://do nothing
+        break;
     }
 }
 
+void SubscriptionImpl::release(const SequenceSet& messageIds) {
+    Mutex::ScopedLock l(lock);
+    manager.getSession().messageRelease(messageIds);
+    if (settings.acceptMode == ACCEPT_MODE_EXPLICIT)
+        unaccepted.remove(messageIds);
+}
+
 Session SubscriptionImpl::getSession() const { return manager.getSession(); }
 
 SubscriptionManager&SubscriptionImpl:: getSubscriptionManager() const { return manager; }
@@ -102,16 +119,23 @@ void SubscriptionImpl::received(Message& m) {
         listener->received(m);
     }
 
-    if (settings.autoComplete) {
+    if (settings.completionMode == COMPLETE_ON_DELIVERY) {
         manager.getSession().markCompleted(m.getId(), false, false);
     }
     if (settings.autoAck) {
         if (unaccepted.size() >= settings.autoAck) {
             async(manager.getSession()).messageAccept(unaccepted);
-            unaccepted.clear();
-            if (settings.autoComplete) {
+            switch (settings.completionMode) {
+              case COMPLETE_ON_ACCEPT:
+                manager.getSession().markCompleted(unaccepted, true);
+                break;
+              case COMPLETE_ON_DELIVERY:
                 manager.getSession().sendCompletion();
+                break;
+              default://do nothing
+                break;
             }
+            unaccepted.clear();
         }
     }
 }