#include <RtpsUdpReceiveStrategy.h>

Inheritance diagram for OpenDDS::DCPS::RtpsUdpReceiveStrategy:

[legend]

Collaboration diagram for OpenDDS::DCPS::RtpsUdpReceiveStrategy:

[legend]

Classes
struct	MessageReceiver

Public Types
typedef std::pair< SequenceNumber, RTPS::FragmentNumberSet >	SeqFragPair

Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

Public Member Functions
	RtpsUdpReceiveStrategy (RtpsUdpDataLink *link, const GuidPrefix_t &local_prefix, ThreadStatusManager &thread_status_manager)

virtual int	handle_input (ACE_HANDLE fd)

bool	remove_frags_from_bitmap (CORBA::Long bitmap[], CORBA::ULong num_bits, const SequenceNumber &base, const GUID_t &pub_id, ACE_CDR::ULong &samples_requested)

void	remove_fragments (const SequenceRange &range, const GUID_t &pub_id)

typedef	OPENDDS_VECTOR (SeqFragPair) FragmentInfo

void	clear_completed_fragments (const GUID_t &pub_id)

bool	has_fragments (const SequenceRange &range, const GUID_t &pub_id, FragmentInfo *frag_info=0)

const ReceivedDataSample *	withhold_data_from (const GUID_t &sub_id)

void	do_not_withhold_data_from (const GUID_t &sub_id)

virtual void	begin_transport_header_processing ()
	Begin Current Transport Header Processing. More...

virtual void	end_transport_header_processing ()
	End Current Transport Header Processing. More...

Public Member Functions inherited from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >
virtual	~TransportReceiveStrategy ()

int	start ()

void	stop ()

int	handle_dds_input (ACE_HANDLE fd)

virtual void	relink (bool do_suspend=true)

const RtpsTransportHeader &	received_header () const

RtpsTransportHeader &	received_header ()

const RtpsSampleHeader &	received_sample_header () const

RtpsSampleHeader &	received_sample_header ()

ACE_Message_Block *	to_msgblock (const ReceivedDataSample &sample)

Public Member Functions inherited from OpenDDS::DCPS::TransportStrategy
virtual	~TransportStrategy ()

Public Member Functions inherited from OpenDDS::DCPS::RcObject
virtual	~RcObject ()

virtual void	_add_ref ()

virtual void	_remove_ref ()

long	ref_count () const

WeakObject *	_get_weak_object () const

Public Member Functions inherited from OpenDDS::DCPS::RcEventHandler
	RcEventHandler ()

ACE_Event_Handler::Reference_Count	add_reference ()

ACE_Event_Handler::Reference_Count	remove_reference ()

Public Member Functions inherited from ACE_Event_Handler
virtual	~ACE_Event_Handler (void)

virtual ACE_HANDLE	get_handle (void) const

virtual void	set_handle (ACE_HANDLE)

virtual int	priority (void) const

virtual void	priority (int priority)

