OpenDDS::DCPS::RtpsSampleHeader Class Reference

Adapt the TransportReceiveStrategy for RTPS's "sample" (submessage) Header. More...

#include <RtpsSampleHeader.h>

Collaboration diagram for OpenDDS::DCPS::RtpsSampleHeader:

[legend]List of all members.

Public Member Functions
	RtpsSampleHeader ()
	RtpsSampleHeader (ACE_Message_Block &mb)
RtpsSampleHeader &	operator= (ACE_Message_Block &mn)
void	pdu_remaining (size_t size)
size_t	marshaled_size ()
ACE_UINT32	message_length ()
bool	valid () const
bool	into_received_data_sample (ReceivedDataSample &rds)
bool	more_fragments () const
Static Public Member Functions
static size_t	max_marshaled_size ()
static bool	partial (const ACE_Message_Block &)
static SequenceRange	split (const ACE_Message_Block &orig, size_t size, ACE_Message_Block *&head, ACE_Message_Block *&tail)
static void	populate_data_sample_submessages (RTPS::SubmessageSeq &subm, const DataSampleElement &dsle, bool requires_inline_qos)
static void	populate_data_control_submessages (RTPS::SubmessageSeq &subm, const TransportSendControlElement &tsce, bool requires_inline_qos)
static void	populate_inline_qos (const TransportSendListener::InlineQosData &qos_data, RTPS::ParameterList &plist)
static bool	control_message_supported (char message_id)
Public Attributes
RTPS::Submessage	submessage_
Static Public Attributes
static const ACE_CDR::UShort	FRAG_SIZE = 1024
Private Member Functions
void	init (ACE_Message_Block &mb)
Static Private Member Functions
static void	process_iqos (DataSampleHeader &opendds, const OpenDDS::RTPS::ParameterList &iqos)
Private Attributes
bool	valid_
bool	frag_
size_t	marshaled_size_
size_t	message_length_

---

Detailed Description

Adapt the TransportReceiveStrategy for RTPS's "sample" (submessage) Header.

Definition at line 30 of file RtpsSampleHeader.h.

---

Constructor & Destructor Documentation

ACE_INLINE OpenDDS::DCPS::RtpsSampleHeader::RtpsSampleHeader

(

)

Definition at line 12 of file RtpsSampleHeader.inl.

  : valid_(false)
  , frag_(false)
  , marshaled_size_(0)
  , message_length_(0)
{
}

ACE_INLINE OpenDDS::DCPS::RtpsSampleHeader::RtpsSampleHeader

(

ACE_Message_Block &

mb

)

[explicit]

Definition at line 21 of file RtpsSampleHeader.inl.

References init().

  : valid_(false)
  , frag_(false)
  , marshaled_size_(0)
  , message_length_(0)
{
  init(mb);
}

---

Member Function Documentation

ACE_INLINE bool OpenDDS::DCPS::RtpsSampleHeader::control_message_supported

(

char

message_id

)

[static]

Definition at line 72 of file RtpsSampleHeader.inl.

References OpenDDS::DCPS::DISPOSE_INSTANCE, OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE, OpenDDS::DCPS::INSTANCE_REGISTRATION, and OpenDDS::DCPS::UNREGISTER_INSTANCE.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element().

{
  switch (message_id) {
  case INSTANCE_REGISTRATION:
  case UNREGISTER_INSTANCE:
  case DISPOSE_INSTANCE:
  case DISPOSE_UNREGISTER_INSTANCE:
    return true;
  default:
    return false;
  }
}

void OpenDDS::DCPS::RtpsSampleHeader::init

(

ACE_Message_Block &

mb

)

[private]

Definition at line 60 of file RtpsSampleHeader.cpp.

References OpenDDS::RTPS::ACKNACK, OpenDDS::DCPS::Serializer::ALIGN_CDR, CASE_SMKIND, OpenDDS::RTPS::DATA, OpenDDS::RTPS::DATA_FRAG, OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_E, FLAG_K_IN_DATA, frag_, OpenDDS::RTPS::GAP, OpenDDS::RTPS::HEARTBEAT, OpenDDS::RTPS::HEARTBEAT_FRAG, OpenDDS::RTPS::INFO_DST, OpenDDS::RTPS::INFO_REPLY, OpenDDS::RTPS::INFO_REPLY_IP4, OpenDDS::RTPS::INFO_SRC, OpenDDS::RTPS::INFO_TS, marshaled_size_, message_length_, OpenDDS::RTPS::NACK_FRAG, OpenDDS::RTPS::PAD, OpenDDS::DCPS::Serializer::skip(), OpenDDS::RTPS::SMHDR_SZ, submessage_, and valid_.

Referenced by operator=(), and RtpsSampleHeader().

