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

#include <RtpsSampleHeader.h>

Collaboration diagram for OpenDDS::DCPS::RtpsSampleHeader:

[legend]

Public Member Functions
	RtpsSampleHeader ()

	RtpsSampleHeader (ACE_Message_Block &mb)

RtpsSampleHeader &	operator= (ACE_Message_Block &mn)

void	pdu_remaining (size_t size)

size_t	get_serialized_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	get_max_serialized_size ()

static bool	partial (const ACE_Message_Block &)

static SequenceRange	split (const ACE_Message_Block &orig, size_t size, Message_Block_Ptr &head, Message_Block_Ptr &tail)

static bool	payload_byte_order (const ReceivedDataSample &rds)

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_

bool	data_

size_t	serialized_size_

size_t	message_length_

Detailed Description

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

Definition at line 32 of file RtpsSampleHeader.h.

Constructor & Destructor Documentation

◆ RtpsSampleHeader() [1/2]

ACE_INLINE OpenDDS::DCPS::RtpsSampleHeader::RtpsSampleHeader ( )

Definition at line 14 of file RtpsSampleHeader.inl.

References ACE_INLINE.

   : valid_(false)
   , frag_(false)
   , data_(false)
   , serialized_size_(0)
   , message_length_(0)
 {
 }

◆ RtpsSampleHeader() [2/2]

ACE_INLINE OpenDDS::DCPS::RtpsSampleHeader::RtpsSampleHeader ( ACE_Message_Block & mb )

explicit

Definition at line 24 of file RtpsSampleHeader.inl.

References ACE_INLINE, and init().

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

Member Function Documentation

◆ control_message_supported()

ACE_INLINE bool OpenDDS::DCPS::RtpsSampleHeader::control_message_supported ( char message_id )

static

Definition at line 77 of file RtpsSampleHeader.inl.

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

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::customize_queue_element_helper(), and OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element_non_reliable_i().

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

◆ get_max_serialized_size()

static size_t OpenDDS::DCPS::RtpsSampleHeader::get_max_serialized_size ( )

inlinestatic

Definition at line 39 of file RtpsSampleHeader.h.

39 { return 4; }

◆ get_serialized_size()

ACE_INLINE size_t OpenDDS::DCPS::RtpsSampleHeader::get_serialized_size ( )

Definition at line 59 of file RtpsSampleHeader.inl.

References ACE_INLINE, and serialized_size_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::handle_input().

 {
   return serialized_size_;
 }

◆ init()

void OpenDDS::DCPS::RtpsSampleHeader::init ( ACE_Message_Block & mb )

private

Definition at line 59 of file RtpsSampleHeader.cpp.

