#include <RtpsUdpDataLink.h>

Inheritance diagram for OpenDDS::DCPS::RtpsUdpDataLink:

[legend]

Collaboration diagram for OpenDDS::DCPS::RtpsUdpDataLink:

[legend]

Classes
struct	Bundle

struct	CountKeeper

struct	CountMapPair

struct	CountMapping

class	DeliverHeldData

struct	InterestingRemote
	Data structure representing an "interesting" remote entity for static discovery. More...

struct	MultiSendBuffer

struct	ReaderInfo

struct	ReaderInfoSetHolder

struct	RemoteInfo

class	ReplayDurableData

class	RtpsReader

class	RtpsWriter

struct	WriterInfo

Public Member Functions
	RtpsUdpDataLink (const RtpsUdpTransport_rch &transport, const GuidPrefix_t &local_prefix, const RtpsUdpInst_rch &config, const ReactorTask_rch &reactor_task, InternalTransportStatistics &transport_statistics, ACE_Thread_Mutex &transport_statistics_mutex)

	~RtpsUdpDataLink ()

bool	add_delayed_notification (TransportQueueElement *element)

RemoveResult	remove_sample (const DataSampleElement *sample)

void	remove_all_msgs (const GUID_t &pub_id)

RtpsUdpInst_rch	config () const

ACE_Reactor *	get_reactor ()

ReactorInterceptor_rch	get_reactor_interceptor () const

bool	reactor_is_shut_down ()

ACE_SOCK_Dgram &	unicast_socket ()

ACE_SOCK_Dgram_Mcast &	multicast_socket ()

bool	open (const ACE_SOCK_Dgram &unicast_socket)

void	received (const RTPS::DataSubmessage &data, const GuidPrefix_t &src_prefix, const NetworkAddress &remote_addr)

void	received (const RTPS::GapSubmessage &gap, const GuidPrefix_t &src_prefix, bool directed, const NetworkAddress &remote_addr)

void	received (const RTPS::HeartBeatSubmessage &heartbeat, const GuidPrefix_t &src_prefix, bool directed, const NetworkAddress &remote_addr)

void	received (const RTPS::HeartBeatFragSubmessage &hb_frag, const GuidPrefix_t &src_prefix, bool directed, const NetworkAddress &remote_addr)

void	received (const RTPS::AckNackSubmessage &acknack, const GuidPrefix_t &src_prefix, const NetworkAddress &remote_addr)

void	received (const RTPS::NackFragSubmessage &nackfrag, const GuidPrefix_t &src_prefix, const NetworkAddress &remote_addr)

const GuidPrefix_t &	local_prefix () const

void	remove_locator_and_bundling_cache (const GUID_t &remote_id)

NetworkAddress	get_last_recv_address (const GUID_t &remote_id)

void	update_locators (const GUID_t &remote_id, AddrSet &unicast_addresses, AddrSet &multicast_addresses, bool requires_inline_qos, bool add_ref)

AddrSet	get_addresses (const GUID_t &local, const GUID_t &remote) const

AddrSet	get_addresses (const GUID_t &local) const
	Given a 'local' id, return the set of address for all remote peers. More...

void	filterBestEffortReaders (const ReceivedDataSample &ds, RepoIdSet &selected, RepoIdSet &withheld)

int	make_reservation (const GUID_t &remote_publication_id, const GUID_t &local_subscription_id, const TransportReceiveListener_wrch &receive_listener, bool reliable)

bool	associated (const GUID_t &local, const GUID_t &remote, bool local_reliable, bool remote_reliable, bool local_durable, bool remote_durable, const MonotonicTime_t &participant_discovered_at, ACE_CDR::ULong participant_flags, SequenceNumber max_sn, const TransportClient_rch &client, AddrSet &unicast_addresses, AddrSet &multicast_addresses, const NetworkAddress &last_addr_hint, bool requires_inline_qos)

void	disassociated (const GUID_t &local, const GUID_t &remote)

void	register_for_reader (const GUID_t &writerid, const GUID_t &readerid, const AddrSet &addresses, DiscoveryListener *listener)

void	unregister_for_reader (const GUID_t &writerid, const GUID_t &readerid)

void	register_for_writer (const GUID_t &readerid, const GUID_t &writerid, const AddrSet &addresses, DiscoveryListener *listener)

void	unregister_for_writer (const GUID_t &readerid, const GUID_t &writerid)

void	client_stop (const GUID_t &localId)

virtual void	pre_stop_i ()

DCPS::RcHandle< ICE::Agent >	get_ice_agent () const

virtual DCPS::WeakRcHandle< ICE::Endpoint >	get_ice_endpoint () const

virtual bool	is_leading (const GUID_t &writer_id, const GUID_t &reader_id) const

Security::SecurityConfig_rch	security_config () const

Security::HandleRegistry_rch	handle_registry () const

DDS::Security::ParticipantCryptoHandle	local_crypto_handle () const

void	local_crypto_handle (DDS::Security::ParticipantCryptoHandle pch)

RtpsUdpTransport_rch	transport ()

void	enable_response_queue ()

void	disable_response_queue (bool send_immediately)

bool	requires_inline_qos (const GUIDSeq_var &peers)

EventDispatcher_rch	event_dispatcher ()

RcHandle< JobQueue >	get_job_queue () const

typedef	OPENDDS_VECTOR (Bundle) BundleVec

Public Member Functions inherited from OpenDDS::DCPS::DataLink
	DataLink (const TransportImpl_rch &impl, Priority priority, bool is_loopback, bool is_active)
	Only called by our TransportImpl object. More...

virtual	~DataLink ()

int	handle_exception (ACE_HANDLE)
	Reactor invokes this after being notified in schedule_stop or cancel_release. More...

void	schedule_stop (const MonotonicTimePoint &schedule_to_stop_at)

void	stop ()
	The stop method is used to stop the DataLink prior to shutdown. More...

void	resume_send ()

virtual int	make_reservation (const GUID_t &remote_subscription_id, const GUID_t &local_publication_id, const TransportSendListener_wrch &send_listener, bool reliable)

void	release_reservations (GUID_t remote_id, GUID_t local_id, DataLinkSetMap &released_locals)

void	schedule_delayed_release ()

const TimeDuration &	datalink_release_delay () const

void	remove_listener (const GUID_t &local_id)

void	send_start ()

void	send (TransportQueueElement *element)

void	send_stop (GUID_t repoId)

int	data_received (ReceivedDataSample &sample, const GUID_t &readerId=GUID_UNKNOWN)

void	data_received_include (ReceivedDataSample &sample, const RepoIdSet &incl)

DataLinkIdType	id () const
	Obtain a unique identifier for this DataLink object. More...

void	transport_shutdown ()

void	notify (ConnectionNotice notice)

void	release_resources ()

void	terminate_send ()

void	terminate_send_if_suspended ()

bool	is_target (const GUID_t &remote_id)

void	clear_associations ()

int	handle_timeout (const ACE_Time_Value &tv, const void *arg)

int	handle_close (ACE_HANDLE h, ACE_Reactor_Mask m)

void	set_dscp_codepoint (int cp, ACE_SOCK &socket)

Priority &	transport_priority ()

Priority	transport_priority () const

bool &	is_loopback ()

bool	is_loopback () const

bool &	is_active ()

bool	is_active () const

bool	cancel_release ()

ACE_Message_Block *	create_control (char submessage_id, DataSampleHeader &header, Message_Block_Ptr data)

SendControlStatus	send_control (const DataSampleHeader &header, Message_Block_Ptr data)

GUIDSeq *	target_intersection (const GUID_t &pub_id, const GUIDSeq &in, size_t &n_subs)

TransportImpl_rch	impl () const

void	default_listener (const TransportReceiveListener_wrch &trl)

TransportReceiveListener_wrch	default_listener () const

bool	add_on_start_callback (const TransportClient_wrch &client, const GUID_t &remote)

void	remove_on_start_callback (const TransportClient_wrch &client, const GUID_t &remote)

void	invoke_on_start_callbacks (bool success)

bool	invoke_on_start_callbacks (const GUID_t &local, const GUID_t &remote, bool success)

void	remove_startup_callbacks (const GUID_t &local, const GUID_t &remote)

void	set_scheduling_release (bool scheduling_release)

virtual void	send_final_acks (const GUID_t &readerid)

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_input (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_output (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_exit (ACE_Process *)

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)

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::InternalDataReaderListener< NetworkInterfaceAddress >
	InternalDataReaderListener ()

	InternalDataReaderListener (JobQueue_rch job_queue)

void	job_queue (JobQueue_rch job_queue)

virtual void	on_data_available (InternalDataReader_rch reader)=0

void	schedule (InternalDataReader_rch reader)

Static Public Member Functions
static bool	separate_message (EntityId_t entity)

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)

Private Types
typedef CORBA::ULong	FragmentNumberValue

typedef RcHandle< ReaderInfo >	ReaderInfo_rch

typedef RcHandle< ReaderInfoSetHolder >	ReaderInfoSetHolder_rch

typedef RcHandle< RtpsWriter >	RtpsWriter_rch

typedef RcHandle< WriterInfo >	WriterInfo_rch

typedef RcHandle< RtpsReader >	RtpsReader_rch

typedef std::pair< GUID_t, InterestingRemote >	CallbackType

Private Member Functions
void	on_data_available (RcHandle< InternalDataReader< NetworkInterfaceAddress > > reader)

AddrSet	get_addresses_i (const GUID_t &local, const GUID_t &remote) const

AddrSet	get_addresses_i (const GUID_t &local) const

virtual void	stop_i ()

virtual TransportQueueElement *	customize_queue_element (TransportQueueElement *element)

virtual void	release_reservations_i (const GUID_t &remote_id, const GUID_t &local_id)

RtpsUdpSendStrategy_rch	send_strategy ()

RtpsUdpReceiveStrategy_rch	receive_strategy ()

typedef	OPENDDS_MAP_CMP (GUID_t, RemoteInfo, GUID_tKeyLessThan) RemoteInfoMap

void	update_last_recv_addr (const GUID_t &src, const NetworkAddress &addr, const MonotonicTimePoint &now=MonotonicTimePoint::now())

ACE_Message_Block *	alloc_msgblock (size_t size, ACE_Allocator *data_allocator)

ACE_Message_Block *	submsgs_to_msgblock (const RTPS::SubmessageSeq &subm)

RcHandle< SingleSendBuffer >	get_writer_send_buffer (const GUID_t &pub_id)

typedef	OPENDDS_MAP (FragmentNumberValue, RTPS::FragmentNumberSet) RequestedFragMap

typedef	OPENDDS_MAP (SequenceNumber, RequestedFragMap) RequestedFragSeqMap

typedef	OPENDDS_MAP_CMP (GUID_t, ReaderInfo_rch, GUID_tKeyLessThan) ReaderInfoMap

typedef	OPENDDS_SET (ReaderInfo_rch) ReaderInfoSet

typedef	OPENDDS_MAP (SequenceNumber, ReaderInfoSetHolder_rch) SNRIS

typedef	OPENDDS_MAP_CMP (GUID_t, RtpsWriter_rch, GUID_tKeyLessThan) RtpsWriterMap

typedef	OPENDDS_MAP_CMP (GUID_t, WriterInfo_rch, GUID_tKeyLessThan) WriterInfoMap

typedef	OPENDDS_SET (WriterInfo_rch) WriterInfoSet

typedef	OPENDDS_VECTOR (MetaSubmessageVec::iterator) MetaSubmessageIterVec

typedef	OPENDDS_MAP_CMP (GUID_t, MetaSubmessageIterVec, GUID_tKeyLessThan) DestMetaSubmessageMap

typedef	OPENDDS_MAP (AddressCacheEntryProxy, DestMetaSubmessageMap) AddrDestMetaSubmessageMap

typedef	OPENDDS_VECTOR (MetaSubmessageIterVec) MetaSubmessageIterVecVec

typedef	OPENDDS_SET (CORBA::Long) CountSet

typedef	OPENDDS_MAP_CMP (EntityId_t, CountSet, EntityId_tKeyLessThan) IdCountSet

typedef	OPENDDS_MAP (CORBA::Long, CountMapPair) CountMap

typedef	OPENDDS_MAP_CMP (EntityId_t, CountMapping, EntityId_tKeyLessThan) IdCountMapping

void	build_meta_submessage_map (MetaSubmessageVec &meta_submessages, AddrDestMetaSubmessageMap &addr_map)

void	bundle_mapped_meta_submessages (const Encoding &encoding, AddrDestMetaSubmessageMap &addr_map, BundleVec &bundles, CountKeeper &counts)

void	queue_submessages (MetaSubmessageVec &meta_submessages)

void	update_required_acknack_count (const GUID_t &local_id, const GUID_t &remote_id, CORBA::Long current)

void	bundle_and_send_submessages (MetaSubmessageVec &meta_submessages)

	OPENDDS_VECTOR (MetaSubmessageVec) fsq_vec_

void	harvest_send_queue (const MonotonicTimePoint &now)

void	flush_send_queue (const MonotonicTimePoint &now)

void	flush_send_queue_i ()

typedef	OPENDDS_MAP_CMP (GUID_t, RtpsReader_rch, GUID_tKeyLessThan) RtpsReaderMap

typedef	OPENDDS_MULTIMAP_CMP (GUID_t, RtpsReader_rch, GUID_tKeyLessThan) RtpsReaderMultiMap

void	durability_resend (TransportQueueElement *element, size_t &cumulative_send_count)

void	durability_resend (TransportQueueElement *element, const RTPS::FragmentNumberSet &fragmentSet, size_t &cumulative_send_count)

template<typename T , typename FN >
void	datawriter_dispatch (const T &submessage, const GuidPrefix_t &src_prefix, const FN &func)

template<typename T , typename FN >
void	datareader_dispatch (const T &submessage, const GuidPrefix_t &src_prefix, bool directed, const FN &func)

void	send_heartbeats (const MonotonicTimePoint &now)

void	check_heartbeats (const MonotonicTimePoint &now)

typedef	OPENDDS_MULTIMAP_CMP (GUID_t, InterestingRemote, GUID_tKeyLessThan) InterestingRemoteMapType

TransportQueueElement *	customize_queue_element_non_reliable_i (TransportQueueElement *element, bool requires_inline_qos, MetaSubmessageVec &meta_submessages, bool &deliver_after_send, ACE_Guard< ACE_Thread_Mutex > &guard)

void	send_heartbeats_manual_i (const TransportSendControlElement *tsce, MetaSubmessageVec &meta_submessages)

typedef	OPENDDS_MAP_CMP (EntityId_t, CORBA::Long, EntityId_tKeyLessThan) CountMapType

void	accumulate_addresses (const GUID_t &local, const GUID_t &remote, AddrSet &addresses, bool prefer_unicast=false) const

Static Private Member Functions
static void	extend_bitmap_range (RTPS::FragmentNumberSet &fnSet, CORBA::ULong extent, ACE_CDR::ULong &cumulative_bits_added)

static bool	include_fragment (const TransportQueueElement &element, const DisjointSequence &fragments, SequenceNumber &lastFragment)

Private Attributes
ReactorTask_rch	reactor_task_

RcHandle< JobQueue >	job_queue_

EventDispatcher_rch	event_dispatcher_

GuidPrefix_t	local_prefix_

RemoteInfoMap	locators_

LocatorCache	locator_cache_

BundlingCache	bundling_cache_

ACE_SOCK_Dgram	unicast_socket_

ACE_SOCK_Dgram_Mcast	multicast_socket_

MessageBlockAllocator	mb_allocator_

DataBlockAllocator	db_allocator_

Dynamic_Cached_Allocator_With_Overflow< ACE_Thread_Mutex >	custom_allocator_

Dynamic_Cached_Allocator_With_Overflow< ACE_Thread_Mutex >	bundle_allocator_

unique_ptr< DataBlockLockPool >	db_lock_pool_

OpenDDS::DCPS::RtpsUdpDataLink::MultiSendBuffer	multi_buff_

RtpsWriterMap	writers_

TransactionalRtpsSendQueue	sq_

ACE_Thread_Mutex	fsq_mutex_

size_t	fsq_vec_size_

RcHandle< SporadicEvent >	harvest_send_queue_sporadic_

RcHandle< SporadicEvent >	flush_send_queue_sporadic_

RepoIdSet	pending_reliable_readers_

RtpsReaderMap	readers_

RtpsReaderMultiMap	readers_of_writer_

WriterToSeqReadersMap	writer_to_seq_best_effort_readers_

ACE_Thread_Mutex	readers_lock_

ACE_Thread_Mutex	writers_lock_

ACE_Thread_Mutex	locators_lock_

CORBA::Long	best_effort_heartbeat_count_

RcHandle< PeriodicEvent >	heartbeat_

RcHandle< PeriodicEvent >	heartbeatchecker_

InterestingRemoteMapType	interesting_readers_

InterestingRemoteMapType	interesting_writers_

CountMapType	heartbeat_counts_

const size_t	max_bundle_size_

InternalTransportStatistics &	transport_statistics_

ACE_Thread_Mutex &	transport_statistics_mutex_

ACE_Thread_Mutex	security_mutex_

Security::SecurityConfig_rch	security_config_

Security::HandleRegistry_rch	handle_registry_

DDS::Security::ParticipantCryptoHandle	local_crypto_handle_

RcHandle< ICE::Agent >	ice_agent_

RcHandle< InternalDataReader< NetworkInterfaceAddress > >	network_interface_address_reader_

MulticastManager	multicast_manager_

Static Private Attributes
static bool	force_inline_qos_ = false
	static member used by testing code to force inline qos More...

Friends
class	::DDS_TEST

Additional Inherited Members
Public Types inherited from OpenDDS::DCPS::DataLink
enum	ConnectionNotice { DISCONNECTED, RECONNECTED, LOST }

typedef WeakRcHandle< TransportClient >	TransportClient_wrch

typedef std::pair< TransportClient_wrch, GUID_t >	OnStartCallback

Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

Public Types inherited from OpenDDS::DCPS::InternalDataReaderListener< NetworkInterfaceAddress >
typedef RcHandle< InternalDataReader< NetworkInterfaceAddress > >	InternalDataReader_rch

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 OpenDDS::DCPS::DataLink
typedef ACE_Guard< LockType >	GuardType

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::DataLink
int	start (const TransportSendStrategy_rch &send_strategy, const TransportStrategy_rch &receive_strategy, bool invoke_all=true)

void	send_start_i ()

virtual void	send_i (TransportQueueElement *element, bool relink=true)

void	send_stop_i (GUID_t repoId)

GUIDSeq *	peer_ids (const GUID_t &local_id) const

void	network_change () const

void	replay_durable_data (const GUID_t &local_pub_id, const GUID_t &remote_sub_id) const

TransportSendStrategy_rch	get_send_strategy ()

typedef	OPENDDS_MAP_CMP (GUID_t, TransportClient_wrch, GUID_tKeyLessThan) RepoToClientMap

typedef	OPENDDS_MAP_CMP (GUID_t, RepoToClientMap, GUID_tKeyLessThan) OnStartCallbackMap

typedef	OPENDDS_MAP_CMP (GUID_t, RepoIdSet, GUID_tKeyLessThan) PendingOnStartsMap

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

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

