#include <MulticastDataLink.h>

Inheritance diagram for OpenDDS::DCPS::MulticastDataLink:

[legend]

Collaboration diagram for OpenDDS::DCPS::MulticastDataLink:

[legend]

Public Member Functions
	MulticastDataLink (const MulticastTransport_rch &transport, const MulticastSessionFactory_rch &session_factory, MulticastPeer local_peer, const MulticastInst_rch &config, const ReactorTask_rch &reactor_task, bool is_active)

virtual	~MulticastDataLink ()

MulticastTransport_rch	transport ()

MulticastPeer	local_peer () const

MulticastSendStrategy *	send_strategy ()

MulticastReceiveStrategy *	receive_strategy ()

SingleSendBuffer *	send_buffer ()

MulticastInst_rch	config ()

ReactorTask_rch	reactor_task ()

ACE_Reactor *	get_reactor ()

ACE_Proactor *	get_proactor ()

ACE_SOCK_Dgram_Mcast &	socket ()

bool	join (const ACE_INET_Addr &group_address)

MulticastSession_rch	find_or_create_session (MulticastPeer remote_peer)

MulticastSession_rch	find_session (MulticastPeer remote_peer)

bool	check_header (const TransportHeader &header)

bool	check_header (const DataSampleHeader &header)

void	sample_received (ReceivedDataSample &sample)

bool	reassemble (ReceivedDataSample &data, const TransportHeader &header)

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

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

void	client_stop (const GUID_t &localId)

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

virtual	~DataLink ()

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

void	schedule_stop (const MonotonicTimePoint &schedule_to_stop_at)

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

void	resume_send ()

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

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

void	schedule_delayed_release ()

const TimeDuration &	datalink_release_delay () const

void	remove_listener (const GUID_t &local_id)

void	send_start ()

void	send (TransportQueueElement *element)

void	send_stop (GUID_t repoId)

virtual RemoveResult	remove_sample (const DataSampleElement *sample)

virtual void	remove_all_msgs (const GUID_t &pub_id)

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

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

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

void	transport_shutdown ()

void	notify (ConnectionNotice notice)

virtual void	pre_stop_i ()

void	release_resources ()

void	terminate_send ()

void	terminate_send_if_suspended ()

bool	is_target (const GUID_t &remote_id)

void	clear_associations ()

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

int	handle_close (ACE_HANDLE h, ACE_Reactor_Mask m)

void	set_dscp_codepoint (int cp, ACE_SOCK &socket)

Priority &	transport_priority ()

Priority	transport_priority () const

bool &	is_loopback ()

bool	is_loopback () const

bool &	is_active ()

bool	is_active () const

bool	cancel_release ()

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

SendControlStatus	send_control (const DataSampleHeader &header, Message_Block_Ptr data)

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

TransportImpl_rch	impl () const

void	default_listener (const TransportReceiveListener_wrch &trl)

TransportReceiveListener_wrch	default_listener () const

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

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

void	invoke_on_start_callbacks (bool success)

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

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

void	set_scheduling_release (bool scheduling_release)

virtual void	send_final_acks (const GUID_t &readerid)

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

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

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

ACE_Event_Handler::Reference_Count	add_reference ()

ACE_Event_Handler::Reference_Count	remove_reference ()

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

virtual ACE_HANDLE	get_handle (void) const

virtual void	set_handle (ACE_HANDLE)

virtual int	priority (void) const

virtual void	priority (int priority)