virtual int	handle_output (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_exception (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_timeout (const ACE_Time_Value &current_time, const void *act=0)

virtual int	handle_exit (ACE_Process *)

virtual int	handle_close (ACE_HANDLE handle, ACE_Reactor_Mask close_mask)

virtual int	handle_signal (int signum, siginfo_t =0, ucontext_t =0)

virtual int	resume_handler (void)

virtual int	handle_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual int	handle_group_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual void	reactor (ACE_Reactor *reactor)

virtual ACE_Reactor *	reactor (void) const

virtual ACE_Reactor_Timer_Interface *	reactor_timer_interface (void) const

Reference_Counting_Policy &	reference_counting_policy (void)

Static Public Member Functions
static ssize_t	receive_bytes_helper (iovec iov[], int n, const ACE_SOCK_Dgram &socket, ACE_INET_Addr &remote_address, DCPS::RcHandle< ICE::Agent > agent, DCPS::WeakRcHandle< ICE::Endpoint > endpoint, RtpsUdpTransport &tport, bool &stop)

Static Public Member Functions inherited from ACE_Event_Handler
static ACE_THR_FUNC_RETURN	read_adapter (void *event_handler)

static int	register_stdin_handler (ACE_Event_Handler eh, ACE_Reactor reactor, ACE_Thread_Manager *thr_mgr, int flags=THR_DETACHED)

static int	remove_stdin_handler (ACE_Reactor reactor, ACE_Thread_Manager thr_mgr)

Static Public Attributes
static const size_t	BUFFER_COUNT = 1u

Static Public Attributes inherited from OpenDDS::DCPS::TransportReceiveConstants
static const size_t	RECEIVE_BUFFERS = DEFAULT_TRANSPORT_RECEIVE_BUFFERS

static const size_t	BUFFER_LOW_WATER = 4096

static const size_t	MESSAGE_BLOCKS = 1000

static const size_t	DATA_BLOCKS = 100

Private Types
typedef TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >	BaseReceiveStrategy

Private Member Functions
bool	getDirectedWriteReaders (RepoIdSet &directedWriteReaders, const RTPS::DataSubmessage &ds) const

const ACE_SOCK_Dgram &	choose_recv_socket (ACE_HANDLE fd) const

virtual ssize_t	receive_bytes (iovec iov[], int n, ACE_INET_Addr &remote_address, ACE_HANDLE fd, bool &stop)
	Only our subclass knows how to do this. More...

virtual void	deliver_sample (ReceivedDataSample &sample, const ACE_INET_Addr &remote_address)
	Called when there is a ReceivedDataSample to be delivered. More...

void	deliver_sample_i (ReceivedDataSample &sample, const RTPS::Submessage &submessage, const NetworkAddress &remote_addr)

virtual int	start_i ()
	Let the subclass start. More...

virtual void	stop_i ()
	Let the subclass stop. More...

virtual bool	check_header (const RtpsTransportHeader &header)
	Check the transport header for suitability. More...

virtual bool	check_header (const RtpsSampleHeader &header)
	Check the data sample header for suitability. More...

virtual bool	reassemble (ReceivedDataSample &data)

virtual bool	reassemble_i (ReceivedDataSample &data, RtpsSampleHeader &rsh)

bool	sec_submsg_to_octets (DDS::OctetSeq &encoded, const RTPS::Submessage &postfix)

void	deliver_from_secure (const RTPS::Submessage &submessage, const NetworkAddress &remote_addr)

bool	decode_payload (ReceivedDataSample &sample, const RTPS::DataSubmessage &submessage)

bool	check_encoded (const EntityId_t &sender)

	OPENDDS_VECTOR (RTPS::Submessage) secure_submessages_

Private Attributes
RtpsUdpDataLink *	link_

SequenceNumber	last_received_

const ReceivedDataSample *	recvd_sample_

RepoIdSet	readers_withheld_

RepoIdSet	readers_selected_

ACE_UINT16	fragment_size_

FragmentRange	frags_

ACE_UINT32	total_frags_

TransportReassembly	reassembly_

MessageReceiver	receiver_

ThreadStatusManager &	thread_status_manager_

ACE_INET_Addr	remote_address_

RTPS::Message	message_

RTPS::SecuritySubmessage	secure_prefix_

ReceivedDataSample	secure_sample_

bool	encoded_rtps_

bool	encoded_submsg_

Additional Inherited Members
Public Attributes inherited from ACE_Event_Handler
	LO_PRIORITY

	HI_PRIORITY

	NULL_MASK

	READ_MASK

	WRITE_MASK

	EXCEPT_MASK

	ACCEPT_MASK

	CONNECT_MASK

	TIMER_MASK

	QOS_MASK

	GROUP_QOS_MASK

	SIGNAL_MASK

	ALL_EVENTS_MASK

	RWE_MASK

	DONT_CALL

	ACE_EVENT_HANDLER_NOT_RESUMED

	ACE_REACTOR_RESUMES_HANDLER

	ACE_APPLICATION_RESUMES_HANDLER

Protected Types inherited from ACE_Event_Handler
typedef ACE_Atomic_Op< ACE_SYNCH_MUTEX, Reference_Count >	Atomic_Reference_Count

Protected Member Functions inherited from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >
	TransportReceiveStrategy (const TransportInst_rch &config, size_t receive_buffers_count=RECEIVE_BUFFERS)

virtual void	finish_message ()

int	skip_bad_pdus ()
	Ignore bad PDUs by skipping over them. More...

void	reset ()

size_t	pdu_remaining () const

size_t	successor_index (size_t index) const
	Manage an index into the receive buffer array. More...

void	update_buffer_index (bool &done)

	OPENDDS_VECTOR (ACE_Message_Block *) receive_buffers_
	Set of receive buffers in use. More...

Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Protected Member Functions inherited from ACE_Event_Handler
	ACE_Event_Handler (ACE_Reactor *=0, int priority=ACE_Event_Handler::LO_PRIORITY)

Protected Attributes inherited from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >
bool	gracefully_disconnected_
	Flag indicates if the GRACEFUL_DISCONNECT message is received. More...

size_t	receive_sample_remaining_
	Bytes remaining in the current DataSample. More...

RtpsTransportHeader	receive_transport_header_
	Current receive TransportHeader. More...

TransportMessageBlockAllocator	mb_allocator_

TransportDataBlockAllocator	db_allocator_

TransportDataAllocator	data_allocator_

ACE_Lock_Adapter< ACE_SYNCH_MUTEX >	receive_lock_
	Locking strategy for the allocators. More...

size_t	buffer_index_
	Current receive buffer index in use. More...

RtpsSampleHeader	data_sample_header_
	Current data sample header. More...

ACE_Message_Block *	payload_

bool	good_pdu_

size_t	pdu_remaining_
	Amount of the current PDU that has not been processed yet. More...

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Definition at line 40 of file RtpsUdpReceiveStrategy.h.

Member Typedef Documentation

◆ BaseReceiveStrategy

typedef TransportReceiveStrategy<RtpsTransportHeader, RtpsSampleHeader> OpenDDS::DCPS::RtpsUdpReceiveStrategy::BaseReceiveStrategy

private

Definition at line 132 of file RtpsUdpReceiveStrategy.h.

◆ SeqFragPair

typedef std::pair<SequenceNumber, RTPS::FragmentNumberSet> OpenDDS::DCPS::RtpsUdpReceiveStrategy::SeqFragPair

Definition at line 65 of file RtpsUdpReceiveStrategy.h.

Constructor & Destructor Documentation

◆ RtpsUdpReceiveStrategy()

OpenDDS::DCPS::RtpsUdpReceiveStrategy::RtpsUdpReceiveStrategy	(	RtpsUdpDataLink *	link,
		const GuidPrefix_t &	local_prefix,
		ThreadStatusManager &	thread_status_manager
	)

Definition at line 31 of file RtpsUdpReceiveStrategy.cpp.

References ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, ACE_NEW_MALLOC, OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::data_allocator_, OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::db_allocator_, OpenDDS::DCPS::Cached_Allocator_With_Overflow< T, ACE_LOCK >::malloc(), ACE_Time_Value::max_time, OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::mb_allocator_, ACE_Message_Block::MB_DATA, OpenDDS::DCPS::RECEIVE_DATA_BUFFER_SIZE, OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::receive_lock_, secure_prefix_, OpenDDS::RTPS::SecuritySubmessage::smHeader, OpenDDS::DCPS::SUBMESSAGE_NONE, OpenDDS::RTPS::SubmessageHeader::submessageId, and ACE_Time_Value::zero.

   : BaseReceiveStrategy(link->config(), BUFFER_COUNT)
   , link_(link)
   , last_received_()
   , recvd_sample_(0)
   , fragment_size_(0)
   , total_frags_(0)
   , reassembly_(link->config()->fragment_reassembly_timeout_)
   , receiver_(local_prefix)
   , thread_status_manager_(thread_status_manager)
 #ifdef OPENDDS_SECURITY
   , secure_sample_()
   , encoded_rtps_(false)
   , encoded_submsg_(false)
 #endif
 {
   // Since BUFFER_COUNT is 1, the index will always be 0
   const size_t INDEX = 0;
 
   if (receive_buffers_[INDEX] == 0) {
     ACE_NEW_MALLOC(
       receive_buffers_[INDEX],
       (ACE_Message_Block*) mb_allocator_.malloc(sizeof(ACE_Message_Block)),
       ACE_Message_Block(
         RECEIVE_DATA_BUFFER_SIZE,           // Buffer size
         ACE_Message_Block::MB_DATA,         // Default
         0,                                  // Start with no continuation
         0,                                  // Let the constructor allocate
         &data_allocator_,                   // Our buffer cache
         &receive_lock_,                     // Our locking strategy
         ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, // Default
         ACE_Time_Value::zero,               // Default
         ACE_Time_Value::max_time,           // Default
         &db_allocator_,                     // Our data block cache
         &mb_allocator_                      // Our message block cache
       ));
   }
 
 #ifdef OPENDDS_SECURITY
   secure_prefix_.smHeader.submessageId = SUBMESSAGE_NONE;
 #endif
 }

Member Function Documentation

◆ begin_transport_header_processing()

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::begin_transport_header_processing ( )

virtual

Begin Current Transport Header Processing.

Reimplemented from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 1065 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::enable_response_queue(), and link_.

 {
   link_->enable_response_queue();
 }

◆ check_encoded()

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_encoded ( const EntityId_t & sender )

private

Definition at line 475 of file RtpsUdpReceiveStrategy.cpp.

References ACE_DEBUG, OpenDDS::DCPS::SecurityDebug::encdec_warn, encoded_rtps_, encoded_submsg_, DDS::Security::ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_PROTECTED, DDS::Security::ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID, DDS::HANDLE_NIL, OpenDDS::DCPS::RtpsUdpDataLink::handle_registry(), OpenDDS::DCPS::GuidConverter::isReader(), link_, LM_WARNING, OpenDDS::DCPS::RtpsUdpDataLink::local_crypto_handle(), OpenDDS::DCPS::make_id(), OPENDDS_STRING, receiver_, OpenDDS::RTPS::security_attributes_to_bitmask(), OpenDDS::DCPS::security_debug, OpenDDS::DCPS::RtpsUdpDataLink::separate_message(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::source_guid_prefix_.

Referenced by deliver_sample_i().

 {
 #ifdef OPENDDS_SECURITY
   using namespace DDS::Security;
   const GUID_t sendGuid = make_id(receiver_.source_guid_prefix_, sender);
   const GuidConverter conv(sendGuid);
 
   if (link_->local_crypto_handle() != DDS::HANDLE_NIL
       && !encoded_rtps_ && !RtpsUdpDataLink::separate_message(sender)) {
     if (security_debug.encdec_warn) {
       ACE_DEBUG((LM_WARNING, "(%P|%t) RtpsUdpReceiveStrategy::check_encoded "
                  "Full message from %C requires protection, dropping\n",
                  OPENDDS_STRING(conv).c_str()));
     }
     return false;
   }
 
   const EndpointSecurityAttributesMask esa = RTPS::security_attributes_to_bitmask(
     conv.isReader() ?
     link_->handle_registry()->get_remote_datareader_security_attributes(sendGuid) :
     link_->handle_registry()->get_remote_datawriter_security_attributes(sendGuid));
   static const EndpointSecurityAttributesMask MASK_PROTECT_SUBMSG =
     ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID | ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_PROTECTED;
   if ((esa & MASK_PROTECT_SUBMSG) == MASK_PROTECT_SUBMSG && !encoded_submsg_) {
     if (security_debug.encdec_warn) {
       ACE_DEBUG((LM_WARNING, "(%P|%t) RtpsUdpReceiveStrategy::check_encoded "
                  "Submessage from %C requires protection, dropping\n",
                  OPENDDS_STRING(conv).c_str()));
     }
     return false;
   }
 #else
   ACE_UNUSED_ARG(sender);
 #endif
   return true;
 }

◆ check_header() [1/2]

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_header ( const RtpsTransportHeader & header )

privatevirtual

Check the transport header for suitability.

Reimplemented from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 1025 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::RTPS::Message::hdr, OpenDDS::DCPS::RtpsTransportHeader::header_, OpenDDS::DCPS::TransportDebug::log_messages, message_, receiver_, remote_address_, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::reset(), secure_prefix_, OpenDDS::RTPS::SecuritySubmessage::smHeader, OpenDDS::DCPS::SUBMESSAGE_NONE, OpenDDS::RTPS::SubmessageHeader::submessageId, OpenDDS::RTPS::Message::submessages, OpenDDS::DCPS::transport_debug, and OpenDDS::DCPS::RtpsTransportHeader::valid().

Referenced by deliver_from_secure(), and handle_input().

 {
   receiver_.reset(remote_address_, header.header_);
   if (transport_debug.log_messages) {
     message_.submessages.length(0);
     message_.hdr = header.header_;
   }
 
 #ifdef OPENDDS_SECURITY
   secure_prefix_.smHeader.submessageId = SUBMESSAGE_NONE;
 #endif
 
   return header.valid();
 }

◆ check_header() [2/2]

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_header ( const RtpsSampleHeader & header )

privatevirtual

Check the data sample header for suitability.

Reimplemented from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 1041 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::RTPS::DATA_FRAG, OpenDDS::RTPS::Submessage::data_frag_sm, fragment_size_, OpenDDS::RTPS::DataFragSubmessage::fragmentsInSubmessage, OpenDDS::RTPS::DataFragSubmessage::fragmentSize, OpenDDS::RTPS::DataFragSubmessage::fragmentStartingNum, frags_, receiver_, OpenDDS::RTPS::DataFragSubmessage::sampleSize, secure_prefix_, OpenDDS::RTPS::SecuritySubmessage::smHeader, OpenDDS::DCPS::RtpsSampleHeader::submessage_, OpenDDS::RTPS::SubmessageHeader::submessageId, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::submsg(), total_frags_, OpenDDS::DCPS::RtpsSampleHeader::valid(), and OpenDDS::RTPS::FragmentNumber_t::value.

 {
 
 #ifdef OPENDDS_SECURITY
   if (secure_prefix_.smHeader.submessageId) {
     return header.valid();
   }
 #endif
 
   receiver_.submsg(header.submessage_);
 
   // save fragmentation details for use in reassemble()
   if (header.valid() && header.submessage_._d() == RTPS::DATA_FRAG) {
     const RTPS::DataFragSubmessage& rtps = header.submessage_.data_frag_sm();
     frags_.first = rtps.fragmentStartingNum.value;
     frags_.second = frags_.first + (rtps.fragmentsInSubmessage - 1);
     fragment_size_ = rtps.fragmentSize;
     total_frags_ = (rtps.sampleSize / rtps.fragmentSize) + (rtps.sampleSize % rtps.fragmentSize ? 1 : 0);
   }
 
   return header.valid();
 }

◆ choose_recv_socket()

const ACE_SOCK_Dgram & OpenDDS::DCPS::RtpsUdpReceiveStrategy::choose_recv_socket ( ACE_HANDLE fd ) const

private

Definition at line 326 of file RtpsUdpReceiveStrategy.cpp.

References ACE_Event_Handler::get_handle(), ACE_IPC_SAP::get_handle(), link_, OpenDDS::DCPS::RtpsUdpDataLink::multicast_socket(), and OpenDDS::DCPS::RtpsUdpDataLink::unicast_socket().

Referenced by receive_bytes().

 {
 #ifdef ACE_HAS_IPV6
   if (fd == link_->ipv6_multicast_socket().get_handle()) {
     return link_->ipv6_multicast_socket();
   }
   if (fd == link_->ipv6_unicast_socket().get_handle()) {
     return link_->ipv6_unicast_socket();
   }
 #endif
   if (fd == link_->multicast_socket().get_handle()) {
     return link_->multicast_socket();
   }
   return link_->unicast_socket();
 }

◆ clear_completed_fragments()

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::clear_completed_fragments ( const GUID_t & pub_id )

Definition at line 1183 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::TransportReassembly::clear_completed(), and reassembly_.

 {
   reassembly_.clear_completed(pub_id);
 }

◆ decode_payload()

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::decode_payload	(	ReceivedDataSample &	sample,
		const RTPS::DataSubmessage &	submessage
	)

private

Definition at line 926 of file RtpsUdpReceiveStrategy.cpp.

References ACE_ERROR, OpenDDS::DCPS::ReceivedDataSample::append(), OpenDDS::DCPS::DataSampleHeader::byte_order_, OpenDDS::DCPS::ReceivedDataSample::clear(), DDS::Security::SecurityException::code, OpenDDS::DCPS::ReceivedDataSample::copy_data(), OpenDDS::DCPS::SecurityDebug::encdec_warn, OpenDDS::STUN::encoding(), DDS::Security::ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_PAYLOAD_PROTECTED, DDS::Security::ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID, OpenDDS::DCPS::equal_guid_prefixes(), OpenDDS::RTPS::SubmessageHeader::flags, OpenDDS::DCPS::GUID_t::guidPrefix, DDS::HANDLE_NIL, OpenDDS::DCPS::RtpsUdpDataLink::handle_registry(), OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::RTPS::DataSubmessage::inlineQos, OpenDDS::DCPS::Encoding::KIND_XCDR1, link_, LM_WARNING, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::local_, DDS::Security::SecurityException::message, DDS::Security::SecurityException::minor_code, OpenDDS::DCPS::RtpsSampleHeader::payload_byte_order(), OpenDDS::DCPS::DataSampleHeader::publication_id_, receiver_, OpenDDS::RTPS::security_attributes_to_bitmask(), OpenDDS::DCPS::RtpsUdpDataLink::security_config(), OpenDDS::DCPS::security_debug, OpenDDS::DCPS::serialized_size(), and OpenDDS::RTPS::DataSubmessage::smHeader.

Referenced by deliver_sample_i().

 {
   using namespace DDS::Security;
 
   static const EndpointSecurityAttributesMask MASK_PROTECT_PAYLOAD =
     ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID | ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_PAYLOAD_PROTECTED;
   const CryptoTransform_var crypto = link_->security_config()->get_crypto_transform();
   DatawriterCryptoHandle writer_crypto_handle;
   EndpointSecurityAttributesMask esa;
 
   if (equal_guid_prefixes(sample.header_.publication_id_.guidPrefix, receiver_.local_)) {
     writer_crypto_handle =
       link_->handle_registry()->get_local_datawriter_crypto_handle(sample.header_.publication_id_);
     esa =
       RTPS::security_attributes_to_bitmask(link_->handle_registry()->get_local_datawriter_security_attributes(sample.header_.publication_id_));
   } else {
     writer_crypto_handle =
       link_->handle_registry()->get_remote_datawriter_crypto_handle(sample.header_.publication_id_);
     esa =
       RTPS::security_attributes_to_bitmask(link_->handle_registry()->get_remote_datawriter_security_attributes(sample.header_.publication_id_));
   }
 
   const bool payload_protected = (esa & MASK_PROTECT_PAYLOAD) == MASK_PROTECT_PAYLOAD;
 
   if (writer_crypto_handle == DDS::HANDLE_NIL || !crypto || !payload_protected) {
     return true;
   }
 
   DDS::OctetSeq encoded = sample.copy_data(), plain, iQos;
 
   const Encoding encoding(Encoding::KIND_XCDR1,
     static_cast<Endianness>(submsg.smHeader.flags & 1));
   size_t iQosSize = 0;
   serialized_size(encoding, iQosSize, submsg.inlineQos);
   iQos.length(static_cast<unsigned int>(iQosSize));
   const char* iQos_raw = reinterpret_cast<const char*>(iQos.get_buffer());
   ACE_Message_Block iQosMb(iQos_raw, iQos.length());
   Serializer ser(&iQosMb, encoding);
   ser << submsg.inlineQos;
 
   SecurityException ex = {"", 0, 0};
   // DDS-Security: since origin authentication for payload is not yet supported
   // the reader's crypto handle is NIL here (could be multiple readers in this
   // participant)
   const bool ok = crypto->decode_serialized_payload(plain, encoded, iQos,
                                                     DDS::HANDLE_NIL,
                                                     writer_crypto_handle, ex);
   if (ok) {
     // The ReceivedDataSample's message block uses the transport's data block so it
     // can't be modified in-place, instead replace it with a new block.
     sample.clear();
     sample.append(reinterpret_cast<const char*>(plain.get_buffer()), plain.length());
 
     if (plain.length() > 1) {
       sample.header_.byte_order_ = RtpsSampleHeader::payload_byte_order(sample);
     }
 
   } else if (security_debug.encdec_warn) {
     ACE_ERROR((LM_WARNING, "(%P|%t) {encdec_warn} RtpsUdpReceiveStrategy: "
                "decode_serialized_payload failed [%d.%d]: %C\n",
                ex.code, ex.minor_code, ex.message.in()));
   }
 
   return ok;
 }

◆ deliver_from_secure()

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_from_secure	(	const RTPS::Submessage &	submessage,
		const NetworkAddress &	remote_addr
	)

private

Definition at line 755 of file RtpsUdpReceiveStrategy.cpp.

Referenced by deliver_sample_i().

 {
   using namespace DDS::Security;
 
   const CryptoTransform_var crypto = link_->security_config()->get_crypto_transform();
   if (!crypto) {
     // security not enabled for this datalink -- this can be reached
     // when a secure message is seen on the same multicast group
     return;
   }
 
   const GUID_t peer = make_id(receiver_.source_guid_prefix_, ENTITYID_PARTICIPANT);
   const ParticipantCryptoHandle peer_pch = equal_guid_prefixes(peer.guidPrefix, receiver_.local_) ?
     link_->local_crypto_handle() :
     link_->handle_registry()->get_remote_participant_crypto_handle(peer);
 
   DDS::OctetSeq encoded_submsg, plain_submsg;
   if (!sec_submsg_to_octets(encoded_submsg, submessage)) {
     if (security_debug.encdec_warn) {
       ACE_ERROR((LM_WARNING, ACE_TEXT("(%P|%t) {encdec_warn} RtpsUdpReceiveStrategy: ")
                  ACE_TEXT("deliver_from_secure failed to encode submessage %C RPCH %d\n"),
                  LogGuid(peer).c_str(), peer_pch));
     }
     return;
   }
   secure_prefix_.smHeader.submessageId = SUBMESSAGE_NONE;
   secure_sample_ = ReceivedDataSample();
 
   DatawriterCryptoHandle dwch = DDS::HANDLE_NIL;
   DatareaderCryptoHandle drch = DDS::HANDLE_NIL;
   SecureSubmessageCategory_t category = INFO_SUBMESSAGE;
   SecurityException ex = {"", 0, 0};
 
   bool ok = crypto->preprocess_secure_submsg(dwch, drch, category, encoded_submsg,
                                              link_->local_crypto_handle(), peer_pch, ex);
 
   if (ok) {
     VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpReceiveStrategy::deliver_from_secure ")
       ACE_TEXT("dwch is %d and drch is %d\n"), dwch, drch), 4);
   }
 
   if (ok && category == DATAWRITER_SUBMESSAGE) {
     ok = crypto->decode_datawriter_submessage(plain_submsg, encoded_submsg,
                                               drch, dwch, ex);
 
   } else if (ok && category == DATAREADER_SUBMESSAGE) {
     ok = crypto->decode_datareader_submessage(plain_submsg, encoded_submsg,
                                               dwch, drch, ex);
 
   } else if (ok && category == INFO_SUBMESSAGE) {
     return;
 
   } else {
     if (security_debug.encdec_warn) {
       ACE_ERROR((LM_WARNING, ACE_TEXT("(%P|%t) {encdec_warn} RtpsUdpReceiveStrategy::deliver_from_secure ")
                  ACE_TEXT("failed remote %C RPCH %d, [%d.%d]: %C\n"),
                  LogGuid(peer).c_str(), peer_pch, ex.code, ex.minor_code, ex.message.in()));
     }
     return;
   }
 
   if (!ok) {
     bool dw = category == DATAWRITER_SUBMESSAGE;
     if (security_debug.encdec_warn) {
       ACE_ERROR((LM_WARNING, ACE_TEXT("(%P|%t) {encdec_warn} RtpsUdpReceiveStrategy::deliver_from_secure ")
                  ACE_TEXT("decode %C submessage failed [%d.%d]: \"%C\" ")
                  ACE_TEXT("(rpch: %u, local d%cch: %u, remote d%cch: %u)\n"),
                  dw ? "writer" : "reader",
                  ex.code, ex.minor_code, ex.message.in(),
                  peer_pch,
                  dw ? 'r' : 'w',
                  dw ? drch : dwch,
                  dw ? 'w' : 'r',
                  dw ? dwch : drch
                  ));
     }
     return;
   }
 
   ACE_Message_Block mb(plain_submsg.length());
   mb.copy(reinterpret_cast<const char*>(plain_submsg.get_buffer()), mb.size());
 
   if (Transport_debug_level > 5) {
     ACE_HEX_DUMP((LM_DEBUG, mb.rd_ptr(), mb.length(),
                   category == DATAWRITER_SUBMESSAGE ?
                   ACE_TEXT("RtpsUdpReceiveStrategy: decoded writer submessage") :
                   ACE_TEXT("RtpsUdpReceiveStrategy: decoded reader submessage")));
   }
 
   RtpsSampleHeader rsh(mb);
   if (check_header(rsh)) {
     ReceivedDataSample plain_sample(mb);
     if (rsh.into_received_data_sample(plain_sample)) {
       if (rsh.more_fragments()) {
         VDBG((LM_DEBUG, "(%P|%t) DBG:   Attempt reassembly of decoded fragments\n"));
         if (reassemble_i(plain_sample, rsh)) {
           VDBG((LM_DEBUG, "(%P|%t) DBG:   Reassembled complete message from decoded\n"));
           encoded_submsg_ = true;
           if (transport_debug.log_messages) {
             // Pop the secure envelope.
             message_.submessages.length(message_.submessages.length() - 3);
             DCPS::push_back(message_.submessages, rsh.submessage_);
           }
           deliver_sample_i(plain_sample, rsh.submessage_, remote_addr);
           return;
         }
       }
       encoded_submsg_ = true;
       if (transport_debug.log_messages) {
         // Pop the secure envelope.
         message_.submessages.length(message_.submessages.length() - 3);
         DCPS::push_back(message_.submessages, rsh.submessage_);
       }
       deliver_sample_i(plain_sample, rsh.submessage_, remote_addr);
     }
   }
 }

◆ deliver_sample()

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample	(	ReceivedDataSample &	sample,
		const ACE_INET_Addr &	remote_address
	)

privatevirtual

Called when there is a ReceivedDataSample to be delivered.

Implements OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 513 of file RtpsUdpReceiveStrategy.cpp.

References ACE_DEBUG, OpenDDS::RTPS::DATA, deliver_sample_i(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::dest_guid_prefix_, encoded_submsg_, link_, LM_DEBUG, OpenDDS::DCPS::RtpsUdpDataLink::local_prefix(), OpenDDS::DCPS::TransportDebug::log_dropped_messages, OpenDDS::DCPS::TransportDebug::log_messages, message_, OpenDDS::DCPS::push_back(), OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::received_sample_header(), receiver_, OpenDDS::RTPS::SEC_POSTFIX, OpenDDS::RTPS::SEC_PREFIX, secure_prefix_, secure_sample_, OpenDDS::RTPS::SecuritySubmessage::smHeader, OpenDDS::DCPS::RtpsSampleHeader::submessage_, OpenDDS::RTPS::SubmessageHeader::submessageId, OpenDDS::RTPS::Message::submessages, and OpenDDS::DCPS::transport_debug.

Referenced by handle_input().

 {
   using namespace RTPS;
 
   if (std::memcmp(receiver_.dest_guid_prefix_, link_->local_prefix(),
                   sizeof(GuidPrefix_t))) {
     // Not our message, we may be on multicast listening to all the others.
     if (transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::deliver_sample - not destination\n"));
     }
     return;
   }
 
   const RtpsSampleHeader& rsh = received_sample_header();
 
   if (transport_debug.log_messages) {
     DCPS::push_back(message_.submessages, rsh.submessage_);
   }
 
 #ifdef OPENDDS_SECURITY
   const SubmessageKind kind = rsh.submessage_._d();
 
   if (secure_prefix_.smHeader.submessageId == SEC_PREFIX && kind != SEC_POSTFIX) {
     // secure envelope in progress, defer processing
     secure_submessages_.push_back(rsh.submessage_);
     if (kind == DATA) {
       secure_sample_ = sample;
     }
     return;
   }
 
   encoded_submsg_ = false;
 #endif
 
   deliver_sample_i(sample, rsh.submessage_, NetworkAddress(remote_address));
 }

◆ deliver_sample_i()

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample_i	(	ReceivedDataSample &	sample,
		const RTPS::Submessage &	submessage,
		const NetworkAddress &	remote_addr
	)

private

Definition at line 552 of file RtpsUdpReceiveStrategy.cpp.

Referenced by deliver_from_secure(), and deliver_sample().

 {
   using namespace RTPS;
   const SubmessageKind kind = submessage._d();
 
   switch (kind) {
   case INFO_SRC:
   case INFO_REPLY_IP4:
   case INFO_DST:
   case INFO_REPLY:
   case INFO_TS:
     // No-op: the INFO_* submessages only modify the state of the
     // MessageReceiver (see check_header()), they are not passed up to DCPS.
     break;
 
   case DATA: {
     receiver_.fill_header(sample.header_);
     const DataSubmessage& data = submessage.data_sm();
     if (!check_encoded(data.writerId)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::deliver_sample_i - decode error\n"));
       }
       break;
     }
 
 #ifdef OPENDDS_SECURITY
     if (!decode_payload(sample, data)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::deliver_sample_i - decode error\n"));
       }
       break;
     }
 #endif
 
     RepoIdSet directedWriteReaders;
     getDirectedWriteReaders(directedWriteReaders, data);
 
     recvd_sample_ = &sample;
     readers_selected_.clear();
     readers_withheld_.clear();
     // If this sample should be withheld from some readers in order to maintain
     // in-order delivery, link_->received() will add it to readers_withheld_ otherwise
     // it will be added to readers_selected_
     link_->received(data, receiver_.source_guid_prefix_, remote_addr);
     recvd_sample_ = 0;
 
     link_->filterBestEffortReaders(sample, readers_selected_, readers_withheld_);
 
     if (data.readerId != ENTITYID_UNKNOWN) {
       GUID_t reader;
       std::memcpy(reader.guidPrefix, link_->local_prefix(),
                   sizeof(GuidPrefix_t));
       reader.entityId = data.readerId;
       if (!readers_withheld_.count(reader) &&
           (directedWriteReaders.empty() || directedWriteReaders.find(reader) != directedWriteReaders.end())) {
         if (Transport_debug_level > 5) {
           ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpReceiveStrategy[%@]::deliver_sample_i - ")
             ACE_TEXT("calling DataLink::data_received for seq: %q to reader %C\n"),
             this, sample.header_.sequence_.getValue(), LogGuid(reader).c_str()));
         }
         link_->data_received(sample, reader);
       }
     } else {
       if (Transport_debug_level > 5) {
         OPENDDS_STRING included_ids;
         bool first = true;
         RepoIdSet::iterator iter = readers_selected_.begin();
         while (iter != readers_selected_.end()) {
           included_ids += (first ? "" : "\n") + LogGuid(*iter).conv_;
           first = false;
           ++iter;
         }
         OPENDDS_STRING excluded_ids;
         first = true;
         RepoIdSet::iterator iter2 = this->readers_withheld_.begin();
         while (iter2 != readers_withheld_.end()) {
           excluded_ids += (first ? "" : "\n") + LogGuid(*iter2).conv_;
           first = false;
           ++iter2;
         }
         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpReceiveStrategy[%@]::deliver_sample_i:")
           ACE_TEXT(" readers_selected ids: %C\n")
           ACE_TEXT(" readers_withheld ids: %C\n"),
           this, included_ids.c_str(), excluded_ids.c_str()));
       }
 
       if (readers_withheld_.empty() && readers_selected_.empty()) {
         if (directedWriteReaders.empty()) {
           if (Transport_debug_level > 5) {
             ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpReceiveStrategy[%@]::deliver_sample_i - ")
               ACE_TEXT("calling DataLink::data_received for seq: %q TO ALL, no exclusion or inclusion\n"),
               this, sample.header_.sequence_.getValue()));
           }
           link_->data_received(sample);
         } else {
           if (Transport_debug_level > 5) {
             ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpReceiveStrategy[%@]::deliver_sample_i - ")
               ACE_TEXT("calling DataLink::data_received_include for seq: %q to directedWriteReaders\n"),
               this, sample.header_.sequence_.getValue()));
           }
           link_->data_received_include(sample, directedWriteReaders);
         }
      } else {
         if (directedWriteReaders.empty()) {
           if (Transport_debug_level > 5) {
             ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpReceiveStrategy[%@]::deliver_sample_i - ")
               ACE_TEXT("calling DataLink::data_received_include for seq: %q to readers_selected_\n"),
               this, sample.header_.sequence_.getValue()));
           }
           link_->data_received_include(sample, readers_selected_);
         } else {
           if (Transport_debug_level > 5) {
             ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpReceiveStrategy[%@]::deliver_sample_i - ")
               ACE_TEXT("calling DataLink::data_received_include for seq: %q to intersection of readers\n"),
               this, sample.header_.sequence_.getValue()));
           }
           set_intersect(directedWriteReaders, readers_selected_, GUID_tKeyLessThan());
           link_->data_received_include(sample, directedWriteReaders);
         }
       }
     }
     break;
   }
   case GAP:
     if (!check_encoded(submessage.gap_sm().writerId)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::deliver_sample_i - decode error\n"));
       }
       break;
     }
     link_->received(submessage.gap_sm(), receiver_.source_guid_prefix_, receiver_.directed_, remote_addr);
     break;
 
   case HEARTBEAT:
     if (!check_encoded(submessage.heartbeat_sm().writerId)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::deliver_sample_i - decode error\n"));
       }
       break;
     }
     link_->received(submessage.heartbeat_sm(), receiver_.source_guid_prefix_, receiver_.directed_, remote_addr);
     if (submessage.heartbeat_sm().smHeader.flags & FLAG_L) {
       // Liveliness has been asserted.  Create a DATAWRITER_LIVELINESS message.
       sample.header_.message_id_ = DATAWRITER_LIVELINESS;
       receiver_.fill_header(sample.header_);
       sample.header_.publication_id_.entityId = submessage.heartbeat_sm().writerId;
       link_->data_received(sample);
     }
     break;
 
   case ACKNACK:
     if (!check_encoded(submessage.acknack_sm().readerId)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::deliver_sample_i - decode error\n"));
       }
       break;
     }
     link_->received(submessage.acknack_sm(), receiver_.source_guid_prefix_, remote_addr);
     break;
 
   case HEARTBEAT_FRAG:
     if (!check_encoded(submessage.hb_frag_sm().writerId)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::deliver_sample_i - decode error\n"));
       }
       break;
     }
     link_->received(submessage.hb_frag_sm(), receiver_.source_guid_prefix_, receiver_.directed_, remote_addr);
     break;
 
   case NACK_FRAG:
     if (!check_encoded(submessage.nack_frag_sm().readerId)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::deliver_sample_i - decode error\n"));
       }
       break;
     }
     link_->received(submessage.nack_frag_sm(), receiver_.source_guid_prefix_, remote_addr);
     break;
 
   /* no case DATA_FRAG: by the time deliver_sample() is called, reassemble()
      has successfully reassembled the fragments and we now have a DATA submsg
    */
 
 #ifdef OPENDDS_SECURITY
   case SEC_PREFIX:
     secure_prefix_ = submessage.security_sm();
     break;
 
   case SEC_POSTFIX:
     deliver_from_secure(submessage, remote_addr);
     break;
 #endif
 
   default:
     break;
   }
 }

