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

7 files changed, 29 insertions, 120 deletions

diff --git a/cpp/src/qpid/client/BlockingQueue.h b/cpp/src/qpid/client/BlockingQueue.h
deleted file mode 100644
index 3ec7dbcf44..0000000000
--- a/cpp/src/qpid/client/BlockingQueue.h
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#ifndef _BlockingQueue_
-#define _BlockingQueue_
-
-#include <queue>
-#include "qpid/sys/Monitor.h"
-
-namespace qpid {
-namespace client {
-
-struct QueueClosed {};
-
-template <class T>
-class BlockingQueue 
-{
-    sys::Monitor lock;
-    std::queue<T> queue;
-    bool closed;
-
-public:
-
-    BlockingQueue() : closed(false) {}
-
-    void reset() 
-    {
-        sys::Monitor::ScopedLock l(lock);
-        closed = true; 
-    }
-
-    T pop()
-    {
-        sys::Monitor::ScopedLock l(lock);
-        while (!closed && queue.empty()) {
-            lock.wait();
-        } 
-        if (closed) {
-            throw QueueClosed();
-        } else {
-            T t = queue.front();
-            queue.pop();
-            return t;
-        }
-    }
-
-    void push(const T& t)
-    {
-        sys::Monitor::ScopedLock l(lock);
-        bool wasEmpty = queue.empty();
-        queue.push(t);
-        if (wasEmpty) {
-            lock.notifyAll();
-        }
-    }
-
-    void close()
-    {
-        sys::Monitor::ScopedLock l(lock);
-        closed = true;
-        lock.notifyAll();
-    }
-    
-    bool empty()
-    {
-        sys::Monitor::ScopedLock l(lock);
-        return queue.empty();
-    }
-};
-
-}}
-
-
-
-#endif
diff --git a/cpp/src/qpid/client/Channel.cpp b/cpp/src/qpid/client/Channel.cpp
index 8b7b7e1118..fbb2e0c6f8 100644
--- a/cpp/src/qpid/client/Channel.cpp
+++ b/cpp/src/qpid/client/Channel.cpp
@@ -187,13 +187,14 @@ bool Channel::get(Message& msg, const Queue& _queue, AckMode ackMode) {
     status.sync();
     session.messageCancel(tag);
 
-    if (incoming.empty()) {
-        return false;
-    } else {
-        msg.populate(*(incoming.pop()));
+    FrameSet::shared_ptr p;
+    if (incoming.tryPop(p)) {
+        msg.populate(*p);
         if (ackMode == AUTO_ACK) msg.acknowledge(session, false, true);
         return true;
     }
+    else
+        return false;
 }
 
 void Channel::publish(Message& msg, const Exchange& exchange,
@@ -263,7 +264,7 @@ void Channel::run() {
                 QPID_LOG(warning, "Dropping unsupported message type: " << content->getMethod());                        
             }
         }
-    } catch (const QueueClosed&) {}
+    } catch (const sys::QueueClosed&) {}
 }
 
 }}
diff --git a/cpp/src/qpid/client/Channel.h b/cpp/src/qpid/client/Channel.h
index 19a3a2de8c..c1f74e0706 100644
--- a/cpp/src/qpid/client/Channel.h
+++ b/cpp/src/qpid/client/Channel.h
@@ -80,7 +80,7 @@ class Channel : private sys::Runnable
     ConsumerMap consumers;
     Session_0_10 session;
     framing::ChannelId channelId;
-    BlockingQueue<framing::FrameSet::shared_ptr> gets;
+    sys::BlockingQueue<framing::FrameSet::shared_ptr> gets;
     framing::Uuid uniqueId;
     uint32_t nameCounter;
     bool active;
diff --git a/cpp/src/qpid/client/Demux.h b/cpp/src/qpid/client/Demux.h
index 0f261f70ba..3c9d4a4857 100644
--- a/cpp/src/qpid/client/Demux.h
+++ b/cpp/src/qpid/client/Demux.h
@@ -24,7 +24,7 @@
 #include <boost/shared_ptr.hpp>
 #include "qpid/framing/FrameSet.h"
 #include "qpid/sys/Mutex.h"
