Inheritance diagram for OpenDDS::RTPS::Spdp::SpdpTransport:

[legend]

Collaboration diagram for OpenDDS::RTPS::Spdp::SpdpTransport:

[legend]

Classes
class	RegisterHandlers

Public Types
typedef size_t	WriteFlags

typedef DCPS::PmfPeriodicTask< SpdpTransport >	SpdpPeriodic

typedef DCPS::PmfSporadicTask< SpdpTransport >	SpdpSporadic

typedef DCPS::PmfMultiTask< SpdpTransport >	SpdpMulti

Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

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

Public Member Functions
	SpdpTransport (DCPS::RcHandle< Spdp > outer)

	~SpdpTransport ()

const ACE_SOCK_Dgram &	choose_recv_socket (ACE_HANDLE h) const

virtual int	handle_input (ACE_HANDLE h)

void	open (const DCPS::ReactorTask_rch &reactor_task, const DCPS::JobQueue_rch &job_queue)

void	register_unicast_socket (ACE_Reactor reactor, ACE_SOCK_Dgram &socket, const char what)

void	register_handlers (const DCPS::ReactorTask_rch &reactor_task)

void	enable_periodic_tasks ()

void	shorten_local_sender_delay_i ()

void	write (WriteFlags flags)

void	write_i (WriteFlags flags)

void	write_i (const DCPS::GUID_t &guid, const ACE_INET_Addr &local_address, WriteFlags flags)

void	send (WriteFlags flags, const ACE_INET_Addr &local_address=ACE_INET_Addr())

const ACE_SOCK_Dgram &	choose_send_socket (const ACE_INET_Addr &addr) const

ssize_t	send (const ACE_INET_Addr &addr, bool relay)

void	close (const DCPS::ReactorTask_rch &reactor_task)

void	dispose_unregister ()

bool	open_unicast_socket (u_short port_common, u_short participant_id)

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

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

ICE::AddressListType	host_addresses () const

void	send (const ACE_INET_Addr &address, const STUN::Message &message)

ACE_INET_Addr	stun_server_address () const

void	ice_connect (const ICE::GuidSetType &guids, const ACE_INET_Addr &addr)

void	ice_disconnect (const ICE::GuidSetType &guids, const ACE_INET_Addr &addr)

	OPENDDS_SET (ACE_INET_Addr) send_addrs_

void	send_local (const DCPS::MonotonicTimePoint &now)

void	send_directed (const DCPS::MonotonicTimePoint &now)

	OPENDDS_LIST (DCPS::GUID_t) directed_guids_

void	process_lease_expirations (const DCPS::MonotonicTimePoint &now)

void	thread_status_task (const DCPS::MonotonicTimePoint &now)

void	process_handshake_deadlines (const DCPS::MonotonicTimePoint &now)

void	process_handshake_resends (const DCPS::MonotonicTimePoint &now)

void	send_relay (const DCPS::MonotonicTimePoint &now)

void	relay_stun_task (const DCPS::MonotonicTimePoint &now)

void	process_relay_sra (ICE::ServerReflexiveStateMachine::StateChange)

void	disable_relay_stun_task ()

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

ACE_Event_Handler::Reference_Count	add_reference ()

ACE_Event_Handler::Reference_Count	remove_reference ()

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

virtual ACE_HANDLE	get_handle (void) const

virtual void	set_handle (ACE_HANDLE)

virtual int	priority (void) const

virtual void	priority (int priority)

virtual int	handle_output (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_exception (ACE_HANDLE fd=ACE_INVALID_HANDLE)

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

virtual int	handle_exit (ACE_Process *)

virtual int	handle_close (ACE_HANDLE handle, ACE_Reactor_Mask close_mask)

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

virtual int	resume_handler (void)

virtual int	handle_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual int	handle_group_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual void	reactor (ACE_Reactor *reactor)

virtual ACE_Reactor *	reactor (void) const

virtual ACE_Reactor_Timer_Interface *	reactor_timer_interface (void) const

Reference_Counting_Policy &	reference_counting_policy (void)

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< DCPS::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)

Public Member Functions inherited from OpenDDS::ICE::Endpoint
virtual	~Endpoint ()

Public Attributes
DCPS::WeakRcHandle< Spdp >	outer_

Header	hdr_

DataSubmessage	data_

DCPS::SequenceNumber	seq_

u_short	uni_port_

ACE_SOCK_Dgram	unicast_socket_

OPENDDS_STRING	multicast_interface_

ACE_INET_Addr	multicast_address_

ACE_SOCK_Dgram_Mcast	multicast_socket_

DCPS::MulticastManager	multicast_manager_

ACE_Message_Block	buff_

ACE_Message_Block	wbuff_

DCPS::RcHandle< SpdpMulti >	local_send_task_

DCPS::RcHandle< SpdpSporadic >	directed_send_task_

DCPS::RcHandle< SpdpSporadic >	lease_expiration_task_

DCPS::RcHandle< SpdpPeriodic >	thread_status_task_

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

DCPS::RcHandle< SpdpSporadic >	handshake_deadline_task_

DCPS::RcHandle< SpdpSporadic >	handshake_resend_task_

DCPS::RcHandle< SpdpSporadic >	relay_spdp_task_

DCPS::FibonacciSequence< TimeDuration >	relay_spdp_task_falloff_

DCPS::RcHandle< SpdpSporadic >	relay_stun_task_

DCPS::FibonacciSequence< TimeDuration >	relay_stun_task_falloff_

ICE::ServerReflexiveStateMachine	relay_srsm_

bool	network_is_unreachable_

bool	ice_endpoint_added_

DCPS::InternalTransportStatistics	transport_statistics_

DCPS::MonotonicTimePoint	last_harvest

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

Static Public Attributes
static const WriteFlags	SEND_MULTICAST = (1 << 0)

static const WriteFlags	SEND_RELAY = (1 << 1)

static const WriteFlags	SEND_DIRECT = (1 << 2)

Additional Inherited Members
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)

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

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 ()

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Definition at line 425 of file Spdp.h.

Member Typedef Documentation

◆ SpdpMulti

typedef DCPS::PmfMultiTask<SpdpTransport> OpenDDS::RTPS::Spdp::SpdpTransport::SpdpMulti

Definition at line 523 of file Spdp.h.

◆ SpdpPeriodic

typedef DCPS::PmfPeriodicTask<SpdpTransport> OpenDDS::RTPS::Spdp::SpdpTransport::SpdpPeriodic

Definition at line 521 of file Spdp.h.

◆ SpdpSporadic

typedef DCPS::PmfSporadicTask<SpdpTransport> OpenDDS::RTPS::Spdp::SpdpTransport::SpdpSporadic

Definition at line 522 of file Spdp.h.

◆ WriteFlags

typedef size_t OpenDDS::RTPS::Spdp::SpdpTransport::WriteFlags

Definition at line 432 of file Spdp.h.

Constructor & Destructor Documentation

◆ SpdpTransport()

OpenDDS::RTPS::Spdp::SpdpTransport::SpdpTransport ( DCPS::RcHandle< Spdp > outer )

explicit

Definition at line 2308 of file Spdp.cpp.

References ACE_ERROR, OpenDDS::DCPS::assign(), OpenDDS::RTPS::DATA, data_, OpenDDS::RTPS::DATA_OCTETS_TO_IQOS, ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE, OpenDDS::DCPS::ENTITYID_SPDP_BUILTIN_PARTICIPANT_WRITER, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::RTPS::DataSubmessage::extraFlags, OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::SubmessageHeader::flags, ACE_OS::getpid(), OpenDDS::RTPS::Header::guidPrefix, hdr_, OpenDDS::RTPS::SequenceNumber_t::high, LM_WARNING, OpenDDS::DCPS::log_level, OpenDDS::RTPS::SequenceNumber_t::low, multicast_address_, multicast_interface_, multicast_socket_, OpenDDS::RTPS::DataSubmessage::octetsToInlineQos, open_unicast_socket(), ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO, ACE_SOCK_Dgram_Mcast::opts(), OpenDDS::RTPS::Header::prefix, OpenDDS::RTPS::PROTOCOLVERSION, OpenDDS::RTPS::DataSubmessage::readerId, OpenDDS::RTPS::DataSubmessage::smHeader, OpenDDS::RTPS::SubmessageHeader::submessageId, OpenDDS::RTPS::SubmessageHeader::submessageLength, uni_port_, OpenDDS::RTPS::Header::vendorId, OpenDDS::RTPS::VENDORID_OPENDDS, OpenDDS::RTPS::Header::version, OpenDDS::DCPS::LogLevel::Warning, OpenDDS::RTPS::DataSubmessage::writerId, and OpenDDS::RTPS::DataSubmessage::writerSN.

   : outer_(outer)
   , buff_(64 * 1024)
   , wbuff_(64 * 1024)
 #ifdef OPENDDS_SECURITY
   , relay_spdp_task_falloff_(outer->config()->sedp_heartbeat_period())
   , relay_stun_task_falloff_(outer->config()->sedp_heartbeat_period())
 #endif
   , network_is_unreachable_(false)
   , ice_endpoint_added_(false)
   , transport_statistics_(DCPS::TransportRegistry::DEFAULT_INST_PREFIX +
                           OPENDDS_STRING("_SPDPTransportInst_") +
                           DCPS::GuidConverter(outer->guid_).uniqueParticipantId() +
                           DCPS::to_dds_string(outer->domain_))
 {
   hdr_.prefix[0] = 'R';
   hdr_.prefix[1] = 'T';
   hdr_.prefix[2] = 'P';
   hdr_.prefix[3] = 'S';
   hdr_.version = PROTOCOLVERSION;
   hdr_.vendorId = VENDORID_OPENDDS;
   DCPS::assign(hdr_.guidPrefix, outer->guid_.guidPrefix);
   data_.smHeader.submessageId = DATA;
   data_.smHeader.flags = FLAG_E | FLAG_D;
   data_.smHeader.submessageLength = 0; // last submessage in the Message
   data_.extraFlags = 0;
   data_.octetsToInlineQos = DATA_OCTETS_TO_IQOS;
   data_.readerId = ENTITYID_UNKNOWN;
   data_.writerId = ENTITYID_SPDP_BUILTIN_PARTICIPANT_WRITER;
   data_.writerSN.high = 0;
   data_.writerSN.low = 0;
 
 #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
   multicast_ipv6_socket_.opts(ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO |
                               ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE);
 #endif
 #endif
 
   multicast_interface_ = outer->disco_->multicast_interface();
 
   const u_short port_common = outer->config_->port_common(outer->domain_);
   multicast_address_ = outer->config_->multicast_address(port_common);
 
 #ifdef ACE_HAS_IPV6
   multicast_ipv6_address_ = outer->config_->ipv6_multicast_address(port_common);
 #endif
 
   send_addrs_.insert(multicast_address_);
 #ifdef ACE_HAS_IPV6
   send_addrs_.insert(multicast_ipv6_address_);
 #endif
 
   typedef RtpsDiscovery::AddrVec::const_iterator iter;
   const RtpsDiscovery::AddrVec addrs = outer->config_->spdp_send_addrs();
   for (iter it = addrs.begin(),
        end = addrs.end(); it != end; ++it) {
     send_addrs_.insert(ACE_INET_Addr(it->c_str()));
   }
 
   u_short participantId = 0;
 
 #ifdef OPENDDS_SAFETY_PROFILE
   const u_short startingParticipantId = participantId;
 #endif
 
   const u_short max_part_id = 119; // RTPS 2.5 9.6.2.3
   while (!open_unicast_socket(port_common, participantId)) {
     if (participantId == max_part_id && log_level >= LogLevel::Warning) {
       ACE_ERROR((LM_WARNING, "(%P|%t) WARNING: Spdp::SpdpTransport: "
         "participant id is going above max %u allowed by RTPS spec\n", max_part_id));
       // As long as it doesn't result in an invalid port, going past this
       // shouldn't cause a problem, but it could be a sign that OpenDDS has a
       // limited number of ports at its disposal. Also another implementation
       // could use this as a hard limit, but that's much less of a concern.
     }
     ++participantId;
   }
 #ifdef ACE_HAS_IPV6
   u_short port = uni_port_;
 
   while (!open_unicast_ipv6_socket(port)) {
     ++port;
   }
 #endif
 
 #ifdef OPENDDS_SAFETY_PROFILE
   if (participantId > startingParticipantId && ACE_OS::getpid() == -1) {
     // Since pids are not available, use the fact that we had to increment
     // participantId to modify the GUID's pid bytes.  This avoids GUID conflicts
     // between processes on the same host which start at the same time
     // (resulting in the same seed value for the random number generator).
     hdr_.guidPrefix[8] = static_cast<CORBA::Octet>(participantId >> 8);
     hdr_.guidPrefix[9] = static_cast<CORBA::Octet>(participantId & 0xFF);
     outer->guid_.guidPrefix[8] = hdr_.guidPrefix[8];
     outer->guid_.guidPrefix[9] = hdr_.guidPrefix[9];
   }
 #endif
 }