Static Protected Member Functions inherited from OpenDDS::DCPS::DataLink
static ACE_UINT64	get_next_datalink_id ()
	Used to provide unique Ids to all DataLink methods. More...

Protected Attributes inherited from OpenDDS::DCPS::DataLink
TransportStrategy_rch	receive_strategy_
	The transport receive strategy object for this DataLink. More...

TransportSendStrategy_rch	send_strategy_
	The transport send strategy object for this DataLink. More...

LockType	strategy_lock_

OnStartCallbackMap	on_start_callbacks_

PendingOnStartsMap	pending_on_starts_

TimeDuration	datalink_release_delay_

unique_ptr< MessageBlockAllocator >	mb_allocator_

unique_ptr< DataBlockAllocator >	db_allocator_

bool	is_loopback_
	Is remote attached to same transport ? More...

bool	is_active_
	Is pub or sub ? More...

bool	started_

SendResponseListener	send_response_listener_
	Listener for TransportSendControlElements created in send_control. More...

Interceptor	interceptor_

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Definition at line 90 of file RtpsUdpDataLink.h.

Member Typedef Documentation

◆ CallbackType

typedef std::pair<GUID_t, InterestingRemote> OpenDDS::DCPS::RtpsUdpDataLink::CallbackType

private

Definition at line 893 of file RtpsUdpDataLink.h.

◆ FragmentNumberValue

typedef CORBA::ULong OpenDDS::DCPS::RtpsUdpDataLink::FragmentNumberValue

private

Definition at line 339 of file RtpsUdpDataLink.h.

◆ ReaderInfo_rch

typedef RcHandle<ReaderInfo> OpenDDS::DCPS::RtpsUdpDataLink::ReaderInfo_rch

private

Definition at line 387 of file RtpsUdpDataLink.h.

◆ ReaderInfoSetHolder_rch

typedef RcHandle<ReaderInfoSetHolder> OpenDDS::DCPS::RtpsUdpDataLink::ReaderInfoSetHolder_rch

private

Definition at line 397 of file RtpsUdpDataLink.h.

◆ RtpsReader_rch

typedef RcHandle<RtpsReader> OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader_rch

private

Definition at line 681 of file RtpsUdpDataLink.h.

◆ RtpsWriter_rch

typedef RcHandle<RtpsWriter> OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter_rch

private

Definition at line 576 of file RtpsUdpDataLink.h.

◆ WriterInfo_rch

typedef RcHandle<WriterInfo> OpenDDS::DCPS::RtpsUdpDataLink::WriterInfo_rch

private

Definition at line 613 of file RtpsUdpDataLink.h.

Constructor & Destructor Documentation

◆ RtpsUdpDataLink()

OpenDDS::DCPS::RtpsUdpDataLink::RtpsUdpDataLink	(	const RtpsUdpTransport_rch &	transport,
		const GuidPrefix_t &	local_prefix,
		const RtpsUdpInst_rch &	config,
		const ReactorTask_rch &	reactor_task,
		InternalTransportStatistics &	transport_statistics,
		ACE_Thread_Mutex &	transport_statistics_mutex
	)

Definition at line 67 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::assign(), OpenDDS::DCPS::ENTITYID_PARTICIPANT, handle_registry_, OpenDDS::DCPS::InternalDataReaderListener< NetworkInterfaceAddress >::job_queue(), job_queue_, local_prefix(), local_prefix_, OpenDDS::DCPS::make_id(), OpenDDS::DCPS::DataLink::receive_strategy_, OpenDDS::DCPS::ref(), security_config_, OpenDDS::DCPS::DataLink::send_strategy_, and TheServiceParticipant.

   : DataLink(transport, // 3 data link "attributes", below, are unused
              0,         // priority
              false,     // is_loopback
              false)     // is_active
   , reactor_task_(reactor_task)
   , job_queue_(make_rch<JobQueue>(reactor_task->get_reactor()))
   , event_dispatcher_(transport->event_dispatcher())
   , mb_allocator_(TheServiceParticipant->association_chunk_multiplier())
   , db_allocator_(TheServiceParticipant->association_chunk_multiplier())
   , custom_allocator_(TheServiceParticipant->association_chunk_multiplier() * config->anticipated_fragments_, RtpsSampleHeader::FRAG_SIZE)
   , bundle_allocator_(TheServiceParticipant->association_chunk_multiplier(), config->max_message_size_)
   , db_lock_pool_(new DataBlockLockPool(static_cast<unsigned long>(TheServiceParticipant->n_chunks())))
   , multi_buff_(this, config->nak_depth_)
   , fsq_vec_size_(0)
   , harvest_send_queue_sporadic_(make_rch<SporadicEvent>(event_dispatcher_, make_rch<PmfNowEvent<RtpsUdpDataLink> >(rchandle_from(this), &RtpsUdpDataLink::harvest_send_queue)))
   , flush_send_queue_sporadic_(make_rch<SporadicEvent>(event_dispatcher_, make_rch<PmfNowEvent<RtpsUdpDataLink> >(rchandle_from(this), &RtpsUdpDataLink::flush_send_queue)))
   , best_effort_heartbeat_count_(0)
   , heartbeat_(make_rch<PeriodicEvent>(event_dispatcher_, make_rch<PmfNowEvent<RtpsUdpDataLink> >(rchandle_from(this), &RtpsUdpDataLink::send_heartbeats)))
   , heartbeatchecker_(make_rch<PeriodicEvent>(event_dispatcher_, make_rch<PmfNowEvent<RtpsUdpDataLink> >(rchandle_from(this), &RtpsUdpDataLink::check_heartbeats)))
   , max_bundle_size_(config->max_message_size_ - RTPS::RTPSHDR_SZ) // default maximum bundled message size is max udp message size (see TransportStrategy) minus RTPS header
   , transport_statistics_(transport_statistics)
   , transport_statistics_mutex_(transport_statistics_mutex)
 #ifdef OPENDDS_SECURITY
   , security_config_(Security::SecurityRegistry::instance()->default_config())
   , local_crypto_handle_(DDS::HANDLE_NIL)
   , ice_agent_(ICE::Agent::instance())
 #endif
   , network_interface_address_reader_(make_rch<InternalDataReader<NetworkInterfaceAddress> >(true, rchandle_from(this)))
 {
 #ifdef OPENDDS_SECURITY
   const GUID_t guid = make_id(local_prefix, ENTITYID_PARTICIPANT);
   handle_registry_ = security_config_->get_handle_registry(guid);
 #endif
 
   send_strategy_ = make_rch<RtpsUdpSendStrategy>(this, local_prefix);
   receive_strategy_ = make_rch<RtpsUdpReceiveStrategy>(this, local_prefix, ref(TheServiceParticipant->get_thread_status_manager()));
   assign(local_prefix_, local_prefix);
 
   this->job_queue(job_queue_);
 }

◆ ~RtpsUdpDataLink()

OpenDDS::DCPS::RtpsUdpDataLink::~RtpsUdpDataLink ( )

Definition at line 114 of file RtpsUdpDataLink.cpp.

References flush_send_queue_sporadic_, and harvest_send_queue_sporadic_.

 {
   harvest_send_queue_sporadic_->cancel();
   flush_send_queue_sporadic_->cancel();
 }

Member Function Documentation

◆ accumulate_addresses()

void OpenDDS::DCPS::RtpsUdpDataLink::accumulate_addresses	(	const GUID_t &	local,
		const GUID_t &	remote,
		AddrSet &	addresses,
		bool	prefer_unicast = `false`
	)		const

private

Definition at line 4677 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, config(), get_ice_endpoint(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::GUID_t::guidPrefix, ice_agent_, interesting_readers_, interesting_writers_, OpenDDS::DCPS::GuidConverter::isReader(), OpenDDS::DCPS::GuidConverter::isWriter(), locator_cache_, locators_, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::max_value, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OPENDDS_ASSERT, readers_lock_, and writers_lock_.

Referenced by build_meta_submessage_map(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i(), and get_addresses_i().

 {
   OPENDDS_ASSERT(local != GUID_UNKNOWN);
   OPENDDS_ASSERT(remote != GUID_UNKNOWN);
 
   const LocatorCacheKey key(remote, local, prefer_unicast);
   LocatorCache::ScopedAccess entry(locator_cache_, key);
   if (!entry.is_new_) {
     addresses.insert(entry.value().addrs_.begin(), entry.value().addrs_.end());
     return;
   }
 
   RtpsUdpInst_rch cfg = config();
   if (!cfg) {
     return;
   }
 
   if (cfg->rtps_relay_only() && std::memcmp(&local.guidPrefix, &remote.guidPrefix, sizeof(GuidPrefix_t)) != 0) {
     if (NetworkAddress() != cfg->rtps_relay_address()) {
       addresses.insert(cfg->rtps_relay_address());
       entry.value().addrs_.insert(cfg->rtps_relay_address());
     }
     return;
   }
 
   AddrSet normal_addrs;
   MonotonicTimePoint normal_addrs_expires = MonotonicTimePoint::max_value;
   NetworkAddress ice_addr;
   bool valid_last_recv_addr = false;
   static const NetworkAddress NO_ADDR;
 
   const RemoteInfoMap::const_iterator pos = locators_.find(remote);
   if (pos != locators_.end()) {
     if (prefer_unicast && pos->second.insert_recv_addr(normal_addrs)) {
       normal_addrs_expires = pos->second.last_recv_time_ + cfg->receive_address_duration_;
       valid_last_recv_addr = (MonotonicTimePoint::now() - pos->second.last_recv_time_) <= cfg->receive_address_duration_;
     } else if (prefer_unicast && !pos->second.unicast_addrs_.empty()) {
       normal_addrs = pos->second.unicast_addrs_;
     } else if (!pos->second.multicast_addrs_.empty()) {
 #ifdef ACE_HAS_IPV6
       if (pos->second.last_recv_addr_ != NO_ADDR) {
         const AddrSet& mc_addrs = pos->second.multicast_addrs_;
         for (AddrSet::const_iterator it = mc_addrs.begin(); it != mc_addrs.end(); ++it) {
           if (it->get_type() == pos->second.last_recv_addr_.get_type()) {
             normal_addrs.insert(*it);
           }
         }
       } else {
         normal_addrs = pos->second.multicast_addrs_;
       }
 #else
       normal_addrs = pos->second.multicast_addrs_;
 #endif
     } else if (pos->second.insert_recv_addr(normal_addrs)) {
       normal_addrs_expires = pos->second.last_recv_time_ + cfg->receive_address_duration_;
       valid_last_recv_addr = (MonotonicTimePoint::now() - pos->second.last_recv_time_) <= cfg->receive_address_duration_;
     } else {
       normal_addrs = pos->second.unicast_addrs_;
     }
   } else {
     const GuidConverter conv(remote);
     if (conv.isReader()) {
       ACE_GUARD(ACE_Thread_Mutex, g, writers_lock_);
       InterestingRemoteMapType::const_iterator ipos = interesting_readers_.find(remote);
       if (ipos != interesting_readers_.end()) {
         normal_addrs = ipos->second.addresses;
       }
     } else if (conv.isWriter()) {
       ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
       InterestingRemoteMapType::const_iterator ipos = interesting_writers_.find(remote);
       if (ipos != interesting_writers_.end()) {
         normal_addrs = ipos->second.addresses;
       }
     }
   }
 
 #ifdef OPENDDS_SECURITY
   WeakRcHandle<ICE::Endpoint> endpoint = get_ice_endpoint();
   if (endpoint) {
     ice_addr = ice_agent_->get_address(endpoint, local, remote);
   }
 #endif
 
   if (ice_addr == NO_ADDR) {
     addresses.insert(normal_addrs.begin(), normal_addrs.end());
     entry.value().addrs_.insert(normal_addrs.begin(), normal_addrs.end());
     entry.value().expires_ = normal_addrs_expires;
     const NetworkAddress relay_addr = cfg->rtps_relay_address();
     if (!valid_last_recv_addr && relay_addr != NO_ADDR) {
       addresses.insert(relay_addr);
       entry.value().addrs_.insert(relay_addr);
     }
     return;
   }
 
   if (normal_addrs.count(ice_addr) == 0) {
     addresses.insert(ice_addr);
     entry.value().addrs_.insert(ice_addr);
     return;
   }
 
   addresses.insert(normal_addrs.begin(), normal_addrs.end());
   entry.value().addrs_.insert(normal_addrs.begin(), normal_addrs.end());
   entry.value().expires_ = normal_addrs_expires;
 }

◆ add_delayed_notification()

bool OpenDDS::DCPS::RtpsUdpDataLink::add_delayed_notification ( TransportQueueElement * element )

Definition at line 127 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::TransportQueueElement::publication_id(), writers_, and writers_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpSendStrategy::add_delayed_notification().

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, writers_lock_, false);
   RtpsWriter_rch writer;
   RtpsWriterMap::iterator iter = writers_.find(element->publication_id());
   if (iter != writers_.end()) {
     writer = iter->second;
   }
 
   g.release();
 
   if (writer) {
     writer->add_elem_awaiting_ack(element);
     return true;
   }
   return false;
 }

◆ alloc_msgblock()

ACE_Message_Block * OpenDDS::DCPS::RtpsUdpDataLink::alloc_msgblock	(	size_t	size,
		ACE_Allocator *	data_allocator
	)

private

Definition at line 1035 of file RtpsUdpDataLink.cpp.

References ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, ACE_NEW_MALLOC_RETURN, db_allocator_, db_lock_pool_, DataBlockLockPool::get_lock(), OpenDDS::DCPS::Cached_Allocator_With_Overflow< T, ACE_LOCK >::malloc(), ACE_Time_Value::max_time, mb_allocator_, ACE_Message_Block::MB_DATA, and ACE_Time_Value::zero.

Referenced by bundle_and_send_submessages(), and submsgs_to_msgblock().

                                                                           {
   ACE_Message_Block* result;
   ACE_NEW_MALLOC_RETURN(result,
     static_cast<ACE_Message_Block*>(
       mb_allocator_.malloc(sizeof(ACE_Message_Block))),
     ACE_Message_Block(size,
                       ACE_Message_Block::MB_DATA,
                       0, // cont
                       0, // data
                       data_allocator,
                       db_lock_pool_->get_lock(), // locking_strategy
                       ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                       ACE_Time_Value::zero,
                       ACE_Time_Value::max_time,
                       &db_allocator_,
                       &mb_allocator_),
     0);
   return result;
 }

◆ associated()

bool OpenDDS::DCPS::RtpsUdpDataLink::associated	(	const GUID_t &	local,
		const GUID_t &	remote,
		bool	local_reliable,
		bool	remote_reliable,
		bool	local_durable,
		bool	remote_durable,
		const MonotonicTime_t &	participant_discovered_at,
		ACE_CDR::ULong	participant_flags,
		SequenceNumber	max_sn,
		const TransportClient_rch &	client,
		AddrSet &	unicast_addresses,
		AddrSet &	multicast_addresses,
		const NetworkAddress &	last_addr_hint,
		bool	requires_inline_qos
	)

Definition at line 558 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::DataLink::add_on_start_callback(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::add_writer(), OpenDDS::DCPS::TransportSendBuffer::capacity(), OpenDDS::DCPS::GUID_t::entityId, heartbeat_counts_, OpenDDS::DCPS::TransactionalRtpsSendQueue::ignore(), OpenDDS::DCPS::DataLink::invoke_on_start_callbacks(), OpenDDS::DCPS::GuidConverter::isReader(), OpenDDS::DCPS::GuidConverter::isWriter(), OpenDDS::DCPS::log_progress(), OpenDDS::DCPS::TransportDebug::log_progress, multi_buff_, pending_reliable_readers_, readers_, readers_lock_, readers_of_writer_, receive_strategy(), sq_, OpenDDS::DCPS::DataLink::strategy_lock_, OpenDDS::DCPS::transport_debug, update_last_recv_addr(), update_locators(), writer_to_seq_best_effort_readers_, writers_, and writers_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpTransport::use_datalink().

 {
   sq_.ignore(local_id, remote_id);
 
   update_locators(remote_id, unicast_addresses, multicast_addresses, requires_inline_qos, true);
   if (last_addr_hint) {
     update_last_recv_addr(remote_id, last_addr_hint);
   }
 
   const GuidConverter conv(local_id);
 
   if (!local_reliable) {
     if (conv.isReader()) {
       ACE_GUARD_RETURN(ACE_Thread_Mutex, g, readers_lock_, true);
       WriterToSeqReadersMap::iterator i = writer_to_seq_best_effort_readers_.find(remote_id);
       if (i == writer_to_seq_best_effort_readers_.end()) {
         writer_to_seq_best_effort_readers_.insert(WriterToSeqReadersMap::value_type(remote_id, SeqReaders(local_id)));
       } else if (i->second.readers.find(local_id) == i->second.readers.end()) {
         i->second.readers.insert(local_id);
       }
     }
     return true;
   }
 
   bool associated = true;
 
   if (conv.isWriter()) {
     if (transport_debug.log_progress) {
       log_progress("RTPS writer/reader association", local_id, remote_id, participant_discovered_at);
     }
 
     if (remote_reliable) {
       ACE_GUARD_RETURN(ACE_Thread_Mutex, g, writers_lock_, true);
       // Insert count if not already there.
       RtpsWriterMap::iterator rw = writers_.find(local_id);
       if (rw == writers_.end()) {
         RtpsUdpDataLink_rch link(this, inc_count());
         CORBA::Long hb_start = 0;
         CountMapType::iterator hbc_it = heartbeat_counts_.find(local_id.entityId);
         if (hbc_it != heartbeat_counts_.end()) {
           hb_start = hbc_it->second;
           heartbeat_counts_.erase(hbc_it);
         }
         RtpsWriter_rch writer = make_rch<RtpsWriter>(client, link, local_id, local_durable,
                                                      max_sn, hb_start, multi_buff_.capacity());
         rw = writers_.insert(RtpsWriterMap::value_type(local_id, writer)).first;
       }
       RtpsWriter_rch writer = rw->second;
       g.release();
       const SequenceNumber writer_max_sn = writer->update_max_sn(remote_id, max_sn);
       writer->add_reader(make_rch<ReaderInfo>(remote_id, remote_durable, participant_discovered_at, participant_flags, writer_max_sn + 1));
     }
     invoke_on_start_callbacks(local_id, remote_id, true);
   } else if (conv.isReader()) {
     if (transport_debug.log_progress) {
       log_progress("RTPS reader/writer association", local_id, remote_id, participant_discovered_at);
     }
     {
       GuardType guard(strategy_lock_);
       if (receive_strategy()) {
         receive_strategy()->clear_completed_fragments(remote_id);
       }
     }
     if (remote_reliable) {
       ACE_GUARD_RETURN(ACE_Thread_Mutex, g, readers_lock_, true);
       RtpsReaderMap::iterator rr = readers_.find(local_id);
       if (rr == readers_.end()) {
         pending_reliable_readers_.erase(local_id);
         RtpsUdpDataLink_rch link(this, inc_count());
         RtpsReader_rch reader = make_rch<RtpsReader>(link, local_id);
         rr = readers_.insert(RtpsReaderMap::value_type(local_id, reader)).first;
       }
       RtpsReader_rch reader = rr->second;
       readers_of_writer_.insert(RtpsReaderMultiMap::value_type(remote_id, rr->second));
       g.release();
       add_on_start_callback(client, remote_id);
       reader->add_writer(make_rch<WriterInfo>(remote_id, participant_discovered_at, participant_flags));
       associated = false;
     } else {
       invoke_on_start_callbacks(local_id, remote_id, true);
     }
   }
 
   return associated;
 }

◆ build_meta_submessage_map()

void OpenDDS::DCPS::RtpsUdpDataLink::build_meta_submessage_map	(	MetaSubmessageVec &	meta_submessages,
		AddrDestMetaSubmessageMap &	addr_map
	)

private

Definition at line 2409 of file RtpsUdpDataLink.cpp.

References accumulate_addresses(), ACE_GUARD, bundling_cache_, get_addresses_i(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::GUIDPREFIX_UNKNOWN, DDS::HANDLE_NIL, LM_DEBUG, local_crypto_handle(), locators_lock_, OpenDDS::DCPS::make_unknown_guid(), OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), separate_message(), and VDBG.

Referenced by bundle_and_send_submessages().

 {
   size_t cache_hits = 0;
   size_t cache_misses = 0;
   size_t addrset_min_size = std::numeric_limits<size_t>::max();
   size_t addrset_max_size = 0;
 
   BundlingCache::ScopedAccess global_access(bundling_cache_);
   const MonotonicTimePoint now = MonotonicTimePoint::now();
 
   // Sort meta_submessages by address set and destination
   for (MetaSubmessageVec::iterator it = meta_submessages.begin(), limit = meta_submessages.end(); it != limit; ++it) {
     if (it->ignore_) {
       continue;
     }
 
     const BundlingCacheKey key(it->src_guid_, it->dst_guid_);
     BundlingCache::ScopedAccess entry(bundling_cache_, key, false, now);
     if (entry.is_new_) {
 
       AddrSet& addrs = entry.value().addrs_;
       ACE_GUARD(ACE_Thread_Mutex, g, locators_lock_);
 
       const bool directed = it->dst_guid_ != GUID_UNKNOWN;
       if (directed) {
         accumulate_addresses(it->src_guid_, it->dst_guid_, addrs, true);
       } else {
         addrs = get_addresses_i(it->src_guid_);
       }
 #ifdef OPENDDS_SECURITY
       if (local_crypto_handle() != DDS::HANDLE_NIL && separate_message(it->src_guid_.entityId)) {
         addrs.insert(BUNDLING_PLACEHOLDER); // removed in bundle_mapped_meta_submessages
       }
 #endif
 #if defined ACE_HAS_CPP11
       entry.recalculate_hash();
 #endif
       ++cache_misses;
     } else {
       ++cache_hits;
     }
 
     const BundlingCache::ScopedAccess& const_entry = entry;
     const AddrSet& addrs = const_entry.value().addrs_;
     addrset_min_size = std::min(addrset_min_size, static_cast<size_t>(addrs.size()));
     addrset_max_size = std::max(addrset_max_size, static_cast<size_t>(addrs.size()));
     if (addrs.empty()) {
       continue;
 #ifdef OPENDDS_SECURITY
     } else if (addrs.size() == 1 && *addrs.begin() == BUNDLING_PLACEHOLDER) {
       continue;
 #endif
     }
 
     DestMetaSubmessageMap& dest_map = addr_map[AddressCacheEntryProxy(const_entry.rch_)];
     if (std::memcmp(&(it->dst_guid_.guidPrefix), &GUIDPREFIX_UNKNOWN, sizeof(GuidPrefix_t)) != 0) {
       MetaSubmessageIterVec& vec = dest_map[make_unknown_guid(it->dst_guid_.guidPrefix)];
       vec.reserve(meta_submessages.size());
       vec.push_back(it);
     } else {
       MetaSubmessageIterVec& vec = dest_map[GUID_UNKNOWN];
       vec.reserve(meta_submessages.size());
       vec.push_back(it);
     }
   }
 
   VDBG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::build_meta_submessage_map()"
                   "- Bundling Cache Stats: hits = %B, misses = %B, min = %B, max = %B\n",
                   cache_hits, cache_misses, addrset_min_size, addrset_max_size));
 }