References OpenDDS::RTPS::ACKNACK, CASE_SMKIND, ACE_Message_Block::cont(), OpenDDS::RTPS::DATA, OpenDDS::RTPS::DATA_FRAG, OpenDDS::DCPS::ENDIAN_BIG, OpenDDS::DCPS::ENDIAN_LITTLE, OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::FLAG_K_IN_DATA, 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, OpenDDS::DCPS::Encoding::KIND_XCDR1, ACE_Message_Block::length(), OpenDDS::RTPS::NACK_FRAG, OpenDDS::RTPS::PAD, ACE_Message_Block::rd_ptr(), OpenDDS::RTPS::SEC_BODY, OpenDDS::RTPS::SEC_POSTFIX, OpenDDS::RTPS::SEC_PREFIX, OpenDDS::DCPS::Serializer::skip(), OpenDDS::RTPS::SMHDR_SZ, OpenDDS::RTPS::SecuritySubmessage::smHeader, OpenDDS::RTPS::SRTPS_POSTFIX, OpenDDS::RTPS::SRTPS_PREFIX, OpenDDS::RTPS::SubmessageHeader::submessageLength, and ACE_Message_Block::total_length().

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 size_t starting_length = mb.total_length();
   Serializer ser(&mb, Encoding::KIND_XCDR1,
     (flags & FLAG_E) ? ENDIAN_LITTLE : ENDIAN_BIG);
 
   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)
 
 #if defined(OPENDDS_SECURITY)
     // Each submessage type introduced by the Security spec is treated
     // as an opaque octet sequence at this layer.
     case SEC_BODY:
     case SEC_PREFIX:
     case SEC_POSTFIX:
     case SRTPS_PREFIX:
     case SRTPS_POSTFIX: {
     SecuritySubmessage submessage;
     if (ser >> submessage) {
       octetsToNextHeader = submessage.smHeader.submessageLength;
       submessage_.security_sm(submessage);
       submessage_._d(kind);
       valid_ = true;
     }
     break;
     }
 #endif
 
   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);
     data_ = (kind == DATA);
 
     // serialized_size_ is # of bytes of submessage we have read from "mb"
     serialized_size_ = starting_length - mb.total_length();
 
     const ACE_CDR::UShort remaining = static_cast<ACE_CDR::UShort>(message_length_ - SMHDR_SZ);
 
     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 = remaining;
 
     } else if (octetsToNextHeader > remaining) {
       valid_ = false;
       return;
     }
 
     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 - serialized_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>(serialized_size_);
       if (octetsToNextHeader + SMHDR_SZ > marshaled) {
         valid_ = ser.skip(octetsToNextHeader + SMHDR_SZ - marshaled);
         serialized_size_ = octetsToNextHeader + SMHDR_SZ;
       }
     }
   }
 }

◆ into_received_data_sample()

bool OpenDDS::DCPS::RtpsSampleHeader::into_received_data_sample ( ReceivedDataSample & rds )

