#include <TcpDataLink.h>

Inheritance diagram for OpenDDS::DCPS::TcpDataLink:

[legend]

Collaboration diagram for OpenDDS::DCPS::TcpDataLink:

[legend]

Public Member Functions
	TcpDataLink (const TcpTransport_rch &transport_impl, const ACE_INET_Addr &remote_address, Priority priority, bool is_loopback, bool is_active)

virtual	~TcpDataLink ()

const ACE_INET_Addr &	remote_address () const
	Accessor for the remote address. More...

int	connect (const TcpConnection_rch &connection, const RcHandle< TcpSendStrategy > &send_strategy, const RcHandle< TcpReceiveStrategy > &receive_strategy)

int	reuse_existing_connection (const TcpConnection_rch &connection)

int	reconnect (const TcpConnection_rch &connection)

TcpConnection_rch	get_connection ()

bool	check_active_client (const GUID_t &local_id)

void	client_stop (const GUID_t &local_id)

virtual void	pre_stop_i ()

void	set_release_pending (bool flag)
	Set release pending flag. More...

bool	is_release_pending () const
	Get release pending flag. More...

void	ack_received (const ReceivedDataSample &sample)

void	request_ack_received (const ReceivedDataSample &sample)

void	drop_pending_request_acks ()

TcpSendStrategy_rch	send_strategy ()

TcpReceiveStrategy_rch	receive_strategy ()

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

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

void	do_association_actions ()

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

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)

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

Protected Member Functions
virtual void	stop_i ()

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

virtual void	send_stop_i (GUID_t repoId)

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

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

Private Member Functions
bool	handle_send_request_ack (TransportQueueElement *element)

void	send_graceful_disconnect_message ()

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

typedef	OPENDDS_VECTOR (TransportQueueElement *) PendingRequestAcks

typedef	OPENDDS_SET_CMP (GUID_t, GUID_tKeyLessThan) RepoIdSetType

Private Attributes
ACE_INET_Addr	remote_address_

WeakRcHandle< TcpConnection >	connection_

bool	graceful_disconnect_sent_

AtomicBool	release_is_pending_

ACE_SYNCH_MUTEX	pending_request_acks_lock_

PendingRequestAcks	pending_request_acks_

RepoIdSetType	stopped_clients_

ACE_Thread_Mutex	stopped_clients_mutex_

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

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 25 of file TcpDataLink.h.

Constructor & Destructor Documentation

◆ TcpDataLink()

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL OpenDDS::DCPS::TcpDataLink::TcpDataLink	(	const TcpTransport_rch &	transport_impl,
		const ACE_INET_Addr &	remote_address,
		Priority	priority,
		bool	is_loopback,
		bool	is_active
	)

Definition at line 30 of file TcpDataLink.cpp.

References DBG_ENTRY_LVL.

   : DataLink(transport_impl, priority, is_loopback, is_active)
   , remote_address_(remote_address)
   , graceful_disconnect_sent_(false)
   , release_is_pending_(false)
 {
   DBG_ENTRY_LVL("TcpDataLink","TcpDataLink",6);
 }

◆ ~TcpDataLink()

OpenDDS::DCPS::TcpDataLink::~TcpDataLink ( )

virtual

Definition at line 44 of file TcpDataLink.cpp.

References DBG_ENTRY_LVL.

 {
   DBG_ENTRY_LVL("TcpDataLink","~TcpDataLink",6);
 }

Member Function Documentation

◆ ack_received()

void OpenDDS::DCPS::TcpDataLink::ack_received ( const ReceivedDataSample & sample )

Definition at line 408 of file TcpDataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::ReceivedDataSample::header_, LM_DEBUG, pending_request_acks_, pending_request_acks_lock_, OpenDDS::DCPS::DataSampleHeader::publication_id_, send_strategy(), OpenDDS::DCPS::DataSampleHeader::sequence_, and OpenDDS::DCPS::Transport_debug_level.

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

 {
   const SequenceNumber sequence = sample.header_.sequence_;
 
   if (sequence == -1) {
     return;
   }
 
   if (Transport_debug_level >= 1) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TcpDataLink::ack_received() received sequence number %q, publiction_id=%C\n"),
       sequence.getValue(), LogGuid(sample.header_.publication_id_).c_str()));
   }
 
   TransportQueueElement* elem=0;
   {
     // find the pending request with the same sequence number.
     ACE_Guard<ACE_SYNCH_MUTEX> guard(pending_request_acks_lock_);
     PendingRequestAcks::iterator it;
     for (it = pending_request_acks_.begin(); it != pending_request_acks_.end(); ++it) {
       if ((*it)->sequence() == sequence && (*it)->publication_id() == sample.header_.publication_id_) {
         elem = *it;
         pending_request_acks_.erase(it);
         break;
       }
     }
   }
 
   if (elem) {
     if (Transport_debug_level >= 1) {
       ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TcpDataLink::ack_received() found matching element %@\n"),
         elem));
     }
     send_strategy()->deliver_ack_request(elem);
   }
   else {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TcpDataLink::ack_received() received unknown sequence number %q\n"),
       sequence.getValue()));
   }
 }