◆ ~SpdpTransport()

OpenDDS::RTPS::Spdp::SpdpTransport::~SpdpTransport ( )

Definition at line 2477 of file Spdp.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), ACE_SOCK::close(), OpenDDS::DCPS::DCPS_debug_level, dispose_unregister(), LM_INFO, multicast_socket_, outer_, and unicast_socket_.

 {
   if (DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, ACE_TEXT("(%P|%t) SpdpTransport::~SpdpTransport\n")));
   }
 
   DCPS::RcHandle<Spdp> outer = outer_.lock();
 
   if (outer) {
     ACE_GUARD(ACE_Thread_Mutex, g, outer->lock_);
     try {
       dispose_unregister();
     } catch (const CORBA::BAD_PARAM&) {}
     outer->eh_shutdown_ = true;
     outer->shutdown_cond_.notify_all();
   }
 
   unicast_socket_.close();
   multicast_socket_.close();
 #ifdef ACE_HAS_IPV6
   unicast_ipv6_socket_.close();
   multicast_ipv6_socket_.close();
 #endif
 }

Member Function Documentation

◆ choose_recv_socket()

const ACE_SOCK_Dgram & OpenDDS::RTPS::Spdp::SpdpTransport::choose_recv_socket ( ACE_HANDLE h ) const

Definition at line 2966 of file Spdp.cpp.

References ACE_IPC_SAP::get_handle(), multicast_socket_, and unicast_socket_.

Referenced by handle_input().

 {
 #ifdef ACE_HAS_IPV6
   if (h == unicast_ipv6_socket_.get_handle()) {
     return unicast_ipv6_socket_;
   }
   if (h == multicast_ipv6_socket_.get_handle()) {
     return multicast_ipv6_socket_;
   }
 #endif
   if (h == multicast_socket_.get_handle()) {
     return multicast_socket_;
   }
 
   return unicast_socket_;
 }

◆ choose_send_socket()

const ACE_SOCK_Dgram & OpenDDS::RTPS::Spdp::SpdpTransport::choose_send_socket ( const ACE_INET_Addr & addr ) const

Definition at line 2909 of file Spdp.cpp.

References ACE_Addr::get_type(), and unicast_socket_.

Referenced by send().

 {
 #ifdef ACE_HAS_IPV6
   if (addr.get_type() == AF_INET6) {
     return unicast_ipv6_socket_;
   }
 #endif
   ACE_UNUSED_ARG(addr);
   return unicast_socket_;
 }

◆ close()

void OpenDDS::RTPS::Spdp::SpdpTransport::close ( const DCPS::ReactorTask_rch & reactor_task )

Definition at line 2603 of file Spdp.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, directed_send_task_, ACE_Event_Handler::DONT_CALL, ACE_IPC_SAP::get_handle(), get_ice_endpoint(), OpenDDS::DCPS::ReactorTask::get_reactor(), handshake_deadline_task_, handshake_resend_task_, ice_endpoint_added_, lease_expiration_task_, LM_INFO, local_send_task_, multicast_socket_, network_interface_address_reader_, outer_, ACE_Event_Handler::reactor(), ACE_Event_Handler::READ_MASK, relay_spdp_task_, relay_stun_task_, ACE_Reactor::remove_handler(), TheServiceParticipant, thread_status_task_, and unicast_socket_.

 {
   if (DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, ACE_TEXT("(%P|%t) SpdpTransport::close\n")));
   }
 
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   TheServiceParticipant->network_interface_address_topic()->disconnect(network_interface_address_reader_);
 
 #ifdef OPENDDS_SECURITY
   DCPS::WeakRcHandle<ICE::Endpoint> endpoint = get_ice_endpoint();
   if (endpoint) {
     outer->ice_agent_->remove_endpoint(endpoint);
     ice_endpoint_added_ = false;
   }
 
   if (handshake_deadline_task_) {
     handshake_deadline_task_->cancel();
   }
   if (handshake_resend_task_) {
     handshake_resend_task_->cancel();
   }
   if (relay_spdp_task_) {
     relay_spdp_task_->cancel();
   }
   if (relay_stun_task_) {
     relay_stun_task_->cancel();
   }
 #endif
   if (local_send_task_) {
     local_send_task_->disable();
   }
   if (directed_send_task_) {
     directed_send_task_->cancel();
   }
   if (lease_expiration_task_) {
     lease_expiration_task_->cancel();
   }
   if (thread_status_task_) {
     thread_status_task_->disable();
   }
 
   ACE_Reactor* reactor = reactor_task->get_reactor();
   const ACE_Reactor_Mask mask =
     ACE_Event_Handler::READ_MASK | ACE_Event_Handler::DONT_CALL;
   reactor->remove_handler(multicast_socket_.get_handle(), mask);
   reactor->remove_handler(unicast_socket_.get_handle(), mask);
 #ifdef ACE_HAS_IPV6
   reactor->remove_handler(multicast_ipv6_socket_.get_handle(), mask);
   reactor->remove_handler(unicast_ipv6_socket_.get_handle(), mask);
 #endif
 }

◆ disable_relay_stun_task()

void OpenDDS::RTPS::Spdp::SpdpTransport::disable_relay_stun_task ( )

Definition at line 4211 of file Spdp.cpp.

References OpenDDS::DCPS::ConnectionRecord::address, OpenDDS::DCPS::ConnectionRecord::guid, OpenDDS::DCPS::ConnectionRecord::latency, outer_, 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.

 {
 #ifndef DDS_HAS_MINIMUM_BIT
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   relay_stun_task_->cancel();
 
   DCPS::ConnectionRecord connection_record;
   std::memset(connection_record.guid, 0, sizeof(connection_record.guid));
   connection_record.protocol = DCPS::RTPS_RELAY_STUN_PROTOCOL;
   connection_record.latency = DCPS::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();
     outer->sedp_->job_queue()->enqueue(DCPS::make_rch<DCPS::WriteConnectionRecords>(outer->bit_subscriber_, false, connection_record));
   }
 
   relay_srsm_ = ICE::ServerReflexiveStateMachine();
 #endif
 }

◆ dispose_unregister()

void OpenDDS::RTPS::Spdp::SpdpTransport::dispose_unregister ( )

Definition at line 2570 of file Spdp.cpp.

References ACE_ERROR, ACE_TEXT(), data_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::Serializer::encoding(), OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::FLAG_K_IN_DATA, OpenDDS::RTPS::FLAG_Q, OpenDDS::RTPS::SubmessageHeader::flags, hdr_, OpenDDS::RTPS::DataSubmessage::inlineQos, LM_ERROR, OpenDDS::DCPS::MUTABLE, outer_, OpenDDS::RTPS::PID_PARTICIPANT_GUID, ACE_Message_Block::reset(), send(), SEND_MULTICAST, SEND_RELAY, seq_, OpenDDS::RTPS::DataSubmessage::smHeader, OpenDDS::RTPS::to_rtps_seqnum(), wbuff_, and OpenDDS::RTPS::DataSubmessage::writerSN.

