Correcting to ensure recovered cluster-durable messages are dequeued from store

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

1 files changed, 8 insertions, 2 deletions

diff --git a/cpp/src/qpid/broker/RecoveryManagerImpl.cpp b/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
index ae5faf9dc5..edddfc6db8 100644
--- a/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
+++ b/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
@@ -45,8 +45,7 @@ class RecoverableMessageImpl : public RecoverableMessage
     intrusive_ptr<Message> msg;
     const uint64_t stagingThreshold;
 public:
-    RecoverableMessageImpl(const intrusive_ptr<Message>& _msg, uint64_t _stagingThreshold) 
-        : msg(_msg), stagingThreshold(_stagingThreshold) {}
+    RecoverableMessageImpl(const intrusive_ptr<Message>& _msg, uint64_t _stagingThreshold); 
     ~RecoverableMessageImpl() {};
     void setPersistenceId(uint64_t id);
     bool loadContent(uint64_t available);
@@ -160,6 +159,13 @@ void RecoveryManagerImpl::recoveryComplete()
     queues.eachQueue(boost::bind(&Queue::recoveryComplete, _1));
 }
 
+RecoverableMessageImpl:: RecoverableMessageImpl(const intrusive_ptr<Message>& _msg, uint64_t _stagingThreshold) : msg(_msg), stagingThreshold(_stagingThreshold) 
+{
+    if (!msg->isPersistent()) {
+        msg->forcePersistent(); // set so that message will get dequeued from store.
+    }
+}
+
 bool RecoverableMessageImpl::loadContent(uint64_t available)
 {
     return !stagingThreshold || available < stagingThreshold;