Definition at line 237 of file RtpsSampleHeader.cpp.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_from_secure(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::handle_input().

 {
   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_ = to_opendds_seqnum(rtps.writerSN);
     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.has_data()) {
         for (CORBA::ULong i = 0; i < rtps.inlineQos.length(); ++i) {
           if (rtps.inlineQos[i]._d() == PID_KEY_HASH) {
             // CDR_BE encapsulation scheme (endianness is not used for key hash)
             rds.replace("\x00\x00\x00\x00", EncapsulationHeader::serialized_size);
             const CORBA::Octet* data = rtps.inlineQos[i].key_hash().value;
             rds.append(reinterpret_cast<const char*>(data), sizeof(DDS::OctetArray16));
             opendds.message_length_ = EncapsulationHeader::serialized_size + sizeof(DDS::OctetArray16);
             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)) {
       // TODO(iguessthislldo: Convert to use Encoding
       // Peek at the byte order from the encapsulation containing the payload.
       opendds.byte_order_ = payload_byte_order(rds);
     }
 
     break;
   }
   case DATA_FRAG: {
     const DataFragSubmessage& rtps = submessage_.data_frag_sm();
     opendds.cdr_encapsulation_ = true;
     opendds.message_length_ = message_length();
     opendds.sequence_ = to_opendds_seqnum(rtps.writerSN);
     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;
 }

◆ message_length()

ACE_INLINE ACE_UINT32 OpenDDS::DCPS::RtpsSampleHeader::message_length ( void )

Definition at line 65 of file RtpsSampleHeader.inl.

References ACE_INLINE, and message_length_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::handle_input().

 {
   return static_cast<ACE_UINT32>(message_length_);
 }

◆ more_fragments()

ACE_INLINE bool OpenDDS::DCPS::RtpsSampleHeader::more_fragments ( void ) const

Definition at line 71 of file RtpsSampleHeader.inl.

References ACE_INLINE, and frag_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_from_secure(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::handle_input().

 {
   return frag_;
 }

◆ operator=()

ACE_INLINE RtpsSampleHeader & OpenDDS::DCPS::RtpsSampleHeader::operator= ( ACE_Message_Block & mn )

Definition at line 35 of file RtpsSampleHeader.inl.

References ACE_INLINE, data_, frag_, init(), and valid_.

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

◆ partial()

static bool OpenDDS::DCPS::RtpsSampleHeader::partial ( const ACE_Message_Block & )

inlinestatic

Definition at line 44 of file RtpsSampleHeader.h.

44 { return false; }

◆ payload_byte_order()

bool OpenDDS::DCPS::RtpsSampleHeader::payload_byte_order ( const ReceivedDataSample & rds )

static

Definition at line 326 of file RtpsSampleHeader.cpp.

References OpenDDS::RTPS::FLAG_E, OpenDDS::DCPS::grow(), header, OpenDDS::RTPS::INFO_TS, OpenDDS::RTPS::INFO_TS_SZ, DDS::Time_t::nanosec, OpenDDS::DCPS::nanoseconds_to_uint32_fractional_seconds(), OpenDDS::DCPS::ReceivedDataSample::peek(), DDS::Time_t::sec, OpenDDS::DCPS::DataSampleHeader::source_timestamp_nanosec_, and OpenDDS::DCPS::DataSampleHeader::source_timestamp_sec_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::decode_payload().

 {
   return rds.peek(1) & RTPS::FLAG_E;
 }

◆ pdu_remaining()

ACE_INLINE void OpenDDS::DCPS::RtpsSampleHeader::pdu_remaining ( size_t size )

Definition at line 53 of file RtpsSampleHeader.inl.

References ACE_INLINE, and message_length_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::handle_input().

 {
   message_length_ = size;
 }

◆ populate_data_control_submessages()

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

static

Definition at line 440 of file RtpsSampleHeader.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DataSampleHeader::byte_order_, OpenDDS::RTPS::DATA, OpenDDS::RTPS::DATA_OCTETS_TO_IQOS, OpenDDS::DCPS::DISPOSE_INSTANCE, OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::DCPS::EntityId_t::entityKind, OpenDDS::RTPS::FLAG_K_IN_DATA, OpenDDS::RTPS::FLAG_Q, OpenDDS::RTPS::SubmessageHeader::flags, OpenDDS::DCPS::grow(), header, OpenDDS::DCPS::TransportSendControlElement::header(), OpenDDS::RTPS::DataSubmessage::inlineQos, OpenDDS::DCPS::INSTANCE_REGISTRATION, OpenDDS::DCPS::TransportSendControlElement::listener(), LM_INFO, OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::TransportSendControlElement::msg_payload(), OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::TransportSendListener::retrieve_inline_qos_data(), OpenDDS::DCPS::DataSampleHeader::sequence_, OpenDDS::RTPS::DataSubmessage::smHeader, OpenDDS::RTPS::to_rtps_seqnum(), and OpenDDS::DCPS::UNREGISTER_INSTANCE.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::customize_queue_element_helper(), and OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element_non_reliable_i().

 {
   using namespace OpenDDS::RTPS;
 
   const DataSampleHeader& header = tsce.header();
   const ACE_CDR::Octet flags = header.byte_order_;
   add_timestamp(subm, flags, header);
 
   static const CORBA::Octet BUILT_IN_WRITER = 0xC2;
 
   DataSubmessage data = {
     {DATA, flags, 0},
     0,
     DATA_OCTETS_TO_IQOS,
     ENTITYID_UNKNOWN,
     header.publication_id_.entityId,
     RTPS::to_rtps_seqnum(header.sequence_),
     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 = DCPS::grow(data.inlineQos) - 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 = DCPS::grow(data.inlineQos) - 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 = DCPS::grow(data.inlineQos) - 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;
   }
 
   CORBA::ULong idx = DCPS::grow(subm) - 1;
   subm[idx].data_sm(data);
 }

◆ populate_data_sample_submessages()

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

static

Definition at line 348 of file RtpsSampleHeader.cpp.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::customize_queue_element_helper(), and OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element_non_reliable_i().

 {
   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));
     i = DCPS::grow(subm);
     subm[i - 1].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));
         i = DCPS::grow(subm);
         subm[i - 1].info_dst_sm(idest);
         break;
       }
     }
   }
 
   DataSubmessage data = {
     {DATA, flags, 0},
     0,
     DATA_OCTETS_TO_IQOS,
     readerId,
     dsle.get_pub_id().entityId,
     RTPS::to_rtps_seqnum(dsle.get_header().sequence_),
     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;
   }
 
   i = DCPS::grow(subm);
   subm[i - 1].data_sm(data);
 }