-#include "BlockingQueue.h"
+#include "qpid/sys/BlockingQueue.h"
 
 #ifndef _Demux_
 #define _Demux_
@@ -44,7 +44,7 @@ class Demux
 {
 public:
     typedef boost::function<bool(const framing::FrameSet&)> Condition;
-    typedef BlockingQueue<framing::FrameSet::shared_ptr> Queue;
+    typedef sys::BlockingQueue<framing::FrameSet::shared_ptr> Queue;
 
     void handle(framing::FrameSet::shared_ptr);
     void close();
diff --git a/cpp/src/qpid/client/Dispatcher.cpp b/cpp/src/qpid/client/Dispatcher.cpp
index ea6fcfb463..37f3941022 100644
--- a/cpp/src/qpid/client/Dispatcher.cpp
+++ b/cpp/src/qpid/client/Dispatcher.cpp
@@ -24,7 +24,7 @@
 #include "qpid/framing/FrameSet.h"
 #include "qpid/framing/MessageTransferBody.h"
 #include "qpid/log/Statement.h"
-#include "BlockingQueue.h"
+#include "qpid/sys/BlockingQueue.h"
 #include "Message.h"
 
 using qpid::framing::FrameSet;
@@ -62,7 +62,7 @@ void Dispatcher::start()
 
 void Dispatcher::run()
 {    
-    BlockingQueue<FrameSet::shared_ptr>& q = queue.empty() ? 
+    sys::BlockingQueue<FrameSet::shared_ptr>& q = queue.empty() ? 
         session.execution().getDemux().getDefault() : 
         session.execution().getDemux().get(queue); 
 
diff --git a/cpp/src/qpid/client/MessageQueue.h b/cpp/src/qpid/client/MessageQueue.h
index 1f5c492910..8ff537701f 100644
--- a/cpp/src/qpid/client/MessageQueue.h
+++ b/cpp/src/qpid/client/MessageQueue.h
@@ -22,28 +22,29 @@
 #ifndef _MessageQueue_
 #define _MessageQueue_
 #include <iostream>
-#include "BlockingQueue.h"
+#include "qpid/sys/BlockingQueue.h"
 #include "MessageListener.h"
 
 namespace qpid {
 namespace client {
 
-    /**
-     * A MessageListener implementation that simply queues up
-     * messages.
-     *
-     * \ingroup clientapi
-     */
-    class MessageQueue : public MessageListener, public BlockingQueue<Message>
+/**
+ * A MessageListener implementation that simply queues up
+ * messages.
+ *
+ * \ingroup clientapi
+ */
+class MessageQueue : public MessageListener,
+                     public sys::BlockingQueue<Message>
+{
+    std::queue<Message> messages;
+  public:
+    void received(Message& msg)
     {
-        std::queue<Message> messages;
-    public:
-        void received(Message& msg)
-        {
-            std::cout << "Adding message to queue: " << msg.getData() << std::endl;
-            push(msg);
-        }
-    };
+        std::cout << "Adding message to queue: " << msg.getData() << std::endl;
+        push(msg);
+    }
+};
 
 }
 }
diff --git a/cpp/src/qpid/client/SessionCore.cpp b/cpp/src/qpid/client/SessionCore.cpp
index 27440465fe..ea877f9b40 100644
--- a/cpp/src/qpid/client/SessionCore.cpp
+++ b/cpp/src/qpid/client/SessionCore.cpp
@@ -120,7 +120,7 @@ SessionCore::~SessionCore() {
     Lock l(state);
     invariant();
     detach(COMMAND_INVALID, "Session deleted");
-    state.waitAll();
+    state.waitWaiters();
 }
 
 void SessionCore::detach(int c, const std::string& t) {