virtual int	handle_input (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_output (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_exit (ACE_Process *)

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

virtual int	resume_handler (void)

virtual int	handle_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual int	handle_group_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual void	reactor (ACE_Reactor *reactor)

virtual ACE_Reactor *	reactor (void) const

virtual ACE_Reactor_Timer_Interface *	reactor_timer_interface (void) const

Reference_Counting_Policy &	reference_counting_policy (void)

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

virtual void	_add_ref ()

virtual void	_remove_ref ()

long	ref_count () const

WeakObject *	_get_weak_object () const

Private Member Functions
typedef	OPENDDS_MAP (MulticastPeer, MulticastSession_rch) MulticastSessionMap

virtual void	stop_i ()

void	syn_received_no_session (MulticastPeer source, const Message_Block_Ptr &data, bool swap_bytes)

void	release_remote_i (const GUID_t &remote)

bool	ready_to_deliver (const ReceivedDataSample &data)

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

typedef WeakRcHandle< TransportClient >	TransportClient_wrch

typedef std::pair< TransportClient_wrch, GUID_t >	OnStartCallback

Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

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)

Public Attributes inherited from ACE_Event_Handler
	LO_PRIORITY

	HI_PRIORITY

	NULL_MASK

	READ_MASK

	WRITE_MASK

	EXCEPT_MASK

	ACCEPT_MASK

	CONNECT_MASK

	TIMER_MASK

	QOS_MASK

	GROUP_QOS_MASK

	SIGNAL_MASK

	ALL_EVENTS_MASK

	RWE_MASK

	DONT_CALL

	ACE_EVENT_HANDLER_NOT_RESUMED

	ACE_REACTOR_RESUMES_HANDLER

	ACE_APPLICATION_RESUMES_HANDLER

Protected Types inherited from OpenDDS::DCPS::DataLink
typedef ACE_Guard< LockType >	GuardType

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

Protected Member Functions inherited from OpenDDS::DCPS::DataLink
int	start (const TransportSendStrategy_rch &send_strategy, const TransportStrategy_rch &receive_strategy, bool invoke_all=true)

void	send_start_i ()

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

void	send_stop_i (GUID_t repoId)

GUIDSeq *	peer_ids (const GUID_t &local_id) const

void	network_change () const

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

TransportSendStrategy_rch	get_send_strategy ()

typedef	OPENDDS_MAP_CMP (GUID_t, TransportClient_wrch, GUID_tKeyLessThan) RepoToClientMap

typedef	OPENDDS_MAP_CMP (GUID_t, RepoToClientMap, GUID_tKeyLessThan) OnStartCallbackMap

typedef	OPENDDS_MAP_CMP (GUID_t, RepoIdSet, GUID_tKeyLessThan) PendingOnStartsMap

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

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

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

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

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

LockType	strategy_lock_

OnStartCallbackMap	on_start_callbacks_

PendingOnStartsMap	pending_on_starts_

TimeDuration	datalink_release_delay_

unique_ptr< MessageBlockAllocator >	mb_allocator_

unique_ptr< DataBlockAllocator >	db_allocator_

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

bool	is_active_
	Is pub or sub ? More...

bool	started_

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

Interceptor	interceptor_

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Definition at line 34 of file MulticastDataLink.h.

Constructor & Destructor Documentation

◆ MulticastDataLink()

OpenDDS::DCPS::MulticastDataLink::MulticastDataLink	(	const MulticastTransport_rch &	transport,
		const MulticastSessionFactory_rch &	session_factory,
		MulticastPeer	local_peer,
		const MulticastInst_rch &	config,
		const ReactorTask_rch &	reactor_task,
		bool	is_active
	)

Definition at line 41 of file MulticastDataLink.cpp.

References OpenDDS::DCPS::DEFAULT_CONFIG_MAX_SAMPLES_PER_PACKET, OpenDDS::DCPS::MulticastInst::DEFAULT_NAK_DEPTH, OpenDDS::DCPS::MulticastSessionFactory::requires_send_buffer(), OpenDDS::DCPS::TransportSendStrategy::send_buffer(), send_buffer_, send_strategy_, and session_factory_.

 : DataLink(transport, 0 /*priority*/, false /*loopback*/, is_active),
   session_factory_(session_factory),
   local_peer_(local_peer),
   reactor_task_(reactor_task),
   send_strategy_(make_rch<MulticastSendStrategy>(this)),
   recv_strategy_(make_rch<MulticastReceiveStrategy>(this))
 {
   // A send buffer may be bound to the send strategy to ensure a
   // configured number of most-recent datagrams are retained:
   if (session_factory_->requires_send_buffer()) {
     const size_t nak_depth = config ? config->nak_depth_ : MulticastInst::DEFAULT_NAK_DEPTH;
     const size_t default_max_samples = DEFAULT_CONFIG_MAX_SAMPLES_PER_PACKET;
     const size_t max_samples_per_packet = config ? config->max_samples_per_packet_ : default_max_samples;
     send_buffer_.reset(new SingleSendBuffer(nak_depth, max_samples_per_packet));
     send_strategy_->send_buffer(send_buffer_.get());
   }
 }

◆ ~MulticastDataLink()

OpenDDS::DCPS::MulticastDataLink::~MulticastDataLink ( )

virtual

Definition at line 65 of file MulticastDataLink.cpp.

References OpenDDS::DCPS::TransportSendStrategy::send_buffer(), send_buffer_, and send_strategy_.

 {
   if (send_buffer_) {
     send_strategy_->send_buffer(0);
   }
 }

Member Function Documentation

◆ check_header() [1/2]

bool OpenDDS::DCPS::MulticastDataLink::check_header ( const TransportHeader & header )

Definition at line 208 of file MulticastDataLink.cpp.

References ACE_GUARD_RETURN, ACE_SYNCH_RECURSIVE_MUTEX, OpenDDS::DCPS::DataLink::is_active(), session_lock_, sessions_, and OpenDDS::DCPS::TransportHeader::source_.

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

 {
   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
       guard,
       this->session_lock_,
       false);
 
   MulticastSessionMap::iterator it(this->sessions_.find(header.source_));
   if (it == this->sessions_.end() && is_active()) {
     return false;
   }
   if (it != this->sessions_.end() && it->second->acked()) {
     return it->second->check_header(header);
   }
 
   return true;
 }

