Transient flow-to-disk messages switched to store from BDB. Only single-queue transient messages are handled at this point, multi-queue bindings are blocked under all circumstances to prevent routing order dependencies from making the outcome inconsistent. BZ525813 - "Move Flow to disk from BDB to journal"

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

3 files changed, 8 insertions, 4 deletions

diff --git a/cpp/src/qpid/broker/Exchange.cpp b/cpp/src/qpid/broker/Exchange.cpp
index 9852f84f5b..757127eef2 100644
--- a/cpp/src/qpid/broker/Exchange.cpp
+++ b/cpp/src/qpid/broker/Exchange.cpp
@@ -81,6 +81,10 @@ void Exchange::doRoute(Deliverable& msg, ConstBindingList b)
     int count = 0;
 
     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)
+            msg.getMessage().blockContentRelease();
+
         for(std::vector<Binding::shared_ptr>::const_iterator i = b->begin(); i != b->end(); i++, count++) {
             msg.deliverTo((*i)->queue);
             if ((*i)->mgmtBinding != 0)
diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 5bfec0f24e..86de96468d 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -712,7 +712,7 @@ bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg,
         msg->addTraceId(traceId);
     }
 
-    if (msg->isPersistent() && store) {
+    if ((msg->isPersistent() || msg->checkContentReleasable()) && store) {
         msg->enqueueAsync(shared_from_this(), store); //increment to async counter -- for message sent to more than one queue
         boost::intrusive_ptr<PersistableMessage> pmsg = boost::static_pointer_cast<PersistableMessage>(msg);
         store->enqueue(ctxt, pmsg, *this);
@@ -743,7 +743,7 @@ bool Queue::dequeue(TransactionContext* ctxt, const QueuedMessage& msg)
             dequeued(msg);
         }
     }
-    if (msg.payload->isPersistent() && store) {
+    if ((msg.payload->isPersistent() || msg.payload->checkContentReleasable()) && store) {
         msg.payload->dequeueAsync(shared_from_this(), store); //increment to async counter -- for message sent to more than one queue
         boost::intrusive_ptr<PersistableMessage> pmsg = boost::static_pointer_cast<PersistableMessage>(msg.payload);
         store->dequeue(ctxt, pmsg, *this);
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index 0acfa1b1ff..3cfaa763ca 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -906,7 +906,7 @@ QPID_AUTO_TEST_CASE(testFlowToDiskBlocking){
     DeliverableMessage dmsg12(msg12);
     mbdFanout3.route(dmsg12, "", 0);
     msg12->tryReleaseContent();
-    BOOST_CHECK_EQUAL(msg12->isContentReleased(), true);
+    BOOST_CHECK_EQUAL(msg12->isContentReleased(), false); // XXXX - consequence of transient msg multi-queue ftd policy-handling limitations, fix in broker at some point!
     BOOST_CHECK_EQUAL(2u, dq3->getMessageCount());
     BOOST_CHECK_EQUAL(2u, dq4->getMessageCount());
     BOOST_CHECK_EQUAL(2u, dq5->getMessageCount());
@@ -924,7 +924,7 @@ QPID_AUTO_TEST_CASE(testFlowToDiskBlocking){
     DeliverableMessage dmsg14(msg14);
     mbdFanout3.route(dmsg14, "", 0);
     msg14->tryReleaseContent();
-    BOOST_CHECK_EQUAL(msg14->isContentReleased(), true);
+    BOOST_CHECK_EQUAL(msg14->isContentReleased(), false); // XXXX - consequence of transient msg multi-queue ftd policy-handling limitations, fix in broker at some point!
     BOOST_CHECK_EQUAL(4u, dq3->getMessageCount());
     BOOST_CHECK_EQUAL(4u, dq4->getMessageCount());
     BOOST_CHECK_EQUAL(4u, dq5->getMessageCount());