diff options
| -rw-r--r-- | cpp/src/qpid/broker/IncompleteMessageList.cpp | 12 | ||||
| -rw-r--r-- | cpp/src/qpid/broker/IncompleteMessageList.h | 2 | ||||
| -rw-r--r-- | cpp/src/tests/IncompleteMessageList.cpp | 37 |
3 files changed, 48 insertions, 3 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; diff --git a/cpp/src/qpid/broker/IncompleteMessageList.h b/cpp/src/qpid/broker/IncompleteMessageList.h index f89c0023b0..98971ebff0 100644 --- a/cpp/src/qpid/broker/IncompleteMessageList.h +++ b/cpp/src/qpid/broker/IncompleteMessageList.h @@ -39,6 +39,7 @@ class IncompleteMessageList sys::Monitor lock; Messages incomplete; Message::MessageCallback callback; + bool closed; public: typedef Message::MessageCallback CompletionListener; @@ -46,6 +47,7 @@ public: IncompleteMessageList(); ~IncompleteMessageList(); + void close(); void add(boost::intrusive_ptr<Message> msg); void process(const CompletionListener& l, bool sync); void each(const CompletionListener& l); diff --git a/cpp/src/tests/IncompleteMessageList.cpp b/cpp/src/tests/IncompleteMessageList.cpp index 925cdbf43e..d9ea70f815 100644 --- a/cpp/src/tests/IncompleteMessageList.cpp +++ b/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() |