◆ check_header() [2/2]

bool OpenDDS::DCPS::MulticastDataLink::check_header ( const DataSampleHeader & header )

Definition at line 227 of file MulticastDataLink.cpp.

References ACE_GUARD_RETURN, ACE_SYNCH_RECURSIVE_MUTEX, OpenDDS::DCPS::DataSampleHeader::message_id_, receive_strategy(), session_lock_, sessions_, and OpenDDS::DCPS::TRANSPORT_CONTROL.

 {
   if (header.message_id_ == TRANSPORT_CONTROL) return true;
 
   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
       guard,
       this->session_lock_,
       false);
 
   // Skip data sample unless there is a session for it.
   return (this->sessions_.count(receive_strategy()->received_header().source_) > 0);
 }

◆ client_stop()

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

Definition at line 461 of file MulticastDataLink.cpp.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL, and send_buffer_.

 {
   if (send_buffer_) {
     send_buffer_->retain_all(localId);
     send_buffer_.reset();
   }
 }

◆ config()

ACE_INLINE MulticastInst_rch OpenDDS::DCPS::MulticastDataLink::config ( )

Definition at line 44 of file MulticastDataLink.inl.

References ACE_INLINE, OPENDDS_TEST_AND_CALL_ASSIGN, and transport().

Referenced by OpenDDS::DCPS::ReliableSession::expire_naks(), join(), OpenDDS::DCPS::ReliableSession::nak_delay(), OpenDDS::DCPS::MulticastSendStrategy::send_bytes_i(), OpenDDS::DCPS::ReliableSession::send_naks(), and syn_received_no_session().

 {
   MulticastInst_rch result;
   OPENDDS_TEST_AND_CALL_ASSIGN(MulticastTransport_rch, transport(), config(), result);
   return result;
 }

◆ find_or_create_session()

MulticastSession_rch OpenDDS::DCPS::MulticastDataLink::find_or_create_session ( MulticastPeer remote_peer )

Definition at line 166 of file MulticastDataLink.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_SYNCH_RECURSIVE_MUTEX, ACE_TEXT(), OpenDDS::DCPS::MulticastSessionFactory::create(), OpenDDS::DCPS::RcHandle< T >::is_nil(), LM_ERROR, session_factory_, session_lock_, sessions_, and transport().

 {
   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
       guard,
       this->session_lock_,
       MulticastSession_rch());
 
   MulticastSessionMap::iterator it(this->sessions_.find(remote_peer));
   if (it != this->sessions_.end()) {
     return it->second;
   }
 
   MulticastSession_rch session;
   MulticastTransport_rch mt = transport();
   if (mt) {
     session = session_factory_->create(mt->reactor_task()->interceptor(), this, remote_peer);
     if (session.is_nil()) {
       ACE_ERROR_RETURN((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("MulticastDataLink::find_or_create_session: ")
           ACE_TEXT("failed to create session for remote peer: %#08x%08x!\n"),
           (unsigned int) (remote_peer >> 32),
           (unsigned int) remote_peer),
           MulticastSession_rch());
     }
 
     std::pair<MulticastSessionMap::iterator, bool> pair = this->sessions_.insert(
         MulticastSessionMap::value_type(remote_peer, session));
     if (pair.first == this->sessions_.end()) {
       ACE_ERROR_RETURN((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("MulticastDataLink::find_or_create_session: ")
           ACE_TEXT("failed to insert session for remote peer: %#08x%08x!\n"),
           (unsigned int) (remote_peer >> 32),
           (unsigned int) remote_peer),
           MulticastSession_rch());
     }
   }
   return session;
 }

◆ find_session()

MulticastSession_rch OpenDDS::DCPS::MulticastDataLink::find_session ( MulticastPeer remote_peer )

Definition at line 151 of file MulticastDataLink.cpp.

References ACE_GUARD_RETURN, ACE_SYNCH_RECURSIVE_MUTEX, session_lock_, and sessions_.

