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

5 files changed, 25 insertions, 27 deletions

diff --git a/qpid/cpp/src/tests/Cluster.cpp b/qpid/cpp/src/tests/Cluster.cpp
index 531a74b0c2..ee13bdd72a 100644
--- a/qpid/cpp/src/tests/Cluster.cpp
+++ b/qpid/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/qpid/cpp/src/tests/Cluster.h b/qpid/cpp/src/tests/Cluster.h
index d14d7c1392..6ff5c21fdb 100644
--- a/qpid/cpp/src/tests/Cluster.h
+++ b/qpid/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 <boost/bind.hpp>
@@ -45,12 +45,10 @@ using namespace boost;
 void null_deleter(void*) {}
 
 template <class T>
-class TestHandler : public Handler<T&>, public ConcurrentQueue<T>
+class TestHandler : public Handler<T&>, public BlockingQueue<T>
 {
   public:
     void handle(T& frame) { push(frame); }
-    bool waitPop(T& x) { return waitPop(x, TIME_SEC); }
-    using ConcurrentQueue<T>::waitPop;
 };
 
 typedef TestHandler<AMQFrame> TestFrameHandler;
diff --git a/qpid/cpp/src/tests/Cluster_child.cpp b/qpid/cpp/src/tests/Cluster_child.cpp
index c03d7396f0..8d0682473b 100644
--- a/qpid/cpp/src/tests/Cluster_child.cpp
+++ b/qpid/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/qpid/cpp/src/tests/ConcurrentQueue.cpp b/qpid/cpp/src/tests/ConcurrentQueue.cpp
index 39155b4ff2..c6ca40e897 100644
--- a/qpid/cpp/src/tests/ConcurrentQueue.cpp
+++ b/qpid/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 T> class DualVectorDualLockQueue {
     typename std::vector<T>::iterator popIter;
 };
 
-template <class T> struct LockedDequeQueue : public ConcurrentQueue<T> {
+template <class T> struct LockedDequeQueue : public BlockingQueue<T> {
     /** size_t ignored, can't pre-allocate space in a dequeue */
     LockedDequeQueue(size_t=0) {};
 };
diff --git a/qpid/cpp/src/tests/InProcessBroker.h b/qpid/cpp/src/tests/InProcessBroker.h
index c5860568db..3f6ff0936e 100644
--- a/qpid/cpp/src/tests/InProcessBroker.h
+++ b/qpid/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 <vector>
@@ -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<AMQFrame> queue;
+        sys::BlockingQueue<AMQFrame> queue;
         sys::Thread thread;
         FrameHandler* inputHandler;
         const char* const receiver;