{
  using namespace OpenDDS::RTPS;

  // valid_ is false here, it will only be set to true if there is a Submessage

  // Manually grab the first two bytes for the SubmessageKind and the byte order
  if (mb.length() == 0) {
    return;
  }

  const SubmessageKind kind = static_cast<SubmessageKind>(*mb.rd_ptr());

  ACE_CDR::Octet flags = 0;

  if (mb.length() > 1) {
    flags = mb.rd_ptr()[1];
  } else if (mb.cont() && mb.cont()->length() > 0) {
    flags = mb.cont()->rd_ptr()[0];
  } else {
    return;
  }

  const bool little_endian = flags & FLAG_E;
  const size_t starting_length = mb.total_length();
  Serializer ser(&mb, ACE_CDR_BYTE_ORDER != little_endian,
                 Serializer::ALIGN_CDR);

  ACE_CDR::UShort octetsToNextHeader = 0;

#define CASE_SMKIND(kind, class, name) case kind: {               \
    class submessage;                                             \
    if (ser >> submessage) {                                      \
      octetsToNextHeader = submessage.smHeader.submessageLength;  \
      submessage_.name##_sm(submessage);                          \
      valid_ = true;                                              \
    }                                                             \
    break;                                                        \
  }

  switch (kind) {
  CASE_SMKIND(PAD, PadSubmessage, pad)
  CASE_SMKIND(ACKNACK, AckNackSubmessage, acknack)
  CASE_SMKIND(HEARTBEAT, HeartBeatSubmessage, heartbeat)
  CASE_SMKIND(GAP, GapSubmessage, gap)
  CASE_SMKIND(INFO_TS, InfoTimestampSubmessage, info_ts)
  CASE_SMKIND(INFO_SRC, InfoSourceSubmessage, info_src)
  CASE_SMKIND(INFO_REPLY_IP4, InfoReplyIp4Submessage, info_reply_ipv4)
  CASE_SMKIND(INFO_DST, InfoDestinationSubmessage, info_dst)
  CASE_SMKIND(INFO_REPLY, InfoReplySubmessage, info_reply)
  CASE_SMKIND(NACK_FRAG, NackFragSubmessage, nack_frag)
  CASE_SMKIND(HEARTBEAT_FRAG, HeartBeatFragSubmessage, hb_frag)
  CASE_SMKIND(DATA, DataSubmessage, data)
  CASE_SMKIND(DATA_FRAG, DataFragSubmessage, data_frag)
  default:
    {
      SubmessageHeader submessage;
      if (ser >> submessage) {
        octetsToNextHeader = submessage.submessageLength;
        submessage_.unknown_sm(submessage);
        valid_ = true;
      }
      break;
    }
  }
#undef CASE_SMKIND

  if (valid_) {

    frag_ = (kind == DATA_FRAG);

    // marshaled_size_ is # of bytes of submessage we have read from "mb"
    marshaled_size_ = starting_length - mb.total_length();

    if (octetsToNextHeader == 0 && kind != PAD && kind != INFO_TS) {
      // see RTPS v2.1 section 9.4.5.1.3
      // In this case the current Submessage extends to the end of Message,
      // so we will use the message_length_ that was set in pdu_remaining().
      octetsToNextHeader =
        static_cast<ACE_CDR::UShort>(message_length_ - SMHDR_SZ);
    }

    if ((kind == DATA && (flags & (FLAG_D | FLAG_K_IN_DATA)))
        || kind == DATA_FRAG) {
      // These Submessages have a payload which we haven't deserialized yet.
      // The TransportReceiveStrategy will know this via message_length().
      // octetsToNextHeader does not count the SubmessageHeader (4 bytes)
      message_length_ = octetsToNextHeader + SMHDR_SZ - marshaled_size_;
    } else {
      // These Submessages _could_ have extra data that we don't know about
      // (from a newer minor version of the RTPS spec).  Either way, indicate
      // to the TransportReceiveStrategy that there is no data payload here.
      message_length_ = 0;
      ACE_CDR::UShort marshaled = static_cast<ACE_CDR::UShort>(marshaled_size_);
      if (octetsToNextHeader + SMHDR_SZ > marshaled) {
        valid_ = ser.skip(octetsToNextHeader + SMHDR_SZ - marshaled);
        marshaled_size_ = octetsToNextHeader + SMHDR_SZ;
      }
    }
  }
}

bool OpenDDS::DCPS::RtpsSampleHeader::into_received_data_sample

(

ReceivedDataSample &

rds

)

Definition at line 213 of file RtpsSampleHeader.cpp.

References OpenDDS::DCPS::DataSampleHeader::byte_order_, OpenDDS::DCPS::DataSampleHeader::cdr_encapsulation_, OpenDDS::RTPS::DATA, OpenDDS::RTPS::DATA_FRAG, OpenDDS::RTPS::Submessage::data_frag_sm, OpenDDS::RTPS::Submessage::data_sm, OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_E, FLAG_K_IN_DATA, FLAG_K_IN_FRAG, OpenDDS::RTPS::FLAG_Q, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::DataSampleHeader::key_fields_only_, OpenDDS::DCPS::DataSampleHeader::message_id_, message_length(), OpenDDS::DCPS::DataSampleHeader::message_length_, OpenDDS::DCPS::DataSampleHeader::more_fragments_, OpenDDS::RTPS::PID_KEY_HASH, process_iqos(), OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::ReceivedDataSample::sample_, OpenDDS::DCPS::SAMPLE_DATA, OpenDDS::DCPS::DataSampleHeader::sequence_, OpenDDS::DCPS::SequenceNumber::setValue(), submessage_, and OpenDDS::DCPS::Transport_debug_level.

