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@721256 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 36 insertions, 1 deletions

diff --git a/qpid/cpp/src/tests/IncompleteMessageList.cpp b/qpid/cpp/src/tests/IncompleteMessageList.cpp
index 925cdbf43e..d9ea70f815 100644
--- a/qpid/cpp/src/tests/IncompleteMessageList.cpp
+++ b/qpid/cpp/src/tests/IncompleteMessageList.cpp
@@ -24,6 +24,8 @@
 #include "qpid/broker/NullMessageStore.h"
 #include "qpid/broker/Queue.h"
 #include "qpid/broker/IncompleteMessageList.h"
+#include "qpid/sys/Runnable.h"
+#include "qpid/sys/Thread.h"
 
 #include "unit_test.h"
 
@@ -93,7 +95,6 @@ QPID_AUTO_TEST_CASE(testProcessWithIncomplete)
     list.process(Checker(3, 5), false);    
 }
 
-
 struct MockStore : public NullMessageStore
 {
     Queue::shared_ptr queue;
@@ -125,4 +126,38 @@ QPID_AUTO_TEST_CASE(testSyncProcessWithIncomplete)
     list.process(Checker(1, 5), true);
 }
 
+struct AsyncProcessor : qpid::sys::Runnable
+{
+    Checker checker;
+    IncompleteMessageList& list;
+
+    AsyncProcessor(uint start, uint end, IncompleteMessageList& list_) : checker(start, end), list(list_) {}
+    
+    void run() 
+    {
+        list.process(checker, true);
+    }
+};
+
+QPID_AUTO_TEST_CASE(testSyncProcessInterruptedOnClose)
+{
+    IncompleteMessageList list;
+    SequenceNumber counter(1);
+    NullMessageStore store;
+    Queue::shared_ptr queue(new Queue("mock-queue"));
+    //fill up list with messages
+    for (int i = 0; i < 5; i++) {
+        boost::intrusive_ptr<Message> msg(new Message(counter++));
+        list.add(msg);
+        if (i == 2) {
+            //mark a message in the middle as incomplete
+            msg->enqueueAsync(queue, &store);
+        }
+    }
+    AsyncProcessor ap(1, 2, list);
+    qpid::sys::Thread thread(ap);
+    list.close();
+    thread.join();
+}
+
 QPID_AUTO_TEST_SUITE_END()