◆ bundle_and_send_submessages()

void OpenDDS::DCPS::RtpsUdpDataLink::bundle_and_send_submessages ( MetaSubmessageVec & meta_submessages )

private

Definition at line 2757 of file RtpsUdpDataLink.cpp.

Referenced by flush_send_queue_i().

 {
   using namespace RTPS;
 
   // Sort meta_submessages based on both locator IPs and INFO_DST GUID destination/s
   AddrDestMetaSubmessageMap addr_map;
   build_meta_submessage_map(meta_submessages, addr_map);
 
   const Encoding encoding(Encoding::KIND_XCDR1);
 
   // Build reasonably-sized submessage bundles based on our destination map
   BundleVec bundles;
   bundles.reserve(meta_submessages.size());
 
   CountKeeper counts;
   bundle_mapped_meta_submessages(encoding, addr_map, bundles, counts);
 
   // Reusable INFO_DST
   InfoDestinationSubmessage idst = {
     {INFO_DST, FLAG_E, INFO_DST_SZ},
     {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
   };
 
   for (IdCountMapping::iterator it = counts.heartbeat_counts_.begin(), limit = counts.heartbeat_counts_.end(); it != limit; ++it) {
     it->second.next_directed_unassigned_ = it->second.map_.begin();
     it->second.next_undirected_unassigned_ = it->second.map_.begin();
     for (CountMap::iterator it2 = it->second.map_.begin(), limit2 = it->second.map_.end(); it2 != limit2; ++it2) {
       if (it2->second.undirected_) {
         ++(it->second.next_directed_unassigned_);
       }
     }
   }
 
   // Allocate buffers, seralize, and send bundles
   GUID_t prev_dst; // used to determine when we need to write a new info_dst
   for (size_t i = 0; i < bundles.size(); ++i) {
     RTPS::Message rtps_message;
     prev_dst = GUID_UNKNOWN;
     Message_Block_Ptr mb_bundle(alloc_msgblock(bundles[i].size_, &bundle_allocator_));
     Serializer ser(mb_bundle.get(), encoding);
     const MetaSubmessageIterVec& bundle_vec = bundles[i].submessages_;
     for (MetaSubmessageIterVec::const_iterator it = bundle_vec.begin(), limit = bundle_vec.end(); it != limit; ++it) {
       MetaSubmessage& res = **it;
       const GUID_t dst = make_unknown_guid(res.dst_guid_);
       if (dst != prev_dst) {
         assign(idst.guidPrefix, dst.guidPrefix);
         ser << idst;
         if (transport_debug.log_messages) {
           append_submessage(rtps_message, idst);
         }
       }
       switch (res.sm_._d()) {
         case HEARTBEAT: {
           CountMapping& mapping = counts.heartbeat_counts_[res.sm_.heartbeat_sm().writerId];
           CountMapPair& map_pair = mapping.map_[res.sm_.heartbeat_sm().count.value];
           if (!map_pair.is_new_assigned_) {
             if (map_pair.undirected_) {
               OPENDDS_ASSERT(mapping.next_undirected_unassigned_ != mapping.map_.end());
               map_pair.new_ = mapping.next_undirected_unassigned_->first;
               ++mapping.next_undirected_unassigned_;
             } else {
               OPENDDS_ASSERT(mapping.next_directed_unassigned_ != mapping.map_.end());
               map_pair.new_ = mapping.next_directed_unassigned_->first;
               ++mapping.next_directed_unassigned_;
               if (res.sm_.heartbeat_sm().smHeader.flags & RTPS::OPENDDS_FLAG_R) {
                 if (res.sm_.heartbeat_sm().count.value != map_pair.new_) {
                   update_required_acknack_count(res.src_guid_, res.dst_guid_, map_pair.new_);
                 }
                 res.sm_.heartbeat_sm().smHeader.flags &= ~RTPS::OPENDDS_FLAG_R;
               }
             }
             map_pair.is_new_assigned_ = true;
           }
           res.sm_.heartbeat_sm().count.value = map_pair.new_;
           const HeartBeatSubmessage& heartbeat = res.sm_.heartbeat_sm();
           if (transport_debug.log_nonfinal_messages && !(heartbeat.smHeader.flags & RTPS::FLAG_F)) {
             const SequenceNumber hb_first = to_opendds_seqnum(heartbeat.firstSN);
             const SequenceNumber hb_last = to_opendds_seqnum(heartbeat.lastSN);
             ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_nonfinal_messages} RtpsUdpDataLink::bundle_and_send_submessages: HEARTBEAT: %C -> %C first %q last %q count %d\n",
               LogGuid(res.src_guid_).c_str(), LogGuid(res.dst_guid_).c_str(), hb_first.getValue(), hb_last.getValue(), heartbeat.count.value));
           }
           break;
         }
         case ACKNACK: {
           const AckNackSubmessage& acknack = res.sm_.acknack_sm();
           if (transport_debug.log_nonfinal_messages && !(acknack.smHeader.flags & RTPS::FLAG_F)) {
             const SequenceNumber ack = to_opendds_seqnum(acknack.readerSNState.bitmapBase);
             ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_nonfinal_messages} RtpsUdpDataLink::bundle_and_send_submessages: ACKNACK: %C -> %C base %q bits %u count %d\n",
               LogGuid(res.src_guid_).c_str(), LogGuid(res.dst_guid_).c_str(), ack.getValue(), acknack.readerSNState.numBits, acknack.count.value));
           }
           break;
         }
         case NACK_FRAG: {
           CountSet& set = counts.nackfrag_counts_[res.sm_.nack_frag_sm().readerId];
           OPENDDS_ASSERT(!set.empty());
           res.sm_.nack_frag_sm().count.value = *set.begin();
           set.erase(set.begin());
           const NackFragSubmessage& nackfrag = res.sm_.nack_frag_sm();
           // All NackFrag messages are technically 'non-final' since they are only used to negatively acknowledge fragments and expect a response
           if (transport_debug.log_nonfinal_messages) {
             const SequenceNumber seq = to_opendds_seqnum(nackfrag.writerSN);
             ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_nonfinal_messages} RtpsUdpDataLink::bundle_and_send_submessages: NACKFRAG: %C -> %C seq %q base %u bits %u\n",
               LogGuid(res.src_guid_).c_str(), LogGuid(res.dst_guid_).c_str(), seq.getValue(), nackfrag.fragmentNumberState.bitmapBase.value, nackfrag.fragmentNumberState.numBits));
           }
           break;
         }
         default: {
           break;
         }
       }
       ser << res.sm_;
       if (transport_debug.log_messages) {
         push_back(rtps_message.submessages, res.sm_);
       }
       prev_dst = dst;
     }
     RtpsUdpSendStrategy_rch ss = send_strategy();
     if (ss) {
       ss->send_rtps_control(rtps_message, *(mb_bundle.get()), bundles[i].proxy_.addrs());
     }
   }
 }

◆ bundle_mapped_meta_submessages()

void OpenDDS::DCPS::RtpsUdpDataLink::bundle_mapped_meta_submessages	(	const Encoding &	encoding,
		AddrDestMetaSubmessageMap &	addr_map,
		BundleVec &	bundles,
		CountKeeper &	counts
	)

private

Definition at line 2561 of file RtpsUdpDataLink.cpp.

References OpenDDS::RTPS::ACKNACK, OpenDDS::RTPS::Submessage::acknack_sm, config(), OpenDDS::RTPS::NackFragSubmessage::count, OpenDDS::RTPS::HeartBeatSubmessage::count, OpenDDS::DCPS::MetaSubmessage::dst_guid_, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::GAP, OpenDDS::RTPS::Submessage::gap_sm, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::RTPS::HEARTBEAT, OpenDDS::DCPS::RtpsUdpDataLink::CountKeeper::heartbeat_counts_, OpenDDS::RTPS::Submessage::heartbeat_sm, OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id(), OpenDDS::RTPS::INFO_DST, OpenDDS::RTPS::INFO_DST_SZ, max_bundle_size_, OpenDDS::RTPS::NACK_FRAG, OpenDDS::RTPS::Submessage::nack_frag_sm, OpenDDS::DCPS::RtpsUdpDataLink::CountKeeper::nackfrag_counts_, OPENDDS_ASSERT, OpenDDS::RTPS::NackFragSubmessage::readerId, OpenDDS::DCPS::MetaSubmessage::sm_, OpenDDS::DCPS::RtpsUdpDataLink::CountMapPair::undirected_, OpenDDS::RTPS::Count_t::value, and OpenDDS::RTPS::HeartBeatSubmessage::writerId.