◆ do_not_withhold_data_from()

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::do_not_withhold_data_from ( const GUID_t & sub_id )

Definition at line 1084 of file RtpsUdpReceiveStrategy.cpp.

References readers_selected_.

 {
   readers_selected_.insert(sub_id);
 }

◆ end_transport_header_processing()

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::end_transport_header_processing ( )

virtual

End Current Transport Header Processing.

Reimplemented from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 1071 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::disable_response_queue(), and link_.

 {
   link_->disable_response_queue(false);
 }

◆ getDirectedWriteReaders()

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::getDirectedWriteReaders	(	RepoIdSet &	directedWriteReaders,
		const RTPS::DataSubmessage &	ds
	)		const

private

Definition at line 1089 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::RTPS::DataSubmessage::inlineQos, OpenDDS::RTPS::ProtocolVersion_t::minor, OpenDDS::RTPS::PID_DIRECTED_WRITE, receiver_, and OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::source_version_.

Referenced by deliver_sample_i().

 {
   directedWriteReaders.clear();
   for (CORBA::ULong i = 0; i < ds.inlineQos.length(); ++i) {
     if (ds.inlineQos[i]._d() == RTPS::PID_DIRECTED_WRITE
         && receiver_.source_version_.minor >= 4) {
       directedWriteReaders.insert(ds.inlineQos[i].guid());
     }
   }
   return !directedWriteReaders.empty();
 }