{
  using namespace OpenDDS::RTPS;
  DataSampleHeader& opendds = rds.header_;

  switch (submessage_._d()) {
  case DATA: {
    const DataSubmessage& rtps = submessage_.data_sm();
    opendds.cdr_encapsulation_ = true;
    opendds.message_length_ = message_length();
    opendds.sequence_.setValue(rtps.writerSN.high, rtps.writerSN.low);
    opendds.publication_id_.entityId = rtps.writerId;
    opendds.message_id_ = SAMPLE_DATA;

    process_iqos(opendds, rtps.inlineQos);

    if (rtps.smHeader.flags & FLAG_K_IN_DATA) {
      opendds.key_fields_only_ = true;
    } else if (!(rtps.smHeader.flags & (FLAG_D | FLAG_K_IN_DATA))) {
      // Interoperability note: the Key may be hiding in the "key hash" param
      // in the InlineQos.  In order to make use of this Key, it mst be 16
      // bytes or less.  We have observed other DDS implementations only send
      // the MD5 hash of a >16 byte key, so we must limit this to Built-in
      // endpoints which are assumed to use GUIDs as keys.
      if ((rtps.writerId.entityKind & 0xC0) == 0xC0 // Only Built-in endpoints
          && (rtps.smHeader.flags & FLAG_Q) && !rds.sample_) {
        for (CORBA::ULong i = 0; i < rtps.inlineQos.length(); ++i) {
          if (rtps.inlineQos[i]._d() == PID_KEY_HASH) {
            rds.sample_ = new ACE_Message_Block(20);
            // CDR_BE encapsuation scheme (endianness is not used for key hash)
            rds.sample_->copy("\x00\x00\x00\x00", 4);
            const CORBA::Octet* data = rtps.inlineQos[i].key_hash().value;
            rds.sample_->copy(reinterpret_cast<const char*>(data), 16);
            opendds.message_length_ =
              static_cast<ACE_UINT32>(rds.sample_->length());
            opendds.key_fields_only_ = true;
            if (Transport_debug_level) {
              ACE_DEBUG((LM_DEBUG,
                         "(%P|%t) RtpsSampleHeader::into_received_data_sample()"
                         " - used KeyHash data as the key-only payload\n"));
            }
            break;
          }
        }
      } else {
      // FUTURE: Handle the case of D = 0 and K = 0
      // used for Coherent Sets in PRESENTATION QoS (see 8.7.5)
        if (Transport_debug_level) {
          ACE_DEBUG((LM_WARNING,
                     "(%P|%t) RtpsSampleHeader::into_received_data_sample() - "
                     "Received a DATA Submessage with D = 0 and K = 0, "
                     "dropping\n"));
        }
        return false;
      }
    }

    if (rtps.smHeader.flags & (FLAG_D | FLAG_K_IN_DATA)) {
      // Peek at the byte order from the encapsulation containing the payload.
      opendds.byte_order_ = rds.sample_->rd_ptr()[1] & FLAG_E;
    }

    break;
  }
  case DATA_FRAG: {
    const DataFragSubmessage& rtps = submessage_.data_frag_sm();
    opendds.cdr_encapsulation_ = true;
    opendds.message_length_ = message_length();
    opendds.sequence_.setValue(rtps.writerSN.high, rtps.writerSN.low);
    opendds.publication_id_.entityId = rtps.writerId;
    opendds.message_id_ = SAMPLE_DATA;
    opendds.key_fields_only_ = (rtps.smHeader.flags & FLAG_K_IN_FRAG);
    // opendds.byte_order_ set in RtpsUdpReceiveStrategy::reassemble().

    process_iqos(opendds, rtps.inlineQos);

    const CORBA::ULong lastFragInSubmsg =
      rtps.fragmentStartingNum.value - 1 + rtps.fragmentsInSubmessage;
    if (lastFragInSubmsg * rtps.fragmentSize < rtps.sampleSize) {
      opendds.more_fragments_ = true;
    }
    break;
  }
  default:
    break;
  }

  return true;
}

ACE_INLINE size_t OpenDDS::DCPS::RtpsSampleHeader::marshaled_size

(

)

Definition at line 54 of file RtpsSampleHeader.inl.

References marshaled_size_.

{
  return marshaled_size_;
}

static size_t OpenDDS::DCPS::RtpsSampleHeader::max_marshaled_size

(

)

[inline, static]

Definition at line 37 of file RtpsSampleHeader.h.

{ return 4; }

ACE_INLINE ACE_UINT32 OpenDDS::DCPS::RtpsSampleHeader::message_length

(

)

Definition at line 60 of file RtpsSampleHeader.inl.

References message_length_.

Referenced by into_received_data_sample().

{
  return static_cast<ACE_UINT32>(message_length_);
}

