1 files changed, 18 insertions, 5 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 8c50f26abd..bc29815e84 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -176,7 +176,7 @@ void Queue::deliver(boost::intrusive_ptr<Message>& msg){
 
 
 void Queue::recover(boost::intrusive_ptr<Message>& msg){
-    push(msg);
+    push(msg, true);
     msg->enqueueComplete(); // mark the message as enqueued
     mgntEnqStats(msg);
 
@@ -545,7 +545,7 @@ void Queue::popMsg(QueuedMessage& qmsg)
     ++dequeueTracker;
 }
 
-void Queue::push(boost::intrusive_ptr<Message>& msg){
+void Queue::push(boost::intrusive_ptr<Message>& msg, bool isRecovery){
     QueueListeners::NotificationSet copy;
     {
         Mutex::ScopedLock locker(messageLock);   
@@ -566,7 +566,13 @@ void Queue::push(boost::intrusive_ptr<Message>& msg){
                 boost::intrusive_ptr<Message> old = i->second->getReplacementMessage(this);
                 if (!old) old = i->second;
                 i->second->setReplacementMessage(msg,this);
-                dequeued(QueuedMessage(qm.queue, old, qm.position));
+                if (isRecovery) {
+                    //can't issue new requests for the store until
+                    //recovery is complete
+                    pendingDequeues.push_back(QueuedMessage(qm.queue, old, qm.position));
+                } else {
+                    dequeue(0, QueuedMessage(qm.queue, old, qm.position));
+                }
             }		 
         }else {
             messages.push_back(qm);
@@ -664,7 +670,7 @@ bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg)
         msg->addTraceId(traceId);
     }
 
-    if (msg->isPersistent() && store && !lastValueQueue) {
+    if (msg->isPersistent() && 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);
@@ -683,7 +689,7 @@ bool Queue::dequeue(TransactionContext* ctxt, const QueuedMessage& msg)
             dequeued(msg);
         }
     }
-    if (msg.payload->isPersistent() && store && !lastValueQueue) {
+    if (msg.payload->isPersistent() && 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);
@@ -976,3 +982,10 @@ void Queue::setQueueEventManager(QueueEvents& mgr)
 {
     eventMgr = &mgr;
 }
+
+void Queue::recoveryComplete()
+{
+    //process any pending dequeues
+    for_each(pendingDequeues.begin(), pendingDequeues.end(), boost::bind(&Queue::dequeue, this, (TransactionContext*) 0, _1));
+    pendingDequeues.clear();
+}