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

3 files changed, 18 insertions, 4 deletions

diff --git a/cpp/src/qpid/broker/DeliveryRecord.cpp b/cpp/src/qpid/broker/DeliveryRecord.cpp
index 1a15373e6a..900016d381 100644
--- a/cpp/src/qpid/broker/DeliveryRecord.cpp
+++ b/cpp/src/qpid/broker/DeliveryRecord.cpp
@@ -43,7 +43,8 @@ DeliveryRecord::DeliveryRecord(const QueuedMessage& _msg,
                                                   cancelled(false),
                                                   completed(false),
                                                   ended(accepted),
-                                                  windowing(_windowing)
+                                                  windowing(_windowing),
+                                                  credit(msg.payload ? msg.payload->getRequiredCredit() : 0)
 {}
 
 void DeliveryRecord::setEnded()
@@ -153,7 +154,7 @@ void DeliveryRecord::reject()
 
 uint32_t DeliveryRecord::getCredit() const
 {
-    return msg.payload ? msg.payload->getRequiredCredit() : 0;
+    return credit;
 }
 
 void DeliveryRecord::acquire(DeliveryIds& results) {
diff --git a/cpp/src/qpid/broker/DeliveryRecord.h b/cpp/src/qpid/broker/DeliveryRecord.h
index f6ffb64697..952e888c03 100644
--- a/cpp/src/qpid/broker/DeliveryRecord.h
+++ b/cpp/src/qpid/broker/DeliveryRecord.h
@@ -64,6 +64,15 @@ class DeliveryRecord
     bool ended;
     const bool windowing;
 
+    /**
+     * Record required credit on construction as the pointer to the
+     * message may be reset once we no longer need to deliver it
+     * (e.g. when it is accepted), but we will still need to be able
+     * to reallocate credit when it is completed (which could happen
+     * after that).
+     */
+    const uint32_t credit;
+
   public:
     DeliveryRecord(
         const QueuedMessage& msg,
diff --git a/cpp/src/qpid/broker/QueuePolicy.cpp b/cpp/src/qpid/broker/QueuePolicy.cpp
index aefb87a392..c967823ecc 100644
--- a/cpp/src/qpid/broker/QueuePolicy.cpp
+++ b/cpp/src/qpid/broker/QueuePolicy.cpp
@@ -48,12 +48,16 @@ bool QueuePolicy::checkLimit(const QueuedMessage& m)
     if (exceeded) {
         if (!policyExceeded) {
             policyExceeded = true;
-            QPID_LOG(info, "Queue size exceeded policy for " << m.queue->getName());
+            if (m.queue) {
+                QPID_LOG(info, "Queue size exceeded policy for " << m.queue->getName());
+            }
         }
     } else {
         if (policyExceeded) {
             policyExceeded = false;
-            QPID_LOG(info, "Queue size within policy for " << m.queue->getName());
+            if (m.queue) {
+                QPID_LOG(info, "Queue size within policy for " << m.queue->getName());
+            }
         }
     }
     return !exceeded;