Referenced by make_reservation(), and release_reservations_i().

 {
   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
       guard,
       this->session_lock_,
       MulticastSession_rch());
 
   MulticastSessionMap::iterator it(this->sessions_.find(remote_peer));
   if (it != this->sessions_.end()) {
     return it->second;
   }
   else return MulticastSession_rch();
 }

◆ get_proactor()

ACE_INLINE ACE_Proactor * OpenDDS::DCPS::MulticastDataLink::get_proactor ( )

Definition at line 65 of file MulticastDataLink.inl.

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

Referenced by OpenDDS::DCPS::MulticastSendStrategy::async_send().

 {
   if (this->reactor_task_ == 0) return 0;
   return this->reactor_task_->get_proactor();
 }

◆ get_reactor()

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

Definition at line 58 of file MulticastDataLink.inl.

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

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

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

◆ join()

bool OpenDDS::DCPS::MulticastDataLink::join ( const ACE_INET_Addr & group_address )

Definition at line 74 of file MulticastDataLink.cpp.

References ACE_DEFAULT_MAX_SOCKET_BUFSIZ, ACE_ERROR_RETURN, ACE_TEXT(), ACE_TEXT_CHAR_TO_TCHAR, ACE_SOCK::close(), config(), ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE, ENOTSUP, ACE_IPC_SAP::get_handle(), ACE_SOCK_Dgram_Mcast::join(), LM_DEBUG, LM_ERROR, ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO, ACE_SOCK_Dgram_Mcast::opts(), recv_strategy_, send_strategy_, OpenDDS::DCPS::set_socket_multicast_ttl(), ACE_OS::setsockopt(), SO_RCVBUF, SO_SNDBUF, socket_, SOL_SOCKET, OpenDDS::DCPS::DataLink::start(), ACE_Utils::truncate_cast(), and VDBG_LVL.

 {
   MulticastInst_rch cfg = config();
   if (!cfg) {
     return false;
   }
 
   const std::string& net_if = cfg->local_address_;
 #ifdef ACE_HAS_MAC_OSX
   socket_.opts(ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO |
                ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE);
 #endif
   if (this->socket_.join(group_address, 1,
       net_if.empty() ? 0 :
           ACE_TEXT_CHAR_TO_TCHAR(net_if.c_str())) != 0) {
     ACE_ERROR_RETURN((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: MulticastDataLink::join: ")
         ACE_TEXT("ACE_SOCK_Dgram_Mcast::join failed %m.\n")),
         false);
   }
   VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) MulticastDataLink::join OK\n")), 6);
 
   ACE_HANDLE handle = this->socket_.get_handle();
 
   if (!OpenDDS::DCPS::set_socket_multicast_ttl(this->socket_, cfg->ttl_)) {
     ACE_ERROR_RETURN((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: ")
         ACE_TEXT("MulticastDataLink::join: ")
         ACE_TEXT("OpenDDS::DCPS::set_socket_multicast_ttl failed.\n")),
         false);
   }
 
   int rcv_buffer_size = ACE_Utils::truncate_cast<int>(cfg->rcv_buffer_size_);
   if (rcv_buffer_size != 0
       && ACE_OS::setsockopt(handle, SOL_SOCKET,
           SO_RCVBUF,
           (char *) &rcv_buffer_size,
           sizeof (int)) < 0) {
     ACE_ERROR_RETURN((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: ")
         ACE_TEXT("MulticastDataLink::join: ")
         ACE_TEXT("ACE_OS::setsockopt RCVBUF failed.\n")),
         false);
   }
 
 #if defined (ACE_DEFAULT_MAX_SOCKET_BUFSIZ)
   int snd_size = ACE_DEFAULT_MAX_SOCKET_BUFSIZ;
 
   if (ACE_OS::setsockopt(handle, SOL_SOCKET,
       SO_SNDBUF,
       (char *) &snd_size,
       sizeof(snd_size)) < 0
       && errno != ENOTSUP) {
     ACE_ERROR_RETURN((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: ")
         ACE_TEXT("MulticastDataLink::join: ")
         ACE_TEXT("ACE_OS::setsockopt SNDBUF failed to set the send buffer size to %d errno %m\n"),
         snd_size),
         false);
   }
 #endif /* ACE_DEFAULT_MAX_SOCKET_BUFSIZ */
 
   if (start(static_rchandle_cast<TransportSendStrategy>(this->send_strategy_),
       static_rchandle_cast<TransportStrategy>(this->recv_strategy_))
       != 0) {
     this->socket_.close();
     ACE_ERROR_RETURN((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: ")
         ACE_TEXT("MulticastDataLink::join: ")
         ACE_TEXT("DataLink::start failed!\n")),
         false);
   }
 
   return true;
 }

