data_input_interface.hh

Go to the documentation of this file.

#ifndef DATA_INPUT_INTERFACE_HH
#define DATA_INPUT_INTERFACE_HH
 
#include <cassert>
#include <vector>
 
#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/foreach.hpp>
#include <boost/scoped_ptr.hpp>
 
#include "input_channel_handler.hh"
 
/**
 * \author cpellegrino
 */
 
class DataInputInterface
{
  size_t m_number_of_workers;
  std::vector<InChannelHandler*> m_channel_handlers;
  boost::asio::io_service m_service;
  boost::scoped_ptr<boost::thread_group> m_threads;
 
 public:
 
  DataInputInterface(size_t number_of_workers = 1)
    :
    m_number_of_workers(number_of_workers),
    m_threads(0)
  {}
 
  ~DataInputInterface()
  {
    stop();
    for (size_t i = 0; i < m_channel_handlers.size(); ++i)
    {
      delete m_channel_handlers[i];
    }
  }
 
  void stop()
  {
    m_service.stop();
 
    if (m_threads)
    {
      m_threads->join_all();
    }
 
    m_threads.reset();
  }
 
  void pause()
  {
    BOOST_FOREACH(InChannelHandler* h, m_channel_handlers)
    {
      h->pause();
    }
  }
 
  void cont()
  {
    BOOST_FOREACH(InChannelHandler* h, m_channel_handlers)
    {
      h->cont();
    }
  }
 
  void start()
  {
    assert(!m_threads && "Calling DataInputInterface::start() on started istance.");
 
    m_threads.reset(new boost::thread_group);
    m_service.reset();
    for (std::size_t i = 0; i < m_number_of_workers; ++i)
    {
      m_threads->create_thread(boost::bind(&boost::asio::io_service::run,
                                          &m_service));
    }
  }
 
  void add_worker()
  {
    if (m_threads)
    {
      m_threads->create_thread(boost::bind(&boost::asio::io_service::run,
                                            &m_service));
    }
 
    ++m_number_of_workers;
  }
 
  void add_channel(unsigned short port, FrameFarm& farm)
  {
    InChannelHandler* new_channel = new InChannelHandler(m_service, port, farm);
    m_channel_handlers.push_back(new_channel);
  }
};
 
#endif // INPUT_INTERFACE_HH