ACE_INLINE bool OpenDDS::DCPS::RtpsSampleHeader::more_fragments

(

)

const

Definition at line 66 of file RtpsSampleHeader.inl.

References frag_.

{
  return frag_;
}

ACE_INLINE RtpsSampleHeader & OpenDDS::DCPS::RtpsSampleHeader::operator=

(

ACE_Message_Block &

mn

)

Definition at line 31 of file RtpsSampleHeader.inl.

References frag_, init(), and valid_.

{
  frag_ = false;
  valid_ = false;
  // message_length_ should not be reset here
  // marshaled_size_ doesn't need to be reset, init() will set it (if valid_)
  init(mb);
  return *this;
}

static bool OpenDDS::DCPS::RtpsSampleHeader::partial

(

const ACE_Message_Block &

)

[inline, static]

Definition at line 42 of file RtpsSampleHeader.h.

{ return false; }

ACE_INLINE void OpenDDS::DCPS::RtpsSampleHeader::pdu_remaining

(

size_t

size

)

Definition at line 48 of file RtpsSampleHeader.inl.

References message_length_.

{
  message_length_ = size;
}

void OpenDDS::DCPS::RtpsSampleHeader::populate_data_control_submessages	(	RTPS::SubmessageSeq &	subm,
		const TransportSendControlElement &	tsce,
		bool	requires_inline_qos
	)			[static]

Definition at line 415 of file RtpsSampleHeader.cpp.

References OpenDDS::DCPS::add_key_hash(), OpenDDS::DCPS::add_timestamp(), OpenDDS::RTPS::DATA, OpenDDS::RTPS::DATA_OCTETS_TO_IQOS, OpenDDS::DCPS::DISPOSE_INSTANCE, OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE, OpenDDS::DCPS::ENTITYID_UNKNOWN, FLAG_K_IN_DATA, OpenDDS::RTPS::FLAG_Q, OpenDDS::DCPS::TransportSendControlElement::header(), header, OpenDDS::DCPS::INSTANCE_REGISTRATION, OpenDDS::DCPS::TransportSendControlElement::listener(), OpenDDS::DCPS::TransportSendControlElement::msg_payload(), populate_inline_qos(), OpenDDS::DCPS::TransportSendListener::retrieve_inline_qos_data(), STATUS_INFO_DISPOSE, STATUS_INFO_DISPOSE_UNREGISTER, STATUS_INFO_REGISTER, STATUS_INFO_UNREGISTER, and OpenDDS::DCPS::UNREGISTER_INSTANCE.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element().

{
  using namespace OpenDDS::RTPS;

  const DataSampleHeader& header = tsce.header();
  const ACE_CDR::Octet flags = header.byte_order_;
  add_timestamp(subm, flags, header);
  CORBA::ULong i = subm.length();

  static const CORBA::Octet BUILT_IN_WRITER = 0xC2;

  DataSubmessage data = {
    {DATA, flags, 0},
    0,
    DATA_OCTETS_TO_IQOS,
    ENTITYID_UNKNOWN,
    header.publication_id_.entityId,
    {header.sequence_.getHigh(), header.sequence_.getLow()},
    ParameterList()
  };
  switch (header.message_id_) {
  case INSTANCE_REGISTRATION: {
    // NOTE: The RTPS spec is not entirely clear about instance registration.
    // We have decided to send a DATA Submessage containing the key and an
    // inlineQoS StatusInfo of zero.
    data.smHeader.flags |= FLAG_K_IN_DATA;
    const int qos_len = data.inlineQos.length();
    data.inlineQos.length(qos_len + 1);
    data.inlineQos[qos_len].status_info(STATUS_INFO_REGISTER);
    break;
  }
  case UNREGISTER_INSTANCE: {
    data.smHeader.flags |= FLAG_K_IN_DATA;
    const int qos_len = data.inlineQos.length();
    data.inlineQos.length(qos_len+1);
    data.inlineQos[qos_len].status_info(STATUS_INFO_UNREGISTER);
    if (header.publication_id_.entityId.entityKind == BUILT_IN_WRITER) {
      add_key_hash(data.inlineQos, tsce.msg_payload());
    }
    break;
  }
  case DISPOSE_INSTANCE: {
    data.smHeader.flags |= FLAG_K_IN_DATA;
    const int qos_len = data.inlineQos.length();
    data.inlineQos.length(qos_len + 1);
    data.inlineQos[qos_len].status_info(STATUS_INFO_DISPOSE);
    if (header.publication_id_.entityId.entityKind == BUILT_IN_WRITER) {
      add_key_hash(data.inlineQos, tsce.msg_payload());
    }
    break;
  }
  case DISPOSE_UNREGISTER_INSTANCE: {
    data.smHeader.flags |= FLAG_K_IN_DATA;
    const int qos_len = data.inlineQos.length();
    data.inlineQos.length(qos_len + 1);
    data.inlineQos[qos_len].status_info(STATUS_INFO_DISPOSE_UNREGISTER);
    if (header.publication_id_.entityId.entityKind == BUILT_IN_WRITER) {
      add_key_hash(data.inlineQos, tsce.msg_payload());
    }
    break;
  }
  // update control_message_supported() when adding new cases here
  default:
    ACE_DEBUG((LM_INFO,
               "RtpsSampleHeader::populate_data_control_submessages(): "
               "Non-sample messages seen, message_id = %d\n",
               header.message_id_));
    break;
  }

  if (requires_inline_qos) {
    TransportSendListener::InlineQosData qos_data;
    tsce.listener()->retrieve_inline_qos_data(qos_data);

    populate_inline_qos(qos_data, data.inlineQos);
  }

  if (data.inlineQos.length() > 0) {
    data.smHeader.flags |= FLAG_Q;
  }

  subm.length(i + 1);
  subm[i].data_sm(data);
}