◆ local_peer()

ACE_INLINE MulticastPeer OpenDDS::DCPS::MulticastDataLink::local_peer ( ) const

Definition at line 20 of file MulticastDataLink.inl.

References ACE_INLINE, and local_peer_.

Referenced by OpenDDS::DCPS::ReliableSession::nak_received(), OpenDDS::DCPS::MulticastSendStrategy::prepare_header_i(), OpenDDS::DCPS::MulticastSession::send_syn(), OpenDDS::DCPS::MulticastSession::send_synack(), OpenDDS::DCPS::MulticastSession::syn_received(), syn_received_no_session(), and OpenDDS::DCPS::MulticastSession::synack_received().

 {
   return this->local_peer_;
 }

◆ make_reservation()

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

virtual

Only called by our TransportImpl object.

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

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 258 of file MulticastDataLink.cpp.

References OpenDDS::DCPS::RepoIdConverter::federationId(), find_session(), OpenDDS::DCPS::DataLink::make_reservation(), and OpenDDS::DCPS::RepoIdConverter::participantId().

 {
   int result = DataLink::make_reservation(rpi, lsi, trl, reliable);
   if (reliable) {
     const MulticastPeer remote_peer = (ACE_INT64)RepoIdConverter(rpi).federationId() << 32
       | RepoIdConverter(rpi).participantId();
     MulticastSession_rch session = find_session(remote_peer);
     if (session) {
       session->add_remote(lsi, rpi);
     }
   } else {
     const MulticastPeer remote_peer = (ACE_INT64)RepoIdConverter(rpi).federationId() << 32
       | RepoIdConverter(rpi).participantId();
     MulticastSession_rch session = find_session(remote_peer);
     if (session) {
       session->add_remote(lsi);
     }
   }
   return result;
 }

◆ OPENDDS_MAP()

typedef OpenDDS::DCPS::MulticastDataLink::OPENDDS_MAP	(	MulticastPeer	,
		MulticastSession_rch
	)

private

◆ reactor_task()

ACE_INLINE ReactorTask_rch OpenDDS::DCPS::MulticastDataLink::reactor_task ( )

Definition at line 52 of file MulticastDataLink.inl.

References ACE_INLINE, and reactor_task_.

 {
   return this->reactor_task_;
 }

◆ ready_to_deliver()

bool OpenDDS::DCPS::MulticastDataLink::ready_to_deliver ( const ReceivedDataSample & data )

private

Definition at line 359 of file MulticastDataLink.cpp.

References ACE_GUARD_RETURN, ACE_SYNCH_RECURSIVE_MUTEX, receive_strategy(), OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::received_header(), session_lock_, sessions_, and OpenDDS::DCPS::TransportHeader::source_.

Referenced by sample_received().

 {
   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
       guard,
       this->session_lock_, false);
 
   const TransportHeader& theader = receive_strategy()->received_header();
 
   MulticastSessionMap::iterator session_it = sessions_.find(theader.source_);
   if (session_it != sessions_.end()) {
     MulticastSession_rch sess_rch(session_it->second);
     guard.release();
     return sess_rch->ready_to_deliver(theader, data);
   }
 
   return true;
 }

◆ reassemble()

bool OpenDDS::DCPS::MulticastDataLink::reassemble	(	ReceivedDataSample &	data,
		const TransportHeader &	header
	)

Definition at line 241 of file MulticastDataLink.cpp.

References ACE_GUARD_RETURN, ACE_SYNCH_RECURSIVE_MUTEX, session_lock_, sessions_, and OpenDDS::DCPS::TransportHeader::source_.

Referenced by OpenDDS::DCPS::MulticastReceiveStrategy::reassemble().

 {
   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
       guard,
       this->session_lock_,
       false);
 
   MulticastSessionMap::iterator it(this->sessions_.find(header.source_));
   if (it == this->sessions_.end()) return false;
   if (it->second->acked()) {
     return it->second->reassemble(data, header);
   }
   return false;
 }

◆ receive_strategy()

ACE_INLINE MulticastReceiveStrategy * OpenDDS::DCPS::MulticastDataLink::receive_strategy ( )

Definition at line 32 of file MulticastDataLink.inl.

References ACE_INLINE, OpenDDS::DCPS::RcHandle< T >::in(), and recv_strategy_.

