Method handle_output

tools/rtpsrelay/RelayHandler.cpp:165–197

}

int RelayHandler::handle_output(ACE_HANDLE handle)
{
  auto& statusManager = TheServiceParticipant->get_thread_status_manager();
  OpenDDS::DCPS::ThreadStatusManager::Event ev(statusManager, WRITE_MASK, handle_to_int(handle));

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

  ACE_GUARD_RETURN(ACE_Thread_Mutex, g, outgoing_mutex_, 0);
  OpenDDS::DCPS::ThreadStatusManager::Event evLocked(statusManager, WRITE_MASK | DONT_CALL, handle_to_int(handle));

  if (!outgoing_.empty()) {
    const auto& out = outgoing_.front();
    size_t total_bytes;

    if (send_i(out, total_bytes) < 0) {
      HANDLER_ERROR((LM_ERROR, "(%P|%t) ERROR: RelayHandler::handle_output %C failed to send to %C: %m\n",
                     name_.c_str(), OpenDDS::DCPS::LogAddr(out.address).c_str()));
      const auto new_now = OpenDDS::DCPS::MonotonicTimePoint::now();
      stats_reporter_.dropped_message(total_bytes, new_now - now, new_now - out.timestamp, now, out.type);
    } else {
      const auto new_now = OpenDDS::DCPS::MonotonicTimePoint::now();
      stats_reporter_.output_message(total_bytes, new_now - now, new_now - out.timestamp, now, out.type);
    }

    outgoing_.pop();
  }

  if (outgoing_.empty()) {
    reactor()->remove_handler(this, WRITE_MASK);
  }

  return 0;
}

void RelayHandler::enqueue_message(const ACE_INET_Addr& addr,
                                   const OpenDDS::DCPS::Lockable_Message_Block_Ptr& msg,