Fix for QPID-2470 - Broker does not honour flow-to-disk policy on recovery

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

8 files changed, 46 insertions, 8 deletions

diff --git a/cpp/src/qpid/broker/Exchange.cpp b/cpp/src/qpid/broker/Exchange.cpp
index 1d3da982d8..fed6698c36 100644
--- a/cpp/src/qpid/broker/Exchange.cpp
+++ b/cpp/src/qpid/broker/Exchange.cpp
@@ -83,8 +83,9 @@ void Exchange::doRoute(Deliverable& msg, ConstBindingList b)
 
     if (b.get()) {
         // Block the content release if the message is transient AND there is more than one binding
-        if (!msg.getMessage().isPersistent() && b->size() > 1)
+        if (!msg.getMessage().isPersistent() && b->size() > 1) {
             msg.getMessage().blockContentRelease();
+        }
 
         for(std::vector<Binding::shared_ptr>::const_iterator i = b->begin(); i != b->end(); i++, count++) {
             msg.deliverTo((*i)->queue);
diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp
index 329451d64e..65106cb99b 100644
--- a/cpp/src/qpid/broker/Message.cpp
+++ b/cpp/src/qpid/broker/Message.cpp
@@ -186,6 +186,7 @@ void Message::decodeContent(framing::Buffer& buffer)
     loaded = true;
 }
 
+// Used for testing only
 void Message::tryReleaseContent()
 {
     if (checkContentReleasable()) {
diff --git a/cpp/src/qpid/broker/PersistableMessage.cpp b/cpp/src/qpid/broker/PersistableMessage.cpp
index 76e9404b5d..62396ad995 100644
--- a/cpp/src/qpid/broker/PersistableMessage.cpp
+++ b/cpp/src/qpid/broker/PersistableMessage.cpp
@@ -170,6 +170,16 @@ bool PersistableMessage::checkContentReleasable()
     return contentReleaseState.requested && !contentReleaseState.blocked;
 }
 
+bool PersistableMessage::isContentReleaseBlocked()
+{
+    return contentReleaseState.blocked;
+}
+
+bool PersistableMessage::isContentReleaseRequested()
+{
+    return contentReleaseState.requested;
+}
+
 }}
 
 
diff --git a/cpp/src/qpid/broker/PersistableMessage.h b/cpp/src/qpid/broker/PersistableMessage.h
index 7d49491dfd..96fb922c1a 100644
--- a/cpp/src/qpid/broker/PersistableMessage.h
+++ b/cpp/src/qpid/broker/PersistableMessage.h
@@ -110,6 +110,8 @@ class PersistableMessage : public Persistable
     void requestContentRelease();
     void blockContentRelease();
     bool checkContentReleasable();
+    bool isContentReleaseBlocked();
+    bool isContentReleaseRequested();
 
     virtual QPID_BROKER_EXTERN bool isPersistent() const = 0;
 
diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 8d9248212f..d3a464a6fe 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -188,10 +188,14 @@ void Queue::recover(boost::intrusive_ptr<Message>& msg){
     msg->enqueueComplete(); // mark the message as enqueued
     mgntEnqStats(msg);
 
-    if (store && !msg->isContentLoaded()) {
+    if (store && (!msg->isContentLoaded() || msg->checkContentReleasable())) {
         //content has not been loaded, need to ensure that lazy loading mode is set:
         //TODO: find a nicer way to do this
         msg->releaseContent(store);
+        // NOTE: The log message in this section are used for flow-to-disk testing (which checks the log for the
+        // presence of this message). Do not change this without also checking these tests.
+        QPID_LOG(debug, "Message id=\"" << msg->getProperties<MessageProperties>()->getMessageId() << "\"; pid=0x" <<
+                        std::hex << msg->getPersistenceId() << std::dec << ": Content released after recovery");
     }
 }
 
diff --git a/cpp/src/qpid/broker/QueuePolicy.cpp b/cpp/src/qpid/broker/QueuePolicy.cpp
index a8aa674c53..c8feaa8a62 100644
--- a/cpp/src/qpid/broker/QueuePolicy.cpp
+++ b/cpp/src/qpid/broker/QueuePolicy.cpp
@@ -29,7 +29,9 @@ using namespace qpid::broker;
 using namespace qpid::framing;
 
 QueuePolicy::QueuePolicy(const std::string& _name, uint32_t _maxCount, uint64_t _maxSize, const std::string& _type) : 
-    maxCount(_maxCount), maxSize(_maxSize), type(_type), count(0), size(0), policyExceeded(false), name(_name) {}
+    maxCount(_maxCount), maxSize(_maxSize), type(_type), count(0), size(0), policyExceeded(false), name(_name) {
+    QPID_LOG(info, "Queue \"" << name << "\": Policy created: type=" << type << "; maxCount=" << maxCount << "; maxSize=" << maxSize);
+}
 
 void QueuePolicy::enqueued(uint64_t _size)
 {
@@ -86,7 +88,7 @@ void QueuePolicy::tryEnqueue(boost::intrusive_ptr<Message> m)
 
 void QueuePolicy::recoverEnqueued(boost::intrusive_ptr<Message> m)
 {
-    enqueued(m->contentSize());
+    tryEnqueue(m);
 }
 
 void QueuePolicy::enqueueAborted(boost::intrusive_ptr<Message> m)
diff --git a/cpp/src/qpid/broker/SemanticState.cpp b/cpp/src/qpid/broker/SemanticState.cpp
index 5148d88e72..d4fa465f37 100644
--- a/cpp/src/qpid/broker/SemanticState.cpp
+++ b/cpp/src/qpid/broker/SemanticState.cpp
@@ -413,8 +413,17 @@ void SemanticState::handle(intrusive_ptr<Message> msg) {
     } else {
         DeliverableMessage deliverable(msg);
         route(msg, deliverable);
-        if (msg->checkContentReleasable()) {
-            msg->releaseContent();
+        if (msg->isContentReleaseRequested()) {
+            // NOTE: The log messages in this section are used for flow-to-disk testing (which checks the log for the
+            // presence of these messages). Do not change these without also checking these tests.
+            if (msg->isContentReleaseBlocked()) {
+                QPID_LOG(debug, "Message id=\"" << msg->getProperties<MessageProperties>()->getMessageId() << "\"; pid=0x" <<
+                                std::hex << msg->getPersistenceId() << std::dec << ": Content release blocked");
+            } else {
+                msg->releaseContent();
+                QPID_LOG(debug, "Message id=\"" << msg->getProperties<MessageProperties>()->getMessageId() << "\"; pid=0x" <<
+                                std::hex << msg->getPersistenceId() << std::dec << ": Content released");
+            }
         }
     }
 }
diff --git a/cpp/src/qpid/broker/TxPublish.cpp b/cpp/src/qpid/broker/TxPublish.cpp
index 4b083033ea..22deb771bd 100644
--- a/cpp/src/qpid/broker/TxPublish.cpp
+++ b/cpp/src/qpid/broker/TxPublish.cpp
@@ -47,8 +47,17 @@ void TxPublish::commit() throw()
 {
     try {
         for_each(prepared.begin(), prepared.end(), Commit(msg));
-        if (msg->checkContentReleasable()) {
-            msg->releaseContent();
+        if (msg->isContentReleaseRequested()) {
+            // NOTE: The log messages in this section are used for flow-to-disk testing (which checks the log for the
+            // presence of these messages). Do not change these without also checking these tests.
+            if (msg->isContentReleaseBlocked()) {
+                QPID_LOG(debug, "Message id=\"" << msg->getProperties<qpid::framing::MessageProperties>()->getMessageId() << "\"; pid=0x" <<
+                                std::hex << msg->getPersistenceId() << std::dec << ": Content release blocked on commit");
+            } else {
+                msg->releaseContent();
+                QPID_LOG(debug, "Message id=\"" << msg->getProperties<qpid::framing::MessageProperties>()->getMessageId() << "\"; pid=0x" <<
+                                std::hex << msg->getPersistenceId() << std::dec << ": Content released on commit");
+            }
         }
     } catch (const std::exception& e) {
         QPID_LOG(error, "Failed to commit: " << e.what());