Referenced by check_header(), OpenDDS::DCPS::ReliableSession::nak_received(), OpenDDS::DCPS::ReliableSession::nakack_received(), ready_to_deliver(), sample_received(), OpenDDS::DCPS::MulticastSession::syn_received(), and OpenDDS::DCPS::MulticastSession::synack_received().

 {
   return this->recv_strategy_.in();
 }

◆ release_remote_i()

void OpenDDS::DCPS::MulticastDataLink::release_remote_i ( const GUID_t & remote )

privatevirtual

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 378 of file MulticastDataLink.cpp.

References ACE_GUARD, ACE_SYNCH_RECURSIVE_MUTEX, OpenDDS::DCPS::RepoIdConverter::federationId(), OpenDDS::DCPS::RepoIdConverter::participantId(), session_lock_, and sessions_.

 {
   ACE_GUARD(ACE_SYNCH_RECURSIVE_MUTEX, guard, session_lock_);
   MulticastPeer remote_source = (ACE_INT64)RepoIdConverter(remote).federationId() << 32
                               | RepoIdConverter(remote).participantId();
   MulticastSessionMap::iterator session_it = sessions_.find(remote_source);
   if (session_it != sessions_.end() && session_it->second->is_reliable()) {
     session_it->second->release_remote(remote);
   }
 }

◆ release_reservations_i()

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

virtual

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 283 of file MulticastDataLink.cpp.

References OpenDDS::DCPS::RepoIdConverter::federationId(), find_session(), and OpenDDS::DCPS::RepoIdConverter::participantId().

 {
   const MulticastPeer remote_peer = (ACE_INT64)RepoIdConverter(remote_id).federationId() << 32
     | RepoIdConverter(remote_id).participantId();
   MulticastSession_rch session = find_session(remote_peer);
   if (session) {
     session->remove_remote(local_id, remote_id);
   }
 }

◆ sample_received()

void OpenDDS::DCPS::MulticastDataLink::sample_received ( ReceivedDataSample & sample )

Definition at line 295 of file MulticastDataLink.cpp.

References ACE_GUARD, ACE_SYNCH_RECURSIVE_MUTEX, OpenDDS::DCPS::ReceivedDataSample::data(), OpenDDS::DCPS::DataLink::data_received(), OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::DataLink::is_active(), OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::MULTICAST_SYN, ACE_Message_Block::rd_ptr(), ready_to_deliver(), receive_strategy(), OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::received_header(), session_lock_, sessions_, OpenDDS::DCPS::TransportHeader::source_, OpenDDS::DCPS::DataSampleHeader::submessage_id_, OpenDDS::DCPS::TransportHeader::swap_bytes(), syn_received_no_session(), and OpenDDS::DCPS::TRANSPORT_CONTROL.

Referenced by OpenDDS::DCPS::MulticastReceiveStrategy::deliver_sample().

 {
   switch (sample.header_.message_id_) {
   case TRANSPORT_CONTROL: {
     // Transport control samples are delivered to all sessions
     // regardless of association status:
     {
       Message_Block_Ptr payload(sample.data());
       char* const ptr = payload ? payload->rd_ptr() : 0;
 
       ACE_GUARD(ACE_SYNCH_RECURSIVE_MUTEX,
           guard,
           this->session_lock_);
 
       const TransportHeader& theader = receive_strategy()->received_header();
 
       if (!is_active() && sample.header_.submessage_id_ == MULTICAST_SYN &&
           sessions_.find(theader.source_) == sessions_.end()) {
         // We have received a SYN but there is no session (yet) for this source.
         // Depending on the data, we may need to send SYNACK.
 
         guard.release();
         syn_received_no_session(theader.source_, payload,
                                 theader.swap_bytes());
 
         guard.acquire();
         MulticastSessionMap::iterator s_itr = sessions_.find(theader.source_);
         if (s_itr != sessions_.end()) {
           s_itr->second->record_header_received(theader);
         }
 
         if (ptr) {
           payload->rd_ptr(ptr);
         }
         return;
       }
 
       MulticastSessionMap temp_sessions(sessions_);
       guard.release();
 
       for (MulticastSessionMap::iterator it(temp_sessions.begin());
           it != temp_sessions.end(); ++it) {
         it->second->control_received(sample.header_.submessage_id_,
                                      payload);
         it->second->record_header_received(theader);
 
         // reset read pointer
         if (ptr) {
           payload->rd_ptr(ptr);
         }
       }
     }
   } break;
 
   default:
 
     if (ready_to_deliver(sample)) {
       data_received(sample);
     }
     break;
   }
 }

◆ send_buffer()

