From da6e2b9f62966ef7d0cb69f58ffe1365af98d676 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Mon, 29 Oct 2007 21:14:44 +0000 Subject: client/BlockingQueue.h, sys/ConcurrentQueue.h: merged to sys/BlockingQueue.h - updated all users qpid/Exception.h: Removed unimplemented clone() function. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@589857 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/src/tests/Cluster.cpp | 8 +++----- cpp/src/tests/Cluster.h | 6 ++---- cpp/src/tests/Cluster_child.cpp | 3 +-- cpp/src/tests/ConcurrentQueue.cpp | 6 +++--- cpp/src/tests/InProcessBroker.h | 29 ++++++++++++++++------------- 5 files changed, 25 insertions(+), 27 deletions(-) (limited to 'cpp/src/tests') diff --git a/cpp/src/tests/Cluster.cpp b/cpp/src/tests/Cluster.cpp index 531a74b0c2..ee13bdd72a 100644 --- a/cpp/src/tests/Cluster.cpp +++ b/cpp/src/tests/Cluster.cpp @@ -36,8 +36,7 @@ BOOST_AUTO_TEST_CASE(testClusterOne) { TestCluster cluster("clusterOne", "amqp:one:1"); AMQFrame send(1, SessionOpenBody(VER)); cluster.handle(send); - AMQFrame received; - BOOST_REQUIRE(cluster.received.waitPop(received)); + AMQFrame received = cluster.received.pop(); BOOST_CHECK_TYPEID_EQUAL(SessionOpenBody, *received.getBody()); BOOST_CHECK_EQUAL(1u, cluster.size()); Cluster::MemberList members = cluster.getMembers(); @@ -62,11 +61,10 @@ BOOST_AUTO_TEST_CASE(testClusterTwo) { // Exchange frames with child. AMQFrame send(1, SessionOpenBody(VER)); cluster.handle(send); - AMQFrame received; - BOOST_REQUIRE(cluster.received.waitPop(received)); + AMQFrame received = cluster.received.pop(); BOOST_CHECK_TYPEID_EQUAL(SessionOpenBody, *received.getBody()); - BOOST_REQUIRE(cluster.received.waitPop(received)); + received=cluster.received.pop(); BOOST_CHECK_TYPEID_EQUAL(SessionAttachedBody, *received.getBody()); if (!nofork) { diff --git a/cpp/src/tests/Cluster.h b/cpp/src/tests/Cluster.h index d14d7c1392..6ff5c21fdb 100644 --- a/cpp/src/tests/Cluster.h +++ b/cpp/src/tests/Cluster.h @@ -20,7 +20,7 @@ */ #include "qpid/cluster/Cluster.h" -#include "qpid/sys/ConcurrentQueue.h" +#include "qpid/sys/BlockingQueue.h" #include "qpid/framing/AMQFrame.h" #include @@ -45,12 +45,10 @@ using namespace boost; void null_deleter(void*) {} template -class TestHandler : public Handler, public ConcurrentQueue +class TestHandler : public Handler, public BlockingQueue { public: void handle(T& frame) { push(frame); } - bool waitPop(T& x) { return waitPop(x, TIME_SEC); } - using ConcurrentQueue::waitPop; }; typedef TestHandler TestFrameHandler; diff --git a/cpp/src/tests/Cluster_child.cpp b/cpp/src/tests/Cluster_child.cpp index c03d7396f0..8d0682473b 100644 --- a/cpp/src/tests/Cluster_child.cpp +++ b/cpp/src/tests/Cluster_child.cpp @@ -35,8 +35,7 @@ static const ProtocolVersion VER; /** Child part of Cluster::clusterTwo test */ void clusterTwo() { TestCluster cluster("clusterTwo", "amqp:child:2"); - AMQFrame frame; - BOOST_REQUIRE(cluster.received.waitPop(frame)); // Frame from parent. + AMQFrame frame = cluster.received.pop(frame); // Frame from parent. BOOST_CHECK_TYPEID_EQUAL(SessionOpenBody, *frame.getBody()); BOOST_CHECK_EQUAL(2u, cluster.size()); // Me and parent diff --git a/cpp/src/tests/ConcurrentQueue.cpp b/cpp/src/tests/ConcurrentQueue.cpp index 39155b4ff2..c6ca40e897 100644 --- a/cpp/src/tests/ConcurrentQueue.cpp +++ b/cpp/src/tests/ConcurrentQueue.cpp @@ -20,10 +20,10 @@ */ /**@file - * Compare alternative implementations for ConcurrentQueue. + * Compare alternative implementations for BlockingQueue. */ -#include "qpid/sys/ConcurrentQueue.h" +#include "qpid/sys/BlockingQueue.h" #include "qpid/sys/Thread.h" #include "qpid/sys/Monitor.h" #include "qpid/sys/Runnable.h" @@ -83,7 +83,7 @@ template class DualVectorDualLockQueue { typename std::vector::iterator popIter; }; -template struct LockedDequeQueue : public ConcurrentQueue { +template struct LockedDequeQueue : public BlockingQueue { /** size_t ignored, can't pre-allocate space in a dequeue */ LockedDequeQueue(size_t=0) {}; }; diff --git a/cpp/src/tests/InProcessBroker.h b/cpp/src/tests/InProcessBroker.h index c5860568db..3f6ff0936e 100644 --- a/cpp/src/tests/InProcessBroker.h +++ b/cpp/src/tests/InProcessBroker.h @@ -26,7 +26,7 @@ #include "qpid/client/Connection.h" #include "qpid/log/Statement.h" #include "qpid/sys/Thread.h" -#include "qpid/sys/ConcurrentQueue.h" +#include "qpid/sys/BlockingQueue.h" #include "qpid/shared_ptr.h" #include @@ -65,26 +65,29 @@ class InProcessConnector : } ~NetworkQueue() { - queue.shutdown(); + queue.close(); thread.join(); } void push(AMQFrame& f) { queue.push(f); } void run() { - AMQFrame f; - while (queue.waitPop(f)) { - Lock l(lock); - if (inputHandler) { - QPID_LOG(debug, QPID_MSG(receiver << " RECV: " << f)); - inputHandler->handle(f); - } - else { - QPID_LOG(debug, QPID_MSG(receiver << " DROP: " << f)); + try { + while(true) { + AMQFrame f = queue.pop(); + if (inputHandler) { + QPID_LOG(debug, QPID_MSG(receiver << " RECV: " << f)); + inputHandler->handle(f); + } + else + QPID_LOG(debug, QPID_MSG(receiver << " DROP: " << f)); } } + catch (const sys::QueueClosed&) { + return; + } } - + void setInputHandler(FrameHandler* h) { Lock l(lock); inputHandler = h; @@ -92,7 +95,7 @@ class InProcessConnector : private: sys::Mutex lock; - sys::ConcurrentQueue queue; + sys::BlockingQueue queue; sys::Thread thread; FrameHandler* inputHandler; const char* const receiver; -- cgit v1.2.1