◆ check_active_client()

bool OpenDDS::DCPS::TcpDataLink::check_active_client ( const GUID_t & local_id )

Definition at line 88 of file TcpDataLink.cpp.

References stopped_clients_, and stopped_clients_mutex_.

 {
   ACE_Guard<ACE_Thread_Mutex> guard(stopped_clients_mutex_);
   return stopped_clients_.count(local_id) == 0;
 }

◆ client_stop()

void OpenDDS::DCPS::TcpDataLink::client_stop ( const GUID_t & local_id )

Definition at line 95 of file TcpDataLink.cpp.

References send_strategy(), stopped_clients_, and stopped_clients_mutex_.

 {
   ACE_Guard<ACE_Thread_Mutex> guard(stopped_clients_mutex_);
   stopped_clients_.insert(local_id);
 
   TcpSendStrategy_rch strategy = send_strategy();
   if (strategy) {
     strategy->remove_all_msgs(local_id);
   }
 }

◆ connect()

int OpenDDS::DCPS::TcpDataLink::connect	(	const TcpConnection_rch &	connection,
		const RcHandle< TcpSendStrategy > &	send_strategy,
		const RcHandle< TcpReceiveStrategy > &	receive_strategy
	)

Called when an established connection object is available for this TcpDataLink. Called by the TcpTransport's connect_datalink() method.

The TcpTransport calls this method when it has an established connection object for us. This call puts this TcpDataLink into the "connected" state.

Definition at line 138 of file TcpDataLink.cpp.

References ACE_ERROR_RETURN, ACE_NONBLOCK, ACE_TEXT(), connection_, DBG_ENTRY_LVL, do_association_actions(), LM_ERROR, OpenDDS::DCPS::rchandle_from(), OpenDDS::DCPS::DataLink::start(), and OpenDDS::DCPS::DataLink::strategy_lock_.

Referenced by OpenDDS::DCPS::TcpTransport::connect_tcp_datalink().

 {
   DBG_ENTRY_LVL("TcpDataLink","connect",6);
 
   {
     GuardType guard(strategy_lock_);
     this->connection_ = connection;
   }
 
   if (connection->peer().enable(ACE_NONBLOCK) == -1) {
     ACE_ERROR_RETURN((LM_ERROR,
                       "(%P|%t) ERROR: TcpDataLink::connect failed to set "
                       "ACE_NONBLOCK %p\n", ACE_TEXT("enable")), -1);
   }
 
   // Let connection know the datalink for callbacks upon reconnect failure.
   connection->set_datalink(rchandle_from(this));
 
   // And lastly, inform our base class (DataLink) that we are now "connected",
   // and it should start the strategy objects.
   if (this->start(send_strategy, receive_strategy, false) != 0) {
     // Our base (DataLink) class failed to start the strategy objects.
     // We need to "undo" some things here before we return -1 to indicate
     // that an error has taken place.
 
     // Drop our reference to the connection object.
     this->connection_.reset();
 
     return -1;
   }
 
   do_association_actions();
   return 0;
 }

◆ do_association_actions()

void OpenDDS::DCPS::TcpDataLink::do_association_actions ( )

Definition at line 498 of file TcpDataLink.cpp.

References connection_, OpenDDS::DCPS::DataLink::invoke_on_start_callbacks(), OpenDDS::DCPS::GuidConverter::isReader(), OpenDDS::DCPS::TransportSendStrategy::link_released(), OpenDDS::DCPS::DataLink::on_start_callbacks_, send_association_msg(), OpenDDS::DCPS::DataLink::send_strategy_, and OpenDDS::DCPS::DataLink::strategy_lock_.