ACE_INLINE SingleSendBuffer * OpenDDS::DCPS::MulticastDataLink::send_buffer ( )

Definition at line 38 of file MulticastDataLink.inl.

References ACE_INLINE, and send_buffer_.

Referenced by OpenDDS::DCPS::ReliableSession::nak_received().

 {
   return this->send_buffer_.get();
 }

◆ send_strategy()

ACE_INLINE MulticastSendStrategy * OpenDDS::DCPS::MulticastDataLink::send_strategy ( )

Definition at line 26 of file MulticastDataLink.inl.

References ACE_INLINE, OpenDDS::DCPS::RcHandle< T >::in(), and send_strategy_.

 {
   return this->send_strategy_.in();
 }

◆ socket()

ACE_INLINE ACE_SOCK_Dgram_Mcast & OpenDDS::DCPS::MulticastDataLink::socket ( void )

Definition at line 72 of file MulticastDataLink.inl.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL, and socket_.

Referenced by OpenDDS::DCPS::MulticastSendStrategy::async_send(), OpenDDS::DCPS::MulticastReceiveStrategy::get_handle(), OpenDDS::DCPS::MulticastReceiveStrategy::receive_bytes(), and OpenDDS::DCPS::MulticastSendStrategy::sync_send().

 {
   return this->socket_;
 }

◆ stop_i()

void OpenDDS::DCPS::MulticastDataLink::stop_i ( )

privatevirtual

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

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 445 of file MulticastDataLink.cpp.

References ACE_GUARD, ACE_SYNCH_RECURSIVE_MUTEX, ACE_SOCK::close(), session_lock_, sessions_, and socket_.

 {
   ACE_GUARD(ACE_SYNCH_RECURSIVE_MUTEX,
       guard,
       this->session_lock_);
 
   for (MulticastSessionMap::iterator it(this->sessions_.begin());
       it != this->sessions_.end(); ++it) {
     it->second->stop();
   }
   this->sessions_.clear();
 
   this->socket_.close();
 }

◆ syn_received_no_session()

void OpenDDS::DCPS::MulticastDataLink::syn_received_no_session	(	MulticastPeer	source,
		const Message_Block_Ptr &	data,
		bool	swap_bytes
	)

private

Definition at line 390 of file MulticastDataLink.cpp.

References ACE_ERROR, ACE_TEXT(), config(), OpenDDS::DCPS::DataLink::create_control(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), header, LM_DEBUG, LM_ERROR, local_peer(), local_peer_, OpenDDS::DCPS::move(), OpenDDS::DCPS::MULTICAST_SYNACK, OpenDDS::DCPS::DataLink::send_control(), OpenDDS::DCPS::SEND_CONTROL_OK, transport(), and VDBG_LVL.

Referenced by sample_received().

 {
   Serializer serializer_read(data.get(), encoding_kind, swap_bytes);
 
   MulticastPeer local_peer;
   serializer_read >> local_peer;
 
   if (local_peer != local_peer_) {
     return;
   }
 
   {
     MulticastInst_rch cfg = config();
     VDBG_LVL((LM_DEBUG, "(%P|%t) MulticastDataLink[%C]::syn_received_no_session "
         "send_synack local %#08x%08x remote %#08x%08x\n",
         cfg ? cfg->name().c_str() : "",
         (unsigned int) (local_peer >> 32),
         (unsigned int) local_peer,
         (unsigned int) (source >> 32),
         (unsigned int) source), 2);
   }
 
   Message_Block_Ptr synack_data(new ACE_Message_Block(sizeof(MulticastPeer)));
 
   Serializer serializer_write(synack_data.get(), encoding_kind);
   serializer_write << source;
 
   DataSampleHeader header;
   Message_Block_Ptr control(
       create_control(MULTICAST_SYNACK, header, move(synack_data)));
 
   if (control == 0) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: ")
                ACE_TEXT("MulticastDataLink::syn_received_no_session: ")
                ACE_TEXT("create_control failed!\n")));
     return;
   }
 
   const int error = send_control(header, move(control));
   if (error != SEND_CONTROL_OK) {
     ACE_ERROR((LM_ERROR, "(%P|%t) MulticastDataLink::syn_received_no_session: "
         "ERROR: send_control failed: %d!\n", error));
     return;
   }
 
   MulticastTransport_rch mt= transport();
   if (mt) {
     mt->passive_connection(local_peer, source);
   }
 }

◆ transport()

ACE_INLINE MulticastTransport_rch OpenDDS::DCPS::MulticastDataLink::transport ( void )

Definition at line 14 of file MulticastDataLink.inl.

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