void OpenDDS::DCPS::RtpsSampleHeader::populate_data_sample_submessages	(	RTPS::SubmessageSeq &	subm,
		const DataSampleElement &	dsle,
		bool	requires_inline_qos
	)			[static]

Definition at line 321 of file RtpsSampleHeader.cpp.

References OpenDDS::DCPS::add_timestamp(), OpenDDS::DCPS::DataSampleHeader::byte_order_, OpenDDS::RTPS::DATA, OpenDDS::RTPS::DATA_OCTETS_TO_IQOS, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_Q, OpenDDS::DCPS::DataSampleElement::get_header(), OpenDDS::DCPS::DataSampleElement::get_num_subs(), OpenDDS::DCPS::DataSampleElement::get_pub_id(), OpenDDS::DCPS::DataSampleElement::get_send_listener(), OpenDDS::DCPS::DataSampleElement::get_sub_id(), OpenDDS::DCPS::SequenceNumber::getHigh(), OpenDDS::DCPS::SequenceNumber::getLow(), OpenDDS::DCPS::GUID_t::guidPrefix, OpenDDS::DCPS::GUIDPREFIX_UNKNOWN, OpenDDS::RTPS::INFO_DST, OpenDDS::RTPS::INFO_DST_SZ, OpenDDS::DCPS::DataSampleHeader::message_id_, populate_inline_qos(), OpenDDS::DCPS::TransportSendListener::retrieve_inline_qos_data(), OpenDDS::DCPS::SAMPLE_DATA, and OpenDDS::DCPS::DataSampleHeader::sequence_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element().

{
  using namespace OpenDDS::RTPS;

  const ACE_CDR::Octet flags = dsle.get_header().byte_order_;
  add_timestamp(subm, flags, dsle.get_header());
  CORBA::ULong i = subm.length();

  EntityId_t readerId = ENTITYID_UNKNOWN;
  if (dsle.get_num_subs() == 1) {
    readerId = dsle.get_sub_id(0).entityId;
    InfoDestinationSubmessage idest;
    idest.smHeader.submessageId = INFO_DST;
    idest.smHeader.flags = flags;
    idest.smHeader.submessageLength = INFO_DST_SZ;
    std::memcpy(idest.guidPrefix, dsle.get_sub_id(0).guidPrefix,
                sizeof(GuidPrefix_t));
    subm.length(i + 1);
    subm[i++].info_dst_sm(idest);
  } else {
    //Not durability resend, but could have inline gaps
    for (CORBA::ULong x = 0; x < i; ++x) {
      if (subm[x]._d() == INFO_DST) {
        //Need to add INFO_DST
        InfoDestinationSubmessage idest;
        idest.smHeader.submessageId = INFO_DST;
        idest.smHeader.flags = flags;
        idest.smHeader.submessageLength = INFO_DST_SZ;
        std::memcpy(idest.guidPrefix, GUIDPREFIX_UNKNOWN,
                    sizeof(GuidPrefix_t));
        subm.length(i + 1);
        subm[i++].info_dst_sm(idest);
        break;
      }
    }
  }

  DataSubmessage data = {
    {DATA, flags, 0},
    0,
    DATA_OCTETS_TO_IQOS,
    readerId,
    dsle.get_pub_id().entityId,
    {dsle.get_header().sequence_.getHigh(), dsle.get_header().sequence_.getLow()},
    ParameterList()
  };
  const char message_id = dsle.get_header().message_id_;
  switch (message_id) {
  case SAMPLE_DATA:
    // Must be a data message
    data.smHeader.flags |= FLAG_D;
    break;
  default:
    ACE_DEBUG((LM_INFO, "(%P|%t) RtpsSampleHeader::populate_submessages(): "
               "Non-sample messages seen, message_id = %d\n", int(message_id)));
    break;
  }

  if (requires_inline_qos) {
    TransportSendListener::InlineQosData qos_data;
    dsle.get_send_listener()->retrieve_inline_qos_data(qos_data);

    populate_inline_qos(qos_data, data.inlineQos);
  }

  if (data.inlineQos.length() > 0) {
    data.smHeader.flags |= FLAG_Q;
  }

  subm.length(i + 1);
  subm[i].data_sm(data);
}

void OpenDDS::DCPS::RtpsSampleHeader::populate_inline_qos	(	const TransportSendListener::InlineQosData &	qos_data,
		RTPS::ParameterList &	plist
	)			[static]