◆ populate_inline_qos()

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

static

Definition at line 531 of file RtpsSampleHeader.cpp.

References OpenDDS::RTPS::BEST_EFFORT, DDS::BEST_EFFORT_RELIABILITY_QOS, ACE_Message_Block::copy(), OpenDDS::DCPS::TransportSendListener::InlineQosData::dw_qos, OpenDDS::DCPS::grow(), OpenDDS::RTPS::ReliabilityQosPolicyRtps::kind, DDS::ReliabilityQosPolicy::kind, OpenDDS::RTPS::ReliabilityQosPolicyRtps::max_blocking_time, DDS::ReliabilityQosPolicy::max_blocking_time, OpenDDS::RTPS::PID_TOPIC_NAME, PROCESS_INLINE_QOS, OpenDDS::DCPS::TransportSendListener::InlineQosData::pub_qos, DDS::DataWriterQos::reliability, OpenDDS::RTPS::RELIABLE, ACE_CDR::swap_2(), ACE_CDR::swap_4(), TheServiceParticipant, OpenDDS::DCPS::TransportSendListener::InlineQosData::topic_name, OpenDDS::RTPS::ReliabilityKind_t::value, ACE_Message_Block::wr_ptr(), and write().

 {
   using namespace OpenDDS::RTPS;
 
   // Always include topic name (per the spec)
   {
     const int idx = DCPS::grow(plist) - 1;
     plist[idx].string_data(qos_data.topic_name.c_str());
     plist[idx]._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);
   if (qos_data.dw_qos.reliability != default_dw_qos.reliability) {
     const int idx = DCPS::grow(plist) - 1;
 
     ReliabilityQosPolicyRtps reliability;
     reliability.max_blocking_time = qos_data.dw_qos.reliability.max_blocking_time;
 
     if (qos_data.dw_qos.reliability.kind == DDS::BEST_EFFORT_RELIABILITY_QOS) {
       reliability.kind.value = RTPS::BEST_EFFORT;
     } else { // default to RELIABLE for writers
       reliability.kind.value = RTPS::RELIABLE;
     }
 
     plist[idx].reliability(reliability);
   }
   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);
 }

◆ process_iqos()

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

staticprivate

Definition at line 187 of file RtpsSampleHeader.cpp.

References DDS::PresentationQosPolicy::access_scope, ACE_DEBUG, DDS::PresentationQosPolicy::coherent_access, OpenDDS::DCPS::DISPOSE_INSTANCE, OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE, OpenDDS::DCPS::DataSampleHeader::historic_sample_, OpenDDS::DCPS::INSTANCE_REGISTRATION, LM_DEBUG, 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, OpenDDS::DCPS::to_dds_string(), and OpenDDS::DCPS::UNREGISTER_INSTANCE.

 {
   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
     }
   }
 }

◆ split()

