You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2008/11/27 19:48:18 UTC
svn commit: r721258 - in /incubator/qpid/trunk/qpid/cpp/src:
qpid/broker/IncompleteMessageList.cpp qpid/broker/IncompleteMessageList.h
tests/IncompleteMessageList.cpp
Author: gsim
Date: Thu Nov 27 10:48:18 2008
New Revision: 721258
URL: http://svn.apache.org/viewvc?rev=721258&view=rev
Log:
Backed out last change; not sure it actually makes sense.
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.h
incubator/qpid/trunk/qpid/cpp/src/tests/IncompleteMessageList.cpp
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.cpp?rev=721258&r1=721257&r2=721258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.cpp Thu Nov 27 10:48:18 2008
@@ -25,20 +25,13 @@
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 @@
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;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.h?rev=721258&r1=721257&r2=721258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/IncompleteMessageList.h Thu Nov 27 10:48:18 2008
@@ -39,7 +39,6 @@
sys::Monitor lock;
Messages incomplete;
Message::MessageCallback callback;
- bool closed;
public:
typedef Message::MessageCallback CompletionListener;
@@ -47,7 +46,6 @@
IncompleteMessageList();
~IncompleteMessageList();
- void close();
void add(boost::intrusive_ptr<Message> msg);
void process(const CompletionListener& l, bool sync);
void each(const CompletionListener& l);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/IncompleteMessageList.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/IncompleteMessageList.cpp?rev=721258&r1=721257&r2=721258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/IncompleteMessageList.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/IncompleteMessageList.cpp Thu Nov 27 10:48:18 2008
@@ -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 @@
list.process(Checker(3, 5), false);
}
+
struct MockStore : public NullMessageStore
{
Queue::shared_ptr queue;
@@ -126,38 +125,4 @@
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()