Referenced by ~SpdpTransport().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   // Send the dispose/unregister SPDP sample
   data_.writerSN = to_rtps_seqnum(seq_);
   data_.smHeader.flags = FLAG_E | FLAG_Q | FLAG_K_IN_DATA;
   data_.inlineQos.length(1);
   static const StatusInfo_t dispose_unregister = { {0, 0, 0, 3} };
   data_.inlineQos[0].status_info(dispose_unregister);
 
   ParameterList plist(1);
   plist.length(1);
   plist[0].guid(outer->guid_);
   plist[0]._d(PID_PARTICIPANT_GUID);
 
   wbuff_.reset();
   DCPS::Serializer ser(&wbuff_, encoding_plain_native);
   DCPS::EncapsulationHeader encap(ser.encoding(), DCPS::MUTABLE);
   if (!(ser << hdr_) || !(ser << data_) || !(ser << encap) || !(ser << plist)) {
     if (DCPS::DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::dispose_unregister() - ")
         ACE_TEXT("failed to serialize headers for dispose/unregister\n")));
     }
     return;
   }
 
   send(SEND_MULTICAST | SEND_RELAY);
 }

◆ enable_periodic_tasks()

void OpenDDS::RTPS::Spdp::SpdpTransport::enable_periodic_tasks ( )

Definition at line 2544 of file Spdp.cpp.

References local_send_task_, outer_, relay_spdp_task_, relay_spdp_task_falloff_, relay_stun_task_, relay_stun_task_falloff_, TheServiceParticipant, OpenDDS::DCPS::ThreadStatusManager::thread_status_interval(), thread_status_task_, OpenDDS::DCPS::ThreadStatusManager::update_thread_status(), and OpenDDS::DCPS::TimeDuration::zero_value.

 {
   if (local_send_task_) {
     local_send_task_->enable(TimeDuration::zero_value);
   }
 
 #ifdef OPENDDS_SECURITY
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   relay_spdp_task_falloff_.set(outer->config_->sedp_heartbeat_period());
   relay_spdp_task_->schedule(TimeDuration::zero_value);
 
   relay_stun_task_falloff_.set(outer->config_->sedp_heartbeat_period());
   relay_stun_task_->schedule(TimeDuration::zero_value);
 #endif
 
 #ifndef DDS_HAS_MINIMUM_BIT
   const DCPS::ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
   if (thread_status_manager.update_thread_status()) {
     thread_status_task_->enable(false, thread_status_manager.thread_status_interval());
   }
 #endif /* DDS_HAS_MINIMUM_BIT */
 }

◆ get_ice_endpoint()

DCPS::WeakRcHandle< ICE::Endpoint > OpenDDS::RTPS::Spdp::SpdpTransport::get_ice_endpoint ( )

Definition at line 3235 of file Spdp.cpp.

References outer_, OpenDDS::DCPS::rchandle_from(), and OpenDDS::DCPS::static_rchandle_cast().

Referenced by close(), handle_input(), open(), and write_i().

 {
 #ifdef OPENDDS_SECURITY
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   return outer && outer->config_->use_ice() ? DCPS::static_rchandle_cast<ICE::Endpoint>(rchandle_from(this)) : DCPS::WeakRcHandle<ICE::Endpoint>();
 #else
   return DCPS::WeakRcHandle<ICE::Endpoint>();
 #endif
 }

◆ handle_input()

int OpenDDS::RTPS::Spdp::SpdpTransport::handle_input ( ACE_HANDLE h )

virtual

Reimplemented from ACE_Event_Handler.

Definition at line 2989 of file Spdp.cpp.

References ACE_CDR_BYTE_ORDER, ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_NOTSUP_RETURN, ACE_TEXT(), OpenDDS::RTPS::append_submessage(), OpenDDS::STUN::Message::block, buff_, choose_recv_socket(), OpenDDS::RTPS::DATA, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::STUN::encoding(), OpenDDS::DCPS::Serializer::encoding(), OpenDDS::DCPS::ENTITYID_PARTICIPANT, OpenDDS::DCPS::ENTITYID_SPDP_BUILTIN_PARTICIPANT_WRITER, OpenDDS::RTPS::FLAG_D, OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::FLAG_K_IN_DATA, OpenDDS::RTPS::SubmessageHeader::flags, ACE_IPC_SAP::get_handle(), get_ice_endpoint(), OpenDDS::RTPS::Spdp::guid(), OpenDDS::RTPS::Header::guidPrefix, OpenDDS::RTPS::Message::hdr, header, OpenDDS::RTPS::INFO_DST, OpenDDS::ICE::ServerReflexiveStateMachine::is_response(), OpenDDS::DCPS::Encoding::kind(), OpenDDS::DCPS::Encoding::KIND_XCDR1, ACE_Message_Block::length(), LM_ERROR, LM_WARNING, OpenDDS::DCPS::make_id(), OpenDDS::DCPS::MCK_RTPS, OpenDDS::DCPS::MCK_STUN, ACE_OS::memcmp(), OpenDDS::DCPS::InternalTransportStatistics::message_count, OpenDDS::DCPS::MUTABLE, outer_, OpenDDS::RTPS::PID_PARTICIPANT_GUID, process_relay_sra(), ACE_Message_Block::rd_ptr(), read(), OpenDDS::ICE::ServerReflexiveStateMachine::receive(), relay_srsm_, ACE_Message_Block::reset(), ACE_Message_Block::size(), OpenDDS::DCPS::Serializer::skip(), OpenDDS::RTPS::SMHDR_SZ, OpenDDS::RTPS::DataSubmessage::smHeader, OpenDDS::RTPS::InfoDestinationSubmessage::smHeader, socket(), ACE_Message_Block::space(), OpenDDS::RTPS::SubmessageHeader::submessageLength, OpenDDS::DCPS::Serializer::swap_bytes(), TheServiceParticipant, OpenDDS::DCPS::EncapsulationHeader::to_encoding(), OpenDDS::DCPS::transport_debug, transport_statistics_, unicast_socket_, OpenDDS::RTPS::valid_size(), ACE_Message_Block::wr_ptr(), and OpenDDS::RTPS::DataSubmessage::writerId.

 {
   DCPS::ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());
 
   const ACE_SOCK_Dgram& socket = choose_recv_socket(h);
   ACE_INET_Addr remote;
   buff_.reset();
 
 #ifdef ACE_LACKS_SENDMSG
   const ssize_t bytes = socket.recv(buff_.wr_ptr(), buff_.space(), remote);
 #else
   ACE_INET_Addr local;
 
   iovec iov[1];
   iov[0].iov_base = buff_.wr_ptr();
 #ifdef _MSC_VER
 #pragma warning(push)
   // iov_len is 32-bit on 64-bit VC++, but we don't want a cast here
   // since on other platforms iov_len is 64-bit
 #pragma warning(disable : 4267)
 #endif
   iov[0].iov_len = buff_.space();
 #ifdef _MSC_VER
 #pragma warning(pop)
 #endif
   const ssize_t bytes = socket.recv(iov, 1, remote, 0
 #if defined(ACE_RECVPKTINFO) || defined(ACE_RECVPKTINFO6)
                                     , &local
 #endif
                                     );
 #endif
 
   if (!valid_size(remote)) {
     ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::handle_input() - invalid address size\n")));
     return 0;
   }
 
   if (bytes > 0) {
     buff_.wr_ptr(bytes);
   } else if (bytes == 0) {
     return 0;
   } else {
     ACE_DEBUG((
           LM_WARNING,
           ACE_TEXT("(%P|%t) WARNING: Spdp::SpdpTransport::handle_input() - ")
           ACE_TEXT("error reading from %C socket %p\n")
           , (h == unicast_socket_.get_handle()) ? "unicast" : "multicast",
           ACE_TEXT("ACE_SOCK_Dgram::recv")));
     return 0;
   }
 
   DCPS::RcHandle<Spdp> outer = outer_.lock();
 
   if (!outer) {
     return 0;
   }
 
   const bool relay_in_use = (outer->config_->rtps_relay_only() || outer->config_->use_rtps_relay());
   const bool remote_matches_relay_addr = (remote == outer->config_->spdp_rtps_relay_address());
   const bool from_relay = relay_in_use && remote_matches_relay_addr;
 
   // Ignore messages from the relay when not using it.
   if (!relay_in_use && remote_matches_relay_addr) {
     return 0;
   }
 
   if ((buff_.size() >= 4) && ACE_OS::memcmp(buff_.rd_ptr(), "RTPS", 4) == 0) {
     RTPS::Message message;
 
     DCPS::Serializer ser(&buff_, encoding_plain_native);
     Header header;
     if (!(ser >> header)) {
       if (DCPS::DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::handle_input() - ")
                   ACE_TEXT("failed to deserialize RTPS header for SPDP\n")));
       }
       return 0;
     }
 
     if (outer->sedp_->transport_inst()->count_messages()) {
       const DCPS::InternalMessageCountKey key(DCPS::NetworkAddress(remote), DCPS::MCK_RTPS, from_relay);
       ACE_GUARD_RETURN(ACE_Thread_Mutex, g, outer->lock_, -1);
       transport_statistics_.message_count[key].recv(bytes);
     }
 
     if (DCPS::transport_debug.log_messages) {
       message.hdr = header;
     }
 
     while (buff_.length() > 3) {
       const char subm = buff_.rd_ptr()[0], flags = buff_.rd_ptr()[1];
       ser.swap_bytes((flags & FLAG_E) != ACE_CDR_BYTE_ORDER);
       const size_t start = buff_.length();
       CORBA::UShort submessageLength = 0;
       switch (subm) {
       case DATA: {
         DataSubmessage data;
         if (!(ser >> data)) {
           if (DCPS::DCPS_debug_level > 0) {
             ACE_ERROR((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::handle_input() - ")
                       ACE_TEXT("failed to deserialize DATA header for SPDP\n")));
           }
           return 0;
         }
         submessageLength = data.smHeader.submessageLength;
 
         if (DCPS::transport_debug.log_messages) {
           append_submessage(message, data);
         }
 
         if (data.writerId != ENTITYID_SPDP_BUILTIN_PARTICIPANT_WRITER) {
           // Not our message: this could be the same multicast group used
           // for SEDP and other traffic.
           break;
         }
 
         ParameterList plist;
         if (data.smHeader.flags & (FLAG_D | FLAG_K_IN_DATA)) {
           DCPS::EncapsulationHeader encap;
           DCPS::Encoding enc;
           if (!(ser >> encap) || !encap.to_encoding(enc, DCPS::MUTABLE) || enc.kind() != Encoding::KIND_XCDR1) {
             if (DCPS::DCPS_debug_level > 0) {
               ACE_ERROR((LM_ERROR,
                         ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::handle_input() - ")
                         ACE_TEXT("failed to deserialize encapsulation header for SPDP\n")));
             }
             return 0;
           }
           ser.encoding(enc);
           if (!(ser >> plist)) {
 
             if (DCPS::DCPS_debug_level > 0) {
               ACE_ERROR((LM_ERROR,
                         ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::handle_input() - ")
                         ACE_TEXT("failed to deserialize data payload for SPDP\n")));
             }
             return 0;
           }
         } else {
           plist.length(1);
           const GUID_t guid = make_id(header.guidPrefix, ENTITYID_PARTICIPANT);
           plist[0].guid(guid);
           plist[0]._d(PID_PARTICIPANT_GUID);
         }
 
         DCPS::RcHandle<Spdp> outer = outer_.lock();
         if (outer) {
           outer->data_received(data, plist, remote);
         }
         break;
       }
       case INFO_DST: {
         if (DCPS::transport_debug.log_messages) {
           InfoDestinationSubmessage sm;
           if (!(ser >> sm)) {
             if (DCPS::DCPS_debug_level > 0) {
               ACE_ERROR((LM_ERROR,
                         ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::handle_input() - ")
                         ACE_TEXT("failed to deserialize INFO_DST header for SPDP\n")));
             }
             return 0;
           }
           submessageLength = sm.smHeader.submessageLength;
           append_submessage(message, sm);
           break;
         }
       }
       // fallthrough
       default:
         SubmessageHeader smHeader;
         if (!(ser >> smHeader)) {
           if (DCPS::DCPS_debug_level > 0) {
             ACE_ERROR((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::handle_input() - ")
                       ACE_TEXT("failed to deserialize SubmessageHeader for SPDP\n")));
           }
           return 0;
         }
         submessageLength = smHeader.submessageLength;
         break;
       }
       if (submessageLength && buff_.length()) {
         const size_t read = start - buff_.length();
         if (read < static_cast<size_t>(submessageLength + SMHDR_SZ)) {
           if (!ser.skip(static_cast<CORBA::UShort>(submessageLength + SMHDR_SZ
                                                    - read))) {
             if (DCPS::DCPS_debug_level > 0) {
               ACE_ERROR((LM_ERROR,
                         ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::handle_input() - ")
                         ACE_TEXT("failed to skip sub message length\n")));
             }
             return 0;
           }
         }
       } else if (!submessageLength) {
         break; // submessageLength of 0 indicates the last submessage
       }
     }
 
   } else if ((buff_.size() >= 4) && (ACE_OS::memcmp(buff_.rd_ptr(), "RTPX", 4) == 0)) {
     // Handle some RTI protocol multicast to the same address
     return 0; // Ignore
   }
 
 #ifdef OPENDDS_SECURITY
   // Assume STUN
   if (!outer->initialized() || outer->shutting_down()) {
     return 0;
   }
 
 #ifndef ACE_RECVPKTINFO
   ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::handle_input() ")
              ACE_TEXT("potential STUN message received but this version of the ACE ")
              ACE_TEXT("library doesn't support the local_address extension in ")
              ACE_TEXT("ACE_SOCK_Dgram::recv\n")));
   ACE_NOTSUP_RETURN(0);
 #else
 
   DCPS::Serializer serializer(&buff_, STUN::encoding);
   STUN::Message message;
   message.block = &buff_;
   if (serializer >> message) {
     if (outer->sedp_->transport_inst()->count_messages()) {
       const DCPS::InternalMessageCountKey key(DCPS::NetworkAddress(remote), DCPS::MCK_STUN, from_relay);
       ACE_GUARD_RETURN(ACE_Thread_Mutex, g, outer->lock_, -1);
       transport_statistics_.message_count[key].recv(bytes);
     }
 
     if (relay_srsm_.is_response(message)) {
       process_relay_sra(relay_srsm_.receive(message));
     } else {
       DCPS::WeakRcHandle<ICE::Endpoint> endpoint = get_ice_endpoint();
       if (endpoint) {
         outer->ice_agent_->receive(endpoint, local, remote, message);
       }
     }
   }
 #endif
 #endif
 
   return 0;
 }