Referenced by OpenDDS::DCPS::TcpTransport::accept_datalink(), connect(), OpenDDS::DCPS::TcpTransport::connect_datalink(), reconnect(), and reuse_existing_connection().

 {
   // We have a connection.
   // Invoke callbacks for readers so we can receive messages and let writers know we are ready.
   typedef std::vector<std::pair<GUID_t, GUID_t> > PairVec;
   PairVec to_call_and_send;
 
   {
     GuardType guard(strategy_lock_);
 
     if (!connection_ || !send_strategy_) {
       return;
     }
 
     for (OnStartCallbackMap::const_iterator it = on_start_callbacks_.begin(); it != on_start_callbacks_.end(); ++it) {
       for (RepoToClientMap::const_iterator it2 = it->second.begin(); it2 != it->second.end(); ++it2) {
         if (GuidConverter(it2->first).isReader()) {
           to_call_and_send.push_back(std::make_pair(it2->first, it->first));
         }
       }
     }
   }
 
   send_strategy_->link_released(false);
 
   for (PairVec::const_iterator it = to_call_and_send.begin(); it != to_call_and_send.end(); ++it) {
     invoke_on_start_callbacks(it->first, it->second, true);
     send_association_msg(it->first, it->second);
   }
 }

◆ drop_pending_request_acks()

void OpenDDS::DCPS::TcpDataLink::drop_pending_request_acks ( )

Definition at line 563 of file TcpDataLink.cpp.

References pending_request_acks_, and pending_request_acks_lock_.

Referenced by OpenDDS::DCPS::TcpConnection::disconnect(), OpenDDS::DCPS::TcpConnection::notify_connection_lost(), OpenDDS::DCPS::TcpReceiveStrategy::reset(), and OpenDDS::DCPS::TcpReceiveStrategy::stop_i().

 {
   ACE_Guard<ACE_SYNCH_MUTEX> guard(pending_request_acks_lock_);
   PendingRequestAcks::iterator it;
   for (it = pending_request_acks_.begin(); it != pending_request_acks_.end(); ++it) {
     (*it)->data_dropped(true);
   }
   pending_request_acks_.clear();
 }

◆ get_connection()

ACE_INLINE OpenDDS::DCPS::TcpConnection_rch OpenDDS::DCPS::TcpDataLink::get_connection ( void )

Definition at line 22 of file TcpDataLink.inl.

References connection_, and OPENDDS_END_VERSIONED_NAMESPACE_DECL.

Referenced by OpenDDS::DCPS::TcpTransport::fresh_link(), OpenDDS::DCPS::TcpSendStrategy::get_handle(), OpenDDS::DCPS::TcpSynchResource::notify_lost_on_backpressure_timeout(), OpenDDS::DCPS::TcpReceiveStrategy::receive_bytes(), OpenDDS::DCPS::TcpSendStrategy::relink(), OpenDDS::DCPS::TcpReceiveStrategy::relink(), OpenDDS::DCPS::TcpSendStrategy::send_bytes_i(), and OpenDDS::DCPS::TcpReceiveStrategy::start_i().

 {
   return this->connection_.lock();
 }

◆ handle_send_request_ack()

bool OpenDDS::DCPS::TcpDataLink::handle_send_request_ack ( TransportQueueElement * element )

privatevirtual

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 389 of file TcpDataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), connection_, OpenDDS::DCPS::TransportQueueElement::data_dropped(), OpenDDS::DCPS::SequenceNumber::getValue(), LM_DEBUG, pending_request_acks_, pending_request_acks_lock_, OpenDDS::DCPS::TransportQueueElement::publication_id(), OpenDDS::DCPS::TransportQueueElement::sequence(), and OpenDDS::DCPS::Transport_debug_level.

 {
   if (Transport_debug_level >= 1) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TcpDataLink::handle_send_request_ack(%@) sequence number %q, publication_id=%C\n"),
       element, element->sequence().getValue(), LogGuid(element->publication_id()).c_str()));
   }
   bool result = false;
   TcpConnection_rch connection(connection_.lock());
   if (connection) {
     ACE_Guard<ACE_SYNCH_MUTEX> guard(pending_request_acks_lock_);
     pending_request_acks_.push_back(element);
   } else {
     element->data_dropped(true);
     result = true;
   }
   return result;
 }

◆ is_release_pending()

bool OpenDDS::DCPS::TcpDataLink::is_release_pending ( ) const

Get release pending flag.

Definition at line 383 of file TcpDataLink.cpp.

References release_is_pending_.

Referenced by OpenDDS::DCPS::TcpConnection::active_reconnect_i().

 {
   return release_is_pending_;
 }

◆ make_reservation() [1/2]

