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@779183 13f79535-47bb-0310-9956-ffa450edef68

9 files changed, 123 insertions, 20 deletions

diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index 6930275361..bbdbf19e92 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/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/qpid/cpp/src/qpid/broker/Queue.h b/qpid/cpp/src/qpid/broker/Queue.h
index 0d5f2043d1..de60362854 100644
--- a/qpid/cpp/src/qpid/broker/Queue.h
+++ b/qpid/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/qpid/cpp/src/qpid/broker/SessionAdapter.cpp b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
index 0ddd546a68..8217a46941 100644
--- a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
+++ b/qpid/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/qpid/cpp/src/qpid/broker/SessionAdapter.h b/qpid/cpp/src/qpid/broker/SessionAdapter.h
index b9f8134856..6f3b598398 100644
--- a/qpid/cpp/src/qpid/broker/SessionAdapter.h
+++ b/qpid/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/qpid/cpp/src/qpid/broker/SessionState.h b/qpid/cpp/src/qpid/broker/SessionState.h
index bdfed87905..880dfad99e 100644
--- a/qpid/cpp/src/qpid/broker/SessionState.h
+++ b/qpid/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);
 
diff --git a/qpid/cpp/src/qpid/cluster/Connection.cpp b/qpid/cpp/src/qpid/cluster/Connection.cpp
index dda4b5435b..77bfbe0c6d 100644
--- a/qpid/cpp/src/qpid/cluster/Connection.cpp
+++ b/qpid/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) {
diff --git a/qpid/cpp/src/qpid/cluster/UpdateClient.cpp b/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
index edd83463d2..7fdbe73926 100644
--- a/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
+++ b/qpid/cpp/src/qpid/cluster/UpdateClient.cpp
@@ -124,7 +124,7 @@ void UpdateClient::update() {
     QPID_LOG(debug, updaterId << " updating state to " << updateeId << " at " << updateeUrl);
     Broker& b = updaterBroker;
     b.getExchanges().eachExchange(boost::bind(&UpdateClient::updateExchange, this, _1));
-    b.getQueues().eachQueue(boost::bind(&UpdateClient::updateQueue, this, _1));
+    b.getQueues().eachQueue(boost::bind(&UpdateClient::updateNonExclusiveQueue, this, _1));
     // Update queue is used to transfer acquired messages that are no longer on their original queue.
     session.queueDeclare(arg::queue=UPDATE, arg::autoDelete=true);
     session.sync();
@@ -225,18 +225,35 @@ class MessageUpdater {
     }
 };
 
-void UpdateClient::updateQueue(const boost::shared_ptr<Queue>& q) {
-    QPID_LOG(debug, updaterId << " updating queue " << q->getName());
-    ClusterConnectionProxy proxy(session);
-    proxy.queue(encode(*q));
-    MessageUpdater updater(q->getName(), session, expiry);
+void UpdateClient::updateQueue(client::AsyncSession& s, const boost::shared_ptr<Queue>& q) {
+    broker::Exchange::shared_ptr alternateExchange = q->getAlternateExchange();
+    s.queueDeclare(
+        arg::queue = q->getName(),
+        arg::durable = q->isDurable(),
+        arg::autoDelete = q->isAutoDelete(),
+        arg::alternateExchange = alternateExchange ? alternateExchange->getName() : "",
+        arg::arguments = q->getSettings(),
+        arg::exclusive = q->hasExclusiveOwner()
+    );
+    MessageUpdater updater(q->getName(), s, expiry);
     q->eachMessage(boost::bind(&MessageUpdater::updateQueuedMessage, &updater, _1));
-    q->eachBinding(boost::bind(&UpdateClient::updateBinding, this, q->getName(), _1));
+    q->eachBinding(boost::bind(&UpdateClient::updateBinding, this, s, q->getName(), _1));
 }
 
+void UpdateClient::updateExclusiveQueue(const boost::shared_ptr<broker::Queue>& q) {
+    QPID_LOG(debug, updaterId << " updating exclusive queue " << q->getName() << " on " << shadowSession.getId());
+    updateQueue(shadowSession, q);
+}
 
-void UpdateClient::updateBinding(const std::string& queue, const QueueBinding& binding) {
-    session.exchangeBind(queue, binding.exchange, binding.key, binding.args);
+void UpdateClient::updateNonExclusiveQueue(const boost::shared_ptr<broker::Queue>& q) {
+    if (!q->hasExclusiveOwner()) {
+        QPID_LOG(debug, updaterId << " updating queue " << q->getName());
+        updateQueue(session, q);
+    }//else queue will be updated as part of session state of owning session
+}
+
+void UpdateClient::updateBinding(client::AsyncSession& s, const std::string& queue, const QueueBinding& binding) {
+    s.exchangeBind(queue, binding.exchange, binding.key, binding.args);
 }
 
 void UpdateClient::updateConnection(const boost::intrusive_ptr<Connection>& updateConnection) {
@@ -274,6 +291,9 @@ void UpdateClient::updateSession(broker::SessionHandler& sh) {
 
     // Re-create session state on remote connection.
 
+    QPID_LOG(debug, updaterId << " updating exclusive queues.");
+    ss->getSessionAdapter().eachExclusiveQueue(boost::bind(&UpdateClient::updateExclusiveQueue, this, _1));
+
     // Update consumers. For reasons unknown, boost::bind does not work here with boost 1.33.
     QPID_LOG(debug, updaterId << " updating consumers.");
     ss->getSemanticState().eachConsumer(std::bind1st(std::mem_fun(&UpdateClient::updateConsumer),this));
diff --git a/qpid/cpp/src/qpid/cluster/UpdateClient.h b/qpid/cpp/src/qpid/cluster/UpdateClient.h
index 96e2479955..030566b52d 100644
--- a/qpid/cpp/src/qpid/cluster/UpdateClient.h
+++ b/qpid/cpp/src/qpid/cluster/UpdateClient.h
@@ -81,11 +81,13 @@ class UpdateClient : public sys::Runnable {
     void updateUnacked(const broker::DeliveryRecord&);
 
   private:
-    void updateQueue(const boost::shared_ptr<broker::Queue>&);
+    void updateQueue(client::AsyncSession&, const boost::shared_ptr<broker::Queue>&);
+    void updateNonExclusiveQueue(const boost::shared_ptr<broker::Queue>&);
+    void updateExclusiveQueue(const boost::shared_ptr<broker::Queue>&);
     void updateExchange(const boost::shared_ptr<broker::Exchange>&);
     void updateMessage(const broker::QueuedMessage&);
     void updateMessageTo(const broker::QueuedMessage&, const std::string& queue, client::Session s);
-    void updateBinding(const std::string& queue, const broker::QueueBinding& binding);
+    void updateBinding(client::AsyncSession&, const std::string& queue, const broker::QueueBinding& binding);
     void updateConnection(const boost::intrusive_ptr<Connection>& connection);
     void updateSession(broker::SessionHandler& s);
     void updateTxState(broker::SemanticState& s);
diff --git a/qpid/cpp/src/tests/cluster_test.cpp b/qpid/cpp/src/tests/cluster_test.cpp
index a5f7b91c98..be7d2f9158 100644
--- a/qpid/cpp/src/tests/cluster_test.cpp
+++ b/qpid/cpp/src/tests/cluster_test.cpp
@@ -28,6 +28,7 @@
 #include "qpid/client/Session.h"
 #include "qpid/client/FailoverListener.h"
 #include "qpid/client/FailoverManager.h"
+#include "qpid/client/QueueOptions.h"
 #include "qpid/cluster/Cluster.h"
 #include "qpid/cluster/Cpg.h"
 #include "qpid/cluster/UpdateClient.h"
@@ -770,4 +771,55 @@ QPID_AUTO_TEST_CASE(testHeartbeatCancelledOnFailover)
     fmgr.close();
 }
 
+QPID_AUTO_TEST_CASE(testPolicyUpdate) {
+    ScopedSuppressLogging allQuiet;
+    //tests that the policys internal state is accurate on newly
+    //joined nodes
+    ClusterFixture::Args args;
+    args += "--log-enable", "critical";
+    prepareArgs(args, durableFlag);
+    ClusterFixture cluster(1, args, -1);
+    Client c1(cluster[0], "c1");
+    QueueOptions options;
+    options.setSizePolicy(REJECT, 0, 2);
+    c1.session.queueDeclare("q", arg::arguments=options, arg::durable=durableFlag);
+    c1.session.messageTransfer(arg::content=Message("one", "q"));
+    cluster.add();
+    Client c2(cluster[1], "c2");
+    c2.session.messageTransfer(arg::content=Message("two", "q"));
+    
+    BOOST_CHECK_THROW(c2.session.messageTransfer(arg::content=Message("three", "q")), framing::ResourceLimitExceededException);
+
+    Message received;
+    BOOST_CHECK(c1.subs.get(received, "q"));
+    BOOST_CHECK_EQUAL(received.getData(), std::string("one"));
+    BOOST_CHECK(c1.subs.get(received, "q"));
+    BOOST_CHECK_EQUAL(received.getData(), std::string("two"));
+    BOOST_CHECK(!c1.subs.get(received, "q"));
+}
+
+QPID_AUTO_TEST_CASE(testExclusiveQueueUpdate) {
+    ScopedSuppressLogging allQuiet;
+    //tests that exclusive queues are accurately replicated on newly
+    //joined nodes
+    ClusterFixture::Args args;
+    args += "--log-enable", "critical";
+    prepareArgs(args, durableFlag);
+    ClusterFixture cluster(1, args, -1);
+    Client c1(cluster[0], "c1");
+    c1.session.queueDeclare("q", arg::exclusive=true, arg::autoDelete=true, arg::alternateExchange="amq.fanout");
+    cluster.add();
+    Client c2(cluster[1], "c2");
+    QueueQueryResult result = c2.session.queueQuery("q");
+    BOOST_CHECK_EQUAL(result.getQueue(), std::string("q"));
+    BOOST_CHECK(result.getExclusive());
+    BOOST_CHECK(result.getAutoDelete());
+    BOOST_CHECK(!result.getDurable());
+    BOOST_CHECK_EQUAL(result.getAlternateExchange(), std::string("amq.fanout"));
+    BOOST_CHECK_THROW(c2.session.queueDeclare(arg::queue="q", arg::exclusive=true, arg::passive=true), framing::ResourceLockedException);
+    c1.connection.close();
+    c2.session = c2.connection.newSession();
+    BOOST_CHECK_THROW(c2.session.queueDeclare(arg::queue="q", arg::passive=true), framing::NotFoundException);
+}
+
 QPID_AUTO_TEST_SUITE_END()