Referenced by bundle_and_send_submessages().

 {
   using namespace RTPS;
 
   // Reusable INFO_DST
   InfoDestinationSubmessage idst = {
     {INFO_DST, FLAG_E, INFO_DST_SZ},
     {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
   };
 
   RtpsUdpInst_rch cfg = config();
 
   const bool new_bundle_per_dest_guid = cfg && cfg->rtps_relay_only();
 
   BundleHelper helper(encoding, max_bundle_size_, bundles);
   GUID_t prev_dst; // used to determine when we need to write a new info_dst
   for (AddrDestMetaSubmessageMap::iterator addr_it = addr_map.begin(), limit = addr_map.end(); addr_it != limit; ++addr_it) {
 
     // A new address set always starts a new bundle
     bundles.push_back(Bundle(addr_it->first));
 
     prev_dst = GUID_UNKNOWN;
 
     for (DestMetaSubmessageMap::iterator dest_it = addr_it->second.begin(), limit2 = addr_it->second.end(); dest_it != limit2; ++dest_it) {
 
       if (dest_it->second.empty()) {
         continue;
       }
 
       // Check to see if we're sending separate messages per destination guid
       if (new_bundle_per_dest_guid && bundles.back().submessages_.size()) {
         helper.end_bundle();
         bundles.push_back(Bundle(addr_it->first));
         prev_dst = GUID_UNKNOWN;
       }
 
       for (MetaSubmessageIterVec::iterator resp_it = dest_it->second.begin(), limit3 = dest_it->second.end(); resp_it != limit3; ++resp_it) {
 
         // Check before every meta_submessage to see if we need to prefix a INFO_DST
         if (dest_it->first != prev_dst) {
           // If adding an INFO_DST prefix bumped us over the limit, push the
           // size difference into the next bundle, reset prev_dst, and keep going
           if (!helper.add_to_bundle(idst)) {
             bundles.push_back(Bundle(addr_it->first));
           }
         }
 
         // Attempt to add the submessage meta_submessage to the bundle
         bool result = false, unique = false;
         ACE_UNUSED_ARG(unique);
         MetaSubmessage& res = **resp_it;
         switch (res.sm_._d()) {
           case HEARTBEAT: {
             const EntityId_t id = res.sm_.heartbeat_sm().writerId;
             result = helper.add_to_bundle(res.sm_.heartbeat_sm());
             CountMapPair& map_pair = counts.heartbeat_counts_[id].map_[res.sm_.heartbeat_sm().count.value];
             if (res.dst_guid_ == GUID_UNKNOWN) {
               map_pair.undirected_ = true;
             }
             break;
           }
           case ACKNACK: {
             result = helper.add_to_bundle(res.sm_.acknack_sm());
             break;
           }
           case GAP: {
             result = helper.add_to_bundle(res.sm_.gap_sm());
             break;
           }
           case NACK_FRAG: {
             const EntityId_t id = res.sm_.nack_frag_sm().readerId;
             unique = counts.nackfrag_counts_[id].insert(res.sm_.nack_frag_sm().count.value).second;
             OPENDDS_ASSERT(unique);
             result = helper.add_to_bundle(res.sm_.nack_frag_sm());
             break;
           }
           default: {
             break;
           }
         }
         prev_dst = dest_it->first;
 
         // If adding the submessage bumped us over the limit, push the size
         // difference into the next bundle, reset prev_dst, and keep going
         if (!result) {
           bundles.push_back(Bundle(addr_it->first));
           prev_dst = GUID_UNKNOWN;
         }
         bundles.back().submessages_.push_back(*resp_it);
       }
     }
     helper.end_bundle();
   }
 }

◆ check_heartbeats()

void OpenDDS::DCPS::RtpsUdpDataLink::check_heartbeats ( const MonotonicTimePoint & now )

private

Definition at line 4326 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, config(), OpenDDS::DCPS::RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC, OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::DOES_NOT_EXIST, OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::EXISTS, interesting_writers_, OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::listener, OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::localid, OPENDDS_VECTOR(), readers_lock_, and OpenDDS::DCPS::DiscoveryListener::writer_does_not_exist().

 {
   OPENDDS_VECTOR(CallbackType) writerDoesNotExistCallbacks;
 
   RtpsUdpInst_rch cfg = config();
 
   // Have any interesting writers timed out?
   const MonotonicTimePoint tv(now - 10 * (cfg ? cfg->heartbeat_period_ : TimeDuration(RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC)));
   {
     ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
 
     for (InterestingRemoteMapType::iterator pos = interesting_writers_.begin(), limit = interesting_writers_.end();
          pos != limit;
          ++pos) {
       if (pos->second.status == InterestingRemote::EXISTS && pos->second.last_activity < tv) {
         CallbackType callback(pos->first, pos->second);
         writerDoesNotExistCallbacks.push_back(callback);
         pos->second.status = InterestingRemote::DOES_NOT_EXIST;
       }
     }
   }
 
   OPENDDS_VECTOR(CallbackType)::iterator iter;
   for (iter = writerDoesNotExistCallbacks.begin(); iter != writerDoesNotExistCallbacks.end(); ++iter) {
     const GUID_t& rid = iter->first;
     const InterestingRemote& remote = iter->second;
     remote.listener->writer_does_not_exist(rid, remote.localid);
   }
 }

◆ client_stop()

void OpenDDS::DCPS::RtpsUdpDataLink::client_stop ( const GUID_t & localId )

Definition at line 752 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::TransactionalRtpsSendQueue::ignore_local(), OpenDDS::DCPS::GuidConverter::isReader(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::pre_stop_helper(), readers_, readers_lock_, readers_of_writer_, OpenDDS::DCPS::TransportSendStrategy::remove_all_msgs(), OpenDDS::DCPS::DataLink::send_strategy_, sq_, OpenDDS::DCPS::DataLink::strategy_lock_, writer_to_seq_best_effort_readers_, writers_, and writers_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpTransport::client_stop().

 {
   const GuidConverter conv(localId);
 
   if (conv.isReader()) {
     ACE_GUARD(ACE_Thread_Mutex, gr, readers_lock_);
     RtpsReaderMap::iterator rr = readers_.find(localId);
     if (rr != readers_.end()) {
       for (RtpsReaderMultiMap::iterator iter = readers_of_writer_.begin();
           iter != readers_of_writer_.end();) {
         if (iter->second->id() == localId) {
           readers_of_writer_.erase(iter++);
         } else {
           ++iter;
         }
       }
 
       RtpsReader_rch reader = rr->second;
       readers_.erase(rr);
       gr.release();
 
       reader->pre_stop_helper();
 
     } else {
       for (WriterToSeqReadersMap::iterator w = writer_to_seq_best_effort_readers_.begin();
           w != writer_to_seq_best_effort_readers_.end();) {
         RepoIdSet::iterator r = w->second.readers.find(localId);
         if (r != w->second.readers.end()) {
           w->second.readers.erase(r);
           if (w->second.readers.empty()) {
             writer_to_seq_best_effort_readers_.erase(w++);
             continue;
           }
         }
         ++w;
       }
     }
 
   } else {
     RtpsWriter_rch writer;
     {
       // Don't hold the writers lock when destroying a writer.
       ACE_GUARD(ACE_Thread_Mutex, gw, writers_lock_);
       RtpsWriterMap::iterator pos = writers_.find(localId);
       if (pos != writers_.end()) {
         writer = pos->second;
         writers_.erase(pos);
       }
     }
 
     if (writer) {
       TqeVector to_drop;
       writer->pre_stop_helper(to_drop, true);
 
       TqeVector::iterator drop_it = to_drop.begin();
       while (drop_it != to_drop.end()) {
         (*drop_it)->data_dropped(true);
         ++drop_it;
       }
       writer->remove_all_msgs();
     } else {
       GuardType guard(strategy_lock_);
       if (send_strategy_) {
         send_strategy_->remove_all_msgs(localId);
       }
     }
   }
   sq_.ignore_local(localId);
 }

◆ config()

RtpsUdpInst_rch OpenDDS::DCPS::RtpsUdpDataLink::config ( ) const

Definition at line 121 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::dynamic_rchandle_cast(), and OpenDDS::DCPS::DataLink::impl().

Referenced by accumulate_addresses(), bundle_mapped_meta_submessages(), check_heartbeats(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i(), get_last_recv_address(), on_data_available(), open(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acknack(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_frag_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_nackfrag(), queue_submessages(), register_for_reader(), register_for_writer(), send_heartbeats(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::stop_i(), and update_last_recv_addr().

 {
   return dynamic_rchandle_cast<RtpsUdpTransport>(impl())->config();
 }

◆ customize_queue_element()

TransportQueueElement * OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element ( TransportQueueElement * element )

privatevirtual

Allow derived classes to provide an alternate "customized" queue element for this DataLink (not shared with other links in the DataLinkSet).

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 1305 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, customize_queue_element_non_reliable_i(), OpenDDS::DCPS::TransportQueueElement::data_delivered(), OpenDDS::DCPS::TransportQueueElement::msg(), OpenDDS::DCPS::DataLink::peer_ids(), OpenDDS::DCPS::TransportQueueElement::publication_id(), queue_submessages(), requires_inline_qos(), writers_, and writers_lock_.

 {
   const ACE_Message_Block* msg = element->msg();
   if (!msg) {
     return element;
   }
 
   const GUID_t pub_id = element->publication_id();
   GUIDSeq_var peers = peer_ids(pub_id);
 
   bool require_iq = requires_inline_qos(peers);
 
   ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, writers_lock_, 0);
 
   const RtpsWriterMap::iterator rw = writers_.find(pub_id);
   MetaSubmessageVec meta_submessages;
   RtpsWriter_rch writer;
   TransportQueueElement* result;
   bool deliver_after_send = false;
   if (rw != writers_.end()) {
     writer = rw->second;
     guard.release();
     result = writer->customize_queue_element_helper(element, require_iq, meta_submessages, deliver_after_send);
   } else {
     guard.release();
     result = customize_queue_element_non_reliable_i(element, require_iq, meta_submessages, deliver_after_send, guard);
   }
 
   queue_submessages(meta_submessages);
 
   if (deliver_after_send) {
     element->data_delivered();
   }
 
   return result;
 }

◆ customize_queue_element_non_reliable_i()

TransportQueueElement * OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element_non_reliable_i	(	TransportQueueElement *	element,
		bool	requires_inline_qos,
		MetaSubmessageVec &	meta_submessages,
		bool &	deliver_after_send,
		ACE_Guard< ACE_Thread_Mutex > &	guard
	)

private

Definition at line 1227 of file RtpsUdpDataLink.cpp.

References ACE_Message_Block::cont(), OpenDDS::DCPS::RtpsSampleHeader::control_message_supported(), OpenDDS::DCPS::TransportQueueElement::data_dropped(), OpenDDS::DCPS::DATAWRITER_LIVELINESS, ACE_Message_Block::duplicate(), OpenDDS::DCPS::TransportSendControlElement::header(), OpenDDS::DCPS::TransportDebug::log_messages, OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::move(), OpenDDS::DCPS::TransportQueueElement::msg(), OpenDDS::DCPS::TransportCustomizedElement::original_send_element(), OpenDDS::DCPS::RtpsSampleHeader::populate_data_control_submessages(), OpenDDS::DCPS::RtpsSampleHeader::populate_data_sample_submessages(), OpenDDS::DCPS::TransportQueueElement::publication_id(), ACE_Guard< ACE_LOCK >::release(), OpenDDS::DCPS::unique_ptr< T, Deleter >::release(), OpenDDS::DCPS::unique_ptr< T, Deleter >::reset(), OpenDDS::DCPS::TransportSendElement::sample(), send_heartbeats_manual_i(), send_strategy(), OpenDDS::DCPS::DataLink::send_strategy_, OpenDDS::DCPS::DataLink::strategy_lock_, submsgs_to_msgblock(), and OpenDDS::DCPS::transport_debug.

Referenced by customize_queue_element().

 {
   RTPS::SubmessageSeq subm;
 
   TransportSendElement* tse = dynamic_cast<TransportSendElement*>(element);
   TransportCustomizedElement* tce =
     dynamic_cast<TransportCustomizedElement*>(element);
   TransportSendControlElement* tsce =
     dynamic_cast<TransportSendControlElement*>(element);
 
   Message_Block_Ptr data;
 
   const ACE_Message_Block* msg = element->msg();
 
   // Based on the type of 'element', find and duplicate the data payload
   // continuation block.
   if (tsce) {        // Control message
     if (RtpsSampleHeader::control_message_supported(tsce->header().message_id_)) {
       data.reset(msg->cont()->duplicate());
       // Create RTPS Submessage(s) in place of the OpenDDS DataSampleHeader
       RtpsSampleHeader::populate_data_control_submessages(
                 subm, *tsce, requires_inline_qos);
     } else if (tsce->header().message_id_ == DATAWRITER_LIVELINESS) {
       send_heartbeats_manual_i(tsce, meta_submessages);
       deliver_after_send = true;
       return 0;
     } else {
       guard.release();
       element->data_dropped(true /*dropped_by_transport*/);
       return 0;
     }
 
   } else if (tse) {  // Basic data message
     // {DataSampleHeader} -> {Data Payload}
     data.reset(msg->cont()->duplicate());
     const DataSampleElement* dsle = tse->sample();
     // Create RTPS Submessage(s) in place of the OpenDDS DataSampleHeader
     RtpsSampleHeader::populate_data_sample_submessages(
               subm, *dsle, requires_inline_qos);
 
   } else if (tce) {  // Customized data message
     // {DataSampleHeader} -> {Content Filtering GUIDs} -> {Data Payload}
     data.reset(msg->cont()->cont()->duplicate());
     const DataSampleElement* dsle = tce->original_send_element()->sample();
     // Create RTPS Submessage(s) in place of the OpenDDS DataSampleHeader
     RtpsSampleHeader::populate_data_sample_submessages(
               subm, *dsle, requires_inline_qos);
 
   } else {
     return element;
   }
 
 #ifdef OPENDDS_SECURITY
   const GUID_t pub_id = element->publication_id();
 
   {
     GuardType guard(strategy_lock_);
     if (send_strategy_) {
       send_strategy()->encode_payload(pub_id, data, subm);
     }
   }
 #endif
 
   if (transport_debug.log_messages) {
     send_strategy()->append_submessages(subm);
   }
 
   Message_Block_Ptr hdr(submsgs_to_msgblock(subm));
   hdr->cont(data.release());
   return new RtpsCustomizedElement(element, move(hdr));
 }

◆ datareader_dispatch()

template<typename T , typename FN >

void OpenDDS::DCPS::RtpsUdpDataLink::datareader_dispatch	(	const T &	submessage,
		const GuidPrefix_t &	src_prefix,
		bool	directed,
		const FN &	func
	)

inlineprivate

Definition at line 816 of file RtpsUdpDataLink.h.

References ACE_DEBUG, ACE_GUARD, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_UNKNOWN, func(), LM_DEBUG, OpenDDS::DCPS::TransportDebug::log_dropped_messages, OpenDDS::DCPS::make_id(), OpenDDS::ICE::OPENDDS_VECTOR(), and OpenDDS::DCPS::transport_debug.

Referenced by received().

   {
     const GUID_t local = make_id(local_prefix_, submessage.readerId);
     const GUID_t src = make_id(src_prefix, submessage.writerId);
 
     OPENDDS_VECTOR(RtpsReader_rch) to_call;
     {
       ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
       if (local.entityId == ENTITYID_UNKNOWN) {
         typedef std::pair<RtpsReaderMultiMap::iterator, RtpsReaderMultiMap::iterator> RRMM_IterRange;
         for (RRMM_IterRange iters = readers_of_writer_.equal_range(src); iters.first != iters.second; ++iters.first) {
           to_call.push_back(iters.first->second);
         }
         if (to_call.empty()) {
           if (transport_debug.log_dropped_messages) {
             ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::datawreader_dispatch - %C -> X no local readers\n", LogGuid(src).c_str()));
           }
           return;
         }
       } else {
         const RtpsReaderMap::iterator rr = readers_.find(local);
         if (rr == readers_.end()) {
           if (transport_debug.log_dropped_messages) {
             ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::datareader_dispatch - %C -> %C unknown local reader\n", LogGuid(src).c_str(), LogGuid(local).c_str()));
           }
           return;
         }
         to_call.push_back(rr->second);
       }
     }
     MetaSubmessageVec meta_submessages;
     for (OPENDDS_VECTOR(RtpsReader_rch)::const_iterator it = to_call.begin(); it < to_call.end(); ++it) {
       RtpsReader& reader = **it;
       (reader.*func)(submessage, src, directed, meta_submessages);
     }
     queue_submessages(meta_submessages);
   }

◆ datawriter_dispatch()

template<typename T , typename FN >

void OpenDDS::DCPS::RtpsUdpDataLink::datawriter_dispatch	(	const T &	submessage,
		const GuidPrefix_t &	src_prefix,
		const FN &	func
	)

inlineprivate

Definition at line 789 of file RtpsUdpDataLink.h.

References ACE_DEBUG, ACE_GUARD, func(), LM_DEBUG, OpenDDS::DCPS::TransportDebug::log_dropped_messages, OpenDDS::DCPS::make_id(), OpenDDS::ICE::OPENDDS_VECTOR(), and OpenDDS::DCPS::transport_debug.

Referenced by received().

   {
     const GUID_t local = make_id(local_prefix_, submessage.writerId);
     const GUID_t src = make_id(src_prefix, submessage.readerId);
 
     OPENDDS_VECTOR(RtpsWriter_rch) to_call;
     {
       ACE_GUARD(ACE_Thread_Mutex, g, writers_lock_);
       const RtpsWriterMap::iterator rw = writers_.find(local);
       if (rw == writers_.end()) {
         if (transport_debug.log_dropped_messages) {
           ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::datawriter_dispatch - %C -> %C unknown local writer\n", LogGuid(local).c_str(), LogGuid(src).c_str()));
         }
         return;
       }
       to_call.push_back(rw->second);
     }
     MetaSubmessageVec meta_submessages;
     for (OPENDDS_VECTOR(RtpsWriter_rch)::const_iterator it = to_call.begin(); it < to_call.end(); ++it) {
       RtpsWriter& writer = **it;
       (writer.*func)(submessage, src, meta_submessages);
     }
     queue_submessages(meta_submessages);
   }

◆ disable_response_queue()

void OpenDDS::DCPS::RtpsUdpDataLink::disable_response_queue ( bool send_immediately )

Definition at line 2693 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::TransactionalRtpsSendQueue::end_transaction(), flush_send_queue_i(), flush_send_queue_sporadic_, fsq_mutex_, fsq_vec_size_, sq_, and OpenDDS::DCPS::TimeDuration::zero_value.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::end_transport_header_processing(), and harvest_send_queue().

 {
   MetaSubmessageVec vec;
   ACE_Guard<ACE_Thread_Mutex> fsq_guard(fsq_mutex_);
   sq_.end_transaction(vec);
   if (!vec.empty()) {
     if (fsq_vec_.size() == fsq_vec_size_) {
       fsq_vec_.resize(fsq_vec_.size() + 1);
     }
     fsq_vec_[fsq_vec_size_++].swap(vec);
   }
 
   if (fsq_vec_size_) {
     if (send_immediately) {
       flush_send_queue_i();
     } else {
       flush_send_queue_sporadic_->schedule(TimeDuration::zero_value);
     }
   }
 
 }

◆ disassociated()

void OpenDDS::DCPS::RtpsUdpDataLink::disassociated	(	const GUID_t &	local,
		const GUID_t &	remote
	)

Definition at line 655 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_GUARD, OpenDDS::DCPS::TransactionalRtpsSendQueue::ignore_remote(), LM_DEBUG, locators_, locators_lock_, OPENDDS_ASSERT, release_reservations_i(), remove_locator_and_bundling_cache(), sq_, and OpenDDS::DCPS::Transport_debug_level.

Referenced by OpenDDS::DCPS::RtpsUdpTransport::stop_accepting_or_connecting().

 {
   release_reservations_i(remote_id, local_id);
   remove_locator_and_bundling_cache(remote_id);
   sq_.ignore_remote(remote_id);
 
   ACE_GUARD(ACE_Thread_Mutex, g, locators_lock_);
 
   RemoteInfoMap::iterator pos = locators_.find(remote_id);
   if (pos != locators_.end()) {
     OPENDDS_ASSERT(pos->second.ref_count_ > 0);
 
     --pos->second.ref_count_;
     if (pos->second.ref_count_ == 0) {
       locators_.erase(pos);
     }
   } else if (Transport_debug_level > 3) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::disassociated: local id %C does not have any locators\n", LogGuid(local_id).c_str()));
   }
 }

◆ durability_resend() [1/2]

void OpenDDS::DCPS::RtpsUdpDataLink::durability_resend	(	TransportQueueElement *	element,
		size_t &	cumulative_send_count
	)

private

Definition at line 4014 of file RtpsUdpDataLink.cpp.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i().

 {
   static CORBA::Long buffer[8];
   static const RTPS::FragmentNumberSet none = { {0}, 0, RTPS::LongSeq8(0, buffer) };
   durability_resend(element, none, cumulative_send_count);
 }

◆ durability_resend() [2/2]

void OpenDDS::DCPS::RtpsUdpDataLink::durability_resend	(	TransportQueueElement *	element,
		const RTPS::FragmentNumberSet &	fragmentSet,
		size_t &	cumulative_send_count
	)

private

Definition at line 4022 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_ERROR, OpenDDS::RTPS::FragmentNumberSet::bitmap, OpenDDS::RTPS::FragmentNumberSet::bitmapBase, OpenDDS::DCPS::DisjointSequence::empty(), get_addresses(), OpenDDS::DCPS::SequenceNumber::getValue(), include_fragment(), OpenDDS::DCPS::DisjointSequence::insert(), LM_DEBUG, LM_ERROR, OpenDDS::RTPS::FragmentNumberSet::numBits, OpenDDS::DCPS::TransportQueueElement::publication_id(), send_strategy(), OpenDDS::DCPS::TransportQueueElement::sequence(), OpenDDS::DCPS::TransportQueueElement::subscription_id(), OpenDDS::DCPS::Transport_debug_level, and OpenDDS::RTPS::FragmentNumber_t::value.

 {
   if (Transport_debug_level > 5) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) TRACK RtpsUdpDataLink::durability_resend %q\n", element->sequence().getValue()));
   }
   const AddrSet addrs = get_addresses(element->publication_id(), element->subscription_id());
   if (addrs.empty()) {
     const LogGuid conv(element->subscription_id());
     ACE_ERROR((LM_ERROR,
                "(%P|%t) ERROR: RtpsUdpDataLink::durability_resend() - "
                "no locator for remote %C\n", conv.c_str()));
     return;
   }
 
   TqeVector to_send;
   if (!send_strategy()->fragmentation_helper(element, to_send)) {
     return;
   }
 
   DisjointSequence fragments;
   fragments.insert(fragmentSet.bitmapBase.value, fragmentSet.numBits,
                    fragmentSet.bitmap.get_buffer());
   SequenceNumber lastFragment = 0;
 
   const TqeVector::iterator end = to_send.end();
   for (TqeVector::iterator i = to_send.begin(); i != end; ++i) {
     if (fragments.empty() || include_fragment(**i, fragments, lastFragment)) {
       RTPS::Message message;
       send_strategy()->send_rtps_control(message, *const_cast<ACE_Message_Block*>((*i)->msg()), addrs);
       ++cumulative_send_count;
     }
 
     (*i)->data_delivered();
   }
 }

◆ enable_response_queue()

void OpenDDS::DCPS::RtpsUdpDataLink::enable_response_queue ( )

Definition at line 2687 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::TransactionalRtpsSendQueue::begin_transaction(), and sq_.

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

 {
   sq_.begin_transaction();
 }

◆ event_dispatcher()

EventDispatcher_rch OpenDDS::DCPS::RtpsUdpDataLink::event_dispatcher ( )

inline

Definition at line 248 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acknack().

248 { return event_dispatcher_; }

OpenDDS::DCPS::RtpsUdpDataLink::event_dispatcher_

EventDispatcher_rch event_dispatcher_

Definition: RtpsUdpDataLink.h:272

◆ extend_bitmap_range()

void OpenDDS::DCPS::RtpsUdpDataLink::extend_bitmap_range	(	RTPS::FragmentNumberSet &	fnSet,
		CORBA::ULong	extent,
		ACE_CDR::ULong &	cumulative_bits_added
	)

staticprivate

Extend the FragmentNumberSet to cover the fragments that are missing from our last known fragment to the extent

Parameters

fnSet	FragmentNumberSet for the message sequence number in question
extent	is the highest fragment sequence number for this FragmentNumberSet

Definition at line 2967 of file RtpsUdpDataLink.cpp.

References OpenDDS::RTPS::FragmentNumberSet::bitmap, OpenDDS::RTPS::FragmentNumberSet::bitmapBase, OpenDDS::DCPS::DisjointSequence::fill_bitmap_range(), OpenDDS::RTPS::FragmentNumberSet::numBits, and OpenDDS::RTPS::FragmentNumber_t::value.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::generate_nack_frags_i().

 {
   if (extent < fnSet.bitmapBase.value) {
     return; // can't extend to some number under the base
   }
   // calculate the index to the extent to determine the new_num_bits
   const CORBA::ULong new_num_bits = std::min(CORBA::ULong(256),
                                              extent - fnSet.bitmapBase.value + 1),
                      len = (new_num_bits + 31) / 32;
   if (new_num_bits < fnSet.numBits) {
     return; // bitmap already extends past "extent"
   }
   fnSet.bitmap.length(len);
   // We are missing from one past old bitmap end to the new end
   DisjointSequence::fill_bitmap_range(fnSet.numBits, new_num_bits,
                                       fnSet.bitmap.get_buffer(), len,
                                       fnSet.numBits, samples_requested);
 }

◆ filterBestEffortReaders()

void OpenDDS::DCPS::RtpsUdpDataLink::filterBestEffortReaders	(	const ReceivedDataSample &	ds,
		RepoIdSet &	selected,
		RepoIdSet &	withheld
	)

Definition at line 524 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::DataSampleHeader::publication_id_, readers_lock_, OpenDDS::DCPS::DataSampleHeader::sequence_, and writer_to_seq_best_effort_readers_.

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

 {
   ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
   const GUID_t& writer = ds.header_.publication_id_;
   const SequenceNumber& seq = ds.header_.sequence_;
   WriterToSeqReadersMap::iterator w = writer_to_seq_best_effort_readers_.find(writer);
   if (w != writer_to_seq_best_effort_readers_.end()) {
     if (w->second.seq < seq) {
       w->second.seq = seq;
       selected.insert(w->second.readers.begin(), w->second.readers.end());
     } else {
       withheld.insert(w->second.readers.begin(), w->second.readers.end());
     }
   } // else the writer is not associated with best effort readers
 }

◆ flush_send_queue()

void OpenDDS::DCPS::RtpsUdpDataLink::flush_send_queue ( const MonotonicTimePoint & now )

private

Definition at line 2669 of file RtpsUdpDataLink.cpp.

References flush_send_queue_i(), and fsq_mutex_.

 {
   ACE_Guard<ACE_Thread_Mutex> fsq_guard(fsq_mutex_);
   flush_send_queue_i();
 }

◆ flush_send_queue_i()

void OpenDDS::DCPS::RtpsUdpDataLink::flush_send_queue_i ( )

private

Definition at line 2676 of file RtpsUdpDataLink.cpp.

References bundle_and_send_submessages(), OpenDDS::DCPS::dedup(), and fsq_vec_size_.

Referenced by disable_response_queue(), and flush_send_queue().

 {
   for (size_t idx = 0; idx != fsq_vec_size_; ++idx) {
     dedup(fsq_vec_[idx]);
     bundle_and_send_submessages(fsq_vec_[idx]);
     fsq_vec_[idx].clear();
   }
   fsq_vec_size_ = 0;
 }

◆ get_addresses() [1/2]

AddrSet OpenDDS::DCPS::RtpsUdpDataLink::get_addresses	(	const GUID_t &	local,
		const GUID_t &	remote
	)		const

Given a 'local' id and a 'remote' id of a publication or subscription, return the set of addresses of the remote peers.

Definition at line 4593 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, get_addresses_i(), and locators_lock_.

Referenced by durability_resend(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i(), and OpenDDS::DCPS::RtpsUdpSendStrategy::send_bytes_i_helper().

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, locators_lock_, AddrSet());
   return get_addresses_i(local, remote);
 }

◆ get_addresses() [2/2]

AddrSet OpenDDS::DCPS::RtpsUdpDataLink::get_addresses ( const GUID_t & local ) const

Given a 'local' id, return the set of address for all remote peers.

Definition at line 4600 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, get_addresses_i(), and locators_lock_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, locators_lock_, AddrSet());
   return get_addresses_i(local);
 }

