diff options
author | Rajith Muditha Attapattu <rajith@apache.org> | 2011-05-27 15:44:23 +0000 |
---|---|---|
committer | Rajith Muditha Attapattu <rajith@apache.org> | 2011-05-27 15:44:23 +0000 |
commit | 66765100f4257159622cefe57bed50125a5ad017 (patch) | |
tree | a88ee23bb194eb91f0ebb2d9b23ff423e3ea8e37 /qpid/cpp/src/qpid/sys/BlockingQueue.h | |
parent | 1aeaa7b16e5ce54f10c901d75c4d40f9f88b9db6 (diff) | |
parent | 88b98b2f4152ef59a671fad55a0d08338b6b78ca (diff) | |
download | qpid-python-66765100f4257159622cefe57bed50125a5ad017.tar.gz |
Creating a branch for experimenting with some ideas for JMS client.rajith_jms_client
git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/rajith_jms_client@1128369 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/cpp/src/qpid/sys/BlockingQueue.h')
-rw-r--r-- | qpid/cpp/src/qpid/sys/BlockingQueue.h | 129 |
1 files changed, 129 insertions, 0 deletions
diff --git a/qpid/cpp/src/qpid/sys/BlockingQueue.h b/qpid/cpp/src/qpid/sys/BlockingQueue.h new file mode 100644 index 0000000000..ca6b529930 --- /dev/null +++ b/qpid/cpp/src/qpid/sys/BlockingQueue.h @@ -0,0 +1,129 @@ +#ifndef QPID_SYS_BLOCKINGQUEUE_H +#define QPID_SYS_BLOCKINGQUEUE_H + +/* + * + * 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. + * + */ + +#include "qpid/sys/Waitable.h" + +#include <queue> + +namespace qpid { +namespace sys { + +/** + * A simple blocking queue template + */ +template <class T> +class BlockingQueue +{ + mutable sys::Waitable waitable; + std::queue<T> queue; + +public: + BlockingQueue() {} + ~BlockingQueue() { close(); } + + /** Pop from the queue, block up to timeout if empty. + *@param result Set to value popped from queue. + *@param timeout Defaults to infinite. + *@return true if result was set, false if queue empty after timeout. + */ + bool pop(T& result, Duration timeout=TIME_INFINITE) { + Mutex::ScopedLock l(waitable); + { + Waitable::ScopedWait w(waitable); + if (timeout == TIME_INFINITE) { + while (queue.empty()) waitable.wait(); + } else if (timeout) { + AbsTime deadline(now(),timeout); + while (queue.empty() && deadline > now()) waitable.wait(deadline); + } else { + //ensure zero timeout pop does not miss the fact that + //queue is closed + waitable.checkException(); + } + } + if (queue.empty()) return false; + result = queue.front(); + queue.pop(); + if (!queue.empty()) + waitable.notify(); // Notify another waiter. + return true; + } + + T pop(Duration timeout=TIME_INFINITE) { + T result; + bool ok = pop(result, timeout); + if (!ok) + throw Exception("Timed out waiting on a blocking queue"); + return result; + } + + /** Push a value onto the queue. + * Note it is not an error to push onto a closed queue. + */ + void push(const T& t) { + Mutex::ScopedLock l(waitable); + queue.push(t); + waitable.notify(); // Notify a waiter. + } + + /** + * Close the queue. + *@ex exception to throw to waiting threads. ClosedException by default. + */ + void close(const ExceptionHolder& ex=ExceptionHolder(new ClosedException())) + { + Mutex::ScopedLock l(waitable); + if (!waitable.hasException()) { + waitable.setException(ex); + waitable.notifyAll(); + waitable.waitWaiters(); // Ensure no threads are still waiting. + } + } + + /** Open a closed queue. */ + void open() { + Mutex::ScopedLock l(waitable); + waitable.resetException(); + } + + bool isClosed() const { + Mutex::ScopedLock l(waitable); + return waitable.hasException(); + } + + bool empty() const { + Mutex::ScopedLock l(waitable); + return queue.empty(); + } + size_t size() const { + Mutex::ScopedLock l(waitable); + return queue.size(); + } +}; + +}} + + + +#endif /*!QPID_SYS_BLOCKINGQUEUE_H*/ |