Method enqueue_or_send_message

tools/rtpsrelay/RelayHandler.cpp:731–763

{}

void HorizontalHandler::enqueue_or_send_message(const ACE_INET_Addr& addr,
                                                const StringSet& to_partitions,
                                                const GuidSet& to_guids,
                                                const OpenDDS::DCPS::Lockable_Message_Block_Ptr& msg,
                                                const OpenDDS::DCPS::MonotonicTimePoint& now)
{
  using namespace OpenDDS::DCPS;

  const Encoding encoding(Encoding::KIND_XCDR1);

  RelayHeader relay_header;
  auto& tp = relay_header.to_partitions();
  for (const auto& p : to_partitions) {
    tp.push_back(p);
  }
  auto& tg = relay_header.to_guids();
  for (const auto& g : to_guids) {
    tg.push_back(rtps_guid_to_relay_guid(g));
  }

  const size_t size = serialized_size(encoding, relay_header);
  const size_t total_size = size + msg->length();
  if (total_size > TransportSendStrategy::UDP_MAX_MESSAGE_SIZE) {
    HANDLER_ERROR((LM_ERROR, "(%P|%t) ERROR: HorizontalHandler::enqueue_message %C header and message too large (%B > %B)\n", name_.c_str(), total_size, static_cast<size_t>(TransportSendStrategy::UDP_MAX_MESSAGE_SIZE)));
    return;
  }

  Lockable_Message_Block_Ptr header_block(new ACE_Message_Block(size));
  Serializer ser(header_block.get(), encoding);
  ser << relay_header;
  header_block.lockable_cont(msg);
  RelayHandler::enqueue_message(addr, header_block, now, MessageType::Rtps);
}

CORBA::ULong HorizontalHandler::process_message(const ACE_INET_Addr&,
                                                const OpenDDS::DCPS::MonotonicTimePoint& now,