◆ handle_input()

int OpenDDS::DCPS::RtpsUdpReceiveStrategy::handle_input ( ACE_HANDLE fd )

virtual

Reimplemented from ACE_Event_Handler.

Definition at line 77 of file RtpsUdpReceiveStrategy.cpp.

 {
   ThreadStatusManager::Event ev(thread_status_manager_);
 
   // Since BUFFER_COUNT is 1, the index will always be 0
   const size_t INDEX = 0;
 
   ACE_Message_Block* const cur_rb = receive_buffers_[INDEX];
   cur_rb->reset();
 
   iovec iov;
 #ifdef _MSC_VER
 #pragma warning(push)
 // iov_len is 32-bit on 64-bit VC++, but we don't want a cast here
 // since on other platforms iov_len is 64-bit
 #pragma warning(disable : 4267)
 #endif
   iov.iov_len = cur_rb->space();
 #ifdef _MSC_VER
 #pragma warning(pop)
 #endif
   iov.iov_base = cur_rb->wr_ptr();
 
   ACE_INET_Addr remote_address;
   bool stop = false;
   ssize_t bytes_remaining = receive_bytes(&iov,
                                           1,
                                           remote_address,
                                           fd,
                                           stop);
 
   if (stop) {
     return 0;
   }
 
   if (bytes_remaining < 0) {
     relink();
     return -1;
   }
 
   cur_rb->wr_ptr(bytes_remaining);
 
   if (bytes_remaining == 0) {
     if (gracefully_disconnected_) {
       return -1;
     } else {
       relink();
       return -1;
     }
   }
 
   if (!pdu_remaining_) {
     receive_transport_header_.length_ = static_cast<ACE_UINT32>(bytes_remaining);
   }
 
   receive_transport_header_ = *cur_rb;
   if (!receive_transport_header_.valid()) {
     cur_rb->reset();
     if (DCPS_debug_level > 0) {
       ACE_DEBUG((LM_WARNING, ACE_TEXT("(%P|%t) WARNING: RtpsUdpReceiveStrategy::handle_input: TransportHeader invalid.\n")));
     }
     return 0;
   }
 
   bytes_remaining = receive_transport_header_.length_;
   if (!check_header(receive_transport_header_)) {
     return 0;
   }
 
   {
     const ScopedHeaderProcessing shp(*this);
     while (bytes_remaining > 0) {
       data_sample_header_.pdu_remaining(bytes_remaining);
       data_sample_header_ = *cur_rb;
       bytes_remaining -= data_sample_header_.get_serialized_size();
       if (!check_header(data_sample_header_)) {
         return 0;
       }
       ReceivedDataSample rds = data_sample_header_.message_length() ? ReceivedDataSample(*cur_rb) : ReceivedDataSample();
       if (data_sample_header_.into_received_data_sample(rds)) {
 
         if (data_sample_header_.more_fragments() || receive_transport_header_.last_fragment()) {
           VDBG((LM_DEBUG,"(%P|%t) DBG:   Attempt reassembly of fragments\n"));
 
           if (reassemble(rds)) {
             VDBG((LM_DEBUG,"(%P|%t) DBG:   Reassembled complete message\n"));
             deliver_sample(rds, remote_address);
           }
           // If reassemble() returned false, it takes ownership of the data
           // just like deliver_sample() does.
 
         } else {
           deliver_sample(rds, remote_address);
         }
       }
       cur_rb->rd_ptr(data_sample_header_.message_length());
       bytes_remaining -= data_sample_header_.message_length();
 
       // For the reassembly algorithm, the 'last_fragment_' header bit only
       // applies to the first DataSampleHeader in the TransportHeader
       receive_transport_header_.last_fragment(false);
     }
   }
 
   // If newly selected buffer index still has a reference count, we'll need to allocate a new one for the read
   if (receive_buffers_[INDEX]->data_block()->reference_count() > 1) {
 
     if (log_level >= LogLevel::Info) {
       ACE_DEBUG((LM_INFO, "(%P|%t) INFO: RtpsUdpReceiveStrategy::handle_input: reallocating primary receive buffer based on reference count\n"));
     }
 
     ACE_DES_FREE(
       receive_buffers_[INDEX],
       mb_allocator_.free,
       ACE_Message_Block);
 
     ACE_NEW_MALLOC_RETURN(
       receive_buffers_[INDEX],
       (ACE_Message_Block*) mb_allocator_.malloc(sizeof(ACE_Message_Block)),
       ACE_Message_Block(
         RECEIVE_DATA_BUFFER_SIZE,           // Buffer size
         ACE_Message_Block::MB_DATA,         // Default
         0,                                  // Start with no continuation
         0,                                  // Let the constructor allocate
         &data_allocator_,                   // Our buffer cache
         &receive_lock_,                     // Our locking strategy
         ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, // Default
         ACE_Time_Value::zero,               // Default
         ACE_Time_Value::max_time,           // Default
         &db_allocator_,                     // Our data block cache
         &mb_allocator_                      // Our message block cache
       ),
       -1);
   }
 
   return 0;
 }

