Method enqueue_message

tools/rtpsrelay/RelayHandler.cpp:199–227

}

void RelayHandler::enqueue_message(const ACE_INET_Addr& addr,
                                   const OpenDDS::DCPS::Lockable_Message_Block_Ptr& msg,
                                   const OpenDDS::DCPS::MonotonicTimePoint& now,
                                   MessageType type)
{
  const Element out(addr, msg, now, type);
  if (config_.synchronous_output()) {
    size_t total_bytes;

    if (send_i(out, total_bytes) < 0) {
      HANDLER_ERROR((LM_ERROR, "(%P|%t) ERROR: RelayHandler::enqueue_message %C failed to send to %C: %m\n",
                     name_.c_str(), OpenDDS::DCPS::LogAddr(out.address).c_str()));
      stats_reporter_.dropped_message(total_bytes, OpenDDS::DCPS::TimeDuration::zero_value, OpenDDS::DCPS::TimeDuration::zero_value, now, out.type);
    } else {
      stats_reporter_.output_message(total_bytes, OpenDDS::DCPS::TimeDuration::zero_value, OpenDDS::DCPS::TimeDuration::zero_value, now, out.type);
    }

  } else {
    ACE_GUARD(ACE_Thread_Mutex, g, outgoing_mutex_);

    const auto empty = outgoing_.empty();

    outgoing_.push(out);
    stats_reporter_.max_queue_size(outgoing_.size(), now);
    if (empty) {
      reactor()->register_handler(this, WRITE_MASK);
    }
  }
}

ssize_t RelayHandler::send_i(const Element& out,
                             size_t& total_bytes)