Referenced by config(), find_or_create_session(), OpenDDS::DCPS::MulticastSession::syn_received(), and syn_received_no_session().

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

Member Data Documentation

◆ local_peer_

MulticastPeer OpenDDS::DCPS::MulticastDataLink::local_peer_

private

Definition at line 87 of file MulticastDataLink.h.

Referenced by local_peer(), and syn_received_no_session().

◆ reactor_task_

ReactorTask_rch OpenDDS::DCPS::MulticastDataLink::reactor_task_

private

Definition at line 89 of file MulticastDataLink.h.

Referenced by get_proactor(), get_reactor(), and reactor_task().

◆ readers_selected_

RepoIdSet OpenDDS::DCPS::MulticastDataLink::readers_selected_

private

Definition at line 109 of file MulticastDataLink.h.

◆ readers_withheld_

RepoIdSet OpenDDS::DCPS::MulticastDataLink::readers_withheld_

private

Definition at line 109 of file MulticastDataLink.h.

◆ recv_strategy_

MulticastReceiveStrategy_rch OpenDDS::DCPS::MulticastDataLink::recv_strategy_

private

Definition at line 92 of file MulticastDataLink.h.

Referenced by join(), and receive_strategy().

◆ send_buffer_

unique_ptr<SingleSendBuffer> OpenDDS::DCPS::MulticastDataLink::send_buffer_

private

Definition at line 94 of file MulticastDataLink.h.

Referenced by client_stop(), MulticastDataLink(), send_buffer(), and ~MulticastDataLink().

◆ send_strategy_

MulticastSendStrategy_rch OpenDDS::DCPS::MulticastDataLink::send_strategy_

private

Definition at line 91 of file MulticastDataLink.h.

Referenced by join(), MulticastDataLink(), send_strategy(), and ~MulticastDataLink().

◆ session_factory_

MulticastSessionFactory_rch OpenDDS::DCPS::MulticastDataLink::session_factory_

private

Definition at line 85 of file MulticastDataLink.h.

Referenced by find_or_create_session(), and MulticastDataLink().

◆ session_lock_

ACE_SYNCH_RECURSIVE_MUTEX OpenDDS::DCPS::MulticastDataLink::session_lock_

private

Definition at line 98 of file MulticastDataLink.h.

Referenced by check_header(), find_or_create_session(), find_session(), ready_to_deliver(), reassemble(), release_remote_i(), sample_received(), and stop_i().

◆ sessions_

MulticastSessionMap OpenDDS::DCPS::MulticastDataLink::sessions_

private

Definition at line 101 of file MulticastDataLink.h.

Referenced by check_header(), find_or_create_session(), find_session(), ready_to_deliver(), reassemble(), release_remote_i(), sample_received(), and stop_i().

◆ socket_

ACE_SOCK_Dgram_Mcast OpenDDS::DCPS::MulticastDataLink::socket_

private

Definition at line 96 of file MulticastDataLink.h.

Referenced by join(), socket(), and stop_i().

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

Private Attributes
MulticastSessionFactory_rch	session_factory_

MulticastPeer	local_peer_

ReactorTask_rch	reactor_task_

MulticastSendStrategy_rch	send_strategy_

MulticastReceiveStrategy_rch	recv_strategy_

unique_ptr< SingleSendBuffer >	send_buffer_

ACE_SOCK_Dgram_Mcast	socket_

ACE_SYNCH_RECURSIVE_MUTEX	session_lock_

MulticastSessionMap	sessions_

RepoIdSet	readers_selected_

RepoIdSet	readers_withheld_

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ MulticastDataLink()

◆ ~MulticastDataLink()

Member Function Documentation

◆ check_header() [1/2]

◆ check_header() [2/2]

◆ client_stop()

◆ config()

◆ find_or_create_session()

◆ find_session()

◆ get_proactor()

◆ get_reactor()

◆ join()

◆ local_peer()

◆ make_reservation()

◆ OPENDDS_MAP()

◆ reactor_task()

◆ ready_to_deliver()

◆ reassemble()

◆ receive_strategy()

◆ release_remote_i()

◆ release_reservations_i()

◆ sample_received()

◆ send_buffer()

◆ send_strategy()

◆ socket()

◆ stop_i()

◆ syn_received_no_session()

◆ transport()

Member Data Documentation

◆ local_peer_

◆ reactor_task_

◆ readers_selected_

◆ readers_withheld_

◆ recv_strategy_

◆ send_buffer_

◆ send_strategy_

◆ session_factory_

◆ session_lock_

◆ sessions_

◆ socket_