ParaRingBuffer.h

#pragma once
#include <boost/circular_buffer.hpp>
#include <boost/thread.hpp>
#include "util/mutex.h"
#include <queue>


namespace ParaEngine
{
    template <class T>
    class CParaRingBuffer
    {
    public:
        typedef boost::circular_buffer<T> container_type;
        typedef typename container_type::size_type size_type;
        typedef typename container_type::value_type value_type;

        enum BufferStatus
        {
            BufferOverFlow = 0,
            BufferFull = 1,
            BufferNormal = 2,
            BufferEmpty = 3,
            BufferFirst = 3,
        };

        explicit CParaRingBuffer(size_type capacity) : m_container(capacity) {}

    public:
        BufferStatus try_push(const value_type& item) 
        {
            Lock lock(m_mutex);
            if(m_container.full())
            {
                return BufferOverFlow;
            }
            else
            {
                BufferStatus bufferStatus =  m_container.empty() ? BufferFirst : BufferNormal;
                m_container.push_back(item);
                return m_container.full() ? BufferFull : bufferStatus;
            }
        }

        BufferStatus try_push_get_front(const value_type& item, value_type** ppFrontItem) 
        {
            Lock lock(m_mutex);
            if(m_container.full())
            {
                return BufferOverFlow;
            }
            else
            {
                BufferStatus bufferStatus =  m_container.empty() ? BufferFirst : BufferNormal;
                m_container.push_back(item);
                *ppFrontItem = &(m_container.front());
                return m_container.full() ? BufferFull : bufferStatus;
            }
        }

        void push_front(const value_type& item) 
        {
            Lock lock(m_mutex);
            m_container.push_front(item);
        }

        BufferStatus try_push_array(const value_type* pItems, int nCount) 
        {
            Lock lock(m_mutex);
            if(m_container.capacity()<nCount)
            {
                return BufferFull;
            }
            else
            {
                BufferStatus bufferStatus =  m_container.empty() ? BufferFirst : BufferNormal;
                for (int i=0; i<nCount;++i)
                {
                    m_container.push_back(*(pItems+i));
                }
                return m_container.full() ? BufferFull : bufferStatus;
            }
        }

        value_type * try_front()
        {
            Lock lock(m_mutex);
            return m_container.empty() ? NULL : &(m_container.front());
        }

        bool try_pop(value_type& item) 
        {
            Lock lock(m_mutex);
            if(m_container.empty())
            {
                return false;
            }
            else
            {
                item = m_container.front();
                m_container.pop_front();
                return true;
            }
        }

        bool try_next(value_type** ppItem) 
        {
            Lock lock(m_mutex);
            if(m_container.empty())
            {
                return false;
            }
            else
            {
                m_container.pop_front();
                if(ppItem!=0 &&  !m_container.empty())
                {
                    *ppItem  = &(m_container.front());
                }
                return true;
            }
        }

        size_type size() const {
            Lock lock(m_mutex);
            return m_container.size();
        };
        bool empty() const {
            Lock lock(m_mutex);
            return m_container.empty();
        };
        bool full() const {
            Lock lock(m_mutex);
            return m_container.full();
        };
    public:
        container_type m_container;
        mutex m_mutex;
    };

    template<typename Data>
    class concurrent_queue
    {
    protected:
        typedef boost::circular_buffer<Data> container_type;
        typedef typename container_type::size_type size_type;
        typedef typename container_type::value_type value_type;

        mutable boost::mutex m_mutex;
        boost::condition_variable m_condition_variable;

        bool m_use_event;

        container_type m_container;
    public:
        enum BufferStatus
        {
            BufferOverFlow = 0,
            BufferFull = 1,
            BufferNormal = 2,
            BufferEmpty = 3,
            BufferFirst = 3,
        };

        explicit concurrent_queue(size_type capacity) : m_container(capacity),m_use_event(true) {}
        
        void SetUseEvent(bool bUseEvent)
        {
            boost::mutex::scoped_lock lock(m_mutex);
            m_use_event = bUseEvent;
        }

        BufferStatus try_push(value_type& item) 
        {
            boost::mutex::scoped_lock lock(m_mutex);
            BufferStatus bufferStatus = m_container.empty() ? BufferFirst : BufferNormal;
            if(m_container.full())
            {
                bufferStatus = BufferOverFlow;
            }
            else
            {
                m_container.push_back(item);
                bufferStatus = m_container.full() ? BufferFull : bufferStatus;
            }
            lock.unlock();
            if(m_use_event)
                m_condition_variable.notify_one();
            return bufferStatus;
        }

        void push(value_type& data)
        {
            boost::mutex::scoped_lock lock(m_mutex);
            m_container.push_back(data);
            lock.unlock();
            if(m_use_event)
                m_condition_variable.notify_one();
        }

        void push_front(value_type & data)
        {
            boost::mutex::scoped_lock lock(m_mutex);
            m_container.push_front(data);
            lock.unlock();
            m_condition_variable.notify_one();
        }

        bool empty() const
        {
            boost::mutex::scoped_lock lock(m_mutex);
            return m_container.empty();
        }

        bool try_pop(value_type& popped_value)
        {
            boost::mutex::scoped_lock lock(m_mutex);
            if(m_container.empty())
            {
                return false;
            }

            popped_value=m_container.front();
            m_container.pop_front();
            return true;
        }

        void wait_and_pop(value_type& popped_value)
        {
            boost::mutex::scoped_lock lock(m_mutex);
            while(m_container.empty())
            {
                m_condition_variable.wait(lock);
            }
            popped_value=m_container.front();
            m_container.pop_front();
        }
    };
}