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

5 files changed, 34 insertions, 2 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 6930275361..bbdbf19e92 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -1006,3 +1006,16 @@ void Queue::insertSequenceNumbers(const std::string& key)
     insertSeqNo = !seqNoKey.empty();
     QPID_LOG(debug, "Inserting sequence numbers as " << key);
 }
+
+void Queue::enqueued(const QueuedMessage& m)
+{
+    if (m.payload) {
+        if (policy.get()) policy->tryEnqueue(m);
+        mgntEnqStats(m.payload);
+        if (m.payload->isPersistent()) {
+            enqueue ( 0, m.payload );
+        }
+    } else {
+        QPID_LOG(warning, "Queue informed of enqueued message that has no payload");
+    }
+}
diff --git a/cpp/src/qpid/broker/Queue.h b/cpp/src/qpid/broker/Queue.h
index 0d5f2043d1..de60362854 100644
--- a/cpp/src/qpid/broker/Queue.h
+++ b/cpp/src/qpid/broker/Queue.h
@@ -124,7 +124,7 @@ namespace qpid {
             QueuedMessage getFront();
             QueuedMessage& checkLvqReplace(QueuedMessage& msg);
             void clearLVQIndex(const QueuedMessage& msg);
-            
+
             inline void mgntEnqStats(const boost::intrusive_ptr<Message>& msg)
             {
                 if (mgmtObject != 0) {
@@ -251,6 +251,14 @@ namespace qpid {
             void dequeueCommitted(const QueuedMessage& msg);
 
             /**
+             * Inform queue of messages that were enqueued, have since
+             * been acquired but not yet accepted or released (and
+             * thus are still logically on the queue) - used in
+             * clustered broker.  
+             */ 
+            void enqueued(const QueuedMessage& msg);
+            
+            /**
              * Gets the next available message 
              */
             QPID_BROKER_EXTERN QueuedMessage get();
diff --git a/cpp/src/qpid/broker/SessionAdapter.cpp b/cpp/src/qpid/broker/SessionAdapter.cpp
index 0ddd546a68..8217a46941 100644
--- a/cpp/src/qpid/broker/SessionAdapter.cpp
+++ b/cpp/src/qpid/broker/SessionAdapter.cpp
@@ -284,7 +284,6 @@ void SessionAdapter::QueueHandlerImpl::destroyExclusiveQueues()
         exclusiveQueues.erase(exclusiveQueues.begin());
     }
 }
-
     
 bool SessionAdapter::QueueHandlerImpl::isLocal(const ConnectionToken* t) const 
 { 
diff --git a/cpp/src/qpid/broker/SessionAdapter.h b/cpp/src/qpid/broker/SessionAdapter.h
index b9f8134856..6f3b598398 100644
--- a/cpp/src/qpid/broker/SessionAdapter.h
+++ b/cpp/src/qpid/broker/SessionAdapter.h
@@ -28,6 +28,7 @@
 #include "qpid/framing/reply_exceptions.h"
 #include "qpid/framing/StructHelper.h"
 
+#include <algorithm>
 #include <vector>
 #include <boost/function.hpp>
 #include <boost/shared_ptr.hpp>
@@ -68,6 +69,12 @@ class Queue;
     FileHandler* getFileHandler() { throw framing::NotImplementedException("Class not implemented"); }
     StreamHandler* getStreamHandler() { throw framing::NotImplementedException("Class not implemented"); }
 
+    template <class F> void eachExclusiveQueue(F f) 
+    { 
+        queueImpl.eachExclusiveQueue(f);
+    }
+
+
   private:
     //common base for utility methods etc that are specific to this adapter
     struct HandlerHelper : public HandlerImpl 
@@ -130,6 +137,10 @@ class Queue;
         bool isLocal(const ConnectionToken* t) const; 
 
         void destroyExclusiveQueues();
+        template <class F> void eachExclusiveQueue(F f) 
+        { 
+            std::for_each(exclusiveQueues.begin(), exclusiveQueues.end(), f);
+        }
     };
 
     class MessageHandlerImpl :
diff --git a/cpp/src/qpid/broker/SessionState.h b/cpp/src/qpid/broker/SessionState.h
index bdfed87905..880dfad99e 100644
--- a/cpp/src/qpid/broker/SessionState.h
+++ b/cpp/src/qpid/broker/SessionState.h
@@ -107,6 +107,7 @@ class SessionState : public qpid::SessionState,
     // Used by cluster to create replica sessions.
     SemanticState& getSemanticState() { return semanticState; }
     boost::intrusive_ptr<Message> getMessageInProgress() { return msgBuilder.getMessage(); }
+    SessionAdapter& getSessionAdapter() { return adapter; }
 
     bool processSendCredit(uint32_t msgs);