#include <RtpsUdpTransport.h>

Inheritance diagram for OpenDDS::DCPS::RtpsUdpTransport:

Collaboration diagram for OpenDDS::DCPS::RtpsUdpTransport:

[legend]

Classes
struct	IceEndpoint

Public Member Functions
	RtpsUdpTransport (const RtpsUdpInst_rch &inst)

RtpsUdpInst_rch	config () const

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

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

virtual void	rtps_relay_only_now (bool flag)

virtual void	use_rtps_relay_now (bool flag)

virtual void	use_ice_now (bool flag)

ICE::ServerReflexiveStateMachine &	relay_srsm ()

void	process_relay_sra (ICE::ServerReflexiveStateMachine::StateChange)

void	disable_relay_stun_task ()

virtual void	update_locators (const GUID_t &, const TransportLocatorSeq &)

virtual void	get_last_recv_locator (const GUID_t &, TransportLocator &)

void	rtps_relay_address_change ()

void	append_transport_statistics (TransportStatisticsSequence &seq)

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

virtual void	unbind_link (DataLink *link)
	Remove any pending_release mappings. More...

bool	release_link_resources (DataLink *link)

TransportInst_rch	config () const

ACE_Reactor_Timer_Interface *	timer () const
	Interface to the transport's reactor for scheduling timers. More...

ACE_Reactor *	reactor () const

ACE_thread_t	reactor_owner () const

bool	is_shut_down () const

void	create_reactor_task (bool useAsyncSend=false, const OPENDDS_STRING &name="")

void	dump ()
	Diagnostic aid. More...

OPENDDS_STRING	dump_to_str ()

void	report ()

ReactorTask_rch	reactor_task ()

EventDispatcher_rch	event_dispatcher ()

int	acquire ()

int	tryacquire ()

int	release ()

int	remove ()

bool	connection_info (TransportLocator &local_info, ConnectionInfoFlags flags) const

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

Private Types
typedef ACE_Thread_Mutex	ThreadLockType

typedef ACE_Guard< ThreadLockType >	GuardThreadType

typedef ACE_SYNCH_MUTEX	LockType
	This protects the connections_ data member. More...

typedef ACE_Guard< LockType >	GuardType

typedef PmfSporadicTask< RtpsUdpTransport >	Sporadic

Private Member Functions
virtual AcceptConnectResult	connect_datalink (const RemoteTransport &remote, const ConnectionAttribs &attribs, const TransportClient_rch &client)

virtual AcceptConnectResult	accept_datalink (const RemoteTransport &remote, const ConnectionAttribs &attribs, const TransportClient_rch &client)

virtual void	stop_accepting_or_connecting (const TransportClient_wrch &client, const GUID_t &remote_id, bool disassociate, bool association_failed)

bool	configure_i (const RtpsUdpInst_rch &config)

void	client_stop (const GUID_t &localId)

virtual void	shutdown_i ()

virtual void	register_for_reader (const GUID_t &participant, const GUID_t &writerid, const GUID_t &readerid, const TransportLocatorSeq &locators, DiscoveryListener *listener)

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

virtual void	register_for_writer (const GUID_t &, const GUID_t &, const GUID_t &, const TransportLocatorSeq &, DiscoveryListener *)

virtual void	unregister_for_writer (const GUID_t &, const GUID_t &, const GUID_t &)

virtual bool	connection_info_i (TransportLocator &info, ConnectionInfoFlags flags) const

void	get_connection_addrs (const TransportBLOB &data, AddrSet uc_addrs, AddrSet mc_addrs=0, bool requires_inline_qos=0, unsigned int blob_bytes_read=0) const

virtual void	release_datalink (DataLink *link)

virtual OPENDDS_STRING	transport_type () const

RtpsUdpDataLink_rch	make_datalink (const GuidPrefix_t &local_prefix)

bool	use_datalink (const GUID_t &local_id, const GUID_t &remote_id, const TransportBLOB &remote_data, const TransportBLOB &discovery_locator, const MonotonicTime_t &participant_discovered_at, ACE_CDR::ULong participant_flags, bool local_reliable, bool remote_reliable, bool local_durable, bool remote_durable, SequenceNumber max_sn, const TransportClient_rch &client)

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

void	relay_stun_task (const MonotonicTimePoint &now)

void	start_ice ()

void	stop_ice ()

Friends
class	RtpsUdpSendStrategy

class	RtpsUdpReceiveStrategy

Additional Inherited Members
Public Attributes inherited from OpenDDS::DCPS::TransportImpl
LockType	lock_
	Lock to protect the config_ and reactor_task_ data members. More...

WeakRcHandle< TransportInst >	config_

ReactorTask_rch	reactor_task_

EventDispatcher_rch	event_dispatcher_
	smart ptr to the associated DL cleanup task More...

unique_ptr< Monitor >	monitor_
	Monitor object for this entity. More...

Protected Types inherited from OpenDDS::DCPS::TransportImpl
typedef ACE_SYNCH_MUTEX	LockType

typedef ACE_Guard< LockType >	GuardType

Protected Member Functions inherited from OpenDDS::DCPS::TransportImpl
	TransportImpl (TransportInst_rch config)

bool	open ()

typedef	OPENDDS_MULTIMAP (TransportClient_wrch, DataLink_rch) PendConnMap

void	add_pending_connection (const TransportClient_rch &client, DataLink_rch link)

void	shutdown ()

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

Protected Attributes inherited from OpenDDS::DCPS::TransportImpl
LockType	pending_connections_lock_
	Lock to protect the pending_connections_ data member. More...

PendConnMap	pending_connections_

AtomicBool	is_shut_down_
	Id of the last link established. More...

Detailed Description

Definition at line 30 of file RtpsUdpTransport.h.

Member Typedef Documentation

◆ GuardThreadType

typedef ACE_Guard<ThreadLockType> OpenDDS::DCPS::RtpsUdpTransport::GuardThreadType

private

Definition at line 127 of file RtpsUdpTransport.h.

◆ GuardType

typedef ACE_Guard<LockType> OpenDDS::DCPS::RtpsUdpTransport::GuardType

private

Definition at line 132 of file RtpsUdpTransport.h.

◆ LockType

typedef ACE_SYNCH_MUTEX OpenDDS::DCPS::RtpsUdpTransport::LockType

private

This protects the connections_ data member.

Definition at line 131 of file RtpsUdpTransport.h.

◆ Sporadic

typedef PmfSporadicTask<RtpsUdpTransport> OpenDDS::DCPS::RtpsUdpTransport::Sporadic

private

Definition at line 180 of file RtpsUdpTransport.h.

◆ ThreadLockType

typedef ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpTransport::ThreadLockType

private

Definition at line 126 of file RtpsUdpTransport.h.

Constructor & Destructor Documentation

◆ RtpsUdpTransport()

OpenDDS::DCPS::RtpsUdpTransport::RtpsUdpTransport ( const RtpsUdpInst_rch & inst )

Definition at line 34 of file RtpsUdpTransport.cpp.

References OpenDDS::DCPS::assign(), configure_i(), OpenDDS::DCPS::GUIDPREFIX_UNKNOWN, if(), local_prefix_, and OpenDDS::DCPS::TransportImpl::open().

   : TransportImpl(inst)
 #if defined(OPENDDS_SECURITY)
   , local_crypto_handle_(DDS::HANDLE_NIL)
 #endif
 #ifdef OPENDDS_SECURITY
   , ice_endpoint_(make_rch<IceEndpoint>(ref(*this)))
   , relay_stun_task_falloff_(TimeDuration::zero_value)
   , ice_agent_(ICE::Agent::instance())
 #endif
   , transport_statistics_(inst->name())
 {
   assign(local_prefix_, GUIDPREFIX_UNKNOWN);
   if (!(configure_i(inst) && open())) {
     throw Transport::UnableToCreate();
   }
 }

