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, 3 insertions, 48 deletions
diff --git a/cpp/src/qpid/broker/IncompleteMessageList.cpp b/cpp/src/qpid/broker/IncompleteMessageList.cpp index fbeec7afae..2077e633ec 100644 --- a/cpp/src/qpid/broker/IncompleteMessageList.cpp +++ b/cpp/src/qpid/broker/IncompleteMessageList.cpp @@ -25,20 +25,13 @@ namespace qpid { namespace broker { IncompleteMessageList::IncompleteMessageList() : - callback(boost::bind(&IncompleteMessageList::enqueueComplete, this, _1)), closed(false) + callback(boost::bind(&IncompleteMessageList::enqueueComplete, this, _1)) {} 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) @@ -64,9 +57,8 @@ void IncompleteMessageList::process(const CompletionListener& listen, bool sync) sys::Mutex::ScopedUnlock u(lock); msg->flush(); // Can re-enter IncompleteMessageList::enqueueComplete } - while (!msg->isEnqueueComplete() && !closed) + while (!msg->isEnqueueComplete()) 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 98971ebff0..f89c0023b0 100644 --- a/cpp/src/qpid/broker/IncompleteMessageList.h +++ b/cpp/src/qpid/broker/IncompleteMessageList.h @@ -39,7 +39,6 @@ class IncompleteMessageList sys::Monitor lock; Messages incomplete; Message::MessageCallback callback; - bool closed; public: typedef Message::MessageCallback CompletionListener; @@ -47,7 +46,6 @@ 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 d9ea70f815..925cdbf43e 100644 --- a/cpp/src/tests/IncompleteMessageList.cpp +++ b/cpp/src/tests/IncompleteMessageList.cpp @@ -24,8 +24,6 @@ #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" @@ -95,6 +93,7 @@ QPID_AUTO_TEST_CASE(testProcessWithIncomplete) list.process(Checker(3, 5), false); } + struct MockStore : public NullMessageStore { Queue::shared_ptr queue; @@ -126,38 +125,4 @@ 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() |