udp_layer.cc

#include <netdb.h>
#include <unistd.h>
#include <chrono>

#include <Port.hh>

#include "loggers.hh"

#include "udp_layer_factory.hh"

udp_layer::udp_layer(const std::string & p_type, const std::string & param) : layer(p_type), PORT(p_type.c_str()), _params(), _saddr{0}, _daddr{0}, _fd(-1), _time_key("udp_layer::Handle_Fd_Event_Readable") {
  loggers::get_instance().log(">>> udp_layer::udp_layer: %s, %s", to_string().c_str(), param.c_str());
  
  // Setup parameters
  params::convert(_params, param);
  params::const_iterator it = _params.find("src_ip");
  if (it == _params.cend()) {
    _params.insert(std::pair<std::string, std::string>(std::string("src_ip"), "127.0.0.1"));
  }
  it = _params.find("src_port");
  if (it == _params.cend()) {
    _params.insert(std::pair<std::string, std::string>(std::string("src_port"), "0")); // Dynamic binding requested
  }
  it = _params.find("dst_ip");
  if (it == _params.cend()) {
    _params.insert(std::pair<std::string, std::string>(std::string("dst_ip"), "127.0.0.1"));
  }
  it = _params.find("dst_port");
  if (it == _params.cend()) {
    _params.insert(std::pair<std::string, std::string>(std::string("dst_port"), "12345"));
  }

  // Initialize the socket
  _saddr.sin_family = AF_INET;
  _saddr.sin_addr.s_addr = htonl(INADDR_ANY);
  loggers::get_instance().log("udp_layer::udp_layer: Port to listen=%d", std::atoi(_params["src_port"].c_str()));
  _saddr.sin_port = htons(std::atoi(_params["src_port"].c_str()));
  // Create socket
  _fd = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
  if (_fd == -1) {
    loggers::get_instance().error("udp_layer::udp_layer: Failed to create socket");
  }
  loggers::get_instance().log("udp_layer::udp_layer: socket id: %d", _fd);
  int reuse = 1;
  if (::setsockopt(_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0) {
    loggers::get_instance().warning("udp_layer::udp_layer: Failed to set SO_REUSEADDR");
  }
  // Bind it
  if(::bind(_fd, (struct sockaddr *)&_saddr, sizeof(_saddr)) < 0) {
    close();
    loggers::get_instance().error("udp_layer::udp_layer: Failed to bind socket");
  }
  // Pass the device file handler to the polling procedure
  Handler_Add_Fd_Read(_fd);
  
  _daddr.sin_family = AF_INET;
  _daddr.sin_addr.s_addr = htonl(get_host_id(_params["dst_ip"]));
  _daddr.sin_port = htons(std::atoi(_params["dst_port"].c_str()));
  
}

udp_layer::~udp_layer() {
  loggers::get_instance().log(">>> udp_layer::~udp_layer");

  close();
}

void udp_layer::close() {
  loggers::get_instance().log(">>> udp_layer::close: %d", _fd);

  if (_fd != -1) {
    ::close(_fd);
    _fd = -1;
  }
}

void udp_layer::send_data(OCTETSTRING& data, params& params) {
  loggers::get_instance().log_msg(">>> udp_layer::send_data: ", data);
  
  int result = ::sendto(_fd, (const char*)static_cast<const unsigned char*>(data), data.lengthof(), 0, (struct sockaddr*)&_daddr, sizeof(_daddr));
  loggers::get_instance().log("udp_layer::send_data: #bytes sent: %d to %s:%d", result, ::inet_ntoa(_daddr.sin_addr), ntohs(_daddr.sin_port));
}

void udp_layer::receive_data(OCTETSTRING& data, params& params) {
  loggers::get_instance().log_msg(">>> udp_layer::receive_data: ", data);

  receive_to_all_layers(data, params);
}

void udp_layer::Handle_Fd_Event_Readable(int fd) {
  loggers::get_instance().log(">>> udp_layer::Handle_Fd_Event_Readable: %d", fd);

  unsigned char buffer[2048] = {0};
  struct sockaddr_in from = {0};
  socklen_t len = sizeof(struct sockaddr_in); // Length of sender's address
  params params;
  int result = ::recvfrom(_fd, buffer, 1024, 0, (struct sockaddr *)&from, &len); // FIXME Add support of EAGAIN
  loggers::get_instance().log("udp_layer::receive_data: src_port = %s:%d, payload length = %d", ::inet_ntoa(from.sin_addr), ntohs(from.sin_port), result);
  params.insert(std::pair<std::string, std::string>(std::string("timestamp"), std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count())));
  if (result < 0) {
    loggers::get_instance().warning("udp_layer::receive_data: Failed to read data, discard them");
    return;
  }
  
  float duration;
  loggers::get_instance().set_start_time(_time_key);
  OCTETSTRING os(result, buffer);
  receive_data(os, params); // TODO Check execution time for decoding operation
  loggers::get_instance().set_stop_time(_time_key, duration);
}

unsigned long udp_layer::get_host_id(const std::string& p_host_name) {
  loggers::get_instance().log(">>> udp_layer::get_host_id");

  if (p_host_name.empty()) {
    loggers::get_instance().warning("udp_layer::get_host_id: Wrong parameter");
    return INADDR_ANY;
  }
  
  unsigned long ip_addr = 0;
  if(p_host_name.compare("255.255.255.255") == 0) {
    loggers::get_instance().warning("udp_layer::get_host_id: Host ip is 255.255.255.255");
    ip_addr = 0xffffffff;
  } else {
    in_addr_t addr = ::inet_addr(p_host_name.c_str());
    if (addr != (in_addr_t)-1) { // host name in XX:XX:XX:XX form
      ip_addr = addr;
    } else { // host name in domain.com form
      struct hostent* hptr;
      if ((hptr = ::gethostbyname(p_host_name.c_str())) == 0) {
        close();
        loggers::get_instance().error("udp_layer::get_host_id: Invalid host name: %s", p_host_name.c_str());
      }
      ip_addr = *((unsigned long*)hptr->h_addr_list[0]);
    }
  }
  
  loggers::get_instance().log("udp_layer::get_host_id: Host name: %s, Host address: %u", p_host_name.c_str(), ip_addr);
  
  return htonl(ip_addr);
}

udp_layer_factory udp_layer_factory::_f;