Member Function Documentation

◆ accept_datalink()

TransportImpl::AcceptConnectResult OpenDDS::DCPS::RtpsUdpTransport::accept_datalink	(	const RemoteTransport &	remote,
		const ConnectionAttribs &	attribs,
		const TransportClient_rch &	client
	)

privatevirtual

accept_datalink() is called from TransportClient to initiate an association as the passive peer. A DataLink may be returned if one is already connected and ready to use, otherwise passively wait for a physical connection from the active side (either in the form of a connection event or handshaking message). Upon completion of the physical connection, the transport calls back to TransportClient::use_datalink().

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 240 of file RtpsUdpTransport.cpp.

 {
   bit_sub_ = client->get_builtin_subscriber_proxy();
 
   GuardThreadType guard_links(links_lock_);
 
   if (is_shut_down()) {
     return AcceptConnectResult();
   }
 
   if (!link_) {
     link_ = make_datalink(attribs.local_id_.guidPrefix);
     if (!link_) {
       return AcceptConnectResult();
     }
   }
   RtpsUdpDataLink_rch link = link_;
 
   if (use_datalink(attribs.local_id_, remote.repo_id_, remote.blob_, remote.discovery_blob_, remote.participant_discovered_at_,
                    remote.context_,
                    attribs.local_reliable_, remote.reliable_,
                    attribs.local_durable_, remote.durable_, attribs.max_sn_, client)) {
     return AcceptConnectResult(link);
   }
 
   GuardType guard(connections_lock_);
   add_pending_connection(client, link);
   VDBG_LVL((LM_DEBUG, "(%P|%t) RtpsUdpTransport::accept_datalink pending.\n"), 2);
   return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
 }

◆ append_transport_statistics()

void OpenDDS::DCPS::RtpsUdpTransport::append_transport_statistics ( TransportStatisticsSequence & seq )

virtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 576 of file RtpsUdpTransport.cpp.

References ACE_GUARD, OpenDDS::DCPS::append(), OpenDDS::DCPS::InternalTransportStatistics::clear(), transport_statistics_, and transport_statistics_mutex_.

 {
   ACE_GUARD(ACE_Thread_Mutex, g, transport_statistics_mutex_);
   append(seq, transport_statistics_);
   transport_statistics_.clear();
 }

◆ client_stop()

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

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 719 of file RtpsUdpTransport.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::client_stop(), link_, links_lock_, and ACE_Guard< ACE_LOCK >::release().

 {
   GuardThreadType guard_links(links_lock_);
   const RtpsUdpDataLink_rch link = link_;
   guard_links.release();
   if (link) {
     link->client_stop(localId);
   }
 }

◆ config()

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

Definition at line 53 of file RtpsUdpTransport.cpp.

References OpenDDS::DCPS::TransportImpl::config(), and OpenDDS::DCPS::dynamic_rchandle_cast().

Referenced by connection_info_i(), get_connection_addrs(), get_ice_endpoint(), OpenDDS::DCPS::RtpsUdpTransport::IceEndpoint::handle_input(), make_datalink(), register_for_reader(), register_for_writer(), relay_stun_task(), rtps_relay_address_change(), rtps_relay_only_now(), shutdown_i(), update_locators(), use_ice_now(), and use_rtps_relay_now().

 {
   return dynamic_rchandle_cast<RtpsUdpInst>(TransportImpl::config());
 }

◆ configure_i()

bool OpenDDS::DCPS::RtpsUdpTransport::configure_i ( const RtpsUdpInst_rch & config )

private

Definition at line 584 of file RtpsUdpTransport.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), ACE_IPC_SAP::control(), OpenDDS::DCPS::TransportImpl::create_reactor_task(), OpenDDS::DCPS::DataLink::default_listener(), ACE_SOCK::get_local_addr(), OpenDDS::DCPS::ReactorTask::get_reactor(), OpenDDS::DCPS::ReactorTask::interceptor(), OpenDDS::DCPS::LogAddr::ip(), IPPROTO_IP, OpenDDS::DCPS::NetworkAddress::is_any(), job_queue_, link_, links_lock_, LM_ERROR, make_datalink(), ACE_SOCK_Dgram::open(), PF_INET, OpenDDS::DCPS::rchandle_from(), OpenDDS::DCPS::TransportImpl::reactor(), OpenDDS::DCPS::TransportImpl::reactor_task(), OpenDDS::DCPS::TransportImpl::reactor_task_, relay_stun_task(), relay_stun_task_, relay_stun_task_falloff_, relay_stun_task_falloff_mutex_, ACE_SOCK::set_option(), start_ice(), TheServiceParticipant, unicast_socket_, and OpenDDS::DCPS::TimeDuration::zero_value.

Referenced by RtpsUdpTransport().

 {
   if (!config) {
     return false;
   }
 
   // Override with DCPSDefaultAddress.
   if (config->local_address() == NetworkAddress() &&
       TheServiceParticipant->default_address() != NetworkAddress()) {
     config->local_address(TheServiceParticipant->default_address());
   }
   if (config->multicast_interface_.empty() &&
       TheServiceParticipant->default_address().to_addr() != ACE_INET_Addr()) {
     config->multicast_interface_ = DCPS::LogAddr::ip(TheServiceParticipant->default_address().to_addr());
   }
 
   // Open the socket here so that any addresses/ports left
   // unspecified in the RtpsUdpInst are known by the time we get to
   // connection_info_i().  Opening the sockets here also allows us to
   // detect and report errors during DataReader/Writer setup instead
   // of during association.
 
   ACE_INET_Addr address = config->local_address().to_addr();
 
   if (unicast_socket_.open(address, PF_INET) != 0) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("RtpsUdpTransport::configure_i: open:")
                       ACE_TEXT("%m\n")),
                      false);
   }
 
 #ifdef ACE_WIN32
   // By default Winsock will cause reads to fail with "connection reset"
   // when UDP sends result in ICMP "port unreachable" messages.
   // The transport framework is not set up for this since returning <= 0
   // from our receive_bytes causes the framework to close down the datalink
   // which in this case is used to receive from multiple peers.
   {
     BOOL recv_udp_connreset = FALSE;
     unicast_socket_.control(SIO_UDP_CONNRESET, &recv_udp_connreset);
   }
 #endif
 
   if (unicast_socket_.get_local_addr(address) != 0) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("RtpsUdpTransport::configure_i: get_local_addr:")
                       ACE_TEXT("%m\n")),
                      false);
   }
 
   config->local_address(NetworkAddress(address));
 
 #ifdef ACE_RECVPKTINFO
   int sockopt = 1;
   if (unicast_socket_.set_option(IPPROTO_IP, ACE_RECVPKTINFO, &sockopt, sizeof sockopt) == -1) {
     ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: RtpsUdpTransport::configure_i: set_option: %m\n")), false);
   }
 #endif
 
 #ifdef ACE_HAS_IPV6
   address = config->ipv6_local_address().to_addr();
 
   if (ipv6_unicast_socket_.open(address, PF_INET6) != 0) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("RtpsUdpTransport::configure_i: open:")
                       ACE_TEXT("%m\n")),
                      false);
   }
 
 #ifdef ACE_WIN32
   // By default Winsock will cause reads to fail with "connection reset"
   // when UDP sends result in ICMP "port unreachable" messages.
   // The transport framework is not set up for this since returning <= 0
   // from our receive_bytes causes the framework to close down the datalink
   // which in this case is used to receive from multiple peers.
   {
     BOOL recv_udp_connreset = FALSE;
     ipv6_unicast_socket_.control(SIO_UDP_CONNRESET, &recv_udp_connreset);
   }
 #endif
 
   if (ipv6_unicast_socket_.get_local_addr(address) != 0) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("RtpsUdpTransport::configure_i: get_local_addr:")
                       ACE_TEXT("%m\n")),
                      false);
   }
 
   NetworkAddress temp(address);
   if (address.is_ipv4_mapped_ipv6() && temp.is_any()) {
     temp = NetworkAddress(address.get_port_number(), "::");
   }
   config->ipv6_local_address(temp);
 
 #ifdef ACE_RECVPKTINFO6
   if (ipv6_unicast_socket_.set_option(IPPROTO_IPV6, ACE_RECVPKTINFO6, &sockopt, sizeof sockopt) == -1) {
     ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: RtpsUdpTransport::configure_i: set_option: %m\n")), false);
   }
 #endif
 #endif
 
   create_reactor_task(false, "RtpsUdpTransport" + config->name());
 
   ACE_Reactor* reactor = reactor_task_->get_reactor();
   job_queue_ = DCPS::make_rch<DCPS::JobQueue>(reactor);
 
 #ifdef OPENDDS_SECURITY
   if (config->use_ice()) {
     start_ice();
   }
 
   relay_stun_task_= make_rch<Sporadic>(TheServiceParticipant->time_source(), reactor_task()->interceptor(), rchandle_from(this), &RtpsUdpTransport::relay_stun_task);
 #endif
 
   if (config->opendds_discovery_default_listener_) {
     GuardThreadType guard_links(links_lock_);
     link_ = make_datalink(config->opendds_discovery_guid_.guidPrefix);
     link_->default_listener(*config->opendds_discovery_default_listener_);
   }
 
 #ifdef OPENDDS_SECURITY
   {
     ACE_Guard<ThreadLockType> guard(relay_stun_task_falloff_mutex_);
     relay_stun_task_falloff_.set(config->heartbeat_period_);
   }
   relay_stun_task_->schedule(TimeDuration::zero_value);
 #endif
 
   return true;
 }