◆ has_fragments()

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::has_fragments	(	const SequenceRange &	range,
		const GUID_t &	pub_id,
		FragmentInfo *	frag_info = `0`
	)

Definition at line 1189 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::RTPS::FragmentNumberSet::bitmap, OpenDDS::RTPS::FragmentNumberSet::bitmapBase, OpenDDS::DCPS::TransportReassembly::get_gaps(), OpenDDS::DCPS::TransportReassembly::has_frags(), OpenDDS::RTPS::FragmentNumberSet::numBits, OPENDDS_VECTOR(), reassembly_, and OpenDDS::RTPS::FragmentNumber_t::value.

 {
   for (SequenceNumber sn = range.first; sn <= range.second; ++sn) {
     ACE_UINT32 total_frags = 0;
     if (reassembly_.has_frags(sn, pub_id, total_frags)) {
       if (frag_info) {
         if (total_frags > 256) {
           static const CORBA::Long empty_buffer[8] = { 0, 0, 0, 0, 0, 0, 0, 0 };
           OPENDDS_VECTOR(CORBA::Long) buffer(total_frags + 31 / 32, 0);
           ACE_UINT32 numBits = 0;
           size_t idx = 0;
           const ACE_UINT32 base = reassembly_.get_gaps(sn, pub_id, &buffer[0], static_cast<CORBA::ULong>(buffer.size()), numBits);
           const CORBA::ULong end = base + numBits;
           for (CORBA::ULong i = base; i <= end; i += 256) {
             const CORBA::ULong remain = end - i;
             const CORBA::ULong len = std::min(remain, static_cast<CORBA::ULong>(256));
             const CORBA::ULong len32 = (len + 31) / 32;
             const CORBA::ULong len8 = len32 * 4;
             if (std::memcmp(&buffer[idx], &empty_buffer[0], len8) != 0) {
               std::pair<SequenceNumber, RTPS::FragmentNumberSet> p;
               p.first = sn;
               p.second = RTPS::FragmentNumberSet();
               frag_info->push_back(p);
               RTPS::FragmentNumberSet& missing_frags = frag_info->back().second;
               missing_frags.numBits = len;
               missing_frags.bitmapBase.value = i;
               missing_frags.bitmap.length(len32);
               std::memcpy(missing_frags.bitmap.get_buffer(), &buffer[idx], len8);
             }
             idx += 8;
           }
         } else {
           std::pair<SequenceNumber, RTPS::FragmentNumberSet> p;
           p.first = sn;
           p.second = RTPS::FragmentNumberSet();
           frag_info->push_back(p);
           RTPS::FragmentNumberSet& missing_frags = frag_info->back().second;
           missing_frags.numBits = 0; // make sure this is a valid number before passing to get_gaps
           missing_frags.bitmap.length(8); // start at max length
           missing_frags.bitmapBase.value =
             reassembly_.get_gaps(sn, pub_id, missing_frags.bitmap.get_buffer(),
                                  8, missing_frags.numBits);
           // reduce length in case get_gaps() didn't need all that room
           missing_frags.bitmap.length((missing_frags.numBits + 31) / 32);
         }
       } else {
         return true;
       }
     }
   }
   return frag_info ? !frag_info->empty() : false;
 }

