concurrent_queue.hpp

Go to the documentation of this file.

#pragma once
 
#if defined(CORELINK_USE_CONCURRENT_QUEUE)
 
#include "commons/base_includes.hpp"
#include "commons/platform_macros.hpp"
#include "commons/sequence_container_includes.hpp"
#include "commons/reader_writer_lock_shim.hpp"
 
namespace corelink
{
    namespace utils
    {
        namespace containers
        {
            template<typename item_type>
            class CORELINK_EXPORT concurrent_queue
                    : public std::enable_shared_from_this<concurrent_queue<item_type>>
            {
            private:
                using queue_type = std::deque<item_type>;
 
                queue_type m_queue_container;
                commons::reader_writer_mutex m_sync_mutex;
 
            public:
                concurrent_queue()
                = default;
 
                explicit concurrent_queue(size_t initial_size)
                {
                    m_queue_container.resize(initial_size);
                }
 
                virtual ~concurrent_queue()
                {
                    clear();
                }
 
                concurrent_queue<item_type> &push(const item_type &value)
                {
                    commons::writer_lock scope_lock(m_sync_mutex);
                    m_queue_container.emplace_back(value);
                    return *this;
                }
 
                item_type &peek()
                {
                    commons::reader_lock scope_lock(m_sync_mutex);
                    return m_queue_container.front();
                }
 
                inline void pop()
                {
                    commons::writer_lock scope_lock(m_sync_mutex);
                    if (!m_queue_container.empty())
                    {
                        m_queue_container.pop_front();
                    }
                }
 
                concurrent_queue<item_type> &clear()
                {
                    commons::writer_lock scope_lock(m_sync_mutex);
                    m_queue_container.clear();
                    return *this;
                }
 
                inline size_t size()
                {
                    commons::reader_lock scoped_lock(m_sync_mutex);
                    return m_queue_container.size();
                }
 
                inline bool empty()
                {
                    commons::reader_lock scoped_lock(m_sync_mutex);
                    return m_queue_container.empty();
                }
            };
        }
    }
}
#endif //CORELINK_USE_CONCURRENT_QUEUE