◆ host_addresses()

ICE::AddressListType OpenDDS::RTPS::Spdp::SpdpTransport::host_addresses ( ) const

virtual

Implements OpenDDS::ICE::Endpoint.

Definition at line 3247 of file Spdp.cpp.

References AF_INET, OpenDDS::DCPS::get_interface_addrs(), ACE_SOCK::get_local_addr(), ACE_INET_Addr::get_port_number(), ACE_INET_Addr::is_any(), and unicast_socket_.

 {
   ICE::AddressListType addresses;
   ACE_INET_Addr addr;
 
   unicast_socket_.get_local_addr(addr);
   if (addr != ACE_INET_Addr()) {
     if (addr.is_any()) {
       ICE::AddressListType addrs;
       DCPS::get_interface_addrs(addrs);
       for (ICE::AddressListType::iterator pos = addrs.begin(), limit = addrs.end(); pos != limit; ++pos) {
         if (pos->get_type() == AF_INET) {
           pos->set_port_number(addr.get_port_number());
           addresses.push_back(*pos);
         }
       }
     } else {
       addresses.push_back(addr);
     }
   }
 
 #ifdef ACE_HAS_IPV6
   unicast_ipv6_socket_.get_local_addr(addr);
   if (addr != ACE_INET_Addr()) {
     if (addr.is_any()) {
       ICE::AddressListType addrs;
       DCPS::get_interface_addrs(addrs);
       for (ICE::AddressListType::iterator pos = addrs.begin(), limit = addrs.end(); pos != limit; ++pos) {
         if (pos->get_type() == AF_INET6) {
           pos->set_port_number(addr.get_port_number());
           addresses.push_back(*pos);
         }
       }
     } else {
       addresses.push_back(addr);
     }
   }
 #endif
 
   return addresses;
 }

◆ ice_connect()

void OpenDDS::RTPS::Spdp::SpdpTransport::ice_connect	(	const ICE::GuidSetType &	guids,
		const ACE_INET_Addr &	addr
	)

virtual

Reimplemented from OpenDDS::ICE::Endpoint.

Definition at line 3361 of file Spdp.cpp.

References outer_.

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   outer->sedp_->job_queue()->enqueue(DCPS::make_rch<IceConnect>(outer, guids, addr, true));
 }

◆ ice_disconnect()

void OpenDDS::RTPS::Spdp::SpdpTransport::ice_disconnect	(	const ICE::GuidSetType &	guids,
		const ACE_INET_Addr &	addr
	)

virtual

Reimplemented from OpenDDS::ICE::Endpoint.

Definition at line 3383 of file Spdp.cpp.

References outer_.

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   outer->sedp_->job_queue()->enqueue(DCPS::make_rch<IceConnect>(outer, guids, addr, false));
 }

◆ on_data_available()

void OpenDDS::RTPS::Spdp::SpdpTransport::on_data_available ( DCPS::RcHandle< DCPS::InternalDataReader< DCPS::NetworkInterfaceAddress > > reader )

Definition at line 3603 of file Spdp.cpp.

References ACE_GUARD, DDS::ANY_INSTANCE_STATE, DDS::ANY_SAMPLE_STATE, DDS::ANY_VIEW_STATE, DDS::LENGTH_UNLIMITED, multicast_address_, multicast_interface_, multicast_manager_, multicast_socket_, network_interface_address_reader_, outer_, OpenDDS::DCPS::MulticastManager::process(), ACE_Event_Handler::reactor(), and shorten_local_sender_delay_i().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   ACE_GUARD(ACE_Thread_Mutex, g, outer->lock_);
   if (outer->shutting_down()) {
     return;
   }
 
   if (outer->shutdown_flag_) {
     return;
   }
 
   DCPS::InternalDataReader<DCPS::NetworkInterfaceAddress>::SampleSequence samples;
   DCPS::InternalSampleInfoSequence infos;
 
   network_interface_address_reader_->take(samples, infos, DDS::LENGTH_UNLIMITED, DDS::ANY_SAMPLE_STATE, DDS::ANY_VIEW_STATE, DDS::ANY_INSTANCE_STATE);
 
   if (multicast_manager_.process(samples,
                                  infos,
                                  multicast_interface_,
                                  reactor(),
                                  this,
                                  DCPS::NetworkAddress(multicast_address_),
                                  multicast_socket_
 #ifdef ACE_HAS_IPV6
                                  , DCPS::NetworkAddress(multicast_ipv6_address_),
                                  multicast_ipv6_socket_
 #endif
                                  )) {
     shorten_local_sender_delay_i();
   }
 }

◆ open()

void OpenDDS::RTPS::Spdp::SpdpTransport::open	(	const DCPS::ReactorTask_rch &	reactor_task,
		const DCPS::JobQueue_rch &	job_queue
	)

Definition at line 2411 of file Spdp.cpp.