◆ OPENDDS_VECTOR() [1/2]

typedef OpenDDS::DCPS::RtpsUdpReceiveStrategy::OPENDDS_VECTOR ( SeqFragPair )

Referenced by has_fragments().

◆ OPENDDS_VECTOR() [2/2]

OpenDDS::DCPS::RtpsUdpReceiveStrategy::OPENDDS_VECTOR ( RTPS::Submessage )

private

◆ reassemble()

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::reassemble ( ReceivedDataSample & data )

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 1101 of file RtpsUdpReceiveStrategy.cpp.

References reassemble_i(), and OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::received_sample_header().

Referenced by handle_input().

 {
   RtpsSampleHeader& rsh = received_sample_header();
   return reassemble_i(data, rsh);
 }

◆ reassemble_i()

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::reassemble_i	(	ReceivedDataSample &	data,
		RtpsSampleHeader &	rsh
	)

privatevirtual

Definition at line 1107 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::DataSampleHeader::byte_order_, OpenDDS::RTPS::DATA, OpenDDS::RTPS::Submessage::data_frag_sm, OpenDDS::RTPS::DATA_OCTETS_TO_IQOS, OpenDDS::RTPS::Submessage::data_sm, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::fill_header(), OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::FLAG_K_IN_DATA, OpenDDS::DCPS::ReceivedDataSample::fragment_size_, fragment_size_, frags_, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::DataLink::is_target(), OpenDDS::DCPS::DataSampleHeader::key_fields_only_, link_, OpenDDS::DCPS::ReceivedDataSample::peek(), OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::TransportReassembly::reassemble(), reassembly_, receiver_, OpenDDS::DCPS::RtpsSampleHeader::submessage_, and total_frags_.

Referenced by deliver_from_secure(), and reassemble().

 {
   using namespace RTPS;
   receiver_.fill_header(data.header_); // set publication_id_.guidPrefix
   data.fragment_size_ = fragment_size_;
   if (link_->is_target(data.header_.publication_id_) && reassembly_.reassemble(frags_, data, total_frags_)) {
 
     // Reassembly was successful, replace DataFrag with Data.  This doesn't have
     // to be a fully-formed DataSubmessage, just enough for this class to use
     // in deliver_sample() which ends up calling RtpsUdpDataLink::received().
     // In particular we will need the SequenceNumber, but ignore the iQoS.
 
     // Peek at the byte order from the encapsulation containing the payload.
     data.header_.byte_order_ = data.peek(1) & FLAG_E;
 
     const DataFragSubmessage& dfsm = rsh.submessage_.data_frag_sm();
 
     const CORBA::Octet data_flags = (data.header_.byte_order_ ? FLAG_E : 0)
       | (data.header_.key_fields_only_ ? FLAG_K_IN_DATA : FLAG_D);
     const DataSubmessage dsm = {
       {DATA, data_flags, 0}, 0, DATA_OCTETS_TO_IQOS,
       dfsm.readerId, dfsm.writerId, dfsm.writerSN, ParameterList()};
     rsh.submessage_.data_sm(dsm);
     return true;
   }
   return false;
 }

◆ receive_bytes()

ssize_t OpenDDS::DCPS::RtpsUdpReceiveStrategy::receive_bytes	(	iovec	iov[],
		int	n,
		ACE_INET_Addr &	remote_address,
		ACE_HANDLE	fd,
		bool &	stop
	)

privatevirtual

Only our subclass knows how to do this.

Implements OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 343 of file RtpsUdpReceiveStrategy.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), choose_recv_socket(), DDS::Security::SecurityException::code, OpenDDS::DCPS::SecurityDebug::encdec_warn, encoded_rtps_, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_PARTICIPANT, OpenDDS::DCPS::equal_guid_prefixes(), OpenDDS::DCPS::RtpsUdpDataLink::get_ice_agent(), OpenDDS::DCPS::RtpsUdpDataLink::get_ice_endpoint(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::GUID_t::guidPrefix, DDS::HANDLE_NIL, OpenDDS::DCPS::RtpsUdpDataLink::handle_registry(), link_, LM_DEBUG, LM_WARNING, OpenDDS::DCPS::RtpsUdpReceiveStrategy::MessageReceiver::local_, OpenDDS::DCPS::RtpsUdpDataLink::local_crypto_handle(), OpenDDS::DCPS::TransportDebug::log_dropped_messages, DDS::Security::SecurityException::message, DDS::Security::SecurityException::minor_code, DDS::Security::OPENDDS_EXCEPTION_CODE_NO_KEY, DDS::Security::OPENDDS_EXCEPTION_MINOR_CODE_NO_KEY, receive_bytes_helper(), receiver_, ACE_SOCK_Dgram::recv(), remote_address_, OpenDDS::RTPS::RTPSHDR_SZ, OpenDDS::DCPS::RtpsUdpDataLink::security_config(), OpenDDS::DCPS::security_debug, OpenDDS::RTPS::SMHDR_SZ, socket(), OpenDDS::RTPS::SRTPS_PREFIX, OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::stop(), OpenDDS::DCPS::RtpsUdpDataLink::transport(), and OpenDDS::DCPS::transport_debug.

Referenced by handle_input().

 {
   const ACE_SOCK_Dgram& socket = choose_recv_socket(fd);
 #ifdef ACE_LACKS_SENDMSG
   ACE_UNUSED_ARG(stop);
   char buffer[0x10000];
   ssize_t scatter = socket.recv(buffer, sizeof buffer, remote_address);
   char* iter = buffer;
   for (int i = 0; scatter > 0 && i < n; ++i) {
     const size_t chunk = std::min(static_cast<size_t>(iov[i].iov_len), // int on LynxOS
                                   static_cast<size_t>(scatter));
     std::memcpy(iov[i].iov_base, iter, chunk);
     scatter -= chunk;
     iter += chunk;
   }
   const ssize_t ret = (scatter < 0) ? scatter : (iter - buffer);
 #else
   const ssize_t ret = receive_bytes_helper(iov, n, socket, remote_address,
 #ifdef OPENDDS_SECURITY
                                            link_->get_ice_agent(), link_->get_ice_endpoint(),
 #endif
                                            *link_->transport(), stop);
 #endif
   remote_address_ = remote_address;
 
 #ifdef OPENDDS_SECURITY
   if (stop) {
     return ret;
   }
 
   using namespace DDS::Security;
   const ParticipantCryptoHandle receiver = link_->local_crypto_handle();
   if (ret > 0 && receiver != DDS::HANDLE_NIL) {
     encoded_rtps_ = false;
 
     GUID_t peer = GUID_UNKNOWN;
 
     const CryptoTransform_var crypto = link_->security_config()->get_crypto_transform();
     if (!crypto) {
       return recv_err("no crypto plugin", remote_address, peer, stop);
     }
 
     if (ret < RTPS::RTPSHDR_SZ + RTPS::SMHDR_SZ) {
       return recv_err("message too short", remote_address, peer, stop);
     }
 
     const unsigned int encLen = static_cast<unsigned int>(ret);
     DDS::OctetSeq encoded(encLen);
     encoded.length(encLen);
     unsigned char* const encBuf = encoded.get_buffer();
     size_t copied = 0;
     for (int i = 0; i < n && copied < encLen; ++i) {
       const size_t chunk = std::min(static_cast<size_t>(iov[i].iov_len),
                                     static_cast<size_t>(encLen - copied));
       std::memcpy(encBuf + copied, iov[i].iov_base, chunk);
       copied += chunk;
     }
 
     if (copied != encLen) {
       return recv_err("received bytes didn't fit in iovec array", remote_address, peer, stop);
     }
 
     if (encoded[RTPS::RTPSHDR_SZ] != RTPS::SRTPS_PREFIX) {
       return ret;
     }
 
     static const int GuidPrefixOffset = 8; // "RTPS", Version(2), Vendor(2)
     std::memcpy(peer.guidPrefix, encBuf + GuidPrefixOffset, sizeof peer.guidPrefix);
     peer.entityId = RTPS::ENTITYID_PARTICIPANT;
     const ParticipantCryptoHandle sender = equal_guid_prefixes(peer.guidPrefix, receiver_.local_) ?
       link_->local_crypto_handle() :
       link_->handle_registry()->get_remote_participant_crypto_handle(peer);
     if (sender == DDS::HANDLE_NIL) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::receive_bytes - decode error from %C\n", LogGuid(peer).c_str()));
       }
       if (security_debug.encdec_warn) {
         ACE_ERROR((LM_WARNING, ACE_TEXT("(%P|%t) {encdec_warn} RtpsUdpReceiveStrategy::receive_bytes: ")
                    ACE_TEXT("decode_rtps_message no remote participant crypto handle for %C, dropping\n"),
                    LogGuid(peer).c_str()));
       }
       stop = true;
       return ret;
     }
 
     DDS::OctetSeq plain;
     SecurityException ex = {"", 0, 0};
     if (!crypto->decode_rtps_message(plain, encoded, receiver, sender, ex)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpReceiveStrategy::receive_bytes - decode error from %C\n", LogGuid(peer).c_str()));
       }
       if (security_debug.encdec_warn) {
         ACE_ERROR((LM_WARNING, "(%P|%t) {encdec_warn} decode_rtps_message SecurityException [%d.%d]: %C\n",
                    ex.code, ex.minor_code, ex.message.in()));
       }
       if (ex.code == OPENDDS_EXCEPTION_CODE_NO_KEY && ex.minor_code == OPENDDS_EXCEPTION_MINOR_CODE_NO_KEY) {
         if (security_debug.encdec_warn) {
           ACE_ERROR((LM_WARNING, ACE_TEXT("(%P|%t) {encdec_warn} RtpsUdpReceiveStrategy::receive_bytes: ")
             ACE_TEXT("decode_rtps_message remote participant has crypto handle but no key, dropping\n")));
         }
         stop = true;
         return ret;
       }
       return recv_err("decode_rtps_message failed", remote_address, peer, stop);
     }
 
     copied = 0;
     const size_t plainLen = plain.length();
     const unsigned char* const plainBuf = plain.get_buffer();
     for (int i = 0; i < n && copied < plainLen; ++i) {
       const size_t chunk = std::min(static_cast<size_t>(iov[i].iov_len),
                                     plainLen - copied);
       std::memcpy(iov[i].iov_base, plainBuf + copied, chunk);
       copied += chunk;
     }
 
     if (copied != plainLen) {
       return recv_err("plaintext doesn't fit in iovec array", remote_address, peer, stop);
     }
 
     encoded_rtps_ = true;
     return plainLen;
   }
 #endif
 
   return ret;
 }