◆ get_addresses_i() [1/2]

AddrSet OpenDDS::DCPS::RtpsUdpDataLink::get_addresses_i	(	const GUID_t &	local,
		const GUID_t &	remote
	)		const

private

Definition at line 4607 of file RtpsUdpDataLink.cpp.

References accumulate_addresses().

Referenced by build_meta_submessage_map(), and get_addresses().

 {
   AddrSet retval;
 
   accumulate_addresses(local, remote, retval, true);
 
   return retval;
 }

◆ get_addresses_i() [2/2]

AddrSet OpenDDS::DCPS::RtpsUdpDataLink::get_addresses_i ( const GUID_t & local ) const

private

Definition at line 4617 of file RtpsUdpDataLink.cpp.

References accumulate_addresses(), OpenDDS::DCPS::GuidConverter::isWriter(), OpenDDS::DCPS::DataLink::peer_ids(), ACE_Guard< ACE_LOCK >::release(), writers_, and writers_lock_.

 {
   AddrSet retval;
   bool use_peers = true;
 
   // For reliable writers, use remote_reader_guids()
   const GuidConverter conv(local);
   if (conv.isWriter()) {
     RtpsWriter_rch writer;
     ACE_Guard<ACE_Thread_Mutex> guard(writers_lock_);
     RtpsWriterMap::const_iterator pos = writers_.find(local);
     if (pos != writers_.end()) {
       writer = pos->second;
     }
     guard.release();
     if (writer) {
       RcHandle<ConstSharedRepoIdSet> addr_guids = writer->get_remote_reader_guids();
       if (addr_guids) {
         for (RepoIdSet::const_iterator it = addr_guids->guids_.begin(),
           limit = addr_guids->guids_.end(); it != limit; ++it) {
           accumulate_addresses(local, *it, retval);
 
         }
         use_peers = false;
       }
     }
   }
 
   if (use_peers) {
     const GUIDSeq_var peers = peer_ids(local);
     if (peers.ptr()) {
       for (CORBA::ULong i = 0; i < peers->length(); ++i) {
         accumulate_addresses(local, peers[i], retval);
       }
     }
   }
 
   return retval;
 }

◆ get_ice_agent()

RcHandle< ICE::Agent > OpenDDS::DCPS::RtpsUdpDataLink::get_ice_agent ( ) const

Definition at line 4786 of file RtpsUdpDataLink.cpp.

References ice_agent_.

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

 {
   return ice_agent_;
 }

◆ get_ice_endpoint()

WeakRcHandle< ICE::Endpoint > OpenDDS::DCPS::RtpsUdpDataLink::get_ice_endpoint ( ) const

virtual

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 4793 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::TransportImpl::get_ice_endpoint(), and OpenDDS::DCPS::DataLink::impl().

Referenced by accumulate_addresses(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::receive_bytes().

 {
   TransportImpl_rch ti = impl();
   return ti ? ti->get_ice_endpoint() : WeakRcHandle<ICE::Endpoint>();
 }

◆ get_job_queue()

RcHandle<JobQueue> OpenDDS::DCPS::RtpsUdpDataLink::get_job_queue ( ) const

inline

Definition at line 249 of file RtpsUdpDataLink.h.

249 { return job_queue_; }

OpenDDS::DCPS::RtpsUdpDataLink::job_queue_

RcHandle< JobQueue > job_queue_

Definition: RtpsUdpDataLink.h:271

◆ get_last_recv_address()

NetworkAddress OpenDDS::DCPS::RtpsUdpDataLink::get_last_recv_address ( const GUID_t & remote_id )

Definition at line 466 of file RtpsUdpDataLink.cpp.

References config(), locators_, locators_lock_, and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now().

Referenced by OpenDDS::DCPS::RtpsUdpTransport::get_last_recv_locator().

 {
   ACE_Guard<ACE_Thread_Mutex> guard(locators_lock_);
   const RemoteInfoMap::const_iterator pos = locators_.find(remote_id);
   if (pos != locators_.end()) {
     RtpsUdpInst_rch cfg = config();
     const TimeDuration threshold = cfg ? cfg->receive_address_duration_ : TimeDuration();
     const bool valid_last_recv_addr = (MonotonicTimePoint::now() - pos->second.last_recv_time_) <= threshold;
     return valid_last_recv_addr ? pos->second.last_recv_addr_ : NetworkAddress();
   }
   return NetworkAddress();
 }

◆ get_reactor()

ACE_INLINE ACE_Reactor * OpenDDS::DCPS::RtpsUdpDataLink::get_reactor ( void )

Definition at line 16 of file RtpsUdpDataLink.inl.

References ACE_INLINE, OpenDDS::DCPS::ReactorTask::get_reactor(), and reactor_task_.

Referenced by on_data_available().

 {
   if (!reactor_task_) return 0;
   return reactor_task_->get_reactor();
 }

◆ get_reactor_interceptor()

ACE_INLINE ReactorInterceptor_rch OpenDDS::DCPS::RtpsUdpDataLink::get_reactor_interceptor ( ) const

Definition at line 23 of file RtpsUdpDataLink.inl.

References ACE_INLINE, OpenDDS::DCPS::ReactorTask::interceptor(), and reactor_task_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::start_i(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::stop_i().

 {
   if (!reactor_task_) return ReactorInterceptor_rch();
   return reactor_task_->interceptor();
 }

◆ get_writer_send_buffer()

RcHandle< SingleSendBuffer > OpenDDS::DCPS::RtpsUdpDataLink::get_writer_send_buffer ( const GUID_t & pub_id )

private

Definition at line 978 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, writers_, and writers_lock_.

 {
   RcHandle<SingleSendBuffer> result;
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, writers_lock_, result);
 
   const RtpsWriterMap::iterator wi = writers_.find(pub_id);
   if (wi != writers_.end()) {
     result = wi->second->get_send_buff();
   }
   return result;
 }

◆ handle_registry()

Security::HandleRegistry_rch OpenDDS::DCPS::RtpsUdpDataLink::handle_registry ( ) const

inline

Definition at line 229 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_encoded(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::decode_payload(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_from_secure(), OpenDDS::DCPS::RtpsUdpSendStrategy::encode_payload(), OpenDDS::DCPS::RtpsUdpSendStrategy::encode_reader_submessage(), OpenDDS::DCPS::RtpsUdpSendStrategy::encode_submessages(), OpenDDS::DCPS::RtpsUdpSendStrategy::encode_writer_submessage(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::receive_bytes().

   {
     ACE_Guard<ACE_Thread_Mutex> guard(security_mutex_);
     return handle_registry_;
   }

◆ harvest_send_queue()

void OpenDDS::DCPS::RtpsUdpDataLink::harvest_send_queue ( const MonotonicTimePoint & now )

private

Definition at line 2660 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::TransactionalRtpsSendQueue::begin_transaction(), disable_response_queue(), OpenDDS::DCPS::TransactionalRtpsSendQueue::ready_to_send(), and sq_.

 {
   sq_.begin_transaction();
   sq_.ready_to_send();
 
   disable_response_queue(true);
 }

◆ include_fragment()

bool OpenDDS::DCPS::RtpsUdpDataLink::include_fragment	(	const TransportQueueElement &	element,
		const DisjointSequence &	fragments,
		SequenceNumber &	lastFragment
	)

staticprivate

Definition at line 4060 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::DisjointSequence::contains_any(), OpenDDS::DCPS::TransportQueueElement::is_fragment(), and OpenDDS::DCPS::RtpsCustomizedElement::last_fragment().

Referenced by durability_resend().

 {
   if (!element.is_fragment()) {
     return true;
   }
 
   const RtpsCustomizedElement* const rce = dynamic_cast<const RtpsCustomizedElement*>(&element);
   if (!rce) {
     return true;
   }
 
   const SequenceRange thisElement(lastFragment + 1, rce->last_fragment());
   lastFragment = thisElement.second;
   return fragments.contains_any(thisElement);
 }

◆ is_leading()

bool OpenDDS::DCPS::RtpsUdpDataLink::is_leading	(	const GUID_t &	writer_id,
		const GUID_t &	reader_id
	)		const

virtual

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 4799 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, writers_, and writers_lock_.

 {
   RtpsWriterMap::mapped_type writer;
 
   {
     ACE_GUARD_RETURN(ACE_Thread_Mutex, g, writers_lock_, false);
     RtpsWriterMap::const_iterator pos = writers_.find(writer_id);
     if (pos == writers_.end()) {
       return false;
     }
     writer = pos->second;
   }
 
   return writer->is_leading(reader_id);
 }

◆ local_crypto_handle() [1/2]

ACE_INLINE DDS::Security::ParticipantCryptoHandle OpenDDS::DCPS::RtpsUdpDataLink::local_crypto_handle ( ) const

Definition at line 64 of file RtpsUdpDataLink.inl.

References ACE_INLINE, local_crypto_handle_, and security_mutex_.

Referenced by build_meta_submessage_map(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_encoded(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_from_secure(), OpenDDS::DCPS::RtpsUdpSendStrategy::encode_rtps_message(), OpenDDS::DCPS::RtpsUdpTransport::local_crypto_handle(), OpenDDS::DCPS::RtpsUdpTransport::make_datalink(), OpenDDS::DCPS::RtpsUdpSendStrategy::pre_send_packet(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::receive_bytes().

 {
   ACE_Guard<ACE_Thread_Mutex> guard(security_mutex_);
   return local_crypto_handle_;
 }

◆ local_crypto_handle() [2/2]

ACE_INLINE void OpenDDS::DCPS::RtpsUdpDataLink::local_crypto_handle ( DDS::Security::ParticipantCryptoHandle pch )

Definition at line 71 of file RtpsUdpDataLink.inl.

References local_crypto_handle_, OPENDDS_END_VERSIONED_NAMESPACE_DECL, and security_mutex_.

 {
   ACE_Guard<ACE_Thread_Mutex> guard(security_mutex_);
   local_crypto_handle_ = h;
 }

◆ local_prefix()

const GuidPrefix_t& OpenDDS::DCPS::RtpsUdpDataLink::local_prefix ( ) const

inline

Definition at line 155 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample_i(), OpenDDS::DCPS::RtpsUdpSendStrategy::encode_submessages(), and RtpsUdpDataLink().

155 { return local_prefix_; }

OpenDDS::DCPS::RtpsUdpDataLink::local_prefix_

GuidPrefix_t local_prefix_

Definition: RtpsUdpDataLink.h:277

◆ make_reservation()

int OpenDDS::DCPS::RtpsUdpDataLink::make_reservation	(	const GUID_t &	remote_publication_id,
		const GUID_t &	local_subscription_id,
		const TransportReceiveListener_wrch &	receive_listener,
		bool	reliable
	)

virtual

Only called by our TransportImpl object.

Return Codes: 0 means successful reservation made. -1 means failure.

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 541 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::DataLink::make_reservation(), pending_reliable_readers_, readers_, readers_lock_, and ACE_Guard< ACE_LOCK >::release().

 {
   ACE_Guard<ACE_Thread_Mutex> guard(readers_lock_);
   if (reliable) {
     RtpsReaderMap::iterator rr = readers_.find(lsi);
     if (rr == readers_.end()) {
       pending_reliable_readers_.insert(lsi);
     }
   }
   guard.release();
   return DataLink::make_reservation(rpi, lsi, trl, reliable);
 }

◆ multicast_socket()

ACE_INLINE ACE_SOCK_Dgram_Mcast & OpenDDS::DCPS::RtpsUdpDataLink::multicast_socket ( )

Definition at line 43 of file RtpsUdpDataLink.inl.

References ACE_INLINE, and multicast_socket_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::choose_recv_socket(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::stop_i().

 {
   return multicast_socket_;
 }

◆ on_data_available()

void OpenDDS::DCPS::RtpsUdpDataLink::on_data_available ( RcHandle< InternalDataReader< NetworkInterfaceAddress > > reader )

private

Definition at line 427 of file RtpsUdpDataLink.cpp.

References config(), get_reactor(), multicast_manager_, multicast_socket_, OpenDDS::DCPS::DataLink::network_change(), network_interface_address_reader_, OpenDDS::DCPS::MulticastManager::process(), and receive_strategy().

 {
   InternalDataReader<NetworkInterfaceAddress>::SampleSequence samples;
   InternalSampleInfoSequence infos;
 
   network_interface_address_reader_->take(samples, infos);
 
   RtpsUdpInst_rch cfg = config();
   if (!cfg || !cfg->use_multicast_) {
     return;
   }
 
   multicast_manager_.process(samples,
                              infos,
                              cfg->multicast_interface_,
                              get_reactor(),
                              receive_strategy().in(),
                              cfg->multicast_group_address(),
                              multicast_socket_
 #ifdef ACE_HAS_IPV6
                              , cfg->ipv6_multicast_group_address(),
                              ipv6_multicast_socket_
 #endif
                              );
 
   if (!samples.empty()) {
     // FUTURE: This is propagating info to discovery.  Write instead.
     network_change();
   }
 }

◆ open()

bool OpenDDS::DCPS::RtpsUdpDataLink::open ( const ACE_SOCK_Dgram & unicast_socket )

Definition at line 293 of file RtpsUdpDataLink.cpp.

References ACE_ERROR, ACE_TEXT(), config(), OpenDDS::DCPS::DCPS_debug_level, ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE, ENOTSUP, LM_ERROR, multi_buff_, multicast_socket_, network_interface_address_reader_, ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO, ACE_SOCK_Dgram_Mcast::opts(), OpenDDS::DCPS::DataLink::receive_strategy_, send_strategy(), OpenDDS::DCPS::DataLink::send_strategy_, ACE_SOCK::set_option(), OpenDDS::DCPS::set_socket_multicast_ttl(), SO_RCVBUF, SO_SNDBUF, SOL_SOCKET, OpenDDS::DCPS::DataLink::start(), stop_i(), TheServiceParticipant, unicast_socket(), and unicast_socket_.

Referenced by OpenDDS::DCPS::RtpsUdpTransport::make_datalink().

 {
   unicast_socket_ = unicast_socket;
 #ifdef ACE_HAS_IPV6
   ipv6_unicast_socket_ = ipv6_unicast_socket;
 #endif
 
   RtpsUdpInst_rch cfg = config();
   if (!cfg) {
     return false;
   }
 
   if (cfg->use_multicast_) {
 #ifdef ACE_HAS_MAC_OSX
     multicast_socket_.opts(ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO |
                            ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE);
 #ifdef ACE_HAS_IPV6
     ipv6_multicast_socket_.opts(ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO |
                                 ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE);
 #endif
 #endif
   }
 
   if (cfg->use_multicast_) {
     if (!set_socket_multicast_ttl(unicast_socket_, cfg->ttl_)) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: ")
                    ACE_TEXT("RtpsUdpDataLink::open: ")
                    ACE_TEXT("failed to set TTL: %d\n"),
                    cfg->ttl_));
       }
       return false;
     }
 #ifdef ACE_HAS_IPV6
     if (!set_socket_multicast_ttl(ipv6_unicast_socket_, cfg->ttl_)) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: ")
                    ACE_TEXT("RtpsUdpDataLink::open: ")
                    ACE_TEXT("failed to set TTL: %d\n"),
                    cfg->ttl_));
       }
       return false;
     }
 #endif
   }
 
   if (cfg->send_buffer_size_ > 0) {
     const int snd_size = cfg->send_buffer_size_;
     if (unicast_socket_.set_option(SOL_SOCKET,
                                 SO_SNDBUF,
                                 (void *) &snd_size,
                                 sizeof(snd_size)) < 0
         && errno != ENOTSUP) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: ")
                    ACE_TEXT("RtpsUdpDataLink::open: failed to set the send buffer size to %d errno %m\n"),
                    snd_size));
       }
       return false;
     }
 #ifdef ACE_HAS_IPV6
     if (ipv6_unicast_socket_.set_option(SOL_SOCKET,
                                         SO_SNDBUF,
                                         (void *) &snd_size,
                                         sizeof(snd_size)) < 0
         && errno != ENOTSUP) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: ")
                    ACE_TEXT("RtpsUdpDataLink::open: failed to set the send buffer size to %d errno %m\n"),
                    snd_size));
       }
       return false;
     }
 #endif
   }
 
   if (cfg->rcv_buffer_size_ > 0) {
     const int rcv_size = cfg->rcv_buffer_size_;
     if (unicast_socket_.set_option(SOL_SOCKET,
                                 SO_RCVBUF,
                                 (void *) &rcv_size,
                                 sizeof(int)) < 0
         && errno != ENOTSUP) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: ")
                    ACE_TEXT("RtpsUdpDataLink::open: failed to set the receive buffer size to %d errno %m\n"),
                    rcv_size));
       }
       return false;
     }
 #ifdef ACE_HAS_IPV6
     if (ipv6_unicast_socket_.set_option(SOL_SOCKET,
                                         SO_RCVBUF,
                                         (void *) &rcv_size,
                                         sizeof(int)) < 0
         && errno != ENOTSUP) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: ")
                    ACE_TEXT("RtpsUdpDataLink::open: failed to set the receive buffer size to %d errno %m\n"),
                    rcv_size));
       }
       return false;
     }
 #endif
   }
 
   send_strategy()->send_buffer(&multi_buff_);
 
   if (start(send_strategy_,
             receive_strategy_, false) != 0) {
     stop_i();
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: ")
                  ACE_TEXT("UdpDataLink::open: start failed!\n")));
     }
     return false;
   }
 
   TheServiceParticipant->network_interface_address_topic()->connect(network_interface_address_reader_);
 
   return true;
 }

