Method handle_input

tools/rtpsrelay/RelayHandler.cpp:109–163

}

int RelayHandler::handle_input(ACE_HANDLE handle)
{
  OpenDDS::DCPS::ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager(), READ_MASK, handle_to_int(handle));

  const auto now = OpenDDS::DCPS::MonotonicTimePoint::now();

  ACE_INET_Addr remote;
  int inlen = 65536; // Default to maximum datagram size.

#ifdef FIONREAD
  if (ACE_OS::ioctl (handle,
                     FIONREAD,
                     &inlen) == -1) {
    HANDLER_ERROR((LM_ERROR, "(%P|%t) ERROR: RelayHandler::handle_input %C failed to get available byte count: %m\n", name_.c_str()));
    return 0;
  }
#else
  ACE_UNUSED_ARG(handle);
#endif

  if (inlen < 0) {
    HANDLER_ERROR((LM_ERROR, "(%P|%t) ERROR: RelayHandler::handle_input %C available byte count is negative\n", name_.c_str()));
    return 0;
  }

  // Allocate at least one byte so that recv cannot return early.
  const auto n = static_cast<size_t>(std::max(inlen, 1));
  OpenDDS::DCPS::Lockable_Message_Block_Ptr buffer(new ACE_Message_Block(n), message_block_locking_);

  const auto bytes = socket_.recv(buffer->wr_ptr(), buffer->space(), remote);

  if (bytes < 0) {
    if (errno == ECONNRESET) {
      // Sending to a non-existent client may result in an ICMP message that is delievered as connection reset.
      return 0;
    }

    HANDLER_ERROR((LM_ERROR, "(%P|%t) ERROR: RelayHandler::handle_input %C failed to recv: %m\n", name_.c_str()));
    return 0;
  } else if (bytes == 0) {
    // Okay.  Empty datagram.
    HANDLER_WARNING((LM_WARNING, "(%P|%t) WARNING: RelayHandler::handle_input %C received an empty datagram from %C\n",
                     name_.c_str(), OpenDDS::DCPS::LogAddr(remote).c_str()));
    return 0;
  }

  buffer->length(static_cast<size_t>(bytes));
  MessageType type = MessageType::Unknown;
  const CORBA::ULong generated_messages = process_message(remote, now, buffer, type);
  stats_reporter_.max_gain(generated_messages, now);
  stats_reporter_.input_message(static_cast<size_t>(bytes),
    OpenDDS::DCPS::MonotonicTimePoint::now() - now, now, type);

  return 0;
}

int RelayHandler::handle_output(ACE_HANDLE handle)
{