◆ receive_bytes_helper()

ssize_t OpenDDS::DCPS::RtpsUdpReceiveStrategy::receive_bytes_helper	(	iovec	iov[],
		int	n,
		const ACE_SOCK_Dgram &	socket,
		ACE_INET_Addr &	remote_address,
		DCPS::RcHandle< ICE::Agent >	agent,
		DCPS::WeakRcHandle< ICE::Endpoint >	endpoint,
		RtpsUdpTransport &	tport,
		bool &	stop
	)

static

Definition at line 216 of file RtpsUdpReceiveStrategy.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_NOTSUP_RETURN, OpenDDS::STUN::Message::block, ACE_Message_Block::cont(), OpenDDS::DCPS::SecurityDebug::encdec_warn, OpenDDS::STUN::encoding(), ACE_INET_Addr::get_addr_size(), ACE_Addr::get_size(), ACE_Message_Block::length(), LM_ERROR, LM_WARNING, OpenDDS::DCPS::MCK_RTPS, OpenDDS::DCPS::MCK_STUN, ACE_SOCK_Dgram::recv(), ACE_Message_Block::release(), OpenDDS::DCPS::security_debug, and OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >::stop().

Referenced by OpenDDS::DCPS::RtpsUdpTransport::IceEndpoint::handle_input(), and receive_bytes().

 {
   ACE_INET_Addr local_address;
   const ssize_t ret = socket.recv(iov, n, remote_address, 0
 #if defined(ACE_RECVPKTINFO) || defined(ACE_RECVPKTINFO6)
                                   , &local_address
 #endif
   );
 
   if (ret == -1) {
     return ret;
   }
 
   if (remote_address.get_size() > remote_address.get_addr_size()) {
     ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: RtpsUdpReceiveStrategy::receive_bytes_helper - invalid address size\n"));
     return 0;
   }
 
   if (n > 0 && ret > 0 && iov[0].iov_len >= 4 && std::memcmp(iov[0].iov_base, "RTPS", 4) == 0) {
     RtpsUdpInst_rch cfg = tport.config();
     if (cfg && cfg->count_messages()) {
       const NetworkAddress ra(remote_address);
       const InternalMessageCountKey key(ra, MCK_RTPS, ra == cfg->rtps_relay_address());
       ACE_GUARD_RETURN(ACE_Thread_Mutex, g, tport.transport_statistics_mutex_, -1);
       tport.transport_statistics_.message_count[key].recv(ret);
     }
     return ret;
   }
 
 #ifdef OPENDDS_SECURITY
   // Assume STUN
 # ifndef ACE_RECVPKTINFO
   ACE_ERROR((LM_ERROR, "ERROR: RtpsUdpReceiveStrategy::receive_bytes_helper potential STUN message "
              "received but this version of the ACE library doesn't support the local_address "
              "extension in ACE_SOCK_Dgram::recv\n"));
   ACE_UNUSED_ARG(stop);
   ACE_NOTSUP_RETURN(-1);
 # else
 
   stop = true;
   size_t bytes = ret;
   size_t block_size = std::min(bytes, static_cast<size_t>(iov[0].iov_len));
   ACE_Message_Block* head = new ACE_Message_Block(static_cast<const char*>(iov[0].iov_base), block_size);
   head->length(block_size);
   bytes -= block_size;
 
   ACE_Message_Block* tail = head;
   for (int i = 1; i < n && bytes != 0; ++i) {
     block_size = std::min(bytes, static_cast<size_t>(iov[i].iov_len));
     ACE_Message_Block* mb = new ACE_Message_Block(static_cast<const char*>(iov[i].iov_base), block_size);
     mb->length(block_size);
     tail->cont(mb);
     tail = mb;
     bytes -= block_size;
   }
 
   DCPS::Serializer serializer(head, STUN::encoding);
   STUN::Message message;
   message.block = head;
   if (serializer >> message) {
     RtpsUdpInst_rch cfg = tport.config();
     if (cfg && cfg->count_messages()) {
       const NetworkAddress ra(remote_address);
       const InternalMessageCountKey key(ra, MCK_STUN, ra == cfg->rtps_relay_address());
       ACE_GUARD_RETURN(ACE_Thread_Mutex, g, tport.transport_statistics_mutex_, -1);
       tport.transport_statistics_.message_count[key].recv(ret);
     }
 
     if (tport.relay_srsm().is_response(message)) {
       tport.process_relay_sra(tport.relay_srsm().receive(message));
 #ifdef OPENDDS_SECURITY
     } else if (endpoint) {
       ice_agent->receive(endpoint, local_address, remote_address, message);
 #endif
     }
   }
   head->release();
 # endif
 #else
   ACE_UNUSED_ARG(stop);
 #endif
 
   return ret;
 }

◆ remove_fragments()

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::remove_fragments	(	const SequenceRange &	range,
		const GUID_t &	pub_id
	)

Remove any saved fragments. We do not expect to receive any more fragments with sequence numbers in "range" from publication "pub_id".

Definition at line 1174 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::TransportReassembly::data_unavailable(), and reassembly_.

 {
   for (SequenceNumber sn = range.first; sn <= range.second; ++sn) {
     reassembly_.data_unavailable(sn, pub_id);
   }
 }

◆ remove_frags_from_bitmap()

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::remove_frags_from_bitmap	(	CORBA::Long	bitmap[],
		CORBA::ULong	num_bits,
		const SequenceNumber &	base,
		const GUID_t &	pub_id,
		ACE_CDR::ULong &	samples_requested
	)

For each "1" bit in the bitmap, change it to a "0" if there are fragments from publication "pub_id" for the sequence number represented by that position in the bitmap. Returns true if the bitmap was changed.

Definition at line 1136 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::TransportReassembly::has_frags(), and reassembly_.

 {
   bool modified = false;
   for (CORBA::ULong i = 0, x = 0, bit = 0; i < num_bits; ++i, ++bit) {
     if (bit == 32) bit = 0;
 
     if (bit == 0) {
       x = static_cast<CORBA::ULong>(bitmap[i / 32]);
       if (x == 0) {
         // skip an entire Long if it's all 0's (adds 32 due to ++i)
         i += 31;
         bit = 31;
         //FUTURE: this could be generalized with something like the x86 "bsr"
         //        instruction using compiler intrinsics, VC++ _BitScanReverse()
         //        and GCC __builtin_clz()
         continue;
       }
     }
 
     const CORBA::ULong mask = 1 << (31 - bit);
     if (x & mask) {
       const bool has_frags = reassembly_.has_frags(base + i, pub_id);
       if (has_frags) {
         x &= ~mask;
         bitmap[i / 32] = x;
         modified = true;
         --cumulative_bits_added;
       }
     }
   }
   return modified;
 }

◆ sec_submsg_to_octets()

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::sec_submsg_to_octets	(	DDS::OctetSeq &	encoded,
		const RTPS::Submessage &	postfix
	)

private

Definition at line 875 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::align(), OpenDDS::DCPS::Serializer::align_r(), OpenDDS::RTPS::DATA, OpenDDS::RTPS::DATA_FRAG, OpenDDS::DCPS::ReceivedDataSample::data_length(), OpenDDS::STUN::encoding(), OpenDDS::DCPS::ENDIAN_BIG, OpenDDS::DCPS::Encoding::KIND_XCDR1, ACE_Message_Block::length(), ACE_Message_Block::rd_ptr(), secure_prefix_, secure_sample_, OpenDDS::DCPS::serialized_size(), OpenDDS::RTPS::SMHDR_SZ, and OpenDDS::DCPS::ReceivedDataSample::write_data().

Referenced by deliver_from_secure().

 {
   const Encoding encoding(Encoding::KIND_XCDR1, ENDIAN_BIG);
   size_t size = serialized_size(encoding, secure_prefix_);
 
   for (size_t i = 0; i < secure_submessages_.size(); ++i) {
     serialized_size(encoding, size, secure_submessages_[i]);
     const RTPS::SubmessageKind kind = secure_submessages_[i]._d();
     if (kind == RTPS::DATA || kind == RTPS::DATA_FRAG) {
       size += secure_sample_.data_length();
     }
     align(size, RTPS::SMHDR_SZ);
   }
   serialized_size(encoding, size, postfix);
 
   ACE_Message_Block mb(size);
   Serializer ser(&mb, encoding);
   if (!(ser << secure_prefix_)) {
     return false;
   }
 
   if (!ser.align_r(RTPS::SMHDR_SZ)) {
     return false;
   }
 
   for (size_t i = 0; i < secure_submessages_.size(); ++i) {
     if (!(ser << secure_submessages_[i])) {
       return false;
     }
     const RTPS::SubmessageKind kind = secure_submessages_[i]._d();
     if (kind == RTPS::DATA || kind == RTPS::DATA_FRAG) {
       if (!secure_sample_.write_data(ser)) {
         return false;
       }
     }
     if (!ser.align_r(RTPS::SMHDR_SZ)) {
       return false;
     }
   }
   if (!(ser << postfix)) {
     return false;
   }
 
   encoded.length(static_cast<unsigned int>(mb.length()));
   std::memcpy(encoded.get_buffer(), mb.rd_ptr(), mb.length());
   secure_submessages_.resize(0);
 
   return true;
 }

