message_queue.cpp

Go to the documentation of this file.

// In-Memory message buffer
// Author: Max Schwarz <max.schwarz@ais.uni-bonn.de>

#include "message_queue.h"

namespace rosbag_fancy
{

MessageQueue::MessageQueue(uint64_t byteLimit)
 : m_byteLimit{byteLimit}
{
}

bool MessageQueue::push(const rosbag_fancy::MessageQueue::Message& msg)
{
        uint64_t len = msg.size();
        if(m_bytesInQueue + len > m_byteLimit)
                return false;

        {
                std::unique_lock<std::mutex> lock(m_mutex);
                m_queue.push(msg);
                m_bytesInQueue += len;
                m_msgsInQueue++;

                m_cond.notify_all();
        }

        return true;
}

boost::optional<MessageQueue::Message> MessageQueue::pop()
{
        std::unique_lock<std::mutex> lock(m_mutex);

        if(m_shuttingDown)
                return {};

        while(m_queue.empty())
        {
                m_cond.wait(lock);
                if(m_shuttingDown)
                        return {};
        }

        auto msg = m_queue.front();
        uint64_t len = msg.size();

        m_bytesInQueue -= len;
        m_msgsInQueue--;
        m_queue.pop();

        return msg;
}

void MessageQueue::shutdown()
{
        m_shuttingDown = true;
        m_cond.notify_all();
}

}