References directed_send_task_, OpenDDS::DCPS::DataReaderQosBuilder::durability_transient_local(), OpenDDS::DCPS::ReactorInterceptor::execute_or_enqueue(), get_ice_endpoint(), OpenDDS::DCPS::ReactorTask::get_reactor(), handshake_deadline_task_, handshake_resend_task_, ice_endpoint_added_, OpenDDS::DCPS::ReactorTask::interceptor(), OpenDDS::DCPS::InternalDataReaderListener< DCPS::NetworkInterfaceAddress >::job_queue(), lease_expiration_task_, local_send_task_, network_interface_address_reader_, outer_, process_handshake_deadlines(), process_handshake_resends(), process_lease_expirations(), OpenDDS::DCPS::rchandle_from(), ACE_Event_Handler::reactor(), OpenDDS::DCPS::ref(), relay_spdp_task_, relay_stun_task(), relay_stun_task_, OpenDDS::DCPS::DataReaderQosBuilder::reliability_reliable(), send_directed(), send_local(), send_relay(), TheServiceParticipant, thread_status_task(), and thread_status_task_.

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
 #ifdef OPENDDS_SECURITY
   // Add the endpoint before any sending and receiving occurs.
   DCPS::WeakRcHandle<ICE::Endpoint> endpoint = get_ice_endpoint();
   if (endpoint) {
     outer->ice_agent_->add_endpoint(endpoint);
     ice_endpoint_added_ = true;
     outer->ice_agent_->add_local_agent_info_listener(endpoint, outer->guid_, DCPS::static_rchandle_cast<ICE::AgentInfoListener>(outer));
   }
 #endif
 
   reactor(reactor_task->get_reactor());
   reactor_task->interceptor()->execute_or_enqueue(DCPS::make_rch<RegisterHandlers>(rchandle_from(this), reactor_task));
 
 #ifdef OPENDDS_SECURITY
   // Now that the endpoint is added, SEDP can write the SPDP info.
   if (outer->is_security_enabled()) {
     outer->write_secure_updates();
   }
 #endif
 
   local_send_task_ = DCPS::make_rch<SpdpMulti>(reactor_task->interceptor(), outer->config_->resend_period(), rchandle_from(this), &SpdpTransport::send_local);
 
   if (outer->config_->periodic_directed_spdp()) {
     directed_send_task_ =
       DCPS::make_rch<SpdpSporadic>(TheServiceParticipant->time_source(), reactor_task->interceptor(),
                                    rchandle_from(this), &SpdpTransport::send_directed);
   }
 
   lease_expiration_task_ =
     DCPS::make_rch<SpdpSporadic>(TheServiceParticipant->time_source(), reactor_task->interceptor(),
                                  rchandle_from(this), &SpdpTransport::process_lease_expirations);
 
 #ifdef OPENDDS_SECURITY
   handshake_deadline_task_ =
     DCPS::make_rch<SpdpSporadic>(TheServiceParticipant->time_source(), reactor_task->interceptor(),
                                  rchandle_from(this), &SpdpTransport::process_handshake_deadlines);
   handshake_resend_task_ =
     DCPS::make_rch<SpdpSporadic>(TheServiceParticipant->time_source(), reactor_task->interceptor(),
                                  rchandle_from(this), &SpdpTransport::process_handshake_resends);
 
   relay_spdp_task_ =
     DCPS::make_rch<SpdpSporadic>(TheServiceParticipant->time_source(), reactor_task->interceptor(),
                                  rchandle_from(this), &SpdpTransport::send_relay);
   relay_stun_task_ =
     DCPS::make_rch<SpdpSporadic>(TheServiceParticipant->time_source(), reactor_task->interceptor(),
                                  rchandle_from(this), &SpdpTransport::relay_stun_task);
 #endif
 
 #ifndef DDS_HAS_MINIMUM_BIT
   // internal thread bit reporting
   if (TheServiceParticipant->get_thread_status_manager().update_thread_status()) {
     thread_status_task_ = DCPS::make_rch<SpdpPeriodic>(reactor_task->interceptor(), ref(*this), &SpdpTransport::thread_status_task);
   }
 #endif /* DDS_HAS_MINIMUM_BIT */
 
   this->job_queue(job_queue);
   network_interface_address_reader_ = DCPS::make_rch<DCPS::InternalDataReader<DCPS::NetworkInterfaceAddress> >(DCPS::DataReaderQosBuilder().reliability_reliable().durability_transient_local(), rchandle_from(this));
   TheServiceParticipant->network_interface_address_topic()->connect(network_interface_address_reader_);
 }

◆ open_unicast_socket()

bool OpenDDS::RTPS::Spdp::SpdpTransport::open_unicast_socket	(	u_short	port_common,
		u_short	participant_id
	)

Definition at line 3400 of file Spdp.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_ERROR_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, ENOTSUP, OpenDDS::DCPS::LogLevel::Error, ACE_SOCK::get_local_addr(), ACE_INET_Addr::get_port_number(), IPPROTO_IP, LM_DEBUG, LM_ERROR, LM_INFO, LM_WARNING, OpenDDS::DCPS::log_level, ACE_SOCK_Dgram::open(), outer_, PF_INET, ACE_SOCK::set_option(), ACE_INET_Addr::set_port_number(), OpenDDS::DCPS::set_socket_multicast_ttl(), SO_RCVBUF, SO_SNDBUF, SOL_SOCKET, uni_port_, and unicast_socket_.

Referenced by SpdpTransport().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) {
     throw std::runtime_error("couldn't get Spdp");
   }
 
   ACE_INET_Addr local_addr = outer->config_->spdp_local_address();
   const bool fixed_port = local_addr.get_port_number();
 
   if (fixed_port) {
     uni_port_ = local_addr.get_port_number();
   } else if (!outer->config_->spdp_request_random_port()) {
     const ACE_UINT32 port = static_cast<ACE_UINT32>(port_common) + outer->config_->d1() +
       outer->config_->pg() * participant_id;
     if (port > 65535) {
       if (log_level >= LogLevel::Error) {
         ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: Spdp::SpdpTransport::open_unicast_socket: "
                    "port %u is too high\n", port));
       }
       throw std::runtime_error("failed to open unicast port for SPDP (port too high)");
     }
     uni_port_ = static_cast<unsigned short>(port);
     local_addr.set_port_number(uni_port_);
   }
 
   if (unicast_socket_.open(local_addr, PF_INET) != 0) {
     if (fixed_port) {
       if (log_level >= LogLevel::Error) {
         ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: Spdp::SpdpTransport::open_unicast_socket: "
                    "failed to open %C %p.\n",
                    LogAddr(local_addr).c_str(), ACE_TEXT("ACE_SOCK_Dgram::open")));
       }
       throw std::runtime_error("failed to open unicast port for SPDP");
     }
     if (DCPS::DCPS_debug_level > 3) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Spdp::SpdpTransport::open_unicast_socket() - ")
                  ACE_TEXT("failed to open %C %p.  ")
                  ACE_TEXT("Trying next participantId...\n"),
                  DCPS::LogAddr(local_addr).c_str(), ACE_TEXT("ACE_SOCK_Dgram::open")));
     }
     return false;
   }
 
   if (!fixed_port && outer->config_->spdp_request_random_port()) {
     ACE_INET_Addr addr;
     if (unicast_socket_.get_local_addr(addr) == 0) {
       uni_port_ = addr.get_port_number();
     }
   }
 
   if (DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO,
                ACE_TEXT("(%P|%t) Spdp::SpdpTransport::open_unicast_socket() - ")
                ACE_TEXT("opened unicast socket on port %d\n"),
                uni_port_));
   }
 
   if (!DCPS::set_socket_multicast_ttl(unicast_socket_, outer->config_->ttl())) {
     if (DCPS::DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::open_unicast_socket() - ")
                 ACE_TEXT("failed to set TTL value to %d ")
                 ACE_TEXT("for port:%hu %p\n"),
                 outer->config_->ttl(), uni_port_, ACE_TEXT("DCPS::set_socket_multicast_ttl:")));
     }
     throw std::runtime_error("failed to set TTL");
   }
 
   const int send_buffer_size = outer->config()->send_buffer_size();
   if (send_buffer_size > 0) {
     if (unicast_socket_.set_option(SOL_SOCKET,
                                    SO_SNDBUF,
                                    (void *) &send_buffer_size,
                                    sizeof(send_buffer_size)) < 0
         && errno != ENOTSUP) {
       if (DCPS::DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::open_unicast_socket() - failed to set the send buffer size to %d errno %m\n"), send_buffer_size));
       }
       throw std::runtime_error("failed to set send buffer size");
     }
   }
 
   const int recv_buffer_size = outer->config()->recv_buffer_size();
   if (recv_buffer_size > 0) {
     if (unicast_socket_.set_option(SOL_SOCKET,
                                    SO_RCVBUF,
                                    (void *) &recv_buffer_size,
                                    sizeof(recv_buffer_size)) < 0
         && errno != ENOTSUP) {
       if (DCPS::DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::open_unicast_socket() - failed to set the recv buffer size to %d errno %m\n"), recv_buffer_size));
       }
       throw std::runtime_error("failed to set recv buffer size");
     }
   }
 
 #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: Spdp::SpdpTransport::open_unicast_socket: set_option: %m\n")), false);
   }
 #endif
 
   return true;
 }

◆ OPENDDS_LIST()

OpenDDS::RTPS::Spdp::SpdpTransport::OPENDDS_LIST ( DCPS::GUID_t )

◆ OPENDDS_SET()

OpenDDS::RTPS::Spdp::SpdpTransport::OPENDDS_SET ( ACE_INET_Addr )

Referenced by send().

◆ process_handshake_deadlines()

void OpenDDS::RTPS::Spdp::SpdpTransport::process_handshake_deadlines ( const DCPS::MonotonicTimePoint & now )

Definition at line 4321 of file Spdp.cpp.

References outer_.

Referenced by open().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   outer->process_handshake_deadlines(now);
 }

◆ process_handshake_resends()

void OpenDDS::RTPS::Spdp::SpdpTransport::process_handshake_resends ( const DCPS::MonotonicTimePoint & now )

Definition at line 4329 of file Spdp.cpp.

References outer_.

Referenced by open().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   outer->process_handshake_resends(now);
 }

◆ process_lease_expirations()

void OpenDDS::RTPS::Spdp::SpdpTransport::process_lease_expirations ( const DCPS::MonotonicTimePoint & now )

Definition at line 4278 of file Spdp.cpp.

References outer_.

Referenced by open().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   outer->process_lease_expirations(now);
 }

◆ process_relay_sra()

void OpenDDS::RTPS::Spdp::SpdpTransport::process_relay_sra ( ICE::ServerReflexiveStateMachine::StateChange sc )

Definition at line 4172 of file Spdp.cpp.

Referenced by handle_input(), and relay_stun_task().

 {
 #ifndef DDS_HAS_MINIMUM_BIT
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   DCPS::ConnectionRecord connection_record;
   std::memset(connection_record.guid, 0, sizeof(connection_record.guid));
   connection_record.protocol = DCPS::RTPS_RELAY_STUN_PROTOCOL;
   connection_record.latency = DCPS::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);
       outer->sedp_->job_queue()->enqueue(DCPS::make_rch<DCPS::WriteConnectionRecords>(outer->bit_subscriber_, true, connection_record));
     }
     break;
   case ICE::ServerReflexiveStateMachine::SRSM_Set:
   case ICE::ServerReflexiveStateMachine::SRSM_Change:
     // Lengthen to normal period.
     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);
     outer->sedp_->job_queue()->enqueue(DCPS::make_rch<DCPS::WriteConnectionRecords>(outer->bit_subscriber_, true, connection_record));
     break;
   case ICE::ServerReflexiveStateMachine::SRSM_Unset:
     connection_record.address = DCPS::LogAddr(relay_srsm_.unset_stun_server_address()).c_str();
     outer->sedp_->job_queue()->enqueue(DCPS::make_rch<DCPS::WriteConnectionRecords>(outer->bit_subscriber_, false, connection_record));
     break;
   }
 #else
   ACE_UNUSED_ARG(sc);
 #endif
 }