◆ start_i()

int OpenDDS::DCPS::RtpsUdpReceiveStrategy::start_i ( )

privatevirtual

Let the subclass start.

Implements OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 995 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::ReactorInterceptor::execute_or_enqueue(), ACE_IPC_SAP::get_handle(), ACE_Event_Handler::get_handle(), OpenDDS::DCPS::RtpsUdpDataLink::get_reactor_interceptor(), link_, ACE_Event_Handler::READ_MASK, and OpenDDS::DCPS::RtpsUdpDataLink::unicast_socket().

 {
   ReactorInterceptor_rch ri = link_->get_reactor_interceptor();
   ri->execute_or_enqueue(make_rch<RegisterHandler>(link_->unicast_socket().get_handle(), this, static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #ifdef ACE_HAS_IPV6
   ri->execute_or_enqueue(make_rch<RegisterHandler>(link_->ipv6_unicast_socket().get_handle(), this, static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #endif
 
   return 0;
 }

◆ stop_i()

void OpenDDS::DCPS::RtpsUdpReceiveStrategy::stop_i ( )

privatevirtual

Let the subclass stop.

Implements OpenDDS::DCPS::TransportReceiveStrategy< RtpsTransportHeader, RtpsSampleHeader >.

Definition at line 1007 of file RtpsUdpReceiveStrategy.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::config(), OpenDDS::DCPS::ReactorInterceptor::execute_or_enqueue(), ACE_IPC_SAP::get_handle(), ACE_Event_Handler::get_handle(), OpenDDS::DCPS::RtpsUdpDataLink::get_reactor_interceptor(), link_, OpenDDS::DCPS::RtpsUdpDataLink::multicast_socket(), ACE_Event_Handler::READ_MASK, and OpenDDS::DCPS::RtpsUdpDataLink::unicast_socket().

 {
   ReactorInterceptor_rch ri = link_->get_reactor_interceptor();
   ri->execute_or_enqueue(make_rch<RemoveHandler>(link_->unicast_socket().get_handle(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #ifdef ACE_HAS_IPV6
   ri->execute_or_enqueue(make_rch<RemoveHandler>(link_->ipv6_unicast_socket().get_handle(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #endif
 
   RtpsUdpInst_rch cfg = link_->config();
   if (cfg && cfg->use_multicast_) {
     ri->execute_or_enqueue(make_rch<RemoveHandler>(link_->multicast_socket().get_handle(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #ifdef ACE_HAS_IPV6
     ri->execute_or_enqueue(make_rch<RemoveHandler>(link_->ipv6_multicast_socket().get_handle(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #endif
   }
 }

◆ withhold_data_from()

const ReceivedDataSample * OpenDDS::DCPS::RtpsUdpReceiveStrategy::withhold_data_from ( const GUID_t & sub_id )

Prevent delivery of the currently in-progress data sample to the subscription sub_id. Returns pointer to the in-progress data so it can be stored for later delivery.

Definition at line 1077 of file RtpsUdpReceiveStrategy.cpp.

References readers_withheld_, and recvd_sample_.

 {
   readers_withheld_.insert(sub_id);
   return recvd_sample_;
 }

Member Data Documentation

◆ BUFFER_COUNT

const size_t OpenDDS::DCPS::RtpsUdpReceiveStrategy::BUFFER_COUNT = 1u

static

Definition at line 45 of file RtpsUdpReceiveStrategy.h.

◆ encoded_rtps_

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::encoded_rtps_

private

Definition at line 181 of file RtpsUdpReceiveStrategy.h.

Referenced by check_encoded(), and receive_bytes().

◆ encoded_submsg_

bool OpenDDS::DCPS::RtpsUdpReceiveStrategy::encoded_submsg_

private

Definition at line 181 of file RtpsUdpReceiveStrategy.h.

Referenced by check_encoded(), deliver_from_secure(), and deliver_sample().

◆ fragment_size_

ACE_UINT16 OpenDDS::DCPS::RtpsUdpReceiveStrategy::fragment_size_

private

Definition at line 140 of file RtpsUdpReceiveStrategy.h.

Referenced by check_header(), and reassemble_i().

◆ frags_

FragmentRange OpenDDS::DCPS::RtpsUdpReceiveStrategy::frags_

private

Definition at line 141 of file RtpsUdpReceiveStrategy.h.

Referenced by check_header(), and reassemble_i().

◆ last_received_

SequenceNumber OpenDDS::DCPS::RtpsUdpReceiveStrategy::last_received_

private

Definition at line 135 of file RtpsUdpReceiveStrategy.h.

◆ link_

RtpsUdpDataLink* OpenDDS::DCPS::RtpsUdpReceiveStrategy::link_

private

Definition at line 134 of file RtpsUdpReceiveStrategy.h.

Referenced by begin_transport_header_processing(), check_encoded(), choose_recv_socket(), decode_payload(), deliver_from_secure(), deliver_sample(), deliver_sample_i(), end_transport_header_processing(), reassemble_i(), receive_bytes(), start_i(), and stop_i().

◆ message_

RTPS::Message OpenDDS::DCPS::RtpsUdpReceiveStrategy::message_

private

Definition at line 175 of file RtpsUdpReceiveStrategy.h.

Referenced by check_header(), deliver_from_secure(), and deliver_sample().

◆ readers_selected_

RepoIdSet OpenDDS::DCPS::RtpsUdpReceiveStrategy::readers_selected_

private

Definition at line 138 of file RtpsUdpReceiveStrategy.h.

Referenced by deliver_sample_i(), and do_not_withhold_data_from().

◆ readers_withheld_

RepoIdSet OpenDDS::DCPS::RtpsUdpReceiveStrategy::readers_withheld_

private

Definition at line 138 of file RtpsUdpReceiveStrategy.h.

Referenced by deliver_sample_i(), and withhold_data_from().

◆ reassembly_

TransportReassembly OpenDDS::DCPS::RtpsUdpReceiveStrategy::reassembly_

private

Definition at line 143 of file RtpsUdpReceiveStrategy.h.

Referenced by clear_completed_fragments(), has_fragments(), reassemble_i(), remove_fragments(), and remove_frags_from_bitmap().

◆ receiver_

MessageReceiver OpenDDS::DCPS::RtpsUdpReceiveStrategy::receiver_

private

Definition at line 172 of file RtpsUdpReceiveStrategy.h.

Referenced by check_encoded(), check_header(), decode_payload(), deliver_from_secure(), deliver_sample(), deliver_sample_i(), getDirectedWriteReaders(), reassemble_i(), and receive_bytes().

◆ recvd_sample_

const ReceivedDataSample* OpenDDS::DCPS::RtpsUdpReceiveStrategy::recvd_sample_

private

Definition at line 137 of file RtpsUdpReceiveStrategy.h.

Referenced by deliver_sample_i(), and withhold_data_from().

◆ remote_address_

ACE_INET_Addr OpenDDS::DCPS::RtpsUdpReceiveStrategy::remote_address_

private

Definition at line 174 of file RtpsUdpReceiveStrategy.h.

Referenced by check_header(), and receive_bytes().

◆ secure_prefix_

RTPS::SecuritySubmessage OpenDDS::DCPS::RtpsUdpReceiveStrategy::secure_prefix_

private

Definition at line 178 of file RtpsUdpReceiveStrategy.h.

Referenced by check_header(), deliver_from_secure(), deliver_sample(), deliver_sample_i(), RtpsUdpReceiveStrategy(), and sec_submsg_to_octets().

◆ secure_sample_

ReceivedDataSample OpenDDS::DCPS::RtpsUdpReceiveStrategy::secure_sample_

private

Definition at line 180 of file RtpsUdpReceiveStrategy.h.

Referenced by deliver_from_secure(), deliver_sample(), and sec_submsg_to_octets().

◆ thread_status_manager_

ThreadStatusManager& OpenDDS::DCPS::RtpsUdpReceiveStrategy::thread_status_manager_

private

Definition at line 173 of file RtpsUdpReceiveStrategy.h.

Referenced by handle_input().

◆ total_frags_

ACE_UINT32 OpenDDS::DCPS::RtpsUdpReceiveStrategy::total_frags_

private

Definition at line 142 of file RtpsUdpReceiveStrategy.h.

Referenced by check_header(), and reassemble_i().

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

Classes

Public Types

Public Member Functions

Static Public Member Functions

Static Public Attributes

Private Types

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ BaseReceiveStrategy

◆ SeqFragPair

Constructor & Destructor Documentation

◆ RtpsUdpReceiveStrategy()

Member Function Documentation

◆ begin_transport_header_processing()

◆ check_encoded()

◆ check_header() [1/2]

◆ check_header() [2/2]

◆ choose_recv_socket()

◆ clear_completed_fragments()

◆ decode_payload()

◆ deliver_from_secure()

◆ deliver_sample()

◆ deliver_sample_i()

◆ do_not_withhold_data_from()

◆ end_transport_header_processing()

◆ getDirectedWriteReaders()

◆ handle_input()

◆ has_fragments()

◆ OPENDDS_VECTOR() [1/2]

◆ OPENDDS_VECTOR() [2/2]

◆ reassemble()

◆ reassemble_i()

◆ receive_bytes()

◆ receive_bytes_helper()

◆ remove_fragments()

◆ remove_frags_from_bitmap()

◆ sec_submsg_to_octets()

◆ start_i()

◆ stop_i()

◆ withhold_data_from()

Member Data Documentation

◆ BUFFER_COUNT

◆ encoded_rtps_

◆ encoded_submsg_

◆ fragment_size_

◆ frags_

◆ last_received_

◆ link_

◆ message_

◆ readers_selected_

◆ readers_withheld_

◆ reassembly_

◆ receiver_

◆ recvd_sample_

◆ remote_address_

◆ secure_prefix_

◆ secure_sample_

◆ thread_status_manager_

◆ total_frags_