◆ OPENDDS_MAP() [1/5]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP	(	FragmentNumberValue	,
		RTPS::FragmentNumberSet
	)

private

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::ReaderInfo::expunge_durable_data(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::generate_nack_frags_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acknack(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_reader().

◆ OPENDDS_MAP() [2/5]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP	(	SequenceNumber	,
		RequestedFragMap
	)

private

◆ OPENDDS_MAP() [3/5]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP	(	SequenceNumber	,
		ReaderInfoSetHolder_rch
	)

private

◆ OPENDDS_MAP() [4/5]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP	(	AddressCacheEntryProxy	,
		DestMetaSubmessageMap
	)

private

◆ OPENDDS_MAP() [5/5]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP	(	CORBA::Long	,
		CountMapPair
	)

private

◆ OPENDDS_MAP_CMP() [1/9]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP_CMP	(	GUID_t	,
		RemoteInfo	,
		GUID_tKeyLessThan
	)

private

Referenced by send_heartbeats().

◆ OPENDDS_MAP_CMP() [2/9]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP_CMP	(	GUID_t	,
		ReaderInfo_rch	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MAP_CMP() [3/9]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP_CMP	(	GUID_t	,
		RtpsWriter_rch	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MAP_CMP() [4/9]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP_CMP	(	GUID_t	,
		WriterInfo_rch	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MAP_CMP() [5/9]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP_CMP	(	GUID_t	,
		MetaSubmessageIterVec	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MAP_CMP() [6/9]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP_CMP	(	EntityId_t	,
		CountSet	,
		EntityId_tKeyLessThan
	)

private

◆ OPENDDS_MAP_CMP() [7/9]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP_CMP	(	EntityId_t	,
		CountMapping	,
		EntityId_tKeyLessThan
	)

private

◆ OPENDDS_MAP_CMP() [8/9]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP_CMP	(	GUID_t	,
		RtpsReader_rch	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MAP_CMP() [9/9]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP_CMP	(	EntityId_t	,
		CORBA::Long	,
		EntityId_tKeyLessThan
	)

private

◆ OPENDDS_MULTIMAP_CMP() [1/2]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MULTIMAP_CMP	(	GUID_t	,
		RtpsReader_rch	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MULTIMAP_CMP() [2/2]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MULTIMAP_CMP	(	GUID_t	,
		InterestingRemote	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_SET() [1/3]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_SET ( ReaderInfo_rch )

private

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::pre_stop_helper().

◆ OPENDDS_SET() [2/3]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_SET ( WriterInfo_rch )

private

◆ OPENDDS_SET() [3/3]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_SET ( CORBA::Long )

private

◆ OPENDDS_VECTOR() [1/4]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_VECTOR ( MetaSubmessageVec::iterator )

private

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::acked_by_all_helper_i(), check_heartbeats(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::deliver_held_data(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::generate_nack_frags_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acknack(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_gap_i(), received(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_reader(), and send_heartbeats().

◆ OPENDDS_VECTOR() [2/4]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_VECTOR ( MetaSubmessageIterVec )

private

◆ OPENDDS_VECTOR() [3/4]

typedef OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_VECTOR ( Bundle )

◆ OPENDDS_VECTOR() [4/4]

OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_VECTOR ( MetaSubmessageVec )

private

◆ pre_stop_i()

void OpenDDS::DCPS::RtpsUdpDataLink::pre_stop_i ( )

virtual

Called before release the datalink or before shutdown to let the concrete DataLink to do anything necessary.

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 860 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, DBG_ENTRY_LVL, heartbeat_counts_, OpenDDS::DCPS::DataLink::pre_stop_i(), readers_, readers_lock_, writers_, and writers_lock_.

 {
   DBG_ENTRY_LVL("RtpsUdpDataLink","pre_stop_i",6);
   DataLink::pre_stop_i();
   TqeVector to_drop;
 
   RtpsWriterMap writers;
   {
     ACE_GUARD(ACE_Thread_Mutex, g, writers_lock_);
     writers_.swap(writers);
     for (RtpsWriterMap::const_iterator it = writers.begin(); it != writers.end(); ++it) {
       heartbeat_counts_.erase(it->first.entityId);
     }
   }
 
   RtpsWriterMap::iterator w_iter = writers.begin();
   while (w_iter != writers.end()) {
     w_iter->second->pre_stop_helper(to_drop, true);
     writers.erase(w_iter++);
   }
 
   TqeVector::iterator drop_it = to_drop.begin();
   while (drop_it != to_drop.end()) {
     (*drop_it)->data_dropped(true);
     ++drop_it;
   }
 
   RtpsReaderMap readers;
   {
     ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
     readers = readers_;
   }
 
   RtpsReaderMap::iterator r_iter = readers.begin();
   while (r_iter != readers.end()) {
     r_iter->second->pre_stop_helper();
     ++r_iter;
   }
 }

◆ queue_submessages()

void OpenDDS::DCPS::RtpsUdpDataLink::queue_submessages ( MetaSubmessageVec & meta_submessages )

private

Definition at line 2716 of file RtpsUdpDataLink.cpp.

References config(), OpenDDS::DCPS::TransactionalRtpsSendQueue::enqueue(), harvest_send_queue_sporadic_, and sq_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::add_reader(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::add_writer(), customize_queue_element(), received(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::send_heartbeats(), send_heartbeats(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::send_nack_responses(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::send_preassociation_acknacks().

 {
   if (in.empty()) {
     return;
   }
 
   if (sq_.enqueue(in)) {
     RtpsUdpInst_rch cfg = config();
     if (cfg) {
       harvest_send_queue_sporadic_->schedule(cfg->send_delay_);
     }
   }
 }

◆ reactor_is_shut_down()

ACE_INLINE bool OpenDDS::DCPS::RtpsUdpDataLink::reactor_is_shut_down ( )

Definition at line 30 of file RtpsUdpDataLink.inl.

References ACE_INLINE, OpenDDS::DCPS::ReactorTask::is_shut_down(), and reactor_task_.

 {
   if (!reactor_task_) return true;
   return reactor_task_->is_shut_down();
 }

◆ receive_strategy()

RtpsUdpReceiveStrategy_rch OpenDDS::DCPS::RtpsUdpDataLink::receive_strategy ( )

private

Definition at line 4587 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::dynamic_rchandle_cast(), and OpenDDS::DCPS::DataLink::receive_strategy_.

Referenced by associated(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::gather_ack_nacks_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::generate_nack_frags_i(), on_data_available(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::pre_stop_helper(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_data_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_gap_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_frag_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_i(), received(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::should_nack_fragments().

 {
   return dynamic_rchandle_cast<RtpsUdpReceiveStrategy>(receive_strategy_);
 }

◆ received() [1/6]

void OpenDDS::DCPS::RtpsUdpDataLink::received	(	const RTPS::DataSubmessage &	data,
		const GuidPrefix_t &	src_prefix,
		const NetworkAddress &	remote_addr
	)

Definition at line 1567 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_UNKNOWN, local_prefix_, OpenDDS::DCPS::make_id(), OPENDDS_VECTOR(), pending_reliable_readers_, queue_submessages(), OpenDDS::RTPS::DataSubmessage::readerId, readers_, readers_lock_, readers_of_writer_, receive_strategy(), OpenDDS::DCPS::DataLink::strategy_lock_, update_last_recv_addr(), and OpenDDS::RTPS::DataSubmessage::writerId.

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

 {
   const GUID_t local = make_id(local_prefix_, data.readerId);
   const GUID_t src = make_id(src_prefix, data.writerId);
 
   update_last_recv_addr(src, remote_addr);
 
   OPENDDS_VECTOR(RtpsReader_rch) to_call;
   {
     ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
     if (local.entityId == ENTITYID_UNKNOWN) {
       typedef std::pair<RtpsReaderMultiMap::iterator, RtpsReaderMultiMap::iterator> RRMM_IterRange;
       for (RRMM_IterRange iters = readers_of_writer_.equal_range(src); iters.first != iters.second; ++iters.first) {
         to_call.push_back(iters.first->second);
       }
       if (!pending_reliable_readers_.empty()) {
         GuardType guard(strategy_lock_);
         RtpsUdpReceiveStrategy_rch trs = receive_strategy();
         if (trs) {
           for (RepoIdSet::const_iterator it = pending_reliable_readers_.begin();
                it != pending_reliable_readers_.end(); ++it)
           {
             trs->withhold_data_from(*it);
           }
         }
       }
     } else {
       const RtpsReaderMap::iterator rr = readers_.find(local);
       if (rr != readers_.end()) {
         to_call.push_back(rr->second);
       } else if (pending_reliable_readers_.count(local)) {
         GuardType guard(strategy_lock_);
         RtpsUdpReceiveStrategy_rch trs = receive_strategy();
         if (trs) {
           trs->withhold_data_from(local);
         }
       }
     }
   }
   MetaSubmessageVec meta_submessages;
   for (OPENDDS_VECTOR(RtpsReader_rch)::const_iterator it = to_call.begin(); it < to_call.end(); ++it) {
     (*it)->process_data_i(data, src, meta_submessages);
   }
   queue_submessages(meta_submessages);
 }

◆ received() [2/6]

void OpenDDS::DCPS::RtpsUdpDataLink::received	(	const RTPS::GapSubmessage &	gap,
		const GuidPrefix_t &	src_prefix,
		bool	directed,
		const NetworkAddress &	remote_addr
	)

Definition at line 1780 of file RtpsUdpDataLink.cpp.

References datareader_dispatch(), OpenDDS::DCPS::make_id(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_gap_i(), update_last_recv_addr(), and OpenDDS::RTPS::GapSubmessage::writerId.

 {
   update_last_recv_addr(make_id(src_prefix, gap.writerId), remote_addr);
   datareader_dispatch(gap, src_prefix, directed, &RtpsReader::process_gap_i);
 }

◆ received() [3/6]

void OpenDDS::DCPS::RtpsUdpDataLink::received	(	const RTPS::HeartBeatSubmessage &	heartbeat,
		const GuidPrefix_t &	src_prefix,
		bool	directed,
		const NetworkAddress &	remote_addr
	)

Definition at line 1864 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, datareader_dispatch(), OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::DOES_NOT_EXIST, OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::EXISTS, interesting_writers_, OpenDDS::DCPS::make_id(), OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OPENDDS_VECTOR(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_i(), queue_submessages(), readers_lock_, update_last_recv_addr(), and OpenDDS::RTPS::HeartBeatSubmessage::writerId.

 {
   const GUID_t src = make_id(src_prefix, heartbeat.writerId);
   const MonotonicTimePoint now = MonotonicTimePoint::now();
 
   update_last_recv_addr(src, remote_addr, now);
 
   MetaSubmessageVec meta_submessages;
   OPENDDS_VECTOR(InterestingRemote) callbacks;
   {
     ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
 
     // We received a heartbeat from a writer.
     // We should ACKNACK if the writer is interesting and there is no association.
 
     for (InterestingRemoteMapType::iterator pos = interesting_writers_.lower_bound(src),
            limit = interesting_writers_.upper_bound(src);
          pos != limit;
          ++pos) {
       pos->second.last_activity = now;
       if (pos->second.status == InterestingRemote::DOES_NOT_EXIST) {
         callbacks.push_back(pos->second);
         pos->second.status = InterestingRemote::EXISTS;
       }
     }
   }
   queue_submessages(meta_submessages);
 
   for (size_t i = 0; i < callbacks.size(); ++i) {
     callbacks[i].listener->writer_exists(src, callbacks[i].localid);
   }
 
   datareader_dispatch(heartbeat, src_prefix, directed, &RtpsReader::process_heartbeat_i);
 }

◆ received() [4/6]

void OpenDDS::DCPS::RtpsUdpDataLink::received	(	const RTPS::HeartBeatFragSubmessage &	hb_frag,
		const GuidPrefix_t &	src_prefix,
		bool	directed,
		const NetworkAddress &	remote_addr
	)

Definition at line 2989 of file RtpsUdpDataLink.cpp.

References datareader_dispatch(), OpenDDS::DCPS::make_id(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_frag_i(), update_last_recv_addr(), and OpenDDS::RTPS::HeartBeatFragSubmessage::writerId.

 {
   update_last_recv_addr(make_id(src_prefix, hb_frag.writerId), remote_addr);
   datareader_dispatch(hb_frag, src_prefix, directed, &RtpsReader::process_heartbeat_frag_i);
 }

◆ received() [5/6]

void OpenDDS::DCPS::RtpsUdpDataLink::received	(	const RTPS::AckNackSubmessage &	acknack,
		const GuidPrefix_t &	src_prefix,
		const NetworkAddress &	remote_addr
	)

Definition at line 3061 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, datawriter_dispatch(), OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::DOES_NOT_EXIST, OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::EXISTS, interesting_readers_, local_prefix_, OpenDDS::DCPS::make_id(), OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OPENDDS_VECTOR(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acknack(), OpenDDS::RTPS::AckNackSubmessage::readerId, update_last_recv_addr(), OpenDDS::RTPS::AckNackSubmessage::writerId, and writers_lock_.

 {
   // local side is DW
   const GUID_t local = make_id(local_prefix_, acknack.writerId); // can't be ENTITYID_UNKNOWN
   const GUID_t remote = make_id(src_prefix, acknack.readerId);
   const MonotonicTimePoint now = MonotonicTimePoint::now();
 
   update_last_recv_addr(remote, remote_addr, now);
 
   OPENDDS_VECTOR(DiscoveryListener*) callbacks;
 
   {
     ACE_GUARD(ACE_Thread_Mutex, g, writers_lock_);
     for (InterestingRemoteMapType::iterator pos = interesting_readers_.lower_bound(remote),
            limit = interesting_readers_.upper_bound(remote);
          pos != limit;
          ++pos) {
       pos->second.last_activity = now;
       // Ensure the acknack was for the writer.
       if (local == pos->second.localid) {
         if (pos->second.status == InterestingRemote::DOES_NOT_EXIST) {
           callbacks.push_back(pos->second.listener);
           pos->second.status = InterestingRemote::EXISTS;
         }
       }
     }
   }
 
   for (size_t i = 0; i < callbacks.size(); ++i) {
     callbacks[i]->reader_exists(remote, local);
   }
 
   datawriter_dispatch(acknack, src_prefix, &RtpsWriter::process_acknack);
 }

◆ received() [6/6]

void OpenDDS::DCPS::RtpsUdpDataLink::received	(	const RTPS::NackFragSubmessage &	nackfrag,
		const GuidPrefix_t &	src_prefix,
		const NetworkAddress &	remote_addr
	)

Definition at line 3403 of file RtpsUdpDataLink.cpp.

References datawriter_dispatch(), OpenDDS::DCPS::make_id(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_nackfrag(), OpenDDS::RTPS::NackFragSubmessage::readerId, and update_last_recv_addr().

 {
   update_last_recv_addr(make_id(src_prefix, nackfrag.readerId), remote_addr);
   datawriter_dispatch(nackfrag, src_prefix, &RtpsWriter::process_nackfrag);
 }

◆ register_for_reader()

void OpenDDS::DCPS::RtpsUdpDataLink::register_for_reader	(	const GUID_t &	writerid,
		const GUID_t &	readerid,
		const AddrSet &	addresses,
		DiscoveryListener *	listener
	)

Definition at line 678 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, config(), OpenDDS::DCPS::RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC, OpenDDS::DCPS::GUID_t::entityId, heartbeat_, heartbeat_counts_, interesting_readers_, and writers_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpTransport::register_for_reader().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, writers_lock_);
   const bool enableheartbeat = interesting_readers_.empty();
   interesting_readers_.insert(
     InterestingRemoteMapType::value_type(
       readerid,
       InterestingRemote(writerid, addresses, listener)));
   if (heartbeat_counts_.find(writerid.entityId) == heartbeat_counts_.end()) {
     heartbeat_counts_[writerid.entityId] = 0;
   }
   g.release();
   if (enableheartbeat) {
     RtpsUdpInst_rch cfg = config();
     heartbeat_->enable(cfg ? cfg->heartbeat_period_ : TimeDuration(RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC));
   }
 }

◆ register_for_writer()

void OpenDDS::DCPS::RtpsUdpDataLink::register_for_writer	(	const GUID_t &	readerid,
		const GUID_t &	writerid,
		const AddrSet &	addresses,
		DiscoveryListener *	listener
	)

Definition at line 717 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, config(), OpenDDS::DCPS::RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC, heartbeatchecker_, interesting_writers_, and readers_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpTransport::register_for_writer().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
   bool enableheartbeatchecker = interesting_writers_.empty();
   interesting_writers_.insert(
     InterestingRemoteMapType::value_type(
       writerid,
       InterestingRemote(readerid, addresses, listener)));
   g.release();
   if (enableheartbeatchecker) {
     RtpsUdpInst_rch cfg = config();
     heartbeatchecker_->enable(cfg ? cfg->heartbeat_period_ : TimeDuration(RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC));
   }
 }

◆ release_reservations_i()

void OpenDDS::DCPS::RtpsUdpDataLink::release_reservations_i	(	const GUID_t &	remote_id,
		const GUID_t &	local_id
	)

privatevirtual

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 901 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::GuidConverter::isReader(), OpenDDS::DCPS::GuidConverter::isWriter(), readers_, readers_lock_, readers_of_writer_, remove_locator_and_bundling_cache(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::remove_writer(), writer_to_seq_best_effort_readers_, writers_, and writers_lock_.

Referenced by disassociated().

 {
   TqeVector to_drop;
   using std::pair;
   const GuidConverter conv(local_id);
   if (conv.isWriter()) {
     ACE_GUARD(ACE_Thread_Mutex, g, writers_lock_);
     RtpsWriterMap::iterator rw = writers_.find(local_id);
 
     if (rw != writers_.end()) {
       RtpsWriter_rch writer = rw->second;
       g.release();
       writer->remove_reader(remote_id);
       if (writer->reader_count() == 0) {
         writer->pre_stop_helper(to_drop, false);
       }
       writer->process_acked_by_all();
     }
 
   } else if (conv.isReader()) {
     ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
     RtpsReaderMap::iterator rr = readers_.find(local_id);
 
     if (rr != readers_.end()) {
       for (pair<RtpsReaderMultiMap::iterator, RtpsReaderMultiMap::iterator> iters =
              readers_of_writer_.equal_range(remote_id);
            iters.first != iters.second;) {
         if (iters.first->second->id() == local_id) {
           readers_of_writer_.erase(iters.first++);
         } else {
           ++iters.first;
         }
       }
 
       RtpsReader_rch reader = rr->second;
       g.release();
 
       reader->remove_writer(remote_id);
 
     } else {
       WriterToSeqReadersMap::iterator w = writer_to_seq_best_effort_readers_.find(remote_id);
       if (w != writer_to_seq_best_effort_readers_.end()) {
         RepoIdSet::iterator r = w->second.readers.find(local_id);
         if (r != w->second.readers.end()) {
           w->second.readers.erase(r);
           if (w->second.readers.empty()) {
             writer_to_seq_best_effort_readers_.erase(w);
           }
         }
       }
     }
   }
 
   remove_locator_and_bundling_cache(remote_id);
 
   for (TqeVector::iterator drop_it = to_drop.begin(); drop_it != to_drop.end(); ++drop_it) {
     (*drop_it)->data_dropped(true);
   }
 }

◆ remove_all_msgs()

void OpenDDS::DCPS::RtpsUdpDataLink::remove_all_msgs ( const GUID_t & pub_id )

virtual

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 165 of file RtpsUdpDataLink.cpp.

References ACE_Guard< ACE_LOCK >::release(), writers_, and writers_lock_.

 {
   ACE_Guard<ACE_Thread_Mutex> g(writers_lock_);
   RtpsWriter_rch writer;
   RtpsWriterMap::iterator iter = writers_.find(pub_id);
   if (iter != writers_.end()) {
     writer = iter->second;
   }
 
   g.release();
 
   if (writer) {
     writer->remove_all_msgs();
   }
 }

◆ remove_locator_and_bundling_cache()

void OpenDDS::DCPS::RtpsUdpDataLink::remove_locator_and_bundling_cache ( const GUID_t & remote_id )

Definition at line 459 of file RtpsUdpDataLink.cpp.

References bundling_cache_, locator_cache_, and OpenDDS::DCPS::AddressCache< Key >::remove_id().

Referenced by disassociated(), release_reservations_i(), update_last_recv_addr(), and update_locators().

 {
   locator_cache_.remove_id(remote_id);
   bundling_cache_.remove_id(remote_id);
 }

◆ remove_sample()

RemoveResult OpenDDS::DCPS::RtpsUdpDataLink::remove_sample ( const DataSampleElement * sample )

virtual

This method is essentially an "undo_send()" method. It's goal is to remove all traces of the sample from this DataLink (if the sample is even known to the DataLink).

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 146 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::DataSampleElement::get_pub_id(), ACE_Guard< ACE_LOCK >::release(), OpenDDS::DCPS::REMOVE_NOT_FOUND, writers_, and writers_lock_.

 {
   GUID_t pub_id = sample->get_pub_id();
 
   ACE_Guard<ACE_Thread_Mutex> g(writers_lock_);
   RtpsWriter_rch writer;
   RtpsWriterMap::iterator iter = writers_.find(pub_id);
   if (iter != writers_.end()) {
     writer = iter->second;
   }
 
   g.release();
 
   if (writer) {
     return writer->remove_sample(sample);
   }
   return REMOVE_NOT_FOUND;
 }

◆ requires_inline_qos()

bool OpenDDS::DCPS::RtpsUdpDataLink::requires_inline_qos ( const GUIDSeq_var & peers )

Definition at line 1426 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, force_inline_qos_, locators_, and locators_lock_.

Referenced by customize_queue_element(), OpenDDS::DCPS::RtpsUdpTransport::get_last_recv_locator(), and update_locators().

 {
   if (force_inline_qos_) {
     // Force true for testing purposes
     return true;
   } else {
     if (!peers.ptr()) {
       return false;
     }
     ACE_GUARD_RETURN(ACE_Thread_Mutex, g, locators_lock_, false);
     for (CORBA::ULong i = 0; i < peers->length(); ++i) {
       const RemoteInfoMap::const_iterator iter = locators_.find(peers[i]);
       if (iter != locators_.end() && iter->second.requires_inline_qos_) {
         return true;
       }
     }
     return false;
   }
 }

◆ security_config()

Security::SecurityConfig_rch OpenDDS::DCPS::RtpsUdpDataLink::security_config ( ) const

inline

Definition at line 223 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::decode_payload(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_from_secure(), OpenDDS::DCPS::RtpsUdpSendStrategy::encode_payload(), OpenDDS::DCPS::RtpsUdpSendStrategy::pre_send_packet(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::receive_bytes(), OpenDDS::DCPS::RtpsUdpSendStrategy::security_config(), and OpenDDS::DCPS::RtpsUdpSendStrategy::send_rtps_control().

   {
     ACE_Guard<ACE_Thread_Mutex> guard(security_mutex_);
     return security_config_;
   }

◆ send_heartbeats()

void OpenDDS::DCPS::RtpsUdpDataLink::send_heartbeats ( const MonotonicTimePoint & now )

private

Definition at line 4079 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, config(), OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::DOES_NOT_EXIST, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::EXISTS, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::HEARTBEAT, heartbeat_counts_, OpenDDS::RTPS::Submessage::heartbeat_sm, OpenDDS::RTPS::HEARTBEAT_SZ, OpenDDS::DCPS::insert(), interesting_readers_, OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::listener, OpenDDS::DCPS::RtpsUdpDataLink::InterestingRemote::localid, OPENDDS_MAP_CMP(), OPENDDS_VECTOR(), queue_submessages(), OpenDDS::DCPS::DiscoveryListener::reader_does_not_exist(), OpenDDS::DCPS::MetaSubmessage::sm_, OpenDDS::RTPS::to_rtps_seqnum(), writers_, writers_lock_, and OpenDDS::DCPS::SequenceNumber::ZERO().

 {
   OPENDDS_VECTOR(CallbackType) readerDoesNotExistCallbacks;
 
   RtpsUdpInst_rch cfg = config();
 
   MetaSubmessageVec meta_submessages;
 
   {
     ACE_GUARD(ACE_Thread_Mutex, g, writers_lock_);
 
     const MonotonicTimePoint tv = now - 10 * cfg->heartbeat_period_;
     const MonotonicTimePoint tv3 = now - 3 * cfg->heartbeat_period_;
 
     typedef OPENDDS_MAP_CMP(GUID_t, RcHandle<ConstSharedRepoIdSet>, GUID_tKeyLessThan) WtaMap;
     WtaMap writers_to_advertise;
 
     for (InterestingRemoteMapType::iterator pos = interesting_readers_.begin(),
            limit = interesting_readers_.end();
          pos != limit;
          ++pos) {
       if (pos->second.status == InterestingRemote::DOES_NOT_EXIST ||
           (pos->second.status == InterestingRemote::EXISTS && pos->second.last_activity < tv3)) {
         RcHandle<ConstSharedRepoIdSet>& tg = writers_to_advertise[pos->second.localid];
         if (!tg) {
           tg = make_rch<ConstSharedRepoIdSet>();
         }
         const_cast<RepoIdSet&>(tg->guids_).insert(pos->first);
       }
       if (pos->second.status == InterestingRemote::EXISTS && pos->second.last_activity < tv) {
         CallbackType callback(pos->first, pos->second);
         readerDoesNotExistCallbacks.push_back(callback);
         pos->second.status = InterestingRemote::DOES_NOT_EXIST;
       }
     }
 
     for (WtaMap::const_iterator pos = writers_to_advertise.begin(),
            limit = writers_to_advertise.end();
          pos != limit;
          ++pos) {
       RtpsWriterMap::const_iterator wpos = writers_.find(pos->first);
       if (wpos != writers_.end()) {
         wpos->second->gather_heartbeats(pos->second, meta_submessages);
       } else {
         using namespace OpenDDS::RTPS;
         const int count = ++heartbeat_counts_[pos->first.entityId];
         const HeartBeatSubmessage hb = {
           {HEARTBEAT, FLAG_E, HEARTBEAT_SZ},
           ENTITYID_UNKNOWN, // any matched reader may be interested in this
           pos->first.entityId,
           to_rtps_seqnum(SequenceNumber(1)),
           to_rtps_seqnum(SequenceNumber::ZERO()),
           {count}
         };
 
         for (RepoIdSet::const_iterator it = pos->second->guids_.begin(),
           limit = pos->second->guids_.end(); it != limit; ++it) {
 
           MetaSubmessage meta_submessage(pos->first, *it);
           meta_submessage.sm_.heartbeat_sm(hb);
           meta_submessages.push_back(meta_submessage);
         }
       }
     }
   }
 
   queue_submessages(meta_submessages);
 
   for (OPENDDS_VECTOR(CallbackType)::iterator iter = readerDoesNotExistCallbacks.begin();
        iter != readerDoesNotExistCallbacks.end(); ++iter) {
     const InterestingRemote& remote = iter->second;
     remote.listener->reader_does_not_exist(iter->first, remote.localid);
   }
 }

◆ send_heartbeats_manual_i()

void OpenDDS::DCPS::RtpsUdpDataLink::send_heartbeats_manual_i	(	const TransportSendControlElement *	tsce,
		MetaSubmessageVec &	meta_submessages
	)

private

Definition at line 4357 of file RtpsUdpDataLink.cpp.

References best_effort_heartbeat_count_, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::FLAG_F, OpenDDS::RTPS::FLAG_L, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::RTPS::HEARTBEAT, OpenDDS::RTPS::Submessage::heartbeat_sm, OpenDDS::RTPS::HEARTBEAT_SZ, OpenDDS::DCPS::TransportSendControlElement::publication_id(), OpenDDS::DCPS::TransportSendControlElement::sequence(), OpenDDS::DCPS::MetaSubmessage::sm_, and OpenDDS::RTPS::to_rtps_seqnum().

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

 {
   using namespace OpenDDS::RTPS;
 
   const GUID_t pub_id = tsce->publication_id();
   const HeartBeatSubmessage hb = {
     {HEARTBEAT,
      CORBA::Octet(FLAG_E | FLAG_F | FLAG_L),
      HEARTBEAT_SZ},
     ENTITYID_UNKNOWN, // any matched reader may be interested in this
     pub_id.entityId,
     // This liveliness heartbeat is from a best-effort Writer, the sequence numbers are not used
     to_rtps_seqnum(SequenceNumber(1)),
     to_rtps_seqnum(tsce->sequence()),
     {++best_effort_heartbeat_count_}
   };
 
   MetaSubmessage meta_submessage(pub_id, GUID_UNKNOWN);
   meta_submessage.sm_.heartbeat_sm(hb);
   meta_submessages.push_back(meta_submessage);
 }

◆ send_strategy()

RtpsUdpSendStrategy_rch OpenDDS::DCPS::RtpsUdpDataLink::send_strategy ( )

private

Definition at line 4581 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::dynamic_rchandle_cast(), and OpenDDS::DCPS::DataLink::send_strategy_.

Referenced by bundle_and_send_submessages(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::customize_queue_element_helper(), customize_queue_element_non_reliable_i(), durability_resend(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i(), open(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::RtpsWriter().

 {
   return dynamic_rchandle_cast<RtpsUdpSendStrategy>(send_strategy_);
 }

◆ separate_message()

bool OpenDDS::DCPS::RtpsUdpDataLink::separate_message ( EntityId_t entity )

static

Definition at line 2481 of file RtpsUdpDataLink.cpp.

References OpenDDS::RTPS::ENTITYID_P2P_BUILTIN_PARTICIPANT_STATELESS_READER, OpenDDS::RTPS::ENTITYID_P2P_BUILTIN_PARTICIPANT_STATELESS_WRITER, OpenDDS::RTPS::ENTITYID_P2P_BUILTIN_PARTICIPANT_VOLATILE_SECURE_READER, and OpenDDS::RTPS::ENTITYID_P2P_BUILTIN_PARTICIPANT_VOLATILE_SECURE_WRITER.

Referenced by build_meta_submessage_map(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::check_encoded().

 {
   // submessages generated by these entities may not be combined
   // with other submessages when using full-message protection
   // DDS Security v1.1 8.4.2.4 Table 27 is_rtps_protected
   using namespace RTPS;
   return entity == ENTITYID_P2P_BUILTIN_PARTICIPANT_STATELESS_WRITER
     || entity == ENTITYID_P2P_BUILTIN_PARTICIPANT_STATELESS_READER
     || entity == ENTITYID_P2P_BUILTIN_PARTICIPANT_VOLATILE_SECURE_WRITER
     || entity == ENTITYID_P2P_BUILTIN_PARTICIPANT_VOLATILE_SECURE_READER;
 }

◆ stop_i()

void OpenDDS::DCPS::RtpsUdpDataLink::stop_i ( )

privatevirtual

This announces the "stop" event to our subclass. The "stop" event will occur when this DataLink is handling a release_reservations() call and determines that it has just released all of the remaining reservations on this DataLink. The "stop" event will also occur when the TransportImpl is being shutdown() - we call stop_i() from our transport_shutdown() method to handle this case.

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 963 of file RtpsUdpDataLink.cpp.

References ACE_SOCK::close(), heartbeat_, heartbeatchecker_, multicast_socket_, network_interface_address_reader_, TheServiceParticipant, and unicast_socket_.

Referenced by open().

 {
   TheServiceParticipant->network_interface_address_topic()->disconnect(network_interface_address_reader_);
 
   heartbeat_->disable();
   heartbeatchecker_->disable();
   unicast_socket_.close();
   multicast_socket_.close();
 #ifdef ACE_HAS_IPV6
   ipv6_unicast_socket_.close();
   ipv6_multicast_socket_.close();
 #endif
 }

◆ submsgs_to_msgblock()

ACE_Message_Block * OpenDDS::DCPS::RtpsUdpDataLink::submsgs_to_msgblock ( const RTPS::SubmessageSeq & subm )

private

Definition at line 1056 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::Encoding::align(), OpenDDS::DCPS::Serializer::align_w(), alloc_msgblock(), custom_allocator_, OpenDDS::STUN::encoding(), OpenDDS::DCPS::Encoding::KIND_XCDR1, OpenDDS::DCPS::serialized_size(), and OpenDDS::RTPS::SMHDR_SZ.

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

 {
   // byte swapping is handled in the operator<<() implementation
   const Encoding encoding(Encoding::KIND_XCDR1);
   size_t size = 0;
   for (CORBA::ULong i = 0; i < subm.length(); ++i) {
     encoding.align(size, RTPS::SMHDR_SZ);
     serialized_size(encoding, size, subm[i]);
   }
 
   ACE_Message_Block* hdr = alloc_msgblock(size, &custom_allocator_);
 
   Serializer ser(hdr, encoding);
   for (CORBA::ULong i = 0; i < subm.length(); ++i) {
     ser << subm[i];
     ser.align_w(RTPS::SMHDR_SZ);
   }
   return hdr;
 }

◆ transport()

RtpsUdpTransport_rch OpenDDS::DCPS::RtpsUdpDataLink::transport ( void )

Definition at line 4575 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::dynamic_rchandle_cast(), and OpenDDS::DCPS::DataLink::impl().

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::receive_bytes(), and OpenDDS::DCPS::RtpsUdpSendStrategy::send_single_i().

 {
   return dynamic_rchandle_cast<RtpsUdpTransport>(impl());
 }

◆ unicast_socket()

ACE_INLINE ACE_SOCK_Dgram & OpenDDS::DCPS::RtpsUdpDataLink::unicast_socket ( )

Definition at line 37 of file RtpsUdpDataLink.inl.

References ACE_INLINE, and unicast_socket_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::choose_recv_socket(), OpenDDS::DCPS::RtpsUdpSendStrategy::choose_send_socket(), open(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::start_i(), and OpenDDS::DCPS::RtpsUdpReceiveStrategy::stop_i().

 {
   return unicast_socket_;
 }

◆ unregister_for_reader()

void OpenDDS::DCPS::RtpsUdpDataLink::unregister_for_reader	(	const GUID_t &	writerid,
		const GUID_t &	readerid
	)

Definition at line 700 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, interesting_readers_, and writers_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpTransport::unregister_for_reader().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, writers_lock_);
   for (InterestingRemoteMapType::iterator pos = interesting_readers_.lower_bound(readerid),
          limit = interesting_readers_.upper_bound(readerid);
        pos != limit;
        ) {
     if (pos->second.localid == writerid) {
       interesting_readers_.erase(pos++);
     } else {
       ++pos;
     }
   }
 }

◆ unregister_for_writer()

void OpenDDS::DCPS::RtpsUdpDataLink::unregister_for_writer	(	const GUID_t &	readerid,
		const GUID_t &	writerid
	)

Definition at line 736 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, interesting_writers_, and readers_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpTransport::unregister_for_writer().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, readers_lock_);
   for (InterestingRemoteMapType::iterator pos = interesting_writers_.lower_bound(writerid),
          limit = interesting_writers_.upper_bound(writerid);
        pos != limit;
        ) {
     if (pos->second.localid == readerid) {
       interesting_writers_.erase(pos++);
     } else {
       ++pos;
     }
   }
 }

◆ update_last_recv_addr()

void OpenDDS::DCPS::RtpsUdpDataLink::update_last_recv_addr	(	const GUID_t &	src,
		const NetworkAddress &	addr,
		const MonotonicTimePoint &	now = `MonotonicTimePoint::now()`
	)

private

Definition at line 1531 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, config(), OpenDDS::DCPS::is_more_local(), locators_, locators_lock_, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), and remove_locator_and_bundling_cache().

Referenced by associated(), and received().

 {
   RtpsUdpInst_rch cfg = config();
 
   if (!cfg) {
     return;
   }
 
   if (addr == cfg->rtps_relay_address()) {
     return;
   }
 
   bool remove_cache = false;
   {
     ACE_GUARD(ACE_Thread_Mutex, g, locators_lock_);
     const RemoteInfoMap::iterator pos = locators_.find(src);
     if (pos != locators_.end()) {
       const bool expired = cfg->receive_address_duration_ < (MonotonicTimePoint::now() - pos->second.last_recv_time_);
       const bool allow_update = expired ||
                                 pos->second.last_recv_addr_ == addr ||
                                 is_more_local(pos->second.last_recv_addr_, addr);
       if (allow_update) {
         remove_cache = pos->second.last_recv_addr_ != addr;
         pos->second.last_recv_addr_ = addr;
         pos->second.last_recv_time_ = now;
       }
     }
   }
   if (remove_cache) {
     remove_locator_and_bundling_cache(src);
   }
 }

◆ update_locators()

void OpenDDS::DCPS::RtpsUdpDataLink::update_locators	(	const GUID_t &	remote_id,
		AddrSet &	unicast_addresses,
		AddrSet &	multicast_addresses,
		bool	requires_inline_qos,
		bool	add_ref
	)

Definition at line 480 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::LogAddr::c_str(), OpenDDS::DCPS::DCPS_debug_level, LM_ERROR, LM_INFO, locators_, locators_lock_, OpenDDS::DCPS::RtpsUdpDataLink::RemoteInfo::multicast_addrs_, OpenDDS::DCPS::RtpsUdpDataLink::RemoteInfo::ref_count_, remove_locator_and_bundling_cache(), requires_inline_qos(), OpenDDS::DCPS::RtpsUdpDataLink::RemoteInfo::requires_inline_qos_, and OpenDDS::DCPS::RtpsUdpDataLink::RemoteInfo::unicast_addrs_.

Referenced by associated(), and OpenDDS::DCPS::RtpsUdpTransport::update_locators().

 {
   if (unicast_addresses.empty() && multicast_addresses.empty()) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: RtpsUdpDataLink::update_locators: no addresses for %C\n"), LogGuid(remote_id).c_str()));
     }
   }
 
   remove_locator_and_bundling_cache(remote_id);
 
   ACE_GUARD(ACE_Thread_Mutex, g, locators_lock_);
 
   RemoteInfo& info = locators_[remote_id];
   const bool log_unicast_change = DCPS_debug_level > 3 && info.unicast_addrs_ != unicast_addresses;
   const bool log_multicast_change = DCPS_debug_level > 3 && info.multicast_addrs_ != multicast_addresses;
   info.unicast_addrs_.swap(unicast_addresses);
   info.multicast_addrs_.swap(multicast_addresses);
   info.requires_inline_qos_ = requires_inline_qos;
   if (add_ref) {
     ++info.ref_count_;
   }
 
   g.release();
 
   if (log_unicast_change) {
     for (AddrSet::const_iterator pos = unicast_addresses.begin(), limit = unicast_addresses.end();
          pos != limit; ++pos) {
       ACE_DEBUG((LM_INFO, ACE_TEXT("(%P|%t) RtpsUdpDataLink::update_locators %C is now at %C\n"),
                  LogGuid(remote_id).c_str(), LogAddr(*pos).c_str()));
     }
   }
   if (log_multicast_change) {
     for (AddrSet::const_iterator pos = multicast_addresses.begin(), limit = multicast_addresses.end();
          pos != limit; ++pos) {
       ACE_DEBUG((LM_INFO, ACE_TEXT("(%P|%t) RtpsUdpDataLink::update_locators %C is now at %C\n"),
                  LogGuid(remote_id).c_str(), LogAddr(*pos).c_str()));
     }
   }
 }

◆ update_required_acknack_count()

void OpenDDS::DCPS::RtpsUdpDataLink::update_required_acknack_count	(	const GUID_t &	local_id,
		const GUID_t &	remote_id,
		CORBA::Long	current
	)

private

Definition at line 2741 of file RtpsUdpDataLink.cpp.

References writers_, and writers_lock_.

Referenced by bundle_and_send_submessages().

 {
   RtpsWriter_rch writer;
   {
     ACE_Guard<ACE_Thread_Mutex> guard(writers_lock_);
     RtpsWriterMap::iterator rw = writers_.find(local_id);
     if (rw != writers_.end()) {
       writer = rw->second;
     }
   }
   if (writer) {
     writer->update_required_acknack_count(remote_id, current);
   }
 }

Friends And Related Function Documentation

◆ ::DDS_TEST

friend class ::DDS_TEST

friend

Definition at line 266 of file RtpsUdpDataLink.h.

Member Data Documentation

◆ best_effort_heartbeat_count_

CORBA::Long OpenDDS::DCPS::RtpsUdpDataLink::best_effort_heartbeat_count_

private

Definition at line 860 of file RtpsUdpDataLink.h.

Referenced by send_heartbeats_manual_i().

◆ bundle_allocator_

Dynamic_Cached_Allocator_With_Overflow<ACE_Thread_Mutex> OpenDDS::DCPS::RtpsUdpDataLink::bundle_allocator_

private

Definition at line 314 of file RtpsUdpDataLink.h.

Referenced by bundle_and_send_submessages().

◆ bundling_cache_

BundlingCache OpenDDS::DCPS::RtpsUdpDataLink::bundling_cache_

mutableprivate

Definition at line 302 of file RtpsUdpDataLink.h.

Referenced by build_meta_submessage_map(), remove_locator_and_bundling_cache(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::update_remote_guids_cache_i().

◆ custom_allocator_

Dynamic_Cached_Allocator_With_Overflow<ACE_Thread_Mutex> OpenDDS::DCPS::RtpsUdpDataLink::custom_allocator_

private

Definition at line 313 of file RtpsUdpDataLink.h.

Referenced by submsgs_to_msgblock().

◆ db_allocator_

DataBlockAllocator OpenDDS::DCPS::RtpsUdpDataLink::db_allocator_

private

Definition at line 312 of file RtpsUdpDataLink.h.

Referenced by alloc_msgblock().

◆ db_lock_pool_

unique_ptr<DataBlockLockPool> OpenDDS::DCPS::RtpsUdpDataLink::db_lock_pool_

private

Definition at line 315 of file RtpsUdpDataLink.h.

Referenced by alloc_msgblock().

◆ event_dispatcher_

EventDispatcher_rch OpenDDS::DCPS::RtpsUdpDataLink::event_dispatcher_

private

Definition at line 272 of file RtpsUdpDataLink.h.

◆ flush_send_queue_sporadic_

RcHandle<SporadicEvent> OpenDDS::DCPS::RtpsUdpDataLink::flush_send_queue_sporadic_

private

Definition at line 743 of file RtpsUdpDataLink.h.

Referenced by disable_response_queue(), and ~RtpsUdpDataLink().

◆ force_inline_qos_

bool OpenDDS::DCPS::RtpsUdpDataLink::force_inline_qos_ = false

staticprivate

static member used by testing code to force inline qos

Definition at line 268 of file RtpsUdpDataLink.h.

Referenced by requires_inline_qos().

◆ fsq_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpDataLink::fsq_mutex_

mutableprivate

Definition at line 735 of file RtpsUdpDataLink.h.

Referenced by disable_response_queue(), and flush_send_queue().

◆ fsq_vec_size_

size_t OpenDDS::DCPS::RtpsUdpDataLink::fsq_vec_size_

private

Definition at line 737 of file RtpsUdpDataLink.h.

Referenced by disable_response_queue(), and flush_send_queue_i().

◆ handle_registry_

Security::HandleRegistry_rch OpenDDS::DCPS::RtpsUdpDataLink::handle_registry_

private

Definition at line 933 of file RtpsUdpDataLink.h.

Referenced by RtpsUdpDataLink().

◆ harvest_send_queue_sporadic_

RcHandle<SporadicEvent> OpenDDS::DCPS::RtpsUdpDataLink::harvest_send_queue_sporadic_

private

Definition at line 740 of file RtpsUdpDataLink.h.

Referenced by queue_submessages(), and ~RtpsUdpDataLink().

◆ heartbeat_

RcHandle<PeriodicEvent> OpenDDS::DCPS::RtpsUdpDataLink::heartbeat_

private

Definition at line 862 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::add_reader(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::make_lagger_leader(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::make_leader_lagger(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::pre_stop_helper(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acknack(), register_for_reader(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::send_heartbeats(), and stop_i().

◆ heartbeat_counts_

CountMapType OpenDDS::DCPS::RtpsUdpDataLink::heartbeat_counts_

private

Definition at line 905 of file RtpsUdpDataLink.h.

Referenced by associated(), pre_stop_i(), register_for_reader(), and send_heartbeats().

◆ heartbeatchecker_

RcHandle<PeriodicEvent> OpenDDS::DCPS::RtpsUdpDataLink::heartbeatchecker_

private

Definition at line 863 of file RtpsUdpDataLink.h.

Referenced by register_for_writer(), and stop_i().

◆ ice_agent_

RcHandle<ICE::Agent> OpenDDS::DCPS::RtpsUdpDataLink::ice_agent_

private

Definition at line 935 of file RtpsUdpDataLink.h.

Referenced by accumulate_addresses(), and get_ice_agent().

◆ interesting_readers_

InterestingRemoteMapType OpenDDS::DCPS::RtpsUdpDataLink::interesting_readers_

private

Definition at line 890 of file RtpsUdpDataLink.h.

Referenced by accumulate_addresses(), received(), register_for_reader(), send_heartbeats(), and unregister_for_reader().

◆ interesting_writers_

InterestingRemoteMapType OpenDDS::DCPS::RtpsUdpDataLink::interesting_writers_

private

Definition at line 891 of file RtpsUdpDataLink.h.

Referenced by accumulate_addresses(), check_heartbeats(), received(), register_for_writer(), and unregister_for_writer().

◆ job_queue_

RcHandle<JobQueue> OpenDDS::DCPS::RtpsUdpDataLink::job_queue_

private

Definition at line 271 of file RtpsUdpDataLink.h.

Referenced by RtpsUdpDataLink().

◆ local_crypto_handle_

DDS::Security::ParticipantCryptoHandle OpenDDS::DCPS::RtpsUdpDataLink::local_crypto_handle_

private

Definition at line 934 of file RtpsUdpDataLink.h.

Referenced by local_crypto_handle().

◆ local_prefix_

GuidPrefix_t OpenDDS::DCPS::RtpsUdpDataLink::local_prefix_

private

Definition at line 277 of file RtpsUdpDataLink.h.

Referenced by received(), and RtpsUdpDataLink().

◆ locator_cache_

LocatorCache OpenDDS::DCPS::RtpsUdpDataLink::locator_cache_

mutableprivate

Definition at line 301 of file RtpsUdpDataLink.h.

Referenced by accumulate_addresses(), and remove_locator_and_bundling_cache().

◆ locators_

RemoteInfoMap OpenDDS::DCPS::RtpsUdpDataLink::locators_

private

Definition at line 297 of file RtpsUdpDataLink.h.

Referenced by accumulate_addresses(), disassociated(), get_last_recv_address(), requires_inline_qos(), update_last_recv_addr(), and update_locators().

◆ locators_lock_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpDataLink::locators_lock_

mutableprivate

Definition at line 769 of file RtpsUdpDataLink.h.

Referenced by build_meta_submessage_map(), disassociated(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i(), get_addresses(), get_last_recv_address(), requires_inline_qos(), update_last_recv_addr(), and update_locators().

◆ max_bundle_size_

const size_t OpenDDS::DCPS::RtpsUdpDataLink::max_bundle_size_

private

Definition at line 907 of file RtpsUdpDataLink.h.

Referenced by bundle_mapped_meta_submessages().

◆ mb_allocator_

MessageBlockAllocator OpenDDS::DCPS::RtpsUdpDataLink::mb_allocator_

private

Definition at line 311 of file RtpsUdpDataLink.h.

Referenced by alloc_msgblock().

◆ multi_buff_

OpenDDS::DCPS::RtpsUdpDataLink::MultiSendBuffer OpenDDS::DCPS::RtpsUdpDataLink::multi_buff_

private

Referenced by associated(), and open().

◆ multicast_manager_

MulticastManager OpenDDS::DCPS::RtpsUdpDataLink::multicast_manager_

private

Definition at line 941 of file RtpsUdpDataLink.h.

Referenced by on_data_available().

◆ multicast_socket_

ACE_SOCK_Dgram_Mcast OpenDDS::DCPS::RtpsUdpDataLink::multicast_socket_

private

Definition at line 305 of file RtpsUdpDataLink.h.

Referenced by multicast_socket(), on_data_available(), open(), and stop_i().

◆ network_interface_address_reader_

RcHandle<InternalDataReader<NetworkInterfaceAddress> > OpenDDS::DCPS::RtpsUdpDataLink::network_interface_address_reader_

private

Definition at line 940 of file RtpsUdpDataLink.h.

Referenced by on_data_available(), open(), and stop_i().

◆ pending_reliable_readers_

RepoIdSet OpenDDS::DCPS::RtpsUdpDataLink::pending_reliable_readers_

private

Definition at line 745 of file RtpsUdpDataLink.h.

Referenced by associated(), make_reservation(), and received().

◆ reactor_task_

ReactorTask_rch OpenDDS::DCPS::RtpsUdpDataLink::reactor_task_

private

Definition at line 270 of file RtpsUdpDataLink.h.

Referenced by get_reactor(), get_reactor_interceptor(), and reactor_is_shut_down().

◆ readers_

RtpsReaderMap OpenDDS::DCPS::RtpsUdpDataLink::readers_

private

Definition at line 752 of file RtpsUdpDataLink.h.

Referenced by associated(), client_stop(), make_reservation(), pre_stop_i(), received(), and release_reservations_i().

◆ readers_lock_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpDataLink::readers_lock_

mutableprivate

What was once a single lock for the whole datalink is now split between three (four including ch_lock_):

readers_lock_ protects readers_, readers_of_writer_, pending_reliable_readers_, interesting_writers_, and writer_to_seq_best_effort_readers_ along with anything else that fits the 'reader side activity' of the datalink
writers_lock_ protects writers_, heartbeat_counts_ best_effort_heartbeat_count_, and interesting_readers_ along with anything else that fits the 'writers side activity' of the datalink
locators_lock_ protects locators_ (and therefore calls to get_addresses_i()) for both remote writers and remote readers
send_queues_lock protects thread_send_queues_

Definition at line 767 of file RtpsUdpDataLink.h.

Referenced by accumulate_addresses(), associated(), check_heartbeats(), client_stop(), filterBestEffortReaders(), make_reservation(), pre_stop_i(), received(), register_for_writer(), release_reservations_i(), and unregister_for_writer().

◆ readers_of_writer_

RtpsReaderMultiMap OpenDDS::DCPS::RtpsUdpDataLink::readers_of_writer_

private

Definition at line 755 of file RtpsUdpDataLink.h.

Referenced by associated(), client_stop(), received(), and release_reservations_i().

◆ security_config_

Security::SecurityConfig_rch OpenDDS::DCPS::RtpsUdpDataLink::security_config_

private

Definition at line 932 of file RtpsUdpDataLink.h.

Referenced by RtpsUdpDataLink().

◆ security_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpDataLink::security_mutex_

mutableprivate

Definition at line 931 of file RtpsUdpDataLink.h.

Referenced by local_crypto_handle().

◆ sq_

TransactionalRtpsSendQueue OpenDDS::DCPS::RtpsUdpDataLink::sq_

private

Definition at line 734 of file RtpsUdpDataLink.h.

Referenced by associated(), client_stop(), disable_response_queue(), disassociated(), enable_response_queue(), harvest_send_queue(), and queue_submessages().

◆ transport_statistics_

InternalTransportStatistics& OpenDDS::DCPS::RtpsUdpDataLink::transport_statistics_

private

Definition at line 908 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_frag_i(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_i().

◆ transport_statistics_mutex_

ACE_Thread_Mutex& OpenDDS::DCPS::RtpsUdpDataLink::transport_statistics_mutex_

private

Definition at line 909 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::gather_nack_replies_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_frag_i(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_i().

◆ unicast_socket_

ACE_SOCK_Dgram OpenDDS::DCPS::RtpsUdpDataLink::unicast_socket_

private

Definition at line 304 of file RtpsUdpDataLink.h.

Referenced by open(), stop_i(), and unicast_socket().

◆ writer_to_seq_best_effort_readers_

WriterToSeqReadersMap OpenDDS::DCPS::RtpsUdpDataLink::writer_to_seq_best_effort_readers_

private

Definition at line 757 of file RtpsUdpDataLink.h.

Referenced by associated(), client_stop(), filterBestEffortReaders(), and release_reservations_i().

◆ writers_

RtpsWriterMap OpenDDS::DCPS::RtpsUdpDataLink::writers_

private

Definition at line 583 of file RtpsUdpDataLink.h.

Referenced by add_delayed_notification(), associated(), client_stop(), customize_queue_element(), get_addresses_i(), get_writer_send_buffer(), is_leading(), pre_stop_i(), release_reservations_i(), remove_all_msgs(), remove_sample(), send_heartbeats(), and update_required_acknack_count().

◆ writers_lock_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpDataLink::writers_lock_

mutableprivate

Definition at line 768 of file RtpsUdpDataLink.h.

Referenced by accumulate_addresses(), add_delayed_notification(), associated(), client_stop(), customize_queue_element(), get_addresses_i(), get_writer_send_buffer(), is_leading(), pre_stop_i(), received(), register_for_reader(), release_reservations_i(), remove_all_msgs(), remove_sample(), send_heartbeats(), unregister_for_reader(), and update_required_acknack_count().

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

Classes

Public Member Functions

Static Public Member Functions

Private Types

Private Member Functions

Static Private Member Functions

Private Attributes

Static Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ CallbackType

◆ FragmentNumberValue

◆ ReaderInfo_rch

◆ ReaderInfoSetHolder_rch

◆ RtpsReader_rch

◆ RtpsWriter_rch

◆ WriterInfo_rch

Constructor & Destructor Documentation

◆ RtpsUdpDataLink()

◆ ~RtpsUdpDataLink()

Member Function Documentation

◆ accumulate_addresses()

◆ add_delayed_notification()

◆ alloc_msgblock()

◆ associated()

◆ build_meta_submessage_map()

◆ bundle_and_send_submessages()

◆ bundle_mapped_meta_submessages()

◆ check_heartbeats()

◆ client_stop()

◆ config()

◆ customize_queue_element()

◆ customize_queue_element_non_reliable_i()

◆ datareader_dispatch()

◆ datawriter_dispatch()

◆ disable_response_queue()

◆ disassociated()

◆ durability_resend() [1/2]

◆ durability_resend() [2/2]

◆ enable_response_queue()

◆ event_dispatcher()

◆ extend_bitmap_range()

◆ filterBestEffortReaders()

◆ flush_send_queue()

◆ flush_send_queue_i()

◆ get_addresses() [1/2]

◆ get_addresses() [2/2]

◆ get_addresses_i() [1/2]

◆ get_addresses_i() [2/2]

◆ get_ice_agent()

◆ get_ice_endpoint()

◆ get_job_queue()

◆ get_last_recv_address()

◆ get_reactor()

◆ get_reactor_interceptor()

◆ get_writer_send_buffer()

◆ handle_registry()

◆ harvest_send_queue()

◆ include_fragment()

◆ is_leading()

◆ local_crypto_handle() [1/2]

◆ local_crypto_handle() [2/2]

◆ local_prefix()

◆ make_reservation()

◆ multicast_socket()

◆ on_data_available()

◆ open()

◆ OPENDDS_MAP() [1/5]

◆ OPENDDS_MAP() [2/5]

◆ OPENDDS_MAP() [3/5]

◆ OPENDDS_MAP() [4/5]

◆ OPENDDS_MAP() [5/5]

◆ OPENDDS_MAP_CMP() [1/9]

◆ OPENDDS_MAP_CMP() [2/9]

◆ OPENDDS_MAP_CMP() [3/9]

◆ OPENDDS_MAP_CMP() [4/9]

◆ OPENDDS_MAP_CMP() [5/9]

◆ OPENDDS_MAP_CMP() [6/9]