int OpenDDS::DCPS::TcpDataLink::make_reservation	(	const GUID_t &	remote_subscription_id,
		const GUID_t &	local_publication_id,
		const TransportSendListener_wrch &	send_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 588 of file TcpDataLink.cpp.

References OpenDDS::DCPS::DataLink::make_reservation(), send_association_msg(), stopped_clients_, and stopped_clients_mutex_.

 {
   {
     ACE_Guard<ACE_Thread_Mutex> guard(stopped_clients_mutex_);
     stopped_clients_.erase(local_publication_id);
   }
   const int result = DataLink::make_reservation(remote_subscription_id, local_publication_id, send_listener, reliable);
   send_association_msg(local_publication_id, remote_subscription_id);
   return result;
 }

◆ make_reservation() [2/2]

int OpenDDS::DCPS::TcpDataLink::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 603 of file TcpDataLink.cpp.

References OpenDDS::DCPS::DataLink::make_reservation(), OPENDDS_END_VERSIONED_NAMESPACE_DECL, send_association_msg(), stopped_clients_, and stopped_clients_mutex_.

 {
   {
     ACE_Guard<ACE_Thread_Mutex> guard(stopped_clients_mutex_);
     stopped_clients_.erase(local_subscription_id);
   }
   const int result = DataLink::make_reservation(remote_publication_id, local_subscription_id, receive_listener, reliable);
   send_association_msg(local_subscription_id, remote_publication_id);
 
   return result;
 }

◆ OPENDDS_SET_CMP()

typedef OpenDDS::DCPS::TcpDataLink::OPENDDS_SET_CMP	(	GUID_t	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_VECTOR()

typedef OpenDDS::DCPS::TcpDataLink::OPENDDS_VECTOR ( TransportQueueElement * )

private

◆ pre_stop_i()

void OpenDDS::DCPS::TcpDataLink::pre_stop_i ( )

virtual

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

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 107 of file TcpDataLink.cpp.

References connection_, DBG_ENTRY_LVL, graceful_disconnect_sent_, OpenDDS::DCPS::DataLink::pre_stop_i(), receive_strategy(), and send_graceful_disconnect_message().

 {
   DBG_ENTRY_LVL("TcpDataLink","pre_stop_i",6);
 
   DataLink::pre_stop_i();
 
   TcpReceiveStrategy_rch rs = receive_strategy();
 
   TcpConnection_rch connection(connection_.lock());
 
   if (rs) {
     // If we received the GRACEFUL_DISCONNECT message from peer before we
     // initiate the disconnecting of the datalink, then we will not send
     // the GRACEFUL_DISCONNECT message to the peer.
     bool disconnected = rs->gracefully_disconnected();
 
     if (connection && !graceful_disconnect_sent_ && !disconnected) {
       graceful_disconnect_sent_ = true;
       send_graceful_disconnect_message();
     }
   }
 
   if (connection) {
     connection->shutdown();
   }
 }

◆ receive_strategy()

OpenDDS::DCPS::TcpReceiveStrategy_rch OpenDDS::DCPS::TcpDataLink::receive_strategy ( )

Definition at line 581 of file TcpDataLink.cpp.

References OpenDDS::DCPS::DataLink::receive_strategy_, OpenDDS::DCPS::static_rchandle_cast(), and OpenDDS::DCPS::DataLink::strategy_lock_.

Referenced by OpenDDS::DCPS::TcpConnection::handle_close(), pre_stop_i(), OpenDDS::DCPS::TcpConnection::receive_strategy(), and request_ack_received().

 {
   GuardType guard(strategy_lock_);
   return static_rchandle_cast<OpenDDS::DCPS::TcpReceiveStrategy>(receive_strategy_);
 }

◆ reconnect()

int OpenDDS::DCPS::TcpDataLink::reconnect ( const TcpConnection_rch & connection )

Associate the new connection object with this datalink object. The states of the "old" connection object are copied to the new connection object and the "old" connection object is replaced by the new connection object.

Definition at line 240 of file TcpDataLink.cpp.

References OpenDDS::DCPS::TcpTransport::connect_tcp_datalink(), connection_, DBG_ENTRY_LVL, do_association_actions(), OpenDDS::DCPS::dynamic_rchandle_cast(), OpenDDS::DCPS::DataLink::impl(), OpenDDS::DCPS::RcHandle< T >::in(), LM_ERROR, OpenDDS::DCPS::DataLink::receive_strategy_, OpenDDS::DCPS::RcHandle< T >::reset(), OpenDDS::DCPS::DataLink::send_strategy_, OpenDDS::DCPS::DataLink::strategy_lock_, and VDBG_LVL.

Referenced by OpenDDS::DCPS::TcpTransport::fresh_link().

 {
   DBG_ENTRY_LVL("TcpDataLink","reconnect",6);
 
   TcpConnection_rch existing_connection(connection_.lock());
   // Sanity check - the connection should exist already since we are reconnecting.
   if (!existing_connection) {
     VDBG_LVL((LM_ERROR,
               "(%P|%t) ERROR: TcpDataLink::reconnect old connection is nil.\n")
              , 1);
     return -1;
   }
 
   existing_connection->transfer(connection.in());
 
   bool released = false;
   TcpReceiveStrategy_rch trs;
   TcpSendStrategy_rch tss;
 
   {
     GuardType strategy_guard(strategy_lock_);
 
     trs = dynamic_rchandle_cast<TcpReceiveStrategy>(receive_strategy_);
     tss = dynamic_rchandle_cast<TcpSendStrategy>(send_strategy_);
 
     if (!trs || !tss) {
       // if either are invalid, both should be
       receive_strategy_.reset();
       send_strategy_.reset();
       released = true;
     }
   }
 
   if (released) {
     RcHandle<TcpTransport> transport = dynamic_rchandle_cast<TcpTransport>(impl());
     if (transport) {
       const int result = transport->connect_tcp_datalink(*this, connection);
       if (result == 0) {
         do_association_actions();
       }
       return result;
     }
     return -1;
   }
 
   connection_ = connection;
 
   // Associate the new connection object with the receiveing strategy and disassociate
   // the old connection object with the receiveing strategy.
   int rs_result = trs->reset(existing_connection.in(), connection.in());
 
   // Associate the new connection object with the sending strategy and disassociate
   // the old connection object with the sending strategy.
   int ss_result = tss->reset();
 
   if (rs_result == 0 && ss_result == 0) {
     do_association_actions();
     return 0;
   }
 
   return -1;
 }

◆ remote_address()

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL ACE_INLINE const ACE_INET_Addr & OpenDDS::DCPS::TcpDataLink::remote_address ( ) const

Accessor for the remote address.

Definition at line 15 of file TcpDataLink.inl.

References ACE_INLINE, DBG_ENTRY_LVL, and remote_address_.

Referenced by OpenDDS::DCPS::TcpTransport::release_datalink(), and OpenDDS::DCPS::TcpTransport::unbind_link().

 {
   DBG_ENTRY_LVL("TcpDataLink","remote_address",6);
   return this->remote_address_;
 }

◆ request_ack_received()

void OpenDDS::DCPS::TcpDataLink::request_ack_received ( const ReceivedDataSample & sample )

Definition at line 449 of file TcpDataLink.cpp.

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

 {
   if (sample.header_.sequence_ == -1 && sample.header_.message_length_ == guid_cdr_size) {
     GUID_t local;
     Message_Block_Ptr payload(receive_strategy()->to_msgblock(sample));
     Serializer ser(payload.get(), encoding_unaligned_native);
     if (ser >> local) {
       invoke_on_start_callbacks(local, sample.header_.publication_id_, true);
     }
     return;
   }
 
   DataSampleHeader header_data;
   // The message_id_ is the most important value for the DataSampleHeader.
   header_data.message_id_ = SAMPLE_ACK;
 
   // Other data in the DataSampleHeader are not necessary set. The bogus values
   // can be used.
 
   header_data.byte_order_ = ACE_CDR_BYTE_ORDER;
   header_data.message_length_ = 0;
   header_data.sequence_ = sample.header_.sequence_;
   header_data.publication_id_ = sample.header_.publication_id_;
   header_data.publisher_id_ = sample.header_.publisher_id_;
 
   Message_Block_Ptr message(
     new ACE_Message_Block(header_data.get_max_serialized_size(),
                           ACE_Message_Block::MB_DATA,
                           0, //cont
                           0, //data
                           0, //allocator_strategy
                           0, //locking_strategy
                           ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                           ACE_Time_Value::zero,
                           ACE_Time_Value::max_time,
                           0,
                           0));
 
   *message << header_data;
 
   TransportControlElement* send_element =  new TransportControlElement(move(message));
 
 
   // I don't want to rebuild a connection in order to send
   // a sample ack message
   this->send_i(send_element, false);
 }

◆ reuse_existing_connection()

int OpenDDS::DCPS::TcpDataLink::reuse_existing_connection ( const TcpConnection_rch & connection )

Definition at line 183 of file TcpDataLink.cpp.

References connection_, DBG_ENTRY_LVL, do_association_actions(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::DataLink::is_active_, OpenDDS::DCPS::RcHandle< T >::is_nil(), LM_DEBUG, OpenDDS::DCPS::DataLink::receive_strategy_, OpenDDS::DCPS::TcpReceiveStrategy::reset(), OpenDDS::DCPS::TcpSendStrategy::reset(), OpenDDS::DCPS::DataLink::send_strategy_, and VDBG_LVL.

Referenced by OpenDDS::DCPS::TcpTransport::connect_tcp_datalink().

 {
   DBG_ENTRY_LVL("TcpDataLink","reuse_existing_connection",6);
 
   if (this->is_active_) {
     return -1;
   }
   //Need to check if connection is nil.  If connection is not nil, then connection
   //has previously gone through connection phase so this is a reuse of the connection
   //proceed to determine if we can reuse/reset existing mechanisms or need to start from
   //scratch.
 
   TcpConnection_rch old_connection(this->connection_.lock());
 
   if (old_connection) {
     VDBG_LVL((LM_DEBUG, "(%P|%t) TcpDataLink::reuse_existing_connection - "
                            "trying to reuse existing connection\n"), 0);
     old_connection->transfer(connection.in());
 
     //Connection already exists.
     TransportStrategy_rch brs;
     TransportSendStrategy_rch bss;
 
     if (this->receive_strategy_.is_nil() && this->send_strategy_.is_nil()) {
       return -1;
     } else {
       brs = this->receive_strategy_;
       bss = this->send_strategy_;
 
       this->connection_ = connection;
 
       TcpReceiveStrategy* rs = static_cast<TcpReceiveStrategy*>(brs.in());
 
       TcpSendStrategy* ss = static_cast<TcpSendStrategy*>(bss.in());
 
       // Associate the new connection object with the receiving strategy and disassociate
       // the old connection object with the receiving strategy.
       int rs_result = rs->reset(0, connection.in());
 
       // Associate the new connection object with the sending strategy and disassociate
       // the old connection object with the sending strategy.
       int ss_result = ss->reset(true);
 
       if (rs_result == 0 && ss_result == 0) {
         do_association_actions();
         return 0;
       }
     }
   }
   return -1;
 }

◆ send_association_msg()

void OpenDDS::DCPS::TcpDataLink::send_association_msg	(	const GUID_t &	local,
		const GUID_t &	remote
	)

private

Definition at line 530 of file TcpDataLink.cpp.

References ACE_CDR_BYTE_ORDER, ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, OpenDDS::DCPS::DataSampleHeader::byte_order_, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), OpenDDS::DCPS::DataSampleHeader::get_max_serialized_size(), OpenDDS::DCPS::guid_cdr_size, ACE_Time_Value::max_time, ACE_Message_Block::MB_DATA, OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::DataSampleHeader::message_length_, OpenDDS::DCPS::move(), OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::DataSampleHeader::publisher_id_, OpenDDS::DCPS::REQUEST_ACK, send_i(), OpenDDS::DCPS::DataSampleHeader::sequence_, and ACE_Time_Value::zero.

Referenced by do_association_actions(), and make_reservation().

 {
   DataSampleHeader header_data;
   header_data.message_id_ = REQUEST_ACK;
   header_data.byte_order_ = ACE_CDR_BYTE_ORDER;
   header_data.message_length_ = guid_cdr_size;
   header_data.sequence_ = -1;
   header_data.publication_id_ = local;
   header_data.publisher_id_ = remote;
 
   Message_Block_Ptr message(
     new ACE_Message_Block(header_data.get_max_serialized_size(),
                           ACE_Message_Block::MB_DATA,
                           0, //cont
                           0, //data
                           0, //allocator_strategy
                           0, //locking_strategy
                           ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                           ACE_Time_Value::zero,
                           ACE_Time_Value::max_time,
                           0,
                           0));
 
   *message << header_data;
   Serializer ser(message.get(), encoding_unaligned_native);
   ser << remote;
 
   TransportControlElement* send_element = new TransportControlElement(move(message));
 
   this->send_i(send_element, false);
 }

◆ send_graceful_disconnect_message()

void OpenDDS::DCPS::TcpDataLink::send_graceful_disconnect_message ( )

private

Definition at line 304 of file TcpDataLink.cpp.

References ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, DBG_ENTRY_LVL, OpenDDS::DCPS::DataSampleHeader::get_max_serialized_size(), OpenDDS::DCPS::GRACEFUL_DISCONNECT, ACE_Message_Block::length(), ACE_Time_Value::max_time, ACE_Message_Block::MB_DATA, OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::DataSampleHeader::message_length_, OpenDDS::DCPS::move(), OpenDDS::DCPS::unique_ptr< T, Deleter >::release(), send_i(), OpenDDS::DCPS::DataLink::send_strategy_, OpenDDS::DCPS::TransportSendStrategy::terminate_send(), ACE_Message_Block::wr_ptr(), and ACE_Time_Value::zero.

Referenced by pre_stop_i().

 {
   DBG_ENTRY_LVL("TcpDataLink","send_graceful_disconnect_message",6);
 
   // Will clear all queued messages but still let the disconnect message
   // sent.
   this->send_strategy_->terminate_send(true);
 
   DataSampleHeader header_data;
   // The message_id_ is the most important value for the DataSampleHeader.
   header_data.message_id_ = GRACEFUL_DISCONNECT;
 
   // Other data in the DataSampleHeader are not necessary set. The bogus values
   // can be used.
 
   //header_data.byte_order_
   //  = this->transport_->config()->swap_bytes() ? !TAO_ENCAP_BYTE_ORDER : TAO_ENCAP_BYTE_ORDER;
   //header_data.message_length_ = 0;
   //header_data.sequence_ = 0;
   //DDS::Time_t source_timestamp
   //  = SystemTimePoint::now().to_dds_time();
   //header_data.source_timestamp_sec_ = source_timestamp.sec;
   //header_data.source_timestamp_nanosec_ = source_timestamp.nanosec;
   //header_data.coherency_group_ = 0;
   //header_data.publication_id_ = 0;
 
   // TODO:
   // It seems a bug in the transport implementation that the receiving side can
   // not receive the message when the message has no sample data and is sent
   // in a single packet.
 
   // To work around this problem, I have to add bogus data to chain with the
   // DataSampleHeader to make the receiving work.
 
   Message_Block_Ptr data(
     new ACE_Message_Block(20,
                           ACE_Message_Block::MB_DATA,
                           0, //cont
                           0, //data
                           0, //allocator_strategy
                           0, //locking_strategy
                           ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                           ACE_Time_Value::zero,
                           ACE_Time_Value::max_time,
                           0,
                           0));
   data->wr_ptr(20);
 
   header_data.message_length_ = static_cast<ACE_UINT32>(data->length());
 
   Message_Block_Ptr message(
     new ACE_Message_Block(header_data.get_max_serialized_size(),
                           ACE_Message_Block::MB_DATA,
                           data.release(), //cont
                           0, //data
                           0, //allocator_strategy
                           0, //locking_strategy
                           ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                           ACE_Time_Value::zero,
                           ACE_Time_Value::max_time,
                           0,
                           0));
 
   *message << header_data;
 
   TransportControlElement* send_element = new TransportControlElement(move(message));
 
   // I don't want to rebuild a connection in order to send
   // a graceful disconnect message.
   this->send_i(send_element, false);
 }

◆ send_i()

void OpenDDS::DCPS::TcpDataLink::send_i	(	TransportQueueElement *	element,
		bool	relink = `true`
	)

protectedvirtual

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 68 of file TcpDataLink.cpp.

References OpenDDS::DCPS::TransportQueueElement::data_dropped(), OpenDDS::DCPS::TransportQueueElement::publication_id(), OpenDDS::DCPS::DataLink::send_i(), stopped_clients_, and stopped_clients_mutex_.

Referenced by request_ack_received(), send_association_msg(), and send_graceful_disconnect_message().

 {
   ACE_Guard<ACE_Thread_Mutex> guard(stopped_clients_mutex_);
   if (stopped_clients_.count(element->publication_id())) {
     element->data_dropped(true);
   } else {
     DCPS::DataLink::send_i(element, relink);
   }
 }

◆ send_stop_i()

void OpenDDS::DCPS::TcpDataLink::send_stop_i ( GUID_t repoId )

protectedvirtual

Definition at line 79 of file TcpDataLink.cpp.

References OpenDDS::DCPS::DataLink::send_stop_i(), stopped_clients_, and stopped_clients_mutex_.

 {
   ACE_Guard<ACE_Thread_Mutex> guard(stopped_clients_mutex_);
   if (!stopped_clients_.count(repoId)) {
     DCPS::DataLink::send_stop_i(repoId);
   }
 }

◆ send_strategy()

OpenDDS::DCPS::TcpSendStrategy_rch OpenDDS::DCPS::TcpDataLink::send_strategy ( )

Definition at line 574 of file TcpDataLink.cpp.

References OpenDDS::DCPS::DataLink::send_strategy_, OpenDDS::DCPS::static_rchandle_cast(), and OpenDDS::DCPS::DataLink::strategy_lock_.

Referenced by ack_received(), client_stop(), OpenDDS::DCPS::TcpConnection::handle_close(), OpenDDS::DCPS::TcpConnection::notify_connection_lost(), and OpenDDS::DCPS::TcpConnection::send_strategy().

 {
   GuardType guard(strategy_lock_);
   return static_rchandle_cast<OpenDDS::DCPS::TcpSendStrategy>(send_strategy_);
 }

◆ set_release_pending()

void OpenDDS::DCPS::TcpDataLink::set_release_pending ( bool flag )

Set release pending flag.

Definition at line 377 of file TcpDataLink.cpp.

References release_is_pending_.

Referenced by OpenDDS::DCPS::TcpTransport::find_datalink_i(), and OpenDDS::DCPS::TcpTransport::release_datalink().

 {
   this->release_is_pending_ = flag;
 }

◆ stop_i()

void OpenDDS::DCPS::TcpDataLink::stop_i ( )

protectedvirtual

Called when the DataLink is self-releasing because all of its reservations have been released, or when the TransportImpl is handling a shutdown() call.

Called when the DataLink has been "stopped" for some reason. It could be called from the DataLink::transport_shutdown() method (when the TransportImpl is handling a shutdown() call). Or, it could be called from the DataLink::release_reservations() method, when it discovers that it has just released the last remaining reservations from the DataLink, and the DataLink is in the process of "releasing" itself.

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 56 of file TcpDataLink.cpp.

References connection_, and DBG_ENTRY_LVL.

 {
   DBG_ENTRY_LVL("TcpDataLink","stop_i",6);
 
   TcpConnection_rch connection(this->connection_.lock());
   if (connection) {
     // Tell the connection object to disconnect.
     connection->disconnect();
   }
 }

Member Data Documentation

◆ connection_

WeakRcHandle<TcpConnection> OpenDDS::DCPS::TcpDataLink::connection_

private

Definition at line 96 of file TcpDataLink.h.

Referenced by connect(), do_association_actions(), get_connection(), handle_send_request_ack(), pre_stop_i(), reconnect(), reuse_existing_connection(), and stop_i().

◆ graceful_disconnect_sent_

bool OpenDDS::DCPS::TcpDataLink::graceful_disconnect_sent_

private

Definition at line 97 of file TcpDataLink.h.

Referenced by pre_stop_i().

◆ pending_request_acks_

PendingRequestAcks OpenDDS::DCPS::TcpDataLink::pending_request_acks_

private

Definition at line 101 of file TcpDataLink.h.

Referenced by ack_received(), drop_pending_request_acks(), and handle_send_request_ack().

◆ pending_request_acks_lock_

ACE_SYNCH_MUTEX OpenDDS::DCPS::TcpDataLink::pending_request_acks_lock_

private

Definition at line 100 of file TcpDataLink.h.

Referenced by ack_received(), drop_pending_request_acks(), and handle_send_request_ack().

◆ release_is_pending_

AtomicBool OpenDDS::DCPS::TcpDataLink::release_is_pending_

private

Definition at line 98 of file TcpDataLink.h.

Referenced by is_release_pending(), and set_release_pending().

◆ remote_address_

ACE_INET_Addr OpenDDS::DCPS::TcpDataLink::remote_address_

private

Definition at line 95 of file TcpDataLink.h.

Referenced by remote_address().

◆ stopped_clients_

RepoIdSetType OpenDDS::DCPS::TcpDataLink::stopped_clients_

private

Definition at line 103 of file TcpDataLink.h.

Referenced by check_active_client(), client_stop(), make_reservation(), send_i(), and send_stop_i().

◆ stopped_clients_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::TcpDataLink::stopped_clients_mutex_

mutableprivate

Definition at line 104 of file TcpDataLink.h.

Referenced by check_active_client(), client_stop(), make_reservation(), send_i(), and send_stop_i().

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

Public Member Functions

Protected Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ TcpDataLink()

◆ ~TcpDataLink()

Member Function Documentation

◆ ack_received()

◆ check_active_client()

◆ client_stop()

◆ connect()

◆ do_association_actions()

◆ drop_pending_request_acks()

◆ get_connection()

◆ handle_send_request_ack()

◆ is_release_pending()

◆ make_reservation() [1/2]

◆ make_reservation() [2/2]

◆ OPENDDS_SET_CMP()

◆ OPENDDS_VECTOR()

◆ pre_stop_i()

◆ receive_strategy()

◆ reconnect()

◆ remote_address()

◆ request_ack_received()

◆ reuse_existing_connection()

◆ send_association_msg()

◆ send_graceful_disconnect_message()

◆ send_i()

◆ send_stop_i()

◆ send_strategy()

◆ set_release_pending()

◆ stop_i()

Member Data Documentation

◆ connection_

◆ graceful_disconnect_sent_

◆ pending_request_acks_

◆ pending_request_acks_lock_

◆ release_is_pending_

◆ remote_address_

◆ stopped_clients_

◆ stopped_clients_mutex_