Definition at line 511 of file RtpsSampleHeader.cpp.

References OpenDDS::DCPS::TransportSendListener::InlineQosData::dw_qos, OpenDDS::RTPS::PID_TOPIC_NAME, PROCESS_INLINE_QOS, OpenDDS::DCPS::TransportSendListener::InlineQosData::pub_qos, TheServiceParticipant, and OpenDDS::DCPS::TransportSendListener::InlineQosData::topic_name.

Referenced by populate_data_control_submessages(), and populate_data_sample_submessages().

{
  using namespace OpenDDS::RTPS;

  // Always include topic name (per the spec)
  {
    const int qos_len = plist.length();
    plist.length(qos_len + 1);
    plist[qos_len].string_data(qos_data.topic_name.c_str());
    plist[qos_len]._d(PID_TOPIC_NAME);
  }

  // Conditionally include other QoS inline when the differ from the
  // default value.
  DDS::PublisherQos default_pub_qos =
    TheServiceParticipant->initial_PublisherQos();
  PROCESS_INLINE_QOS(presentation, default_pub_qos, qos_data.pub_qos);
  PROCESS_INLINE_QOS(partition, default_pub_qos, qos_data.pub_qos);

  DDS::DataWriterQos default_dw_qos =
    TheServiceParticipant->initial_DataWriterQos();
  PROCESS_INLINE_QOS(durability, default_dw_qos, qos_data.dw_qos);
  PROCESS_INLINE_QOS(deadline, default_dw_qos, qos_data.dw_qos);
  PROCESS_INLINE_QOS(latency_budget, default_dw_qos, qos_data.dw_qos);
  PROCESS_INLINE_QOS(ownership, default_dw_qos, qos_data.dw_qos);
#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
  PROCESS_INLINE_QOS(ownership_strength, default_dw_qos, qos_data.dw_qos);
#endif
  PROCESS_INLINE_QOS(liveliness, default_dw_qos, qos_data.dw_qos);
  PROCESS_INLINE_QOS(reliability, default_dw_qos, qos_data.dw_qos);
  PROCESS_INLINE_QOS(transport_priority, default_dw_qos, qos_data.dw_qos);
  PROCESS_INLINE_QOS(lifespan, default_dw_qos, qos_data.dw_qos);
  PROCESS_INLINE_QOS(destination_order, default_dw_qos, qos_data.dw_qos);
}

void OpenDDS::DCPS::RtpsSampleHeader::process_iqos	(	DataSampleHeader &	opendds,
		const OpenDDS::RTPS::ParameterList &	iqos
	)			[static, private]

Definition at line 163 of file RtpsSampleHeader.cpp.

References DDS::PresentationQosPolicy::access_scope, DDS::PresentationQosPolicy::coherent_access, OpenDDS::DCPS::DISPOSE_INSTANCE, OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE, OpenDDS::DCPS::DataSampleHeader::historic_sample_, OpenDDS::DCPS::INSTANCE_REGISTRATION, OpenDDS::DCPS::DataSampleHeader::message_id_, DDS::PartitionQosPolicy::name, OPENDDS_STRING, DDS::PresentationQosPolicy::ordered_access, OpenDDS::RTPS::PID_ORIGINAL_WRITER_INFO, OpenDDS::RTPS::PID_PARTITION, OpenDDS::RTPS::PID_PRESENTATION, OpenDDS::RTPS::PID_STATUS_INFO, OpenDDS::RTPS::PID_TOPIC_NAME, STATUS_INFO_DISPOSE, STATUS_INFO_DISPOSE_UNREGISTER, STATUS_INFO_REGISTER, STATUS_INFO_UNREGISTER, OpenDDS::DCPS::to_dds_string(), and OpenDDS::DCPS::UNREGISTER_INSTANCE.

Referenced by into_received_data_sample().

