recipient.hh

Go to the documentation of this file.

#ifndef RECIPIENT_HH
#define RECIPIENT_HH

#include <boost/asio.hpp>
#include <FrameFactory/frame.hh>
#include <boost/circular_buffer.hpp>
#include "log.hh"

/**
 * \author cpellegrino
 */

typedef boost::circular_buffer<Frame> CircularBuffer;

class Recipient
{
  boost::asio::ip::tcp::socket m_sock;
  boost::asio::ip::tcp::endpoint m_endpoint;
  CircularBuffer m_cbuffer;
  bool m_connected;
  bool link_active;
  long heartbeat_c;
  const long heartbeat_interval=5000; //time to wait if the links with the filters are temporarly unavailable
 public:

  Recipient(boost::asio::io_service& service,
            const boost::asio::ip::tcp::endpoint& endpoint,
            size_t circbuff_size)
    :
    m_sock(service),
    m_endpoint(endpoint),
    m_cbuffer(circbuff_size),
    m_connected(false)
  {
    LOG_NOTICE << "Trying to connect to " << m_endpoint;
    connect();
    if (!m_connected) {
      LOG_ERROR << "Connection to " << m_endpoint << " failed";
      link_active=false;   
    }else 
      link_active=true;
    heartbeat_c=0;
  }

  void sock_reset()
  {
    stop();
    connect();
  }

  bool sendIfPossible(const Frame& data)
  {

    //if Recipient receives a EAGAIN message, it stops sending tcp frames for the heartbeat interval
    //otherwise the RAM usage could increase out of control
     if (! link_active){
            m_cbuffer.push_back(data);
            heartbeat_c++;
            if (heartbeat_c == heartbeat_interval){
                    link_active=true;
                    heartbeat_c=0;
            }
            return false;
    }
    if (! m_connected)
    {
      sock_reset();

      if (! m_connected)
      {
        m_cbuffer.push_back(data);
        return false;
      }
    }

    while (m_cbuffer.size())
    {
      if (send(m_cbuffer.front()))
      {
        m_cbuffer.pop_front();
      }
      else
      {
        break;
      }
    }
    if (! send(data))
    {
      m_cbuffer.push_back(data);
    } 
    
    return m_connected;
  }
  
  ~Recipient()
  {
    stop();
  }

  friend class RecipientsHandler;

 private:

  void connect()
  {
    boost::system::error_code ec;
    m_sock.connect(m_endpoint, ec);
    m_connected = !ec;

    if (m_connected) {
      boost::asio::socket_base::send_buffer_size option(67108864);
      m_sock.set_option(option);
      boost::asio::socket_base::non_blocking_io command(true);
      m_sock.io_control(command);
      LOG_NOTICE << "Connection to " << m_endpoint << " succeeded";
      boost::system::error_code ec;
      m_sock.shutdown(boost::asio::ip::tcp::socket::shutdown_receive, ec);
    }
  }

  void stop()
  {
    boost::system::error_code ec;
    m_sock.shutdown(boost::asio::ip::tcp::socket::shutdown_send, ec);
    m_sock.close(ec);
  }
/**
   * Send data
   *
   * \param  data           the Frame to send.
   * \return                true if OK; else false
   */
  bool send(const Frame& data)
  {
    boost::system::error_code ec;
    ssize_t nsent = boost::asio::write(m_sock, boost::asio::buffer(data.data(), data.getFrameLength()), ec);
    m_connected = !ec;

    if (ec) {
      if(ec.value()==11 && nsent<100){ //errno 11 = EAGAIN or EWOULDBLOCK
              link_active=false;
              LOG_ERROR << "The link with " << m_endpoint << 
              " gives EAGAIN: stop sending frame for the heartbeat interval in order to prevent strong RAM usage";
      }else  if(ec.value()!=11)   
            LOG_ERROR << "Error transmitting data to " << m_endpoint << ": " << ec;

    }
    return m_connected;
  }
};

#endif // RECIPIENT_HH