SequenceRange OpenDDS::DCPS::RtpsSampleHeader::split	(	const ACE_Message_Block &	orig,
		size_t	size,
		Message_Block_Ptr &	head,
		Message_Block_Ptr &	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 629 of file RtpsSampleHeader.cpp.

References ACE_CDR_BYTE_ORDER, ACE_ERROR, ACE_Message_Block::cont(), ACE_Message_Block::copy(), OpenDDS::RTPS::DATA, OpenDDS::RTPS::DATA_FRAG, OpenDDS::RTPS::DATA_FRAG_OCTETS_TO_IQOS, OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::FLAG_K_IN_DATA, OpenDDS::RTPS::FLAG_K_IN_FRAG, OpenDDS::RTPS::FLAG_N_IN_DATA, OpenDDS::RTPS::FLAG_N_IN_FRAG, OpenDDS::RTPS::FLAG_Q, ACE_Message_Block::length(), LM_ERROR, OPENDDS_END_VERSIONED_NAMESPACE_DECL, ACE_Message_Block::rd_ptr(), OpenDDS::DCPS::unique_ptr< T, Deleter >::release(), OpenDDS::DCPS::unique_ptr< T, Deleter >::reset(), OpenDDS::RTPS::SMHDR_SZ, ACE_Message_Block::total_length(), OpenDDS::DCPS::Transport_debug_level, OpenDDS::DCPS::unknown_sequence_range, ACE_Message_Block::wr_ptr(), and write().

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

 {
   using namespace RTPS;
   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_sequence_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_sequence_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 = flags & (FLAG_E | FLAG_Q);
     if (flags & FLAG_K_IN_DATA) {
       new_flags |= FLAG_K_IN_FRAG;
     }
     if (flags & FLAG_N_IN_DATA) {
       new_flags |= FLAG_N_IN_FRAG;
     }
 
   }
   head.reset(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);
   if (frags == 0) {
     ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: RtpsSampleHeader::split: "
       "Number of Fragments is Zero: min(%B, %B) / FRAG_SIZE\n",
       max_data, orig_payload));
     return unknown_sequence_range;
   }
   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.reset(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);
 
   Message_Block_Ptr payload_head;
   Message_Block_Ptr payload_tail;
   DataSampleHeader::split_payload(*orig.cont(), frags * FRAG_SIZE,
                                   payload_head, payload_tail);
   head->cont(payload_head.release());
   tail->cont(payload_tail.release());
 
   return SequenceRange(starting_frag + frags - 1,
                        starting_frag + frags + tail_frags - 1);
 }

◆ valid()

ACE_INLINE bool OpenDDS::DCPS::RtpsSampleHeader::valid ( void ) const

Definition at line 47 of file RtpsSampleHeader.inl.

References ACE_INLINE, and valid_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_header().

 {
   return valid_;
 }

Member Data Documentation

◆ data_

bool OpenDDS::DCPS::RtpsSampleHeader::data_

private

Definition at line 74 of file RtpsSampleHeader.h.

Referenced by operator=().

◆ frag_

bool OpenDDS::DCPS::RtpsSampleHeader::frag_

private

Definition at line 74 of file RtpsSampleHeader.h.

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

◆ FRAG_SIZE

const ACE_CDR::UShort OpenDDS::DCPS::RtpsSampleHeader::FRAG_SIZE = 1024

static

Definition at line 98 of file RtpsSampleHeader.h.

◆ message_length_

size_t OpenDDS::DCPS::RtpsSampleHeader::message_length_

private

Definition at line 75 of file RtpsSampleHeader.h.

Referenced by message_length(), and pdu_remaining().

◆ serialized_size_

size_t OpenDDS::DCPS::RtpsSampleHeader::serialized_size_

private

Definition at line 75 of file RtpsSampleHeader.h.

Referenced by get_serialized_size().

◆ submessage_

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

Definition at line 69 of file RtpsSampleHeader.h.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_header(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_from_secure(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::reassemble_i().

◆ valid_

bool OpenDDS::DCPS::RtpsSampleHeader::valid_

private

Definition at line 74 of file RtpsSampleHeader.h.

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

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

Public Member Functions

Static Public Member Functions

Public Attributes

Static Public Attributes

Private Member Functions

Static Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ RtpsSampleHeader() [1/2]

◆ RtpsSampleHeader() [2/2]

Member Function Documentation

◆ control_message_supported()

◆ get_max_serialized_size()

◆ get_serialized_size()

◆ init()

◆ into_received_data_sample()

◆ message_length()

◆ more_fragments()

◆ operator=()

◆ partial()

◆ payload_byte_order()

◆ pdu_remaining()

◆ populate_data_control_submessages()

◆ populate_data_sample_submessages()

◆ populate_inline_qos()

◆ process_iqos()

◆ split()

◆ valid()

Member Data Documentation

◆ data_

◆ frag_

◆ FRAG_SIZE

◆ message_length_

◆ serialized_size_

◆ submessage_

◆ valid_