{
  using namespace OpenDDS::RTPS;
#if defined(OPENDDS_TEST_INLINE_QOS)
  OPENDDS_STRING output("into_received_data_sample(): ");
  output += to_dds_string(iqos.length());
  output += " inline QoS parameters\n";
  for (CORBA::ULong index = 0; index < iqos.length(); ++index) {
    output += "  parameter type = ";
    output += to_dds_string(iqos[index]._d());
    output += "\n";
  }
  ACE_DEBUG((LM_DEBUG, "%C", output.c_str()));
#endif
  for (CORBA::ULong i = 0; i < iqos.length(); ++i) {
    if (iqos[i]._d() == PID_STATUS_INFO) {
      if (iqos[i].status_info() == STATUS_INFO_DISPOSE) {
        opendds.message_id_ = DISPOSE_INSTANCE;
      } else if (iqos[i].status_info() == STATUS_INFO_UNREGISTER) {
        opendds.message_id_ = UNREGISTER_INSTANCE;
      } else if (iqos[i].status_info() == STATUS_INFO_DISPOSE_UNREGISTER) {
        opendds.message_id_ = DISPOSE_UNREGISTER_INSTANCE;
      } else if (iqos[i].status_info() == STATUS_INFO_REGISTER) {
        opendds.message_id_ = INSTANCE_REGISTRATION;
      }
    } else if (iqos[i]._d() == PID_ORIGINAL_WRITER_INFO) {
      opendds.historic_sample_ = true;
#if defined(OPENDDS_TEST_INLINE_QOS)
    } else if (iqos[i]._d() == PID_TOPIC_NAME) {
      ACE_DEBUG((LM_DEBUG, "topic_name = %C\n", iqos[i].string_data()));
    } else if (iqos[i]._d() == PID_PRESENTATION) {
      DDS::PresentationQosPolicy pres_qos = iqos[i].presentation();
      ACE_DEBUG((LM_DEBUG, "presentation qos, access_scope = %d, "
                 "coherent_access = %d, ordered_access = %d\n",
                 pres_qos.access_scope, pres_qos.coherent_access,
                 pres_qos.ordered_access));
    } else if (iqos[i]._d() == PID_PARTITION) {
      DDS::PartitionQosPolicy part_qos = iqos[i].partition();
      ACE_DEBUG((LM_DEBUG, "partition qos(%d): ", part_qos.name.length()));
      for (CORBA::ULong j = 0; j < part_qos.name.length(); j++) {
        ACE_DEBUG((LM_DEBUG, "'%C'  ", part_qos.name[j].in()));
      }
      ACE_DEBUG((LM_DEBUG, "\n"));
#endif
    }
  }
}

SequenceRange OpenDDS::DCPS::RtpsSampleHeader::split	(	const ACE_Message_Block &	orig,
		size_t	size,
		ACE_Message_Block *&	head,
		ACE_Message_Block *&	tail
	)			[static]

Create two new serialized headers (owned by caller), the "head" having at most "size" bytes (header + data) and the "tail" having the rest. Returns a pair containing the largest fragment number in each new header.

Definition at line 597 of file RtpsSampleHeader.cpp.

References OpenDDS::DCPS::DataSampleHeader::alloc_msgblock(), OpenDDS::RTPS::DATA, OpenDDS::RTPS::DATA_FRAG, OpenDDS::RTPS::DATA_FRAG_OCTETS_TO_IQOS, OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_E, FLAG_K_IN_DATA, FLAG_K_IN_FRAG, OpenDDS::RTPS::FLAG_Q, OpenDDS::DCPS::FRAG_SAMPLE_SIZE_OFFSET, FRAG_SIZE, OpenDDS::DCPS::FRAG_START_OFFSET, OpenDDS::DCPS::peek(), OpenDDS::DCPS::SequenceNumber::SEQUENCENUMBER_UNKNOWN(), OpenDDS::RTPS::SMHDR_SZ, OpenDDS::DCPS::DataSampleHeader::split_payload(), OpenDDS::DCPS::Transport_debug_level, and OpenDDS::DCPS::write().

Referenced by OpenDDS::DCPS::RtpsCustomizedElement::fragment().

