ibrcommon/ibrcommon/thread/ThreadSafeQueue.h

Go to the documentation of this file.

/*
 * ThreadSafeQueue.h
 *
 *  Created on: 10.06.2010
 *      Author: morgenro
 */

#ifndef THREADSAFEQUEUE_H_
#define THREADSAFEQUEUE_H_

#include "ibrcommon/thread/MutexLock.h"
#include "ibrcommon/thread/Conditional.h"
#include <queue>

namespace ibrcommon
{
        template <class T>
        class LockedQueue
        {
        public:
                LockedQueue(ibrcommon::Mutex &mutex, std::queue<T> &queue) : _lock(mutex), _queue(queue)
                {
                };

                virtual ~LockedQueue()
                {
                };

                std::queue<T>& operator*()
                {
                        return _queue;
                }

        private:
                ibrcommon::MutexLock _lock;
                std::queue<T> &_queue;
        };

        template <class T>
        class ThreadSafeQueue : public ibrcommon::Conditional
        {
        public:
                ThreadSafeQueue() : _queue(), _unblock(false)
                {};

                virtual ~ThreadSafeQueue()
                {};

                /* Test whether container is empty (public member function) */
                bool empty ( )
                {
                        ibrcommon::MutexLock l(*this);
                        return _queue.empty();
                }

                /* Return size (public member function) */
                size_t size ( )
                {
                        ibrcommon::MutexLock l(*this);
                        return _queue.size();
                }

                /* Access next element (public member function) */
                T front ( )
                {
                        ibrcommon::MutexLock l(*this);
                        return _queue.front();
                }

                const T front ( ) const
                {
                        ibrcommon::MutexLock l(*this);
                        return _queue.front();
                }

                /* Access last element (public member function) */
                T back ( )
                {
                        ibrcommon::MutexLock l(*this);
                        return _queue.back();
                }

                const T back ( ) const
                {
                        ibrcommon::MutexLock l(*this);
                        return _queue.back();
                }

                /* Insert element (public member function) */
                void push ( const T& x )
                {
                        ibrcommon::MutexLock l(*this);
                        _queue.push(x);
                        (*this).signal(true);
                }

                /* Delete next element (public member function) */
                void pop ( )
                {
                        ibrcommon::MutexLock l(*this);
                        _queue.pop();
                }

                T frontpop() throw (ibrcommon::Exception)
                {
                        ibrcommon::MutexLock l(*this);
                        if (_queue.empty())
                        {
                                throw ibrcommon::Exception("pop on empty queue");
                        }

                        T ret = _queue.front();
                        _queue.pop();
                        return ret;
                }

                T blockingpop(size_t timeout = 0) throw (ibrcommon::Exception)
                {
                        ibrcommon::MutexLock l(*this);
                        if (_unblock) throw ibrcommon::Exception("queue unblocked");

                        if (_queue.empty())
                        {
                                if (timeout > 0)
                                {
                                        (*this).wait(timeout);
                                }
                                else
                                {
                                        (*this).wait();
                                }

                                if (_queue.empty() || _unblock) throw ibrcommon::Exception("queue unblocked without an element");
                        }

                        T ret = _queue.front();
                        _queue.pop();
                        return ret;
                }

                void unblock() throw ()
                {
                        ibrcommon::MutexLock l(*this);
                        _unblock = true;
                        (*this).signal(true);
                }

                void reset() throw ()
                {
                        ibrcommon::MutexLock l(*this);
                        _unblock = false;
                }

                LockedQueue<T> LockedAccess()
                {
                        return LockedQueue<T>(*this, _queue);
                }

        protected:
                std::queue<T> _queue;
                bool _unblock;
        };
}

#endif /* THREADSAFEQUEUE_H_ */

---