◆ register_handlers()

void OpenDDS::RTPS::Spdp::SpdpTransport::register_handlers ( const DCPS::ReactorTask_rch & reactor_task )

Definition at line 2524 of file Spdp.cpp.

References ACE_GUARD, OpenDDS::DCPS::ReactorTask::get_reactor(), outer_, ACE_Event_Handler::reactor(), register_unicast_socket(), and unicast_socket_.

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) {
     return;
   }
   ACE_GUARD(ACE_Thread_Mutex, g, outer->lock_);
 
   if (outer->shutdown_flag_) {
     return;
   }
 
   ACE_Reactor* const reactor = reactor_task->get_reactor();
   register_unicast_socket(reactor, unicast_socket_, "IPV4");
 #ifdef ACE_HAS_IPV6
   register_unicast_socket(reactor, unicast_ipv6_socket_, "IPV6");
 #endif
 }

◆ register_unicast_socket()

void OpenDDS::RTPS::Spdp::SpdpTransport::register_unicast_socket	(	ACE_Reactor *	reactor,
		ACE_SOCK_Dgram &	socket,
		const char *	what
	)

Definition at line 2502 of file Spdp.cpp.

References ACE_IPC_SAP::control(), ACE_IPC_SAP::get_handle(), ACE_Event_Handler::READ_MASK, and ACE_Reactor::register_handler().

Referenced by register_handlers().

 {
 #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;
     socket.control(SIO_UDP_CONNRESET, &recv_udp_connreset);
   }
 #endif
 
   if (reactor->register_handler(socket.get_handle(),
                                 this, ACE_Event_Handler::READ_MASK) != 0) {
     throw std::runtime_error(
       (DCPS::String("failed to register ") + what + " unicast input handler").c_str());
   }
 }

◆ relay_stun_task()

void OpenDDS::RTPS::Spdp::SpdpTransport::relay_stun_task ( const DCPS::MonotonicTimePoint & now )

Definition at line 4156 of file Spdp.cpp.

References OpenDDS::ICE::Configuration::instance(), OpenDDS::ICE::ServerReflexiveStateMachine::message(), outer_, process_relay_sra(), relay_srsm_, relay_stun_task_, relay_stun_task_falloff_, OpenDDS::ICE::ServerReflexiveStateMachine::send(), send(), and OpenDDS::ICE::Configuration::server_reflexive_indication_count().

Referenced by open().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   if (outer->config_->use_rtps_relay() || outer->config_->rtps_relay_only()) {
     const ACE_INET_Addr relay_address = outer->config_->spdp_rtps_relay_address();
     if (relay_address != ACE_INET_Addr()) {
       process_relay_sra(relay_srsm_.send(relay_address, ICE::Configuration::instance()->server_reflexive_indication_count(), outer->guid_.guidPrefix));
       send(relay_address, relay_srsm_.message());
       relay_stun_task_falloff_.advance(ICE::Configuration::instance()->server_reflexive_address_period());
       relay_stun_task_->schedule(relay_stun_task_falloff_.get());
     }
   }
 }

◆ send() [1/3]

void OpenDDS::RTPS::Spdp::SpdpTransport::send	(	WriteFlags	flags,
		const ACE_INET_Addr &	local_address = `ACE_INET_Addr()`
	)

Definition at line 2883 of file Spdp.cpp.

References OPENDDS_SET(), outer_, SEND_DIRECT, SEND_MULTICAST, and SEND_RELAY.

Referenced by dispose_unregister(), relay_stun_task(), and write_i().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   if ((flags & SEND_MULTICAST) && !outer->config_->rtps_relay_only()) {
     typedef OPENDDS_SET(ACE_INET_Addr)::const_iterator iter_t;
     for (iter_t iter = send_addrs_.begin(); iter != send_addrs_.end(); ++iter) {
       send(*iter, false);
     }
   }
 
   if (((flags & SEND_DIRECT) && !outer->config_->rtps_relay_only()) &&
       local_address != ACE_INET_Addr()) {
     send(local_address, false);
   }
 
   if ((flags & SEND_RELAY) || outer->config_->rtps_relay_only()) {
     const ACE_INET_Addr relay_address = outer->config_->spdp_rtps_relay_address();
     if (relay_address != ACE_INET_Addr()) {
       send(relay_address, true);
     }
   }
 }

◆ send() [2/3]

ssize_t OpenDDS::RTPS::Spdp::SpdpTransport::send	(	const ACE_INET_Addr &	addr,
		bool	relay
	)

Definition at line 2921 of file Spdp.cpp.

References ACE_ERROR, ACE_TEXT(), choose_send_socket(), OpenDDS::DCPS::DCPS_debug_level, ENETUNREACH, OpenDDS::DCPS::ENTITYID_SPDP_BUILTIN_PARTICIPANT_WRITER, ACE_Message_Block::length(), LM_WARNING, OpenDDS::DCPS::make_id(), OpenDDS::DCPS::MCK_RTPS, OpenDDS::DCPS::InternalTransportStatistics::message_count, network_is_unreachable_, outer_, ACE_Message_Block::rd_ptr(), ACE_SOCK_Dgram::send(), socket(), transport_statistics_, wbuff_, and OpenDDS::DCPS::InternalTransportStatistics::writer_resend_count.

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return -1;
 
 #ifdef OPENDDS_TESTING_FEATURES
   if (outer->sedp_->transport_inst()->should_drop(wbuff_.length())) {
     return wbuff_.length();
   }
 #endif
 
   const ACE_SOCK_Dgram& socket = choose_send_socket(addr);
   const ssize_t res = socket.send(wbuff_.rd_ptr(), wbuff_.length(), addr);
   if (outer->sedp_->transport_inst()->count_messages()) {
     ++transport_statistics_.writer_resend_count[make_id(outer->guid_, ENTITYID_SPDP_BUILTIN_PARTICIPANT_WRITER)];
   }
   if (res < 0) {
     if (outer->sedp_->transport_inst()->count_messages()) {
       const DCPS::InternalMessageCountKey key(DCPS::NetworkAddress(addr), DCPS::MCK_RTPS, relay);
       transport_statistics_.message_count[key].send_fail(wbuff_.length());
     }
     const int err = errno;
     if (err != ENETUNREACH || !network_is_unreachable_) {
       errno = err;
       if (DCPS::DCPS_debug_level > 0) {
         ACE_ERROR((LM_WARNING,
                   ACE_TEXT("(%P|%t) WARNING: Spdp::SpdpTransport::send() - ")
                   ACE_TEXT("destination %C failed send: %m\n"), DCPS::LogAddr(addr).c_str()));
       }
     }
     if (err == ENETUNREACH) {
       network_is_unreachable_ = true;
     }
   } else {
     if (outer->sedp_->transport_inst()->count_messages()) {
       const DCPS::InternalMessageCountKey key(DCPS::NetworkAddress(addr), DCPS::MCK_RTPS, relay);
       transport_statistics_.message_count[key].send(wbuff_.length());
     }
     network_is_unreachable_ = false;
   }
 
   return res;
 }

◆ send() [3/3]

void OpenDDS::RTPS::Spdp::SpdpTransport::send	(	const ACE_INET_Addr &	address,
		const STUN::Message &	message
	)

virtual

Implements OpenDDS::ICE::Endpoint.

Definition at line 3290 of file Spdp.cpp.

References OpenDDS::DCPS::InternalDataReaderListener< DCPS::NetworkInterfaceAddress >::job_queue(), outer_, and OpenDDS::DCPS::rchandle_from().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   DCPS::RcHandle<DCPS::JobQueue> job_queue = outer->sedp_->job_queue();
   if (job_queue) {
     job_queue->enqueue(DCPS::make_rch<SendStun>(rchandle_from(this), address, message));
   }
 }

◆ send_directed()

void OpenDDS::RTPS::Spdp::SpdpTransport::send_directed ( const DCPS::MonotonicTimePoint & now )

Definition at line 4254 of file Spdp.cpp.

References ACE_GUARD, directed_send_task_, outer_, SEND_DIRECT, SEND_RELAY, and write_i().

Referenced by open().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   ACE_GUARD(ACE_Thread_Mutex, g, outer->lock_);
 
   while (!directed_guids_.empty()) {
     const DCPS::GUID_t id = directed_guids_.front();
     directed_guids_.pop_front();
 
     DiscoveredParticipantConstIter pos = outer->participants_.find(id);
     if (pos == outer->participants_.end()) {
       continue;
     }
 
     write_i(id, pos->second.last_recv_address_, SEND_DIRECT | SEND_RELAY);
     directed_guids_.push_back(id);
     directed_send_task_->schedule(outer->config_->resend_period() * (1.0 / directed_guids_.size()));
     break;
   }
 }

◆ send_local()

void OpenDDS::RTPS::Spdp::SpdpTransport::send_local ( const DCPS::MonotonicTimePoint & now )

Definition at line 4249 of file Spdp.cpp.

References SEND_MULTICAST, and write().

Referenced by open().

 {
   write(SEND_MULTICAST);
 }

◆ send_relay()

void OpenDDS::RTPS::Spdp::SpdpTransport::send_relay ( const DCPS::MonotonicTimePoint & now )

Definition at line 4233 of file Spdp.cpp.

References outer_, relay_spdp_task_, relay_spdp_task_falloff_, SEND_RELAY, and write().

Referenced by open().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   if (outer->config_->use_rtps_relay() || outer->config_->rtps_relay_only()) {
     const ACE_INET_Addr relay_address = outer->config_->spdp_rtps_relay_address();
     if (relay_address != ACE_INET_Addr()) {
       write(SEND_RELAY);
       relay_spdp_task_falloff_.advance(outer->config_->spdp_rtps_relay_send_period());
       relay_spdp_task_->schedule(relay_spdp_task_falloff_.get());
     }
   }
 }

◆ shorten_local_sender_delay_i()

void OpenDDS::RTPS::Spdp::SpdpTransport::shorten_local_sender_delay_i ( )

Definition at line 2659 of file Spdp.cpp.

References local_send_task_, and outer_.