{
  using namespace RTPS;
  static const SequenceRange unknown_range(SequenceNumber::SEQUENCENUMBER_UNKNOWN(),
                                           SequenceNumber::SEQUENCENUMBER_UNKNOWN());
  size_t data_offset = 0;
  const char* rd = orig.rd_ptr();
  ACE_CDR::ULong starting_frag, sample_size;
  ACE_CDR::Octet flags;
  bool swap_bytes;

  // Find the start of the DATA | DATA_FRAG submessage in the orig msg block.
  // The submessages from the start of the msg block to this point (data_offset)
  // will be copied to both the head and tail fragments.
  while (true) {
    flags = rd[data_offset + 1];
    swap_bytes = ACE_CDR_BYTE_ORDER != bool(flags & FLAG_E);
    bool found_data = false;

    switch (rd[data_offset]) {
    case DATA:
      if ((flags & (FLAG_D | FLAG_K_IN_DATA)) == 0) {
        if (Transport_debug_level) {
          ACE_ERROR((LM_ERROR, "(%P|%t) RtpsSampleHeader::split() ERROR - "
            "attempting to fragment a Data submessage with no payload.\n"));
        }
        return unknown_range;
      }
      found_data = true;
      starting_frag = 1;
      sample_size = static_cast<ACE_CDR::ULong>(orig.cont()->total_length());
      break;
    case DATA_FRAG:
      found_data = true;
      peek(starting_frag, rd + data_offset + FRAG_START_OFFSET, swap_bytes);
      peek(sample_size, rd + data_offset + FRAG_SAMPLE_SIZE_OFFSET, swap_bytes);
      break;
    }

    if (found_data) {
      break;
    }

    // Scan for next submessage in orig
    ACE_CDR::UShort octetsToNextHeader;
    peek(octetsToNextHeader, rd + data_offset + 2, swap_bytes);

    data_offset += octetsToNextHeader + SMHDR_SZ;
    if (data_offset >= orig.length()) {
      if (Transport_debug_level) {
        ACE_ERROR((LM_ERROR, "(%P|%t) RtpsSampleHeader::split() ERROR - "
          "invalid octetsToNextHeader encountered while fragmenting.\n"));
      }
      return unknown_range;
    }
  }

  // Create the "head" message block (of size "sz") containing DATA_FRAG
  size_t sz = orig.length();
  ACE_CDR::Octet new_flags = flags;
  size_t iqos_offset = data_offset + 8 + DATA_FRAG_OCTETS_TO_IQOS;
  if (rd[data_offset] == DATA) {
    sz += 12; // DATA_FRAG is 12 bytes larger than DATA
    iqos_offset -= 12;
    new_flags &= ~(FLAG_K_IN_DATA | FLAG_K_IN_FRAG);
    if (flags & FLAG_K_IN_DATA) {
      new_flags |= FLAG_K_IN_FRAG;
    }
  }
  head = DataSampleHeader::alloc_msgblock(orig, sz, false);

  head->copy(rd, data_offset);

  head->wr_ptr()[0] = DATA_FRAG;
  head->wr_ptr()[1] = new_flags;
  head->wr_ptr(2);

  std::memset(head->wr_ptr(), 0, 4); // octetsToNextHeader, extraFlags
  head->wr_ptr(4);

  write(head, DATA_FRAG_OCTETS_TO_IQOS, swap_bytes);

  head->copy(rd + data_offset + 8, 16); // readerId, writerId, sequenceNum

  write(head, starting_frag, swap_bytes);
  const size_t max_data = size - sz, orig_payload = orig.cont()->total_length();
  const ACE_CDR::UShort frags =
    static_cast<ACE_CDR::UShort>(std::min(max_data, orig_payload) / FRAG_SIZE);
  write(head, frags, swap_bytes);
  write(head, FRAG_SIZE, swap_bytes);
  write(head, sample_size, swap_bytes);

  if (flags & FLAG_Q) {
    head->copy(rd + iqos_offset, orig.length() - iqos_offset);
  }

  // Create the "tail" message block containing DATA_FRAG with Q=0
  tail = DataSampleHeader::alloc_msgblock(orig, data_offset + 36, false);

  tail->copy(rd, data_offset);

  tail->wr_ptr()[0] = DATA_FRAG;
  tail->wr_ptr()[1] = new_flags & ~FLAG_Q;
  tail->wr_ptr(2);

  std::memset(tail->wr_ptr(), 0, 4); // octetsToNextHeader, extraFlags
  tail->wr_ptr(4);

  write(tail, DATA_FRAG_OCTETS_TO_IQOS, swap_bytes);
  tail->copy(rd + data_offset + 8, 16); // readerId, writerId, sequenceNum

  write(tail, starting_frag + frags, swap_bytes);
  const size_t tail_data = orig_payload - frags * FRAG_SIZE;
  const ACE_CDR::UShort tail_frags =
    static_cast<ACE_CDR::UShort>((tail_data + FRAG_SIZE - 1) / FRAG_SIZE);
  write(tail, tail_frags, swap_bytes);
  write(tail, FRAG_SIZE, swap_bytes);
  write(tail, sample_size, swap_bytes);

  ACE_Message_Block* payload_head = 0;
  ACE_Message_Block* payload_tail;
  DataSampleHeader::split_payload(*orig.cont(), frags * FRAG_SIZE,
                                  payload_head, payload_tail);
  head->cont(payload_head);
  tail->cont(payload_tail);

  return SequenceRange(starting_frag + frags - 1,
                       starting_frag + frags + tail_frags - 1);
}

ACE_INLINE bool OpenDDS::DCPS::RtpsSampleHeader::valid

(

)

const

Definition at line 42 of file RtpsSampleHeader.inl.

References valid_.

{
  return valid_;
}

---

Member Data Documentation

bool OpenDDS::DCPS::RtpsSampleHeader::frag_ [private]

Definition at line 70 of file RtpsSampleHeader.h.

Referenced by init(), more_fragments(), and operator=().

const ACE_CDR::UShort OpenDDS::DCPS::RtpsSampleHeader::FRAG_SIZE = 1024 [static]

Definition at line 94 of file RtpsSampleHeader.h.

Referenced by split().

size_t OpenDDS::DCPS::RtpsSampleHeader::marshaled_size_ [private]

Definition at line 71 of file RtpsSampleHeader.h.

Referenced by init(), and marshaled_size().

size_t OpenDDS::DCPS::RtpsSampleHeader::message_length_ [private]

Definition at line 71 of file RtpsSampleHeader.h.

Referenced by init(), message_length(), and pdu_remaining().

RTPS::Submessage OpenDDS::DCPS::RtpsSampleHeader::submessage_

Definition at line 65 of file RtpsSampleHeader.h.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample(), init(), into_received_data_sample(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::reassemble().

bool OpenDDS::DCPS::RtpsSampleHeader::valid_ [private]

Definition at line 70 of file RtpsSampleHeader.h.

Referenced by init(), operator=(), and valid().

---

The documentation for this class was generated from the following files:

• RtpsSampleHeader.h
• RtpsSampleHeader.cpp
• RtpsSampleHeader.inl

---