Ensure broker doesn't hang waiting for async enqueue to complete on shutdown.

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

1 files changed, 10 insertions, 2 deletions

diff --git a/cpp/src/qpid/broker/IncompleteMessageList.cpp b/cpp/src/qpid/broker/IncompleteMessageList.cpp
index 2077e633ec..fbeec7afae 100644
--- a/cpp/src/qpid/broker/IncompleteMessageList.cpp
+++ b/cpp/src/qpid/broker/IncompleteMessageList.cpp
@@ -25,13 +25,20 @@ namespace qpid {
 namespace broker {
 
 IncompleteMessageList::IncompleteMessageList() :
-    callback(boost::bind(&IncompleteMessageList::enqueueComplete, this, _1))
+    callback(boost::bind(&IncompleteMessageList::enqueueComplete, this, _1)), closed(false)
 {}
 
 IncompleteMessageList::~IncompleteMessageList() 
 {
+    close();
+}
+
+void IncompleteMessageList::close() 
+{
     sys::Mutex::ScopedLock l(lock);
+    closed = true;
     std::for_each(incomplete.begin(), incomplete.end(), boost::bind(&Message::resetEnqueueCompleteCallback, _1));
+    lock.notify();
 }
 
 void IncompleteMessageList::add(boost::intrusive_ptr<Message> msg)
@@ -57,8 +64,9 @@ void IncompleteMessageList::process(const CompletionListener& listen, bool sync)
                     sys::Mutex::ScopedUnlock u(lock);
                     msg->flush(); // Can re-enter IncompleteMessageList::enqueueComplete
                 }
-                while (!msg->isEnqueueComplete())
+                while (!msg->isEnqueueComplete() && !closed)
                     lock.wait();
+                if (closed) return;
             } else {
                 //leave the message as incomplete for now
                 return;