Referenced by on_data_available().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   if (local_send_task_) {
     const TimeDuration quick_resend = outer->config_->resend_period() * outer->quick_resend_ratio_;
     local_send_task_->enable(std::max(quick_resend, outer->min_resend_delay_));
   }
 }

◆ stun_server_address()

ACE_INET_Addr OpenDDS::RTPS::Spdp::SpdpTransport::stun_server_address ( ) const

virtual

Implements OpenDDS::ICE::Endpoint.

Definition at line 3353 of file Spdp.cpp.

References outer_.

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   return outer ? outer->config_->spdp_stun_server_address() : ACE_INET_Addr();
 }

◆ thread_status_task()

void OpenDDS::RTPS::Spdp::SpdpTransport::thread_status_task ( const DCPS::MonotonicTimePoint & now )

Definition at line 4286 of file Spdp.cpp.

References ACE_DEBUG, ACE_GUARD, OpenDDS::DCPS::DCPS_debug_level, last_harvest, LM_DEBUG, DDS::NEW_VIEW_STATE, outer_, TheServiceParticipant, OpenDDS::DCPS::InternalThreadBuiltinTopicData::thread_id, and OpenDDS::DCPS::InternalThreadBuiltinTopicData::utilization.

Referenced by open().

 {
   ACE_UNUSED_ARG(now);
 #ifndef DDS_HAS_MINIMUM_BIT
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   if (DCPS::DCPS_debug_level > 4) {
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) Spdp::SpdpTransport::thread_status_task(): Updating internal thread status BIT.\n"));
   }
 
   ACE_GUARD(ACE_Thread_Mutex, g, outer->lock_);
 
   typedef DCPS::ThreadStatusManager::List List;
   List running;
   List removed;
   TheServiceParticipant->get_thread_status_manager().harvest(last_harvest, running, removed);
   last_harvest = now;
   for (List::const_iterator i = removed.begin(); i != removed.end(); ++i) {
     DCPS::InternalThreadBuiltinTopicData data;
     data.thread_id = i->bit_key().c_str();
     outer->bit_subscriber_->remove_thread_status(data);
   }
   for (List::const_iterator i = running.begin(); i != running.end(); ++i) {
     DCPS::InternalThreadBuiltinTopicData data;
     data.thread_id = i->bit_key().c_str();
     data.utilization = i->utilization(now);
     outer->bit_subscriber_->add_thread_status(data, DDS::NEW_VIEW_STATE, i->timestamp());
   }
 
 #endif /* DDS_HAS_MINIMUM_BIT */
 }

◆ write()

void OpenDDS::RTPS::Spdp::SpdpTransport::write ( WriteFlags flags )

Definition at line 2671 of file Spdp.cpp.

References ACE_GUARD, outer_, and write_i().

Referenced by send_local(), and send_relay().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   ACE_GUARD(ACE_Thread_Mutex, g, outer->lock_);
   write_i(flags);
 }

◆ write_i() [1/2]

void OpenDDS::RTPS::Spdp::SpdpTransport::write_i ( WriteFlags flags )

Definition at line 2681 of file Spdp.cpp.

References ACE_ERROR, ACE_TEXT(), data_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Security::DPDK_ENHANCED, OpenDDS::Security::DPDK_ORIGINAL, OpenDDS::DCPS::Serializer::encoding(), get_ice_endpoint(), hdr_, LM_ERROR, OpenDDS::DCPS::MUTABLE, outer_, ACE_Message_Block::reset(), OpenDDS::RTPS::SEDP_AGENT_INFO_KEY, send(), seq_, OpenDDS::RTPS::SPDP_AGENT_INFO_KEY, OpenDDS::RTPS::ParameterListConverter::to_param_list(), OpenDDS::RTPS::to_rtps_seqnum(), wbuff_, and OpenDDS::RTPS::DataSubmessage::writerSN.

Referenced by send_directed(), and write().

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   if (!outer->config_->undirected_spdp()) {
     return;
   }
 
   const ParticipantData_t pdata = outer->build_local_pdata(
 #ifdef OPENDDS_SECURITY
     true, outer->is_security_enabled() ? Security::DPDK_ENHANCED : Security::DPDK_ORIGINAL
 #endif
   );
 
   data_.writerSN = to_rtps_seqnum(seq_);
   ++seq_;
 
   ParameterList plist;
   if (!ParameterListConverter::to_param_list(pdata, plist)) {
     if (DCPS::DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: ")
         ACE_TEXT("Spdp::SpdpTransport::write() - ")
         ACE_TEXT("failed to convert from SPDPdiscoveredParticipantData ")
         ACE_TEXT("to ParameterList\n")));
     }
     return;
   }
 
 #ifdef OPENDDS_SECURITY
   if (!outer->is_security_enabled()) {
     ICE::AgentInfoMap ai_map;
     DCPS::WeakRcHandle<ICE::Endpoint> sedp_endpoint = outer->sedp_->get_ice_endpoint();
     if (sedp_endpoint) {
       ai_map[SEDP_AGENT_INFO_KEY] = outer->ice_agent_->get_local_agent_info(sedp_endpoint);
     }
     DCPS::WeakRcHandle<ICE::Endpoint> spdp_endpoint = get_ice_endpoint();
     if (spdp_endpoint) {
       ai_map[SPDP_AGENT_INFO_KEY] = outer->ice_agent_->get_local_agent_info(spdp_endpoint);
     }
 
     if (!ParameterListConverter::to_param_list(ai_map, plist)) {
       if (DCPS::DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: ")
                   ACE_TEXT("Spdp::SpdpTransport::write() - ")
                   ACE_TEXT("failed to convert from ICE::AgentInfo ")
                   ACE_TEXT("to ParameterList\n")));
       }
       return;
     }
   }
 #endif
 
   wbuff_.reset();
   DCPS::Serializer ser(&wbuff_, encoding_plain_native);
   DCPS::EncapsulationHeader encap(ser.encoding(), DCPS::MUTABLE);
   if (!(ser << hdr_) || !(ser << data_) || !(ser << encap) || !(ser << plist)) {
     if (DCPS::DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::write() - ")
         ACE_TEXT("failed to serialize headers for SPDP\n")));
     }
     return;
   }
 
   send(flags);
 }

◆ write_i() [2/2]

void OpenDDS::RTPS::Spdp::SpdpTransport::write_i	(	const DCPS::GUID_t &	guid,
		const ACE_INET_Addr &	local_address,
		WriteFlags	flags
	)

Definition at line 2813 of file Spdp.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::assign(), data_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Security::DPDK_ENHANCED, OpenDDS::Security::DPDK_ORIGINAL, OpenDDS::DCPS::Serializer::encoding(), OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::SubmessageHeader::flags, get_ice_endpoint(), OpenDDS::DCPS::GUID_t::guidPrefix, OpenDDS::RTPS::InfoDestinationSubmessage::guidPrefix, hdr_, OpenDDS::RTPS::INFO_DST, LM_ERROR, OpenDDS::DCPS::MUTABLE, outer_, ACE_Message_Block::reset(), OpenDDS::RTPS::SEDP_AGENT_INFO_KEY, send(), seq_, OpenDDS::RTPS::InfoDestinationSubmessage::smHeader, OpenDDS::RTPS::SPDP_AGENT_INFO_KEY, OpenDDS::RTPS::SubmessageHeader::submessageId, OpenDDS::RTPS::SubmessageHeader::submessageLength, OpenDDS::RTPS::ParameterListConverter::to_param_list(), OpenDDS::RTPS::to_rtps_seqnum(), wbuff_, and OpenDDS::RTPS::DataSubmessage::writerSN.

 {
   DCPS::RcHandle<Spdp> outer = outer_.lock();
   if (!outer) return;
 
   const ParticipantData_t pdata = outer->build_local_pdata(
 #ifdef OPENDDS_SECURITY
     true, outer->is_security_enabled() ? Security::DPDK_ENHANCED : Security::DPDK_ORIGINAL
 #endif
   );
 
   data_.writerSN = to_rtps_seqnum(seq_);
   ++seq_;
 
   ParameterList plist;
   if (!ParameterListConverter::to_param_list(pdata, plist)) {
     if (DCPS::DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: ")
         ACE_TEXT("Spdp::SpdpTransport::write_i() - ")
         ACE_TEXT("failed to convert from SPDPdiscoveredParticipantData ")
         ACE_TEXT("to ParameterList\n")));
     }
     return;
   }
 
 #ifdef OPENDDS_SECURITY
   if (!outer->is_security_enabled()) {
     ICE::AgentInfoMap ai_map;
     DCPS::WeakRcHandle<ICE::Endpoint> sedp_endpoint = outer->sedp_->get_ice_endpoint();
     if (sedp_endpoint) {
       ai_map[SEDP_AGENT_INFO_KEY] = outer->ice_agent_->get_local_agent_info(sedp_endpoint);
     }
     DCPS::WeakRcHandle<ICE::Endpoint> spdp_endpoint = get_ice_endpoint();
     if (spdp_endpoint) {
       ai_map[SPDP_AGENT_INFO_KEY] = outer->ice_agent_->get_local_agent_info(spdp_endpoint);
     }
 
     if (!ParameterListConverter::to_param_list(ai_map, plist)) {
       ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: ")
                  ACE_TEXT("Spdp::SpdpTransport::write_i() - ")
                  ACE_TEXT("failed to convert from ICE::AgentInfo ")
                  ACE_TEXT("to ParameterList\n")));
       return;
     }
   }
 #endif
 
   InfoDestinationSubmessage info_dst;
   info_dst.smHeader.submessageId = INFO_DST;
   info_dst.smHeader.flags = FLAG_E;
   info_dst.smHeader.submessageLength = sizeof(guid.guidPrefix);
   DCPS::assign(info_dst.guidPrefix, guid.guidPrefix);
 
   wbuff_.reset();
   DCPS::Serializer ser(&wbuff_, encoding_plain_native);
   DCPS::EncapsulationHeader encap(ser.encoding(), DCPS::MUTABLE);
   if (!(ser << hdr_) || !(ser << info_dst) || !(ser << data_) || !(ser << encap)
       || !(ser << plist)) {
     if (DCPS::DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: Spdp::SpdpTransport::write_i() - ")
         ACE_TEXT("failed to serialize headers for SPDP\n")));
     }
     return;
   }
 
   send(flags, local_address);
 }

