QPID-1488: Ensure policy state (+ store state & mgmt stats) are accurate on newly joined nodes by informing

           the queue of any logically enqueued messages that are currently acquired (but not accepted or
           released).

QPID-1873: Ensure that the various properties of a queue (durability, exclusivity etc) are correctly replicated
           to new cluster members.



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

1 files changed, 4 insertions, 7 deletions

diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
index dda4b5435b..77bfbe0c6d 100644
--- a/cpp/src/qpid/cluster/Connection.cpp
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -330,10 +330,12 @@ void Connection::deliveryRecord(const string& qname,
     broker::QueuedMessage m;
     broker::Queue::shared_ptr queue = findQueue(qname);
     if (!ended) {               // Has a message
-        if (acquired)           // Message is on the update queue
+        if (acquired) {         // Message is on the update queue
             m = getUpdateMessage();
-        else                    // Message at original position in original queue
+            queue->enqueued(m); //inform queue of the message 
+        } else {                // Message at original position in original queue
             m = queue->find(position);
+        }
         if (!m.payload)
             throw Exception(QPID_MSG("deliveryRecord no update message"));
     }
@@ -344,11 +346,6 @@ void Connection::deliveryRecord(const string& qname,
     if (completed) dr.complete();
     if (ended) dr.setEnded();   // Exsitance of message
     semanticState().record(dr); // Part of the session's unacked list.
-
-    // If the message was unacked, the newbie broker must place
-    // it in its messageStore.
-    if ( m.payload && m.payload->isPersistent() && acquired && !ended)
-        queue->enqueue ( 0, m.payload );
 }
 
 void Connection::queuePosition(const string& qname, const SequenceNumber& position) {