◆ connect_datalink()

TransportImpl::AcceptConnectResult OpenDDS::DCPS::RtpsUdpTransport::connect_datalink	(	const RemoteTransport &	remote,
		const ConnectionAttribs &	attribs,
		const TransportClient_rch &	client
	)

privatevirtual

connect_datalink() is called from TransportClient to initiate an association as the active peer. A DataLink may be returned if one is already connected and ready to use, otherwise initiate a connection to the passive side and return from this method. Upon completion of the physical connection, the transport calls back to TransportClient::use_datalink().

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 205 of file RtpsUdpTransport.cpp.

 {
   bit_sub_ = client->get_builtin_subscriber_proxy();
 
   GuardThreadType guard_links(links_lock_);
 
   if (is_shut_down()) {
     return AcceptConnectResult();
   }
 
   if (!link_) {
     link_ = make_datalink(attribs.local_id_.guidPrefix);
     if (!link_) {
       return AcceptConnectResult();
     }
   }
 
   RtpsUdpDataLink_rch link = link_;
 
   if (use_datalink(attribs.local_id_, remote.repo_id_, remote.blob_, remote.discovery_blob_, remote.participant_discovered_at_,
                    remote.context_,
                    attribs.local_reliable_, remote.reliable_,
                    attribs.local_durable_, remote.durable_, attribs.max_sn_, client)) {
     return AcceptConnectResult(link);
   }
 
   GuardType guard(connections_lock_);
   add_pending_connection(client, link);
   VDBG_LVL((LM_DEBUG, "(%P|%t) RtpsUdpTransport::connect_datalink pending.\n"), 2);
   return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
 }

◆ connection_info_i()

bool OpenDDS::DCPS::RtpsUdpTransport::connection_info_i	(	TransportLocator &	local_info,
		ConnectionInfoFlags	flags
	)		const

privatevirtual

Called by our connection_info() method to allow the concrete TransportImpl subclass to do the dirty work since it really is the one that knows how to populate the supplied TransportLocator object.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 393 of file RtpsUdpTransport.cpp.

References config().

 {
   RtpsUdpInst_rch cfg = config();
   if (cfg) {
     cfg->populate_locator(info, flags);
     return true;
   }
   return false;
 }

◆ disable_relay_stun_task()

void OpenDDS::DCPS::RtpsUdpTransport::disable_relay_stun_task ( )

Definition at line 1075 of file RtpsUdpTransport.cpp.

References OpenDDS::DCPS::ConnectionRecord::address, bit_sub_, deferred_connection_records_, OpenDDS::DCPS::JobQueue::enqueue(), OpenDDS::DCPS::ConnectionRecord::guid, job_queue_, OpenDDS::DCPS::ConnectionRecord::latency, OPENDDS_END_VERSIONED_NAMESPACE_DECL, OpenDDS::DCPS::ConnectionRecord::protocol, relay_srsm_, relay_stun_task_, OpenDDS::DCPS::RTPS_RELAY_STUN_PROTOCOL, OpenDDS::ICE::ServerReflexiveStateMachine::stun_server_address(), OpenDDS::DCPS::TimeDuration::to_dds_duration(), and OpenDDS::DCPS::TimeDuration::zero_value.

Referenced by rtps_relay_only_now(), and use_rtps_relay_now().

 {
 #ifndef DDS_HAS_MINIMUM_BIT
   relay_stun_task_->cancel();
 
   DCPS::ConnectionRecord connection_record;
   std::memset(connection_record.guid, 0, sizeof(connection_record.guid));
   connection_record.protocol = RTPS_RELAY_STUN_PROTOCOL;
   connection_record.latency = TimeDuration::zero_value.to_dds_duration();
 
   if (relay_srsm_.stun_server_address() != ACE_INET_Addr()) {
     connection_record.address = DCPS::LogAddr(relay_srsm_.stun_server_address()).c_str();
     deferred_connection_records_.push_back(std::make_pair(false, connection_record));
   }
 
   if (!bit_sub_) {
     return;
   }
 
   if (!deferred_connection_records_.empty()) {
     job_queue_->enqueue(DCPS::make_rch<WriteConnectionRecords>(bit_sub_, deferred_connection_records_));
     deferred_connection_records_.clear();
   }
 
   relay_srsm_ = ICE::ServerReflexiveStateMachine();
 #endif
 }

◆ get_connection_addrs()

void OpenDDS::DCPS::RtpsUdpTransport::get_connection_addrs	(	const TransportBLOB &	data,
		AddrSet *	uc_addrs,
		AddrSet *	mc_addrs = `0`,
		bool *	requires_inline_qos = `0`,
		unsigned int *	blob_bytes_read = `0`
	)		const

private

Definition at line 362 of file RtpsUdpTransport.cpp.

References OpenDDS::RTPS::blob_to_locators(), config(), ACE_INET_Addr::is_multicast(), OpenDDS::DCPS::locator_to_address(), and DDS::RETCODE_OK.

Referenced by register_for_reader(), register_for_writer(), update_locators(), and use_datalink().

 {
   using namespace OpenDDS::RTPS;
   LocatorSeq locators;
   DDS::ReturnCode_t result =
     blob_to_locators(remote, locators, requires_inline_qos, blob_bytes_read);
   if (result != DDS::RETCODE_OK) {
     return;
   }
 
   for (CORBA::ULong i = 0; i < locators.length(); ++i) {
     ACE_INET_Addr addr;
     // If conversion was successful
     if (locator_to_address(addr, locators[i], false) == 0) {
       if (addr.is_multicast()) {
         RtpsUdpInst_rch cfg = config();
         if (cfg && cfg->use_multicast_ && mc_addrs) {
           mc_addrs->insert(NetworkAddress(addr));
         }
       } else if (uc_addrs) {
         uc_addrs->insert(NetworkAddress(addr));
       }
     }
   }
 }

◆ get_ice_agent()

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

Definition at line 60 of file RtpsUdpTransport.cpp.

References ice_agent_.

 {
   return ice_agent_;
 }

◆ get_ice_endpoint()

DCPS::WeakRcHandle< ICE::Endpoint > OpenDDS::DCPS::RtpsUdpTransport::get_ice_endpoint ( )

virtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 67 of file RtpsUdpTransport.cpp.

References config(), and ice_endpoint_.

 {
 #ifdef OPENDDS_SECURITY
   RtpsUdpInst_rch cfg = config();
   return (cfg && cfg->use_ice()) ? static_rchandle_cast<ICE::Endpoint>(ice_endpoint_) : DCPS::WeakRcHandle<ICE::Endpoint>();
 #else
   return DCPS::WeakRcHandle<ICE::Endpoint>();
 #endif
 }

◆ get_last_recv_locator()

void OpenDDS::DCPS::RtpsUdpTransport::get_last_recv_locator	(	const GUID_t &	remote,
		TransportLocator &	tl
	)

virtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 521 of file RtpsUdpTransport.cpp.

References OpenDDS::DCPS::address_to_locator(), OpenDDS::DCPS::TransportLocator::data, OpenDDS::STUN::encoding(), OpenDDS::DCPS::RtpsUdpDataLink::get_last_recv_address(), OpenDDS::RTPS::get_locators_encoding(), OpenDDS::DCPS::TransportImpl::is_shut_down(), link_, links_lock_, OpenDDS::RTPS::message_block_to_sequence(), OpenDDS::DCPS::primitive_serialized_size_boolean(), OpenDDS::DCPS::ref(), OpenDDS::DCPS::RtpsUdpDataLink::requires_inline_qos(), OpenDDS::DCPS::serialized_size(), OpenDDS::DCPS::NetworkAddress::to_addr(), and OpenDDS::DCPS::TransportLocator::transport_type.

 {
   if (is_shut_down()) {
     return;
   }
 
   GuardThreadType guard_links(links_lock_);
 
   bool expects_inline_qos = false;
   NetworkAddress addr;
   if (link_) {
     addr = link_->get_last_recv_address(remote);
     if (addr == NetworkAddress()) {
       return;
     }
     GUIDSeq_var guids(new GUIDSeq);
     GUIDSeq& ref = static_cast<GUIDSeq&>(guids);
     ref.length(1);
     ref[0] = remote;
     expects_inline_qos = link_->requires_inline_qos(guids);
   }
 
   LocatorSeq locators;
   locators.length(1);
   address_to_locator(locators[0], addr.to_addr());
 
   const Encoding& encoding = RTPS::get_locators_encoding();
   size_t size = serialized_size(encoding, locators);
   primitive_serialized_size_boolean(encoding, size);
 
   ACE_Message_Block mb_locator(size);
   Serializer ser_loc(&mb_locator, encoding);
   ser_loc << locators;
   ser_loc << ACE_OutputCDR::from_boolean(expects_inline_qos);
 
   tl.transport_type = "rtps_udp";
   RTPS::message_block_to_sequence(mb_locator, tl.data);
 }

◆ local_crypto_handle()

void OpenDDS::DCPS::RtpsUdpTransport::local_crypto_handle ( DDS::Security::ParticipantCryptoHandle pch )

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 347 of file RtpsUdpTransport.cpp.

References link_, links_lock_, OpenDDS::DCPS::RtpsUdpDataLink::local_crypto_handle(), and local_crypto_handle_.

 {
   RtpsUdpDataLink_rch link;
   {
     ACE_Guard<ACE_Thread_Mutex> guard(links_lock_);
     local_crypto_handle_ = pch;
     link = link_;
   }
   if (link) {
     link->local_crypto_handle(pch);
   }
 }

◆ make_datalink()

RtpsUdpDataLink_rch OpenDDS::DCPS::RtpsUdpTransport::make_datalink ( const GuidPrefix_t & local_prefix )

private

Definition at line 140 of file RtpsUdpTransport.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::assign(), config(), OpenDDS::DCPS::RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC, OpenDDS::DCPS::equal_guid_prefixes(), OpenDDS::DCPS::ReactorInterceptor::execute_or_enqueue(), ACE_IPC_SAP::get_handle(), OpenDDS::DCPS::GUIDPREFIX_UNKNOWN, OpenDDS::DCPS::ReactorTask::interceptor(), LM_ERROR, OpenDDS::DCPS::RtpsUdpDataLink::local_crypto_handle(), local_crypto_handle_, local_prefix_, OpenDDS::DCPS::RtpsUdpDataLink::open(), OPENDDS_ASSERT, OpenDDS::DCPS::rchandle_from(), OpenDDS::DCPS::TransportImpl::reactor_task(), OpenDDS::DCPS::TransportImpl::reactor_task_, ACE_Event_Handler::READ_MASK, OpenDDS::DCPS::ref(), relay_stun_task_, relay_stun_task_falloff_, relay_stun_task_falloff_mutex_, ACE_IPC_SAP::set_handle(), transport_statistics_, transport_statistics_mutex_, unicast_socket_, and OpenDDS::DCPS::TimeDuration::zero_value.

Referenced by accept_datalink(), configure_i(), connect_datalink(), register_for_reader(), and register_for_writer().

 {
   OPENDDS_ASSERT(!equal_guid_prefixes(local_prefix, GUIDPREFIX_UNKNOWN));
 
   RtpsUdpInst_rch cfg = config();
   if (!cfg) {
     return RtpsUdpDataLink_rch();
   }
 
   if (equal_guid_prefixes(local_prefix_, GUIDPREFIX_UNKNOWN)) {
     assign(local_prefix_, local_prefix);
 #ifdef OPENDDS_SECURITY
     {
       ACE_Guard<ThreadLockType> guard(relay_stun_task_falloff_mutex_);
       relay_stun_task_falloff_.set(cfg ? cfg->heartbeat_period_ : TimeDuration(RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC));
     }
     relay_stun_task_->schedule(TimeDuration::zero_value);
 #endif
   }
 
 #if defined(OPENDDS_SECURITY)
   if (cfg->use_ice()) {
     ReactorInterceptor_rch ri = reactor_task_->interceptor();
     ri->execute_or_enqueue(make_rch<RemoveHandler>(unicast_socket_.get_handle(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #ifdef ACE_HAS_IPV6
     ri->execute_or_enqueue(make_rch<RemoveHandler>(ipv6_unicast_socket_.get_handle(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #endif
   }
 #endif
 
   RtpsUdpDataLink_rch link = make_rch<RtpsUdpDataLink>(rchandle_from(this), local_prefix, config(), reactor_task(), ref(transport_statistics_), ref(transport_statistics_mutex_));
 
 #if defined(OPENDDS_SECURITY)
   link->local_crypto_handle(local_crypto_handle_);
 #endif
 
   if (!link->open(unicast_socket_
 #ifdef ACE_HAS_IPV6
                   , ipv6_unicast_socket_
 #endif
                   )) {
 #ifdef ACE_HAS_IPV6
     const ACE_HANDLE v6handle = ipv6_unicast_socket_.get_handle();
 #else
     const ACE_HANDLE v6handle = ACE_INVALID_HANDLE;
 #endif
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: ")
                ACE_TEXT("RtpsUdpTransport::make_datalink: ")
                ACE_TEXT("failed to open DataLink for sockets %d %d\n"),
                unicast_socket_.get_handle(), v6handle
                ));
     return RtpsUdpDataLink_rch();
   }
 
   // RtpsUdpDataLink now owns the socket
   unicast_socket_.set_handle(ACE_INVALID_HANDLE);
 #ifdef ACE_HAS_IPV6
   ipv6_unicast_socket_.set_handle(ACE_INVALID_HANDLE);
 #endif
 
   return link;
 }

◆ process_relay_sra()

void OpenDDS::DCPS::RtpsUdpTransport::process_relay_sra ( ICE::ServerReflexiveStateMachine::StateChange sc )

Definition at line 1023 of file RtpsUdpTransport.cpp.

References OpenDDS::DCPS::ConnectionRecord::address, bit_sub_, deferred_connection_records_, OpenDDS::DCPS::JobQueue::enqueue(), OpenDDS::DCPS::ConnectionRecord::guid, OpenDDS::ICE::Configuration::instance(), job_queue_, OpenDDS::DCPS::ConnectionRecord::latency, OpenDDS::ICE::ServerReflexiveStateMachine::latency(), OpenDDS::ICE::ServerReflexiveStateMachine::latency_available(), OpenDDS::DCPS::ConnectionRecord::protocol, relay_srsm_, relay_stun_task_falloff_, relay_stun_task_falloff_mutex_, OpenDDS::DCPS::RTPS_RELAY_STUN_PROTOCOL, OpenDDS::ICE::ServerReflexiveStateMachine::SRSM_Change, OpenDDS::ICE::ServerReflexiveStateMachine::SRSM_None, OpenDDS::ICE::ServerReflexiveStateMachine::SRSM_Set, OpenDDS::ICE::ServerReflexiveStateMachine::SRSM_Unset, OpenDDS::ICE::ServerReflexiveStateMachine::stun_server_address(), OpenDDS::DCPS::TimeDuration::to_dds_duration(), OpenDDS::ICE::ServerReflexiveStateMachine::unset_stun_server_address(), and OpenDDS::DCPS::TimeDuration::zero_value.

Referenced by relay_stun_task().

 {
 #ifndef DDS_HAS_MINIMUM_BIT
   DCPS::ConnectionRecord connection_record;
   std::memset(connection_record.guid, 0, sizeof(connection_record.guid));
   connection_record.protocol = RTPS_RELAY_STUN_PROTOCOL;
   connection_record.latency = TimeDuration::zero_value.to_dds_duration();
 
   switch (sc) {
   case ICE::ServerReflexiveStateMachine::SRSM_None:
     if (relay_srsm_.latency_available()) {
       connection_record.address = DCPS::LogAddr(relay_srsm_.stun_server_address()).c_str();
       connection_record.latency = relay_srsm_.latency().to_dds_duration();
       relay_srsm_.latency_available(false);
       deferred_connection_records_.push_back(std::make_pair(true, connection_record));
     }
     break;
   case ICE::ServerReflexiveStateMachine::SRSM_Set:
   case ICE::ServerReflexiveStateMachine::SRSM_Change:
     // Lengthen to normal period.
     {
       ACE_Guard<ThreadLockType> guard(relay_stun_task_falloff_mutex_);
       relay_stun_task_falloff_.set(ICE::Configuration::instance()->server_reflexive_address_period());
     }
     connection_record.address = DCPS::LogAddr(relay_srsm_.stun_server_address()).c_str();
     connection_record.latency = relay_srsm_.latency().to_dds_duration();
     relay_srsm_.latency_available(false);
     deferred_connection_records_.push_back(std::make_pair(true, connection_record));
     break;
   case ICE::ServerReflexiveStateMachine::SRSM_Unset:
     {
       connection_record.address = DCPS::LogAddr(relay_srsm_.unset_stun_server_address()).c_str();
       deferred_connection_records_.push_back(std::make_pair(false, connection_record));
       break;
     }
   }
 
   if (!bit_sub_) {
     return;
   }
 
   if (!deferred_connection_records_.empty()) {
     job_queue_->enqueue(DCPS::make_rch<WriteConnectionRecords>(bit_sub_, deferred_connection_records_));
     deferred_connection_records_.clear();
   }
 
 #else
   ACE_UNUSED_ARG(sc);
 #endif
 }

◆ register_for_reader()

void OpenDDS::DCPS::RtpsUdpTransport::register_for_reader	(	const GUID_t &	participant,
		const GUID_t &	writerid,
		const GUID_t &	readerid,
		const TransportLocatorSeq &	locators,
		DiscoveryListener *	listener
	)

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 404 of file RtpsUdpTransport.cpp.

References config(), get_connection_addrs(), OpenDDS::DCPS::GUID_t::guidPrefix, OpenDDS::DCPS::TransportImpl::is_shut_down(), link_, links_lock_, make_datalink(), and OpenDDS::DCPS::RtpsUdpDataLink::register_for_reader().

 {
   if (is_shut_down()) {
     return;
   }
 
   RtpsUdpInst_rch cfg = config();
   if (!cfg) {
     return;
   }
 
   const TransportBLOB* blob = cfg->get_blob(locators);
   if (!blob) {
     return;
   }
 
   GuardThreadType guard_links(links_lock_);
 
   if (!link_) {
     link_ = make_datalink(participant.guidPrefix);
   }
 
   AddrSet uc_addrs;
   get_connection_addrs(*blob, &uc_addrs);
   link_->register_for_reader(writerid, readerid, uc_addrs, listener);
 }

◆ register_for_writer()

void OpenDDS::DCPS::RtpsUdpTransport::register_for_writer	(	const GUID_t &	participant,
		const GUID_t &	readerid,
		const GUID_t &	writerid,
		const TransportLocatorSeq &	locators,
		DiscoveryListener *	listener
	)

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 448 of file RtpsUdpTransport.cpp.

References config(), get_connection_addrs(), OpenDDS::DCPS::GUID_t::guidPrefix, OpenDDS::DCPS::TransportImpl::is_shut_down(), link_, links_lock_, make_datalink(), and OpenDDS::DCPS::RtpsUdpDataLink::register_for_writer().

 {
   if (is_shut_down()) {
     return;
   }
 
   RtpsUdpInst_rch cfg = config();
   if (!cfg) {
     return;
   }
 
   const TransportBLOB* blob = cfg->get_blob(locators);
   if (!blob) {
     return;
   }
 
   GuardThreadType guard_links(links_lock_);
 
   if (!link_) {
     link_ = make_datalink(participant.guidPrefix);
   }
 
   AddrSet uc_addrs;
   get_connection_addrs(*blob, &uc_addrs);
   link_->register_for_writer(readerid, writerid, uc_addrs, listener);
 }

◆ relay_srsm()

ICE::ServerReflexiveStateMachine& OpenDDS::DCPS::RtpsUdpTransport::relay_srsm ( )

inline

Definition at line 42 of file RtpsUdpTransport.h.

42 { return relay_srsm_; }

OpenDDS::DCPS::RtpsUdpTransport::relay_srsm_

ICE::ServerReflexiveStateMachine relay_srsm_

Definition: RtpsUdpTransport.h:185

◆ relay_stun_task()

void OpenDDS::DCPS::RtpsUdpTransport::relay_stun_task ( const MonotonicTimePoint & now )

private

Definition at line 998 of file RtpsUdpTransport.cpp.

References config(), OpenDDS::DCPS::equal_guid_prefixes(), OpenDDS::DCPS::GUIDPREFIX_UNKNOWN, ice_endpoint_, OpenDDS::ICE::Configuration::instance(), links_lock_, local_prefix_, OpenDDS::ICE::ServerReflexiveStateMachine::message(), process_relay_sra(), relay_srsm_, relay_stun_task_, relay_stun_task_falloff_, relay_stun_task_falloff_mutex_, OpenDDS::ICE::ServerReflexiveStateMachine::send(), and OpenDDS::ICE::Configuration::server_reflexive_indication_count().

Referenced by configure_i().

 {
   GuardThreadType guard_links(links_lock_);
 
   RtpsUdpInst_rch cfg = config();
   if (!cfg) {
     return;
   }
 
   const ACE_INET_Addr relay_address = cfg->rtps_relay_address().to_addr();
 
   if ((cfg->use_rtps_relay() || cfg->rtps_relay_only()) &&
       relay_address != ACE_INET_Addr() &&
       !equal_guid_prefixes(local_prefix_, GUIDPREFIX_UNKNOWN)) {
     process_relay_sra(relay_srsm_.send(relay_address, ICE::Configuration::instance()->server_reflexive_indication_count(), local_prefix_));
     ice_endpoint_->send(relay_address, relay_srsm_.message());
     {
       ACE_Guard<ThreadLockType> guard(relay_stun_task_falloff_mutex_);
       relay_stun_task_falloff_.advance(ICE::Configuration::instance()->server_reflexive_address_period());
       relay_stun_task_->schedule(relay_stun_task_falloff_.get());
     }
   }
 }

◆ release_datalink()

void OpenDDS::DCPS::RtpsUdpTransport::release_datalink ( DataLink * link )

privatevirtual

Called by the TransportRegistry when this TransportImpl object is released while the TransportRegistry is handling a release() "event". The DataLink itself calls this method when it thinks it is no longer used for any associations. This occurs during a "remove associations" operation being performed by some TransportClient that uses this TransportImpl. The TransportClient is known to have acquired our reservation_lock_, so there won't be any reserve_datalink() calls being made from any other threads while we perform this release.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 752 of file RtpsUdpTransport.cpp.

 {
   // No-op for rtps_udp: keep the link_ around until the transport is shut down.
 }

◆ rtps_relay_address_change()

void OpenDDS::DCPS::RtpsUdpTransport::rtps_relay_address_change ( )

virtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 562 of file RtpsUdpTransport.cpp.

References config(), OpenDDS::DCPS::RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC, relay_stun_task_, relay_stun_task_falloff_, relay_stun_task_falloff_mutex_, and OpenDDS::DCPS::TimeDuration::zero_value.

 {
 #ifdef OPENDDS_SECURITY
   relay_stun_task_->cancel();
   RtpsUdpInst_rch cfg = config();
   {
     ACE_Guard<ThreadLockType> guard(relay_stun_task_falloff_mutex_);
     relay_stun_task_falloff_.set(cfg ? cfg->heartbeat_period_ : TimeDuration(RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC));
   }
   relay_stun_task_->schedule(TimeDuration::zero_value);
 #endif
 }

◆ rtps_relay_only_now()

void OpenDDS::DCPS::RtpsUdpTransport::rtps_relay_only_now ( bool flag )

virtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 78 of file RtpsUdpTransport.cpp.

References config(), OpenDDS::DCPS::RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC, disable_relay_stun_task(), relay_stun_task_, relay_stun_task_falloff_, relay_stun_task_falloff_mutex_, and OpenDDS::DCPS::TimeDuration::zero_value.

 {
   ACE_UNUSED_ARG(flag);
 
 #ifdef OPENDDS_SECURITY
   RtpsUdpInst_rch cfg = config();
   if (flag) {
     {
       ACE_Guard<ThreadLockType> guard(relay_stun_task_falloff_mutex_);
       relay_stun_task_falloff_.set(cfg ? cfg->heartbeat_period_ : TimeDuration(RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC));
     }
     relay_stun_task_->schedule(TimeDuration::zero_value);
   } else {
     if (!cfg || cfg->use_rtps_relay()) {
       disable_relay_stun_task();
     }
   }
 #endif
 }

◆ shutdown_i()

void OpenDDS::DCPS::RtpsUdpTransport::shutdown_i ( )

privatevirtual

Called during the shutdown() method in order to give the concrete TransportImpl subclass a chance to do something when the shutdown "event" occurs.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 730 of file RtpsUdpTransport.cpp.

References config(), job_queue_, link_, links_lock_, relay_stun_task_, OpenDDS::DCPS::RcHandle< T >::reset(), stop_ice(), and OpenDDS::DCPS::DataLink::transport_shutdown().

 {
 #ifdef OPENDDS_SECURITY
   RtpsUdpInst_rch cfg = config();
 
   if (cfg && cfg->use_ice()) {
     stop_ice();
   }
 
   relay_stun_task_->cancel();
 #endif
 
   job_queue_.reset();
 
   GuardThreadType guard_links(links_lock_);
   if (link_) {
     link_->transport_shutdown();
   }
   link_.reset();
 }

◆ start_ice()

void OpenDDS::DCPS::RtpsUdpTransport::start_ice ( )

private

Definition at line 958 of file RtpsUdpTransport.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::ReactorInterceptor::execute_or_enqueue(), ACE_IPC_SAP::get_handle(), ice_agent_, ice_endpoint_, OpenDDS::DCPS::ReactorTask::interceptor(), link_, links_lock_, LM_INFO, OpenDDS::DCPS::TransportImpl::reactor_task_, ACE_Event_Handler::READ_MASK, and unicast_socket_.

Referenced by configure_i(), and use_ice_now().

 {
   if (DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) RtpsUdpTransport::start_ice\n"));
   }
 
   ice_agent_->add_endpoint(static_rchandle_cast<ICE::Endpoint>(ice_endpoint_));
 
   GuardThreadType guard_links(links_lock_);
 
   if (!link_) {
     ReactorInterceptor_rch ri = reactor_task_->interceptor();
     ri->execute_or_enqueue(make_rch<RegisterHandler>(unicast_socket_.get_handle(), ice_endpoint_.get(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #ifdef ACE_HAS_IPV6
     ri->execute_or_enqueue(make_rch<RegisterHandler>(ipv6_unicast_socket_.get_handle(), ice_endpoint_.get(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #endif
   }
 }

◆ stop_accepting_or_connecting()

void OpenDDS::DCPS::RtpsUdpTransport::stop_accepting_or_connecting	(	const TransportClient_wrch &	client,
		const GUID_t &	remote_id,
		bool	disassociate,
		bool	association_failed
	)

privatevirtual

stop_accepting_or_connecting() is called from TransportClient to terminate the accepting process begun by accept_datalink() or connect_datalink(). This allows the TransportImpl to clean up any resources associated with this pending connection. The TransportClient* passed in to accept or connect is not valid after this method is called.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 275 of file RtpsUdpTransport.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::disassociated(), link_, links_lock_, OpenDDS::DCPS::WeakRcHandle< T >::lock(), OpenDDS::DCPS::TransportImpl::pending_connections_, and OpenDDS::DCPS::TransportImpl::pending_connections_lock_.

 {
   if (disassociate || association_failed) {
     GuardThreadType guard_links(links_lock_);
     if (link_) {
       TransportClient_rch c = client.lock();
       if (c) {
         link_->disassociated(c->get_guid(), remote_id);
       }
     }
   }
 
   {
     GuardType guard(pending_connections_lock_);
     typedef PendConnMap::iterator iter_t;
     const std::pair<iter_t, iter_t> range =
       pending_connections_.equal_range(client);
     for (iter_t iter = range.first; iter != range.second; ++iter) {
       iter->second->remove_on_start_callback(client, remote_id);
     }
     pending_connections_.erase(range.first, range.second);
   }
 }

◆ stop_ice()

void OpenDDS::DCPS::RtpsUdpTransport::stop_ice ( )

private

Definition at line 978 of file RtpsUdpTransport.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::ReactorInterceptor::execute_or_enqueue(), ACE_IPC_SAP::get_handle(), ice_agent_, ice_endpoint_, OpenDDS::DCPS::ReactorTask::interceptor(), link_, links_lock_, LM_INFO, OpenDDS::DCPS::TransportImpl::reactor_task_, ACE_Event_Handler::READ_MASK, and unicast_socket_.

Referenced by shutdown_i(), and use_ice_now().

 {
   if (DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) RtpsUdpTransport::stop_ice\n"));
   }
 
   GuardThreadType guard_links(links_lock_);
 
   if (!link_) {
     ReactorInterceptor_rch ri = reactor_task_->interceptor();
     ri->execute_or_enqueue(make_rch<RemoveHandler>(unicast_socket_.get_handle(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #ifdef ACE_HAS_IPV6
     ri->execute_or_enqueue(make_rch<RemoveHandler>(ipv6_unicast_socket_.get_handle(), static_cast<ACE_Reactor_Mask>(ACE_Event_Handler::READ_MASK)));
 #endif
   }
 
   ice_agent_->remove_endpoint(static_rchandle_cast<ICE::Endpoint>(ice_endpoint_));
 }

◆ transport_type()

virtual OPENDDS_STRING OpenDDS::DCPS::RtpsUdpTransport::transport_type ( ) const

inlineprivatevirtual

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 106 of file RtpsUdpTransport.h.

106 { return "rtps_udp"; }

◆ unregister_for_reader()

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

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 436 of file RtpsUdpTransport.cpp.

References link_, links_lock_, and OpenDDS::DCPS::RtpsUdpDataLink::unregister_for_reader().

 {
   GuardThreadType guard_links(links_lock_);
 
   if (link_) {
     link_->unregister_for_reader(writerid, readerid);
   }
 }

◆ unregister_for_writer()

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

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 480 of file RtpsUdpTransport.cpp.

References link_, links_lock_, and OpenDDS::DCPS::RtpsUdpDataLink::unregister_for_writer().

 {
   GuardThreadType guard_links(links_lock_);
 
   if (link_) {
     link_->unregister_for_writer(readerid, writerid);
   }
 }

◆ update_locators()

void OpenDDS::DCPS::RtpsUdpTransport::update_locators	(	const GUID_t &	remote,
		const TransportLocatorSeq &	locators
	)

virtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 492 of file RtpsUdpTransport.cpp.

References config(), get_connection_addrs(), OpenDDS::DCPS::TransportImpl::is_shut_down(), link_, links_lock_, and OpenDDS::DCPS::RtpsUdpDataLink::update_locators().

 {
   if (is_shut_down()) {
     return;
   }
 
   RtpsUdpInst_rch cfg = config();
   if (!cfg) {
     return;
   }
 
   const TransportBLOB* blob = cfg->get_blob(locators);
   if (!blob) {
     return;
   }
 
   GuardThreadType guard_links(links_lock_);
 
   if (link_) {
     AddrSet uc_addrs, mc_addrs;
     bool requires_inline_qos;
     unsigned int blob_bytes_read;
     get_connection_addrs(*blob, &uc_addrs, &mc_addrs, &requires_inline_qos, &blob_bytes_read);
     link_->update_locators(remote, uc_addrs, mc_addrs, requires_inline_qos, false);
   }
 }

◆ use_datalink()

bool OpenDDS::DCPS::RtpsUdpTransport::use_datalink	(	const GUID_t &	local_id,
		const GUID_t &	remote_id,
		const TransportBLOB &	remote_data,
		const TransportBLOB &	discovery_locator,
		const MonotonicTime_t &	participant_discovered_at,
		ACE_CDR::ULong	participant_flags,
		bool	local_reliable,
		bool	remote_reliable,
		bool	local_durable,
		bool	remote_durable,
		SequenceNumber	max_sn,
		const TransportClient_rch &	client
	)

private

Definition at line 303 of file RtpsUdpTransport.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::associated(), get_connection_addrs(), OpenDDS::DCPS::is_more_local(), and link_.

Referenced by accept_datalink(), and connect_datalink().

 {
   AddrSet uc_addrs, mc_addrs;
   bool requires_inline_qos;
   unsigned int blob_bytes_read;
   get_connection_addrs(remote_data, &uc_addrs, &mc_addrs, &requires_inline_qos, &blob_bytes_read);
 
   NetworkAddress disco_addr_hint;
   if (discovery_locator.length()) {
     AddrSet disco_uc_addrs, disco_mc_addrs;
     bool disco_requires_inline_qos;
     unsigned int disco_blob_bytes_read;
     get_connection_addrs(discovery_locator, &disco_uc_addrs, &disco_mc_addrs, &disco_requires_inline_qos, &disco_blob_bytes_read);
 
     for (AddrSet::const_iterator it = disco_uc_addrs.begin(), limit = disco_uc_addrs.end(); it != limit; ++it) {
       for (AddrSet::const_iterator it2 = uc_addrs.begin(), limit2 = uc_addrs.end(); it2 != limit2; ++it2) {
         if (it->addr_bytes_equal(*it2) && DCPS::is_more_local(disco_addr_hint, *it2)) {
           disco_addr_hint = *it2;
         }
       }
     }
   }
 
   if (link_) {
     return link_->associated(local_id, remote_id, local_reliable, remote_reliable,
                              local_durable, remote_durable,
                              participant_discovered_at, participant_flags, max_sn, client,
                              uc_addrs, mc_addrs, disco_addr_hint, requires_inline_qos);
   }
 
   return true;
 }

◆ use_ice_now()

void OpenDDS::DCPS::RtpsUdpTransport::use_ice_now ( bool flag )

virtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 120 of file RtpsUdpTransport.cpp.

References config(), start_ice(), and stop_ice().

 {
   ACE_UNUSED_ARG(after);
 
 #ifdef OPENDDS_SECURITY
   RtpsUdpInst_rch cfg = config();
   const bool before = cfg && cfg->use_ice();
   if (cfg) {
     cfg->use_ice(after);
   }
 
   if (before && !after) {
     stop_ice();
   } else if (!before && after) {
     start_ice();
   }
 #endif
 }

◆ use_rtps_relay_now()

void OpenDDS::DCPS::RtpsUdpTransport::use_rtps_relay_now ( bool flag )

virtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 99 of file RtpsUdpTransport.cpp.

References config(), OpenDDS::DCPS::RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC, disable_relay_stun_task(), relay_stun_task_, relay_stun_task_falloff_, relay_stun_task_falloff_mutex_, and OpenDDS::DCPS::TimeDuration::zero_value.

 {
   ACE_UNUSED_ARG(flag);
 
 #ifdef OPENDDS_SECURITY
   RtpsUdpInst_rch cfg = config();
   if (flag) {
     {
       ACE_Guard<ThreadLockType> guard(relay_stun_task_falloff_mutex_);
       relay_stun_task_falloff_.set(cfg ? cfg->heartbeat_period_ : TimeDuration(RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC));
     }
     relay_stun_task_->schedule(TimeDuration::zero_value);
   } else {
     if (!cfg || !cfg->rtps_relay_only()) {
       disable_relay_stun_task();
     }
   }
 #endif
 }

Friends And Related Function Documentation

◆ RtpsUdpReceiveStrategy

friend class RtpsUdpReceiveStrategy

friend

Definition at line 197 of file RtpsUdpTransport.h.

◆ RtpsUdpSendStrategy

friend class RtpsUdpSendStrategy

friend

Definition at line 196 of file RtpsUdpTransport.h.

Member Data Documentation

◆ bit_sub_

RcHandle<BitSubscriber> OpenDDS::DCPS::RtpsUdpTransport::bit_sub_

private

Definition at line 135 of file RtpsUdpTransport.h.

Referenced by accept_datalink(), connect_datalink(), disable_relay_stun_task(), and process_relay_sra().

◆ connections_lock_

LockType OpenDDS::DCPS::RtpsUdpTransport::connections_lock_

private

Definition at line 133 of file RtpsUdpTransport.h.

Referenced by accept_datalink(), and connect_datalink().

◆ default_listener_

TransportClient_wrch OpenDDS::DCPS::RtpsUdpTransport::default_listener_

private

Definition at line 149 of file RtpsUdpTransport.h.

◆ deferred_connection_records_

ConnectionRecords OpenDDS::DCPS::RtpsUdpTransport::deferred_connection_records_

private

Definition at line 158 of file RtpsUdpTransport.h.

Referenced by disable_relay_stun_task(), and process_relay_sra().

◆ ice_agent_

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

private

Definition at line 190 of file RtpsUdpTransport.h.

Referenced by get_ice_agent(), start_ice(), and stop_ice().

◆ ice_endpoint_

RcHandle<IceEndpoint> OpenDDS::DCPS::RtpsUdpTransport::ice_endpoint_

private

Definition at line 178 of file RtpsUdpTransport.h.

Referenced by get_ice_endpoint(), relay_stun_task(), start_ice(), and stop_ice().

◆ job_queue_

JobQueue_rch OpenDDS::DCPS::RtpsUdpTransport::job_queue_

private

Definition at line 151 of file RtpsUdpTransport.h.

Referenced by configure_i(), disable_relay_stun_task(), process_relay_sra(), and shutdown_i().

◆ link_

RtpsUdpDataLink_rch OpenDDS::DCPS::RtpsUdpTransport::link_

private

RTPS uses only one link per transport. This link can be safely reused by any clients that belong to the same domain participant (same GUID prefix). Use by a second participant is not possible because the network location returned by connection_info_i() can't be shared among participants.

Definition at line 143 of file RtpsUdpTransport.h.

Referenced by accept_datalink(), client_stop(), configure_i(), connect_datalink(), get_last_recv_locator(), local_crypto_handle(), register_for_reader(), register_for_writer(), shutdown_i(), start_ice(), stop_accepting_or_connecting(), stop_ice(), unregister_for_reader(), unregister_for_writer(), update_locators(), and use_datalink().

◆ links_lock_

ThreadLockType OpenDDS::DCPS::RtpsUdpTransport::links_lock_

private

Definition at line 128 of file RtpsUdpTransport.h.

Referenced by accept_datalink(), client_stop(), configure_i(), connect_datalink(), get_last_recv_locator(), local_crypto_handle(), register_for_reader(), register_for_writer(), relay_stun_task(), shutdown_i(), start_ice(), stop_accepting_or_connecting(), stop_ice(), unregister_for_reader(), unregister_for_writer(), and update_locators().

◆ local_crypto_handle_

DDS::Security::ParticipantCryptoHandle OpenDDS::DCPS::RtpsUdpTransport::local_crypto_handle_

private

Definition at line 155 of file RtpsUdpTransport.h.

Referenced by local_crypto_handle(), and make_datalink().

◆ local_prefix_

GuidPrefix_t OpenDDS::DCPS::RtpsUdpTransport::local_prefix_

private

Definition at line 136 of file RtpsUdpTransport.h.

Referenced by make_datalink(), relay_stun_task(), and RtpsUdpTransport().

◆ relay_srsm_

ICE::ServerReflexiveStateMachine OpenDDS::DCPS::RtpsUdpTransport::relay_srsm_

private

Definition at line 185 of file RtpsUdpTransport.h.

Referenced by disable_relay_stun_task(), process_relay_sra(), and relay_stun_task().

◆ relay_stun_task_

RcHandle<Sporadic> OpenDDS::DCPS::RtpsUdpTransport::relay_stun_task_

private

Definition at line 182 of file RtpsUdpTransport.h.

Referenced by configure_i(), disable_relay_stun_task(), make_datalink(), relay_stun_task(), rtps_relay_address_change(), rtps_relay_only_now(), shutdown_i(), and use_rtps_relay_now().

◆ relay_stun_task_falloff_

FibonacciSequence<TimeDuration> OpenDDS::DCPS::RtpsUdpTransport::relay_stun_task_falloff_

private

Definition at line 183 of file RtpsUdpTransport.h.

Referenced by configure_i(), make_datalink(), process_relay_sra(), relay_stun_task(), rtps_relay_address_change(), rtps_relay_only_now(), and use_rtps_relay_now().

◆ relay_stun_task_falloff_mutex_

ThreadLockType OpenDDS::DCPS::RtpsUdpTransport::relay_stun_task_falloff_mutex_

private

Definition at line 184 of file RtpsUdpTransport.h.

Referenced by configure_i(), make_datalink(), process_relay_sra(), relay_stun_task(), rtps_relay_address_change(), rtps_relay_only_now(), and use_rtps_relay_now().

◆ transport_statistics_

InternalTransportStatistics OpenDDS::DCPS::RtpsUdpTransport::transport_statistics_

private

Definition at line 193 of file RtpsUdpTransport.h.

Referenced by append_transport_statistics(), and make_datalink().

◆ transport_statistics_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpTransport::transport_statistics_mutex_

private

Definition at line 194 of file RtpsUdpTransport.h.

Referenced by append_transport_statistics(), and make_datalink().

◆ unicast_socket_

ACE_SOCK_Dgram OpenDDS::DCPS::RtpsUdpTransport::unicast_socket_

private

Definition at line 145 of file RtpsUdpTransport.h.

Referenced by configure_i(), make_datalink(), start_ice(), and stop_ice().

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

Private Attributes
ThreadLockType	links_lock_

LockType	connections_lock_

RcHandle< BitSubscriber >	bit_sub_

GuidPrefix_t	local_prefix_

RtpsUdpDataLink_rch	link_

ACE_SOCK_Dgram	unicast_socket_

TransportClient_wrch	default_listener_

JobQueue_rch	job_queue_

DDS::Security::ParticipantCryptoHandle	local_crypto_handle_

ConnectionRecords	deferred_connection_records_

RcHandle< IceEndpoint >	ice_endpoint_

RcHandle< Sporadic >	relay_stun_task_

FibonacciSequence< TimeDuration >	relay_stun_task_falloff_

ThreadLockType	relay_stun_task_falloff_mutex_

ICE::ServerReflexiveStateMachine	relay_srsm_

RcHandle< ICE::Agent >	ice_agent_

InternalTransportStatistics	transport_statistics_

ACE_Thread_Mutex	transport_statistics_mutex_

Classes

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ GuardThreadType

◆ GuardType

◆ LockType

◆ Sporadic

◆ ThreadLockType

Constructor & Destructor Documentation

◆ RtpsUdpTransport()

Member Function Documentation

◆ accept_datalink()

◆ append_transport_statistics()

◆ client_stop()

◆ config()

◆ configure_i()

◆ connect_datalink()

◆ connection_info_i()

◆ disable_relay_stun_task()

◆ get_connection_addrs()

◆ get_ice_agent()

◆ get_ice_endpoint()

◆ get_last_recv_locator()

◆ local_crypto_handle()

◆ make_datalink()

◆ process_relay_sra()

◆ register_for_reader()

◆ register_for_writer()

◆ relay_srsm()

◆ relay_stun_task()

◆ release_datalink()

◆ rtps_relay_address_change()

◆ rtps_relay_only_now()

◆ shutdown_i()

◆ start_ice()

◆ stop_accepting_or_connecting()

◆ stop_ice()

◆ transport_type()

◆ unregister_for_reader()

◆ unregister_for_writer()

◆ update_locators()

◆ use_datalink()

◆ use_ice_now()

◆ use_rtps_relay_now()

Friends And Related Function Documentation

◆ RtpsUdpReceiveStrategy

◆ RtpsUdpSendStrategy

Member Data Documentation

◆ bit_sub_

◆ connections_lock_

◆ default_listener_

◆ deferred_connection_records_

◆ ice_agent_

◆ ice_endpoint_

◆ job_queue_

◆ link_

◆ links_lock_

◆ local_crypto_handle_

◆ local_prefix_

◆ relay_srsm_

◆ relay_stun_task_

◆ relay_stun_task_falloff_

◆ relay_stun_task_falloff_mutex_

◆ transport_statistics_

◆ transport_statistics_mutex_

◆ unicast_socket_