Member Data Documentation

◆ buff_

ACE_Message_Block OpenDDS::RTPS::Spdp::SpdpTransport::buff_

Definition at line 520 of file Spdp.h.

Referenced by handle_input().

◆ data_

DataSubmessage OpenDDS::RTPS::Spdp::SpdpTransport::data_

Definition at line 504 of file Spdp.h.

Referenced by dispose_unregister(), SpdpTransport(), and write_i().

◆ directed_send_task_

DCPS::RcHandle<SpdpSporadic> OpenDDS::RTPS::Spdp::SpdpTransport::directed_send_task_

Definition at line 527 of file Spdp.h.

Referenced by close(), open(), and send_directed().

◆ handshake_deadline_task_

DCPS::RcHandle<SpdpSporadic> OpenDDS::RTPS::Spdp::SpdpTransport::handshake_deadline_task_

Definition at line 536 of file Spdp.h.

Referenced by close(), and open().

◆ handshake_resend_task_

DCPS::RcHandle<SpdpSporadic> OpenDDS::RTPS::Spdp::SpdpTransport::handshake_resend_task_

Definition at line 538 of file Spdp.h.

Referenced by close(), and open().

◆ hdr_

Header OpenDDS::RTPS::Spdp::SpdpTransport::hdr_

Definition at line 503 of file Spdp.h.

Referenced by dispose_unregister(), SpdpTransport(), and write_i().

◆ ice_endpoint_added_

bool OpenDDS::RTPS::Spdp::SpdpTransport::ice_endpoint_added_

Definition at line 550 of file Spdp.h.

Referenced by close(), and open().

◆ last_harvest

DCPS::MonotonicTimePoint OpenDDS::RTPS::Spdp::SpdpTransport::last_harvest

Definition at line 553 of file Spdp.h.

Referenced by thread_status_task().

◆ lease_expiration_task_

DCPS::RcHandle<SpdpSporadic> OpenDDS::RTPS::Spdp::SpdpTransport::lease_expiration_task_

Definition at line 530 of file Spdp.h.

Referenced by close(), and open().

◆ local_send_task_

DCPS::RcHandle<SpdpMulti> OpenDDS::RTPS::Spdp::SpdpTransport::local_send_task_

Definition at line 525 of file Spdp.h.

Referenced by close(), enable_periodic_tasks(), open(), and shorten_local_sender_delay_i().

◆ multicast_address_

ACE_INET_Addr OpenDDS::RTPS::Spdp::SpdpTransport::multicast_address_

Definition at line 509 of file Spdp.h.

Referenced by on_data_available(), and SpdpTransport().

◆ multicast_interface_

OPENDDS_STRING OpenDDS::RTPS::Spdp::SpdpTransport::multicast_interface_

Definition at line 508 of file Spdp.h.

Referenced by on_data_available(), and SpdpTransport().

◆ multicast_manager_

DCPS::MulticastManager OpenDDS::RTPS::Spdp::SpdpTransport::multicast_manager_

Definition at line 518 of file Spdp.h.

Referenced by on_data_available().

◆ multicast_socket_

ACE_SOCK_Dgram_Mcast OpenDDS::RTPS::Spdp::SpdpTransport::multicast_socket_

Definition at line 510 of file Spdp.h.

Referenced by choose_recv_socket(), close(), on_data_available(), SpdpTransport(), and ~SpdpTransport().

◆ network_interface_address_reader_

DCPS::RcHandle<DCPS::InternalDataReader<DCPS::NetworkInterfaceAddress> > OpenDDS::RTPS::Spdp::SpdpTransport::network_interface_address_reader_

Definition at line 533 of file Spdp.h.

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

◆ network_is_unreachable_

bool OpenDDS::RTPS::Spdp::SpdpTransport::network_is_unreachable_

Definition at line 549 of file Spdp.h.

Referenced by send().

◆ outer_

DCPS::WeakRcHandle<Spdp> OpenDDS::RTPS::Spdp::SpdpTransport::outer_

◆ relay_spdp_task_

DCPS::RcHandle<SpdpSporadic> OpenDDS::RTPS::Spdp::SpdpTransport::relay_spdp_task_

Definition at line 540 of file Spdp.h.

Referenced by close(), enable_periodic_tasks(), open(), and send_relay().

◆ relay_spdp_task_falloff_

DCPS::FibonacciSequence<TimeDuration> OpenDDS::RTPS::Spdp::SpdpTransport::relay_spdp_task_falloff_

Definition at line 541 of file Spdp.h.

Referenced by enable_periodic_tasks(), and send_relay().

◆ relay_srsm_

ICE::ServerReflexiveStateMachine OpenDDS::RTPS::Spdp::SpdpTransport::relay_srsm_

Definition at line 545 of file Spdp.h.

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

◆ relay_stun_task_

DCPS::RcHandle<SpdpSporadic> OpenDDS::RTPS::Spdp::SpdpTransport::relay_stun_task_

Definition at line 543 of file Spdp.h.

Referenced by close(), disable_relay_stun_task(), enable_periodic_tasks(), open(), and relay_stun_task().

◆ relay_stun_task_falloff_

DCPS::FibonacciSequence<TimeDuration> OpenDDS::RTPS::Spdp::SpdpTransport::relay_stun_task_falloff_

Definition at line 544 of file Spdp.h.

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

◆ SEND_DIRECT

const Spdp::SpdpTransport::WriteFlags OpenDDS::RTPS::Spdp::SpdpTransport::SEND_DIRECT = (1 << 2)

static

Definition at line 435 of file Spdp.h.

Referenced by OpenDDS::RTPS::Spdp::announce_domain_participant_qos(), OpenDDS::RTPS::Spdp::handle_participant_data(), OpenDDS::RTPS::Spdp::process_handshake_resends(), send(), and send_directed().

◆ SEND_MULTICAST

const Spdp::SpdpTransport::WriteFlags OpenDDS::RTPS::Spdp::SpdpTransport::SEND_MULTICAST = (1 << 0)

static

Definition at line 433 of file Spdp.h.

Referenced by OpenDDS::RTPS::Spdp::announce_domain_participant_qos(), dispose_unregister(), send(), and send_local().

◆ SEND_RELAY

const Spdp::SpdpTransport::WriteFlags OpenDDS::RTPS::Spdp::SpdpTransport::SEND_RELAY = (1 << 1)

static

Definition at line 434 of file Spdp.h.

Referenced by OpenDDS::RTPS::Spdp::announce_domain_participant_qos(), dispose_unregister(), OpenDDS::RTPS::Spdp::handle_participant_data(), OpenDDS::RTPS::Spdp::process_handshake_resends(), send(), send_directed(), and send_relay().

◆ seq_

DCPS::SequenceNumber OpenDDS::RTPS::Spdp::SpdpTransport::seq_

Definition at line 505 of file Spdp.h.

Referenced by dispose_unregister(), and write_i().

◆ thread_status_task_

DCPS::RcHandle<SpdpPeriodic> OpenDDS::RTPS::Spdp::SpdpTransport::thread_status_task_

Definition at line 532 of file Spdp.h.

Referenced by close(), enable_periodic_tasks(), and open().

◆ transport_statistics_

DCPS::InternalTransportStatistics OpenDDS::RTPS::Spdp::SpdpTransport::transport_statistics_

Definition at line 552 of file Spdp.h.

Referenced by handle_input(), and send().

◆ uni_port_

u_short OpenDDS::RTPS::Spdp::SpdpTransport::uni_port_

Definition at line 506 of file Spdp.h.

Referenced by open_unicast_socket(), and SpdpTransport().

◆ unicast_socket_

ACE_SOCK_Dgram OpenDDS::RTPS::Spdp::SpdpTransport::unicast_socket_

Definition at line 507 of file Spdp.h.

Referenced by choose_recv_socket(), choose_send_socket(), close(), handle_input(), host_addresses(), open_unicast_socket(), register_handlers(), and ~SpdpTransport().

◆ wbuff_

ACE_Message_Block OpenDDS::RTPS::Spdp::SpdpTransport::wbuff_

Definition at line 520 of file Spdp.h.

Referenced by dispose_unregister(), send(), and write_i().

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

Classes

Public Types

Public Member Functions

Public Attributes

Static Public Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ SpdpMulti

◆ SpdpPeriodic

◆ SpdpSporadic

◆ WriteFlags

Constructor & Destructor Documentation

◆ SpdpTransport()

◆ ~SpdpTransport()

Member Function Documentation

◆ choose_recv_socket()

◆ choose_send_socket()

◆ close()

◆ disable_relay_stun_task()

◆ dispose_unregister()

◆ enable_periodic_tasks()

◆ get_ice_endpoint()

◆ handle_input()

◆ host_addresses()

◆ ice_connect()

◆ ice_disconnect()

◆ on_data_available()

◆ open()

◆ open_unicast_socket()

◆ OPENDDS_LIST()

◆ OPENDDS_SET()

◆ process_handshake_deadlines()

◆ process_handshake_resends()

◆ process_lease_expirations()

◆ process_relay_sra()

◆ register_handlers()

◆ register_unicast_socket()

◆ relay_stun_task()

◆ send() [1/3]

◆ send() [2/3]

◆ send() [3/3]

◆ send_directed()

◆ send_local()

◆ send_relay()

◆ shorten_local_sender_delay_i()

◆ stun_server_address()

◆ thread_status_task()

◆ write()

◆ write_i() [1/2]

◆ write_i() [2/2]

Member Data Documentation

◆ buff_

◆ data_

◆ directed_send_task_

◆ handshake_deadline_task_

◆ handshake_resend_task_

◆ hdr_

◆ ice_endpoint_added_

◆ last_harvest

◆ lease_expiration_task_

◆ local_send_task_

◆ multicast_address_

◆ multicast_interface_

◆ multicast_manager_

◆ multicast_socket_

◆ network_interface_address_reader_

◆ network_is_unreachable_

◆ outer_

◆ relay_spdp_task_

◆ relay_spdp_task_falloff_

◆ relay_srsm_

◆ relay_stun_task_

◆ relay_stun_task_falloff_

◆ SEND_DIRECT

◆ SEND_MULTICAST

◆ SEND_RELAY

◆ seq_

◆ thread_status_task_

◆ transport_statistics_