NPLMessageQueue.h

#pragma once
#include "NPLMessage.h"

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

namespace NPL
{
    class dummy_condition_variable
    {
    public:
        void notify_one(){};

        template <typename T>
        void wait(T type_) {};
    };

    // forward declare
    template <typename Data, typename Condition = boost::condition_variable>
    class concurrent_ptr_queue;

    template<typename Data, typename Condition>
    class concurrent_ptr_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;
        Condition 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_ptr_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;
            }
            item.reset();
            lock.unlock();
            if(m_use_event)
                m_condition_variable.notify_one();
            return bufferStatus;
        }

        BufferStatus try_push_get_front(value_type& item, value_type** ppFrontItem) 
        {
            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);
                *ppFrontItem = &(m_container.front());
                bufferStatus = m_container.full() ? BufferFull : bufferStatus;
            }
            item.reset();
            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);
            data.reset();
            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);
            data.reset();
            lock.unlock();
            if(m_use_event)
                m_condition_variable.notify_one();
        }

        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();
        }

        void wait(int nMessageCount = -1)
        {
            boost::mutex::scoped_lock lock(m_mutex);
            if (nMessageCount >= 0 && (nMessageCount != (int)m_container.size() || nMessageCount>= (int)m_container.capacity()))
                return;
            m_condition_variable.wait(lock);
        }

        value_type peek(size_type nIndex)
        {
            boost::mutex::scoped_lock lock(m_mutex);
            if (nIndex < m_container.size())
            {
                return m_container.at(nIndex);
            }
            return value_type();
        }

        bool try_pop_at(size_type nIndex, value_type& popped_value)
        {
            boost::mutex::scoped_lock lock(m_mutex);
            if (nIndex < m_container.size())
            {
                if (nIndex == 0)
                {
                    popped_value = m_container.front();
                }
                else
                {
                    popped_value = m_container[nIndex];
                    for (size_type i = nIndex; i >= 1; i--)
                    {
                        m_container[i] = m_container[i-1];
                    }
                }
                m_container.pop_front();
                return true;
            }
            return false;
        }

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

        bool try_next(value_type** ppValueFront) 
        {
            boost::mutex::scoped_lock lock(m_mutex);
            if(m_container.empty())
            {
                return false;
            }
            else
            {
                m_container.pop_front();
                if(ppValueFront!=0 &&  !m_container.empty())
                {
                    *ppValueFront = &(m_container.front());
                }
                return true;
            }
        }

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

        size_type size() const
        {
            boost::mutex::scoped_lock lock(m_mutex);
            return m_container.size();
        }

        size_type capacity() const
        {
            return m_container.capacity();
        }

        void set_capacity(size_type new_capacity) 
        {
            boost::mutex::scoped_lock lock(m_mutex);
            m_container.set_capacity(new_capacity);
        }
    };

    class CNPLMessageQueue : public concurrent_ptr_queue<NPLMessage_ptr>
    {
    public:
        CNPLMessageQueue();
        CNPLMessageQueue(int capacity);
        ~CNPLMessageQueue();
    public:

    };


}