input_channel_handler.hh

Go to the documentation of this file.

#ifndef INPUT_CHANNEL_HANDLER_HH
#define INPUT_CHANNEL_HANDLER_HH

#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/atomic.hpp>

#include <FrameFactory/input_buffer_collector.hh>
#include <FrameFactory/frame_farm.hh>

/**
 * \author cpellegrino
 */

class InChannelHandler
{
  boost::asio::ip::udp::socket m_socket;
  CLBDataGram* m_buffer;
  FrameFarm* const m_farm;
  boost::atomic<bool> m_paused;

  void handle(const boost::system::error_code& error, size_t size)
  {
    // The error_code should be handled in principle, but
    // it is not mandatory, since all possibilities should be already
    // handled somewhere else.

    if (size && ! m_paused)
    {
      m_buffer->resize(size);
      m_farm->insert(m_buffer);

      m_buffer = InBufferCollector::getCollector().getEmptyDataGram();
    }

    m_socket.async_receive(
        boost::asio::buffer(m_buffer->raw(), m_buffer->capacity()),
        boost::bind(
            &InChannelHandler::handle,
            this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
  }

 public:

  // Receive buffer size (see SOL_SOCKET/SO_RCVBUF) set to 32MB
  // (default between 32 and 128KB)
  static const int recv_buf_size = 33554432;

  void pause()
  {
    m_paused = true;
  }

  void cont()
  {
    m_paused = false;
  }

  InChannelHandler(boost::asio::io_service& service,
                   unsigned short port,
                   FrameFarm& farm)
    :
    m_socket(service, boost::asio::ip::udp::endpoint(boost::asio::ip::udp::v4(), port)),
    m_buffer(
        InBufferCollector::getCollector().getEmptyDataGram()),
    m_farm(&farm),
    m_paused(false)
  {
    boost::asio::ip::udp::socket::receive_buffer_size option(recv_buf_size);
    m_socket.set_option(option);
    handle(boost::system::error_code(), 0);
  }

  ~InChannelHandler()
  {
    InBufferCollector::getCollector().putDataGram(m_buffer);
  }
};

#endif // INPUT_CHANNEL_HANDLER_HH