#include <TcpTransport.h>

Inheritance diagram for OpenDDS::DCPS::TcpTransport:

Collaboration diagram for OpenDDS::DCPS::TcpTransport:

[legend]

Classes
class	Connector

Public Member Functions
	TcpTransport (const TcpInst_rch &inst)

virtual	~TcpTransport ()

int	fresh_link (TcpConnection_rch connection)

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

TcpInst_rch	config () const

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

bool	release_link_resources (DataLink *link)

TransportInst_rch	config () const

virtual void	register_for_reader (const GUID_t &, const GUID_t &, const GUID_t &, const TransportLocatorSeq &, OpenDDS::DCPS::DiscoveryListener *)

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

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

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

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

virtual void	get_last_recv_locator (const GUID_t &, TransportLocator &)

virtual void	rtps_relay_address_change ()

virtual void	append_transport_statistics (TransportStatisticsSequence &)

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

ACE_Reactor *	reactor () const

ACE_thread_t	reactor_owner () const

bool	is_shut_down () const

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

void	dump ()
	Diagnostic aid. More...

OPENDDS_STRING	dump_to_str ()

void	report ()

virtual WeakRcHandle< ICE::Endpoint >	get_ice_endpoint ()

virtual void	rtps_relay_only_now (bool)

virtual void	use_rtps_relay_now (bool)

virtual void	use_ice_now (bool)

ReactorTask_rch	reactor_task ()

EventDispatcher_rch	event_dispatcher ()

int	acquire ()

int	tryacquire ()

int	release ()

int	remove ()

bool	connection_info (TransportLocator &local_info, ConnectionInfoFlags flags) const

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

virtual void	_add_ref ()

virtual void	_remove_ref ()

long	ref_count () const

WeakObject *	_get_weak_object () const

Private Types
typedef ACE_Hash_Map_Manager_Ex< PriorityKey, TcpDataLink_rch, ACE_Hash< PriorityKey >, ACE_Equal_To< PriorityKey >, ACE_Null_Mutex >	AddrLinkMap
	Map Type: (key) PriorityKey to (value) TcpDataLink_rch. More...

typedef ACE_SYNCH_MUTEX	LockType

typedef ACE_Guard< LockType >	GuardType

typedef ConditionVariable< LockType >	ConditionVariableType

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

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

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

virtual bool	configure_i (const TcpInst_rch &config)

virtual void	client_stop (const GUID_t &local_id)

virtual void	shutdown_i ()

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

virtual void	release_datalink (DataLink *link)
	Called by the DataLink to release itself. More...

virtual std::string	transport_type () const

void	async_connect_failed (const PriorityKey &key)

void	passive_connection (const ACE_INET_Addr &remote_address, const TcpConnection_rch &connection)

bool	find_datalink_i (const PriorityKey &key, TcpDataLink_rch &link)

int	connect_tcp_datalink (TcpDataLink &link, const TcpConnection_rch &connection)
	Common code used by accept_datalink(), passive_connection(), and active completion. More...

PriorityKey	blob_to_key (const TransportBLOB &remote, Priority priority, bool active)

typedef	OPENDDS_MAP (PriorityKey, TcpDataLink_rch) LinkMap

typedef	OPENDDS_MAP (PriorityKey, TcpConnection_rch) ConnectionMap

Private Attributes
unique_ptr< TcpAcceptor >	acceptor_
	Used to accept passive connections on our local_address_. More...

Connector	connector_
	Open TcpConnections using non-blocking connect. More...

AddrLinkMap	links_
	This is the map of connected DataLinks. More...

AddrLinkMap	pending_release_links_

LockType	links_lock_
	This lock is used to protect the links_ data member. More...

ConnectionMap	connections_

LockType	connections_lock_

Atomic< size_t >	last_link_

Friends
class	TcpConnection

class	TcpDataLink

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

WeakRcHandle< TransportInst >	config_

ReactorTask_rch	reactor_task_

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

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

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

typedef ACE_Guard< LockType >	GuardType

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

bool	open ()

typedef	OPENDDS_MULTIMAP (TransportClient_wrch, DataLink_rch) PendConnMap

void	add_pending_connection (const TransportClient_rch &client, DataLink_rch link)

void	shutdown ()

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

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

PendConnMap	pending_connections_

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

Detailed Description

This class provides the "Tcp" transport specific implementation. It creates the acceptor for listening the incoming requests using TCP and maintains a collection of TCP specific connections/datalinks.

Notes about object ownership: 1) Own the datalink objects, passive connection objects, acceptor object and TcpConnectionReplaceTask object(used during reconnecting). 2) Reference to TransportReactorTask object owned by base class.

Definition at line 48 of file TcpTransport.h.

Member Typedef Documentation

◆ AddrLinkMap

typedef ACE_Hash_Map_Manager_Ex<PriorityKey, TcpDataLink_rch, ACE_Hash<PriorityKey>, ACE_Equal_To<PriorityKey>, ACE_Null_Mutex> OpenDDS::DCPS::TcpTransport::AddrLinkMap

private

Map Type: (key) PriorityKey to (value) TcpDataLink_rch.

Definition at line 121 of file TcpTransport.h.

◆ ConditionVariableType

typedef ConditionVariable<LockType> OpenDDS::DCPS::TcpTransport::ConditionVariableType

private

Definition at line 128 of file TcpTransport.h.

◆ GuardType

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

private

Definition at line 127 of file TcpTransport.h.

◆ LockType

typedef ACE_SYNCH_MUTEX OpenDDS::DCPS::TcpTransport::LockType

private

Definition at line 126 of file TcpTransport.h.

Constructor & Destructor Documentation

◆ TcpTransport()

OpenDDS::DCPS::TcpTransport::TcpTransport ( const TcpInst_rch & inst )

explicit

Definition at line 36 of file TcpTransport.cpp.

References configure_i(), DBG_ENTRY_LVL, OpenDDS::DCPS::TransportImpl::open(), and OpenDDS::DCPS::TransportImpl::shutdown().

   : TransportImpl(inst)
   , acceptor_(new TcpAcceptor(RcHandle<TcpTransport>(this, inc_count())))
   , last_link_(0)
 {
   DBG_ENTRY_LVL("TcpTransport","TcpTransport",6);
 
   if (!(configure_i(inst) && open())) {
     this->shutdown();
     throw Transport::UnableToCreate();
   }
 
 }

◆ ~TcpTransport()

OpenDDS::DCPS::TcpTransport::~TcpTransport ( )

virtual

Definition at line 50 of file TcpTransport.cpp.

References DBG_ENTRY_LVL.

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

Member Function Documentation

◆ accept_datalink()

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

privatevirtual

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

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 219 of file TcpTransport.cpp.

References ACE_ERROR, OpenDDS::DCPS::TransportImpl::AcceptConnectResult::ACR_SUCCESS, OpenDDS::DCPS::DataLink::add_on_start_callback(), OpenDDS::DCPS::TransportImpl::add_pending_connection(), OpenDDS::DCPS::PriorityKey::address(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::bind(), OpenDDS::DCPS::TransportImpl::RemoteTransport::blob_, blob_to_key(), OpenDDS::DCPS::LogGuid::c_str(), connect_tcp_datalink(), connections_, connections_lock_, DBG_ENTRY_LVL, OpenDDS::DCPS::TcpDataLink::do_association_actions(), find_datalink_i(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::PriorityKey::is_active(), OpenDDS::DCPS::PriorityKey::is_loopback(), OpenDDS::DCPS::RcHandle< T >::is_nil(), OpenDDS::DCPS::TransportImpl::is_shut_down(), links_, links_lock_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::TransportImpl::ConnectionAttribs::local_id_, OpenDDS::DCPS::PriorityKey::priority(), OpenDDS::DCPS::TransportImpl::ConnectionAttribs::priority_, OpenDDS::DCPS::rchandle_from(), ACE_Guard< ACE_LOCK >::release(), OpenDDS::DCPS::TransportImpl::RemoteTransport::repo_id_, OpenDDS::DCPS::RcHandle< T >::reset(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), and VDBG_LVL.

 {
   DBG_ENTRY_LVL("TcpTransport", "accept_datalink", 6);
 
   if (is_shut_down()) {
     return AcceptConnectResult();
   }
 
 
   VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink local %C "
             "accepting connection from remote %C\n",
             LogGuid(attribs.local_id_).c_str(),
             LogGuid(remote.repo_id_).c_str()), 5);
 
   const PriorityKey key =
     blob_to_key(remote.blob_, attribs.priority_, false /* !active */);
 
   VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink PriorityKey "
             "prio=%d, addr=%C, is_loopback=%d, is_active=%d\n", attribs.priority_,
             LogAddr(key.address()).c_str(), key.is_loopback(), key.is_active()), 2);
 
   TcpDataLink_rch link;
   {
     GuardType guard(links_lock_);
 
     if (find_datalink_i(key, link)) {
       VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink found datalink link[%@]\n", link.in()), 0);
       link->add_on_start_callback(client, remote.repo_id_);
       add_pending_connection(client, link);
       guard.release();
       link->do_association_actions();
       return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
     }
 
     link = make_rch<TcpDataLink>(rchandle_from(this), key.address(), key.priority(),
                                  key.is_loopback(), key.is_active());
     VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink create new link[%@]\n", link.in()), 0);
     if (links_.bind(key, link) != 0 /*OK*/) {
       ACE_ERROR((LM_ERROR,
                  "(%P|%t) ERROR: TcpTransport::accept_datalink "
                  "Unable to bind new TcpDataLink[%@] to "
                  "TcpTransport in links_ map.\n", link.in()));
       return AcceptConnectResult();
     }
 
     link->add_on_start_callback(client, remote.repo_id_);
     add_pending_connection(client, link);
   }
 
   TcpConnection_rch connection;
   {
     GuardType guard(connections_lock_);
     const ConnectionMap::iterator iter = connections_.find(key);
 
     if (iter != connections_.end()) {
       connection = iter->second;
       connections_.erase(iter);
     }
   }
 
   if (connection.is_nil()) {
     return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
   }
 
   if (connect_tcp_datalink(*link, connection) == -1) {
     GuardType guard(links_lock_);
     links_.unbind(key);
     link.reset();
   }
 
   VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink "
             "connected link %@.\n", link.in()), 2);
   return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
 }

◆ async_connect_failed()

void OpenDDS::DCPS::TcpTransport::async_connect_failed ( const PriorityKey & key )

private

Definition at line 171 of file TcpTransport.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DCPS_debug_level, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::find(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::DataLink::invoke_on_start_callbacks(), links_, links_lock_, LM_WARNING, ACE_Guard< ACE_LOCK >::release(), and ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind().

Referenced by OpenDDS::DCPS::TcpConnection::active_open(), and OpenDDS::DCPS::TcpConnection::handle_timeout().

 {
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_WARNING, "(%P|%t) WARNING: Failed to make active connection.\n"));
   }
   GuardType guard(links_lock_);
   TcpDataLink_rch link;
   links_.find(key, link);
   links_.unbind(key);
   guard.release();
 
   if (link.in()) {
     link->invoke_on_start_callbacks(false);
   }
 }

◆ blob_to_key()

PriorityKey OpenDDS::DCPS::TcpTransport::blob_to_key	(	const TransportBLOB &	remote,
		Priority	priority,
		bool	active
	)

private

Definition at line 63 of file TcpTransport.cpp.

References config(), and OpenDDS::DCPS::AssociationData::get_remote_address().

Referenced by accept_datalink(), and connect_datalink().

 {
   const ACE_INET_Addr remote_address = AssociationData::get_remote_address(remote);
   TcpInst_rch cfg = config();
   const bool is_loopback = cfg && (remote_address == cfg->local_address());
   return PriorityKey(priority, remote_address, is_loopback, active);
 }

◆ client_stop()

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

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 390 of file TcpTransport.cpp.

References ACE_Hash_Map_Entry< EXT_ID, INT_ID >::int_id_, links_, links_lock_, ACE_Hash_Map_Iterator_Base_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::next(), and pending_release_links_.

 {
   GuardType guard(links_lock_);
 
   AddrLinkMap::ENTRY* entry;
 
   for (AddrLinkMap::ITERATOR itr(links_); itr.next(entry); itr.advance()) {
     entry->int_id_->client_stop(local_id);
   }
 
   for (AddrLinkMap::ITERATOR itr(pending_release_links_); itr.next(entry); itr.advance()) {
     entry->int_id_->client_stop(local_id);
   }
 }

◆ config()

TcpInst_rch OpenDDS::DCPS::TcpTransport::config ( ) const

Definition at line 57 of file TcpTransport.cpp.

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

Referenced by blob_to_key(), connect_datalink(), connect_tcp_datalink(), connection_info_i(), fresh_link(), and passive_connection().

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

◆ configure_i()

bool OpenDDS::DCPS::TcpTransport::configure_i ( const TcpInst_rch & config )

privatevirtual

Definition at line 319 of file TcpTransport.cpp.

References acceptor_, ACE_ERROR, ACE_ERROR_RETURN, ACE_TEXT(), connector_, OpenDDS::DCPS::TransportImpl::create_reactor_task(), DBG_ENTRY_LVL, OpenDDS::DCPS::get_fully_qualified_hostname(), ACE_INET_Addr::get_port_number(), OpenDDS::DCPS::ReactorTask::get_reactor(), hostname(), LM_DEBUG, LM_ERROR, ACE_Connector< class, class >::open(), OpenDDS::DCPS::TransportImpl::reactor_task(), TheServiceParticipant, and VDBG_LVL.

Referenced by TcpTransport().

 {
   DBG_ENTRY_LVL("TcpTransport", "configure_i", 6);
 
   if (!config) {
     return false;
   }
 
   this->create_reactor_task(false, "TcpTransport" + config->name());
 
   connector_.open(reactor_task()->get_reactor());
 
   // Override with DCPSDefaultAddress.
   if (config->local_address() == ACE_INET_Addr() &&
       TheServiceParticipant->default_address().to_addr() != ACE_INET_Addr()) {
     VDBG_LVL((LM_DEBUG,
               ACE_TEXT("(%P|%t) TcpTransport::configure_i overriding with DCPSDefaultAddress\n")), 2);
 
     config->local_address(TheServiceParticipant->default_address().to_addr());
   }
 
   VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::configure_i opening acceptor for %C on %C\n"),
             config->local_address_string().c_str(), LogAddr(config->local_address()).c_str()), 2);
 
   // Open our acceptor object so that we can accept passive connections
   // on our config->local_address_.
   if (this->acceptor_->open(config->local_address(),
                             this->reactor_task()->get_reactor()) != 0) {
     ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: Acceptor failed to open %C: %p\n"),
                       LogAddr(config->local_address()).c_str(), ACE_TEXT("open")), false);
   }
 
   // update the port number (incase port zero was given).
   ACE_INET_Addr address;
 
   if (this->acceptor_->acceptor().get_local_addr(address) != 0) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TcpTransport::configure_i ")
                ACE_TEXT("- %p"),
                ACE_TEXT("cannot get local addr\n")));
   }
 
   VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::configure_i listening on %C\n"),
             LogAddr(address).c_str()), 2);
 
   const unsigned short port = address.get_port_number();
 
   // As default, the acceptor will be listening on INADDR_ANY but advertise with the fully
   // qualified hostname and actual listening port number.
   if (config->local_address().is_any()) {
     const std::string hostname = get_fully_qualified_hostname();
     config->local_address(port, hostname.c_str());
     if (config->local_address() == ACE_INET_Addr()) {
        ACE_ERROR_RETURN((LM_ERROR,
                           ACE_TEXT("(%P|%t) ERROR: Failed to resolve a local address using fully qualified hostname '%C'\n"),
                           hostname.c_str()),
                           false);
     }
   }
 
   // Now we got the actual listening port. Update the port number in the configuration
   // if it's 0 originally.
   else if (config->local_address().get_port_number() == 0) {
     config->local_address_set_port(port);
   }
 
   // Ahhh...  The sweet smell of success!
   return true;
 }

◆ connect_datalink()

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

privatevirtual

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

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 74 of file TcpTransport.cpp.

References ACE_ERROR, OpenDDS::DCPS::TransportImpl::AcceptConnectResult::ACR_SUCCESS, OpenDDS::DCPS::DataLink::add_on_start_callback(), OpenDDS::DCPS::TransportImpl::add_pending_connection(), OpenDDS::DCPS::PriorityKey::address(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::bind(), OpenDDS::DCPS::TransportImpl::RemoteTransport::blob_, blob_to_key(), config(), ACE_Connector< class, class >::connect(), connector_, DBG_ENTRY_LVL, OpenDDS::DCPS::TcpDataLink::do_association_actions(), EWOULDBLOCK, find_datalink_i(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::DataLink::invoke_on_start_callbacks(), OpenDDS::DCPS::PriorityKey::is_active(), OpenDDS::DCPS::PriorityKey::is_loopback(), OpenDDS::DCPS::TransportImpl::is_shut_down(), links_, links_lock_, LM_DEBUG, LM_ERROR, ACE_Time_Value::msec(), OpenDDS::DCPS::PriorityKey::priority(), OpenDDS::DCPS::TransportImpl::ConnectionAttribs::priority_, OpenDDS::DCPS::rchandle_from(), OpenDDS::DCPS::TransportImpl::RemoteTransport::repo_id_, OpenDDS::DCPS::DataLink::transport_priority(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), ACE_Synch_Options::USE_REACTOR, ACE_Synch_Options::USE_TIMEOUT, and VDBG_LVL.

 {
   DBG_ENTRY_LVL("TcpTransport", "connect_datalink", 6);
 
   if (is_shut_down()) {
     return AcceptConnectResult();
   }
 
   const PriorityKey key =
     blob_to_key(remote.blob_, attribs.priority_, true /*active*/);
 
   VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink PriorityKey "
             "prio=%d, addr=%C, is_loopback=%d, is_active=%d\n",
             key.priority(), LogAddr(key.address()).c_str(), key.is_loopback(),
             key.is_active()), 0);
 
   TcpDataLink_rch link;
   {
     GuardType guard(links_lock_);
 
     if (find_datalink_i(key, link)) {
       VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink found datalink link[%@]\n", link.in()), 0);
       link->add_on_start_callback(client, remote.repo_id_);
       add_pending_connection(client, link);
       link->do_association_actions();
       return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
     }
 
     link = make_rch<TcpDataLink>(rchandle_from(this), key.address(), attribs.priority_,
                                 key.is_loopback(), true /*active*/);
     VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink create new link[%@]\n", link.in()), 0);
     if (links_.bind(key, link) != 0 /*OK*/) {
       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: TcpTransport::connect_datalink "
                  "Unable to bind new TcpDataLink[%@] to "
                  "TcpTransport in links_ map.\n", link.in()));
       return AcceptConnectResult();
     }
 
     link->add_on_start_callback(client, remote.repo_id_);
     add_pending_connection(client, link);
   }
 
   int ret = -1; // Default to failure in case config() is returns an invalid RCH
   errno = EINVAL; // Anything other than EWOULDBLOCK
 
   TcpInst_rch cfg = config();
   if (cfg) {
     TcpConnection_rch connection(make_rch<TcpConnection>(key.address(), link->transport_priority(), cfg));
     connection->set_datalink(link);
 
     TcpConnection* pConn = connection.in();
 
     // Can't make this call while holding onto TransportClient::lock_
     ACE_Time_Value conn_timeout;
     conn_timeout.msec(cfg->active_conn_timeout_period_);
 
     ret = connector_.connect(pConn, key.address(), ACE_Synch_Options(ACE_Synch_Options::USE_REACTOR|ACE_Synch_Options::USE_TIMEOUT, conn_timeout));
   }
 
   if (ret == -1 && errno != EWOULDBLOCK) {
 
     VDBG_LVL((LM_ERROR, "(%P|%t) TcpTransport::connect_datalink error %m.\n"), 2);
     ACE_ERROR((LM_ERROR, "(%P|%t) TcpTransport::connect_datalink error %m.\n"));
     //If the connection fails and, in the interim between releasing
     //lock and re-acquiring to remove the failed link, another association may have found
     //the datalink in links_ (always using find_datalink_i) so must allow the other
     //association to either try to connect again (might succeed for it)
     //or try another transport.  If find_datalink_i was called for this datalink, an
     //on_start_callback will be registered and can be invoked.
     VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink connect failed, remove link[%@]\n", link.in()), 0);
     {
       GuardType guard(links_lock_);
       if (links_.unbind(key, link) != 0 /*OK*/) {
         ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: TcpTransport::connect_datalink "
                    "Unable to unbind failed TcpDataLink[%@] from "
                    "TcpTransport links_ map.\n", link.in()));
       }
     }
     link->invoke_on_start_callbacks(false);
 
     return AcceptConnectResult();
   }
 
   if (ret == 0) {
     // connect() completed synchronously and called TcpConnection::active_open().
     VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink "
               "completed synchronously.\n"), 0);
     return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
   }
 
   VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink pending.\n"), 0);
   return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
 }

◆ connect_tcp_datalink()

int OpenDDS::DCPS::TcpTransport::connect_tcp_datalink	(	TcpDataLink &	link,
		const TcpConnection_rch &	connection
	)

private

Common code used by accept_datalink(), passive_connection(), and active completion.

Code common to make_active_connection() and make_passive_connection().

Definition at line 674 of file TcpTransport.cpp.

References ACE_DEBUG, ACE_TEXT(), config(), OpenDDS::DCPS::TcpDataLink::connect(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, last_link_, LM_DEBUG, OpenDDS::DCPS::Atomic< T >::load(), OpenDDS::DCPS::TransportImpl::reactor_task(), OpenDDS::DCPS::ref(), OpenDDS::DCPS::TcpDataLink::reuse_existing_connection(), and OpenDDS::DCPS::DataLink::transport_priority().

Referenced by accept_datalink(), OpenDDS::DCPS::TcpConnection::active_open(), passive_connection(), and OpenDDS::DCPS::TcpDataLink::reconnect().

 {
   DBG_ENTRY_LVL("TcpTransport", "connect_tcp_datalink", 6);
 
   if (link.reuse_existing_connection(connection) == 0) {
     return 0;
   }
 
   TcpInst_rch cfg = config();
   if (!cfg) {
     return -1;
   }
 
   ++last_link_;
 
   if (DCPS_debug_level > 4) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) TcpTransport::connect_tcp_datalink() [%d] - ")
                ACE_TEXT("creating send strategy with priority %d.\n"),
                last_link_.load(), link.transport_priority()));
   }
 
   connection->id() = last_link_;
 
   TcpSendStrategy_rch send_strategy (
     make_rch<TcpSendStrategy>(last_link_.load(), ref(link),
                              new TcpSynchResource(link,
                                                   cfg->max_output_pause_period_),
                              this->reactor_task(), link.transport_priority()));
 
   TcpReceiveStrategy_rch receive_strategy(
     make_rch<TcpReceiveStrategy>(ref(link), this->reactor_task()));
 
   if (link.connect(connection, send_strategy, receive_strategy) != 0) {
     return -1;
   }
 
   return 0;
 }

◆ connection_info_i()

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

privatevirtual

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

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 470 of file TcpTransport.cpp.

References config(), DBG_ENTRY_LVL, LM_DEBUG, and VDBG_LVL.

 {
   DBG_ENTRY_LVL("TcpTransport", "connection_info_i", 6);
 
   TcpInst_rch cfg = config();
   if (cfg) {
     VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport public address string <%C>\n",
               cfg->get_public_address().c_str()), 2);
 
     cfg->populate_locator(local_info, flags);
     return true;
   }
 
   return false;
 }

◆ find_datalink_i()

bool OpenDDS::DCPS::TcpTransport::find_datalink_i	(	const PriorityKey &	key,
		TcpDataLink_rch &	link
	)

private

Definition at line 189 of file TcpTransport.cpp.

References ACE_TEXT(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::bind(), OpenDDS::DCPS::DataLink::cancel_release(), DBG_ENTRY_LVL, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::find(), OpenDDS::DCPS::RcHandle< T >::in(), links_, LM_DEBUG, pending_release_links_, OpenDDS::DCPS::RcHandle< T >::reset(), OpenDDS::DCPS::TcpDataLink::set_release_pending(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), and VDBG_LVL.

Referenced by accept_datalink(), and connect_datalink().

 {
   DBG_ENTRY_LVL("TcpTransport", "find_datalink_i", 6);
 
   if (links_.find(key, link) == 0 /*OK*/) {
     return true;
   } else if (pending_release_links_.find(key, link) == 0 /*OK*/) {
     if (link->cancel_release()) {
       link->set_release_pending(false);
 
       if (pending_release_links_.unbind(key, link) == 0 /*OK*/
           && links_.bind(key, link) == 0 /*OK*/) {
         VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::find_datalink_i ")
                   ACE_TEXT("found link[%@] in pending release list, cancelled release and moved back to links_.\n"), link.in()), 0);
         return true;
       }
       VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::find_datalink_i ")
                 ACE_TEXT("found link[%@] in pending release list but was unable to shift back to links_.\n"), link.in()), 0);
     } else {
       VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::find_datalink_i ")
                 ACE_TEXT("found link[%@] in pending release list but was unable to cancel release.\n"), link.in()), 0);
     }
     link.reset(); // don't return link to TransportClient
     return false;
   }
 
   return false;
 }

◆ fresh_link()

int OpenDDS::DCPS::TcpTransport::fresh_link ( TcpConnection_rch connection )

This function is called by the TcpReconnectTask thread to check if the passively accepted connection is the re-established connection. If it is, then the "old" connection object in the datalink is replaced by the "new" connection object.

Definition at line 719 of file TcpTransport.cpp.

References config(), DBG_ENTRY_LVL, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::find(), OpenDDS::DCPS::TcpDataLink::get_connection(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::RcHandle< T >::is_nil(), OpenDDS::DCPS::TransportImpl::is_shut_down(), links_, links_lock_, and OpenDDS::DCPS::TcpDataLink::reconnect().

Referenced by passive_connection().

 {
   DBG_ENTRY_LVL("TcpTransport","fresh_link",6);
 
   TcpDataLink_rch link;
   GuardType guard(this->links_lock_);
 
   if (is_shut_down()) {
     return 0;
   }
 
   TcpInst_rch cfg = config();
   if (!cfg) {
     return -1;
   }
 
   PriorityKey key(connection->transport_priority(),
                   connection->get_remote_address(),
                   connection->get_remote_address() == cfg->local_address(),
                   connection->is_connector());
 
   if (this->links_.find(key, link) == 0) {
     TcpConnection_rch old_con = link->get_connection();
 
     // The connection is accepted but may not be associated with the datalink
     // at this point. The thread calling add_associations() will associate
     // the datalink with the connection in make_passive_connection().
     if (old_con.is_nil()) {
       return 0;
     }
 
     if (old_con.in() != connection.in())
       // Replace the "old" connection object with the "new" connection object.
     {
       return link->reconnect(connection);
     }
   }
 
   return 0;
 }

◆ OPENDDS_MAP() [1/2]

typedef OpenDDS::DCPS::TcpTransport::OPENDDS_MAP	(	PriorityKey	,
		TcpDataLink_rch
	)

private

◆ OPENDDS_MAP() [2/2]

typedef OpenDDS::DCPS::TcpTransport::OPENDDS_MAP	(	PriorityKey	,
		TcpConnection_rch
	)

private

◆ passive_connection()

void OpenDDS::DCPS::TcpTransport::passive_connection	(	const ACE_INET_Addr &	remote_address,
		const TcpConnection_rch &	connection
	)

private

Called by the TcpConnection object when it has been created by the acceptor and needs to be attached to a DataLink. The DataLink may or may not already be created and waiting for this passive connection to appear.

This method is called by a TcpConnection object that has been created and opened by our acceptor_ as a result of passively accepting a connection on our local address. Ultimately, the connection object needs to be paired with a DataLink object that is (or will be) expecting this passive connection to be established.

Definition at line 600 of file TcpTransport.cpp.

References ACE_ERROR, ACE_TEXT(), config(), connect_tcp_datalink(), connections_, connections_lock_, DBG_ENTRY_LVL, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::find(), fresh_link(), OpenDDS::DCPS::RcHandle< T >::is_nil(), OpenDDS::DCPS::TransportImpl::is_shut_down(), links_, links_lock_, LM_DEBUG, LM_ERROR, ACE_Guard< ACE_LOCK >::release(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), and VDBG_LVL.

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

 {
   DBG_ENTRY_LVL("TcpTransport", "passive_connection", 6);
 
   if (is_shut_down()) {
     return;
   }
 
   TcpInst_rch cfg = config();
   if (!cfg) {
     return;
   }
 
   const PriorityKey key(connection->transport_priority(),
                         remote_address,
                         remote_address == cfg->local_address(),
                         connection->is_connector());
 
   VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::passive_connection() - ")
             ACE_TEXT("established with %C.\n"),
             LogAddr(remote_address).c_str()), 2);
 
   GuardType connection_guard(connections_lock_);
   TcpDataLink_rch link;
   {
     GuardType guard(links_lock_);
     links_.find(key, link);
   }
 
   if (!link.is_nil()) {
     connection_guard.release();
 
     if (connect_tcp_datalink(*link, connection) == -1) {
       VDBG_LVL((LM_ERROR,
                 ACE_TEXT("(%P|%t) TcpTransport::passive_connection() - ")
                 ACE_TEXT("ERROR: connect_tcp_datalink failed\n")), 5);
       GuardType guard(links_lock_);
       links_.unbind(key);
 
     } else {
       this->fresh_link(connection);
     }
 
     return;
   }
 
   // If we reach this point, this link was not in links_, so the
   // accept_datalink() call hasn't happened yet.  Store in connections_ for the
   // accept_datalink() method to find.
   VDBG_LVL((LM_DEBUG,
             ACE_TEXT("(%P|%t) TcpTransport::passive_connection() - # of before connections: %d\n"),
             connections_.size()), 5);
   const ConnectionMap::iterator where = connections_.find(key);
 
   if (where != connections_.end()) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) TcpTransport::passive_connection() - ")
                ACE_TEXT("ERROR: connection with %C at priority %d already exists, ")
                ACE_TEXT("overwriting previously established connection.\n"),
                LogAddr(remote_address).c_str(),
                connection->transport_priority()));
   }
 
   connections_[key] = connection;
   VDBG_LVL((LM_DEBUG,
             ACE_TEXT("(%P|%t) TcpTransport::passive_connection() - # of after connections: %d\n"),
             connections_.size()), 5);
 
   this->fresh_link(connection);
 }

◆ release_datalink()

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

privatevirtual

Called by the DataLink to release itself.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 487 of file TcpTransport.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::bind(), OpenDDS::DCPS::DataLink::datalink_release_delay(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::DataLink::is_active(), OpenDDS::DCPS::DataLink::is_loopback(), links_, links_lock_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), pending_release_links_, OpenDDS::DCPS::TcpDataLink::remote_address(), OpenDDS::DCPS::DataLink::schedule_delayed_release(), OpenDDS::DCPS::DataLink::schedule_stop(), OpenDDS::DCPS::TcpDataLink::set_release_pending(), OpenDDS::DCPS::DataLink::set_scheduling_release(), OpenDDS::DCPS::TimeDuration::str(), OpenDDS::DCPS::DataLink::transport_priority(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), VDBG_LVL, and OpenDDS::DCPS::TimeDuration::zero_value.

 {
   DBG_ENTRY_LVL("TcpTransport", "release_datalink", 6);
 
   TcpDataLink* tcp_link = static_cast<TcpDataLink*>(link);
 
   if (tcp_link == 0) {
     // Really an assertion failure
     ACE_ERROR((LM_ERROR,
                "(%P|%t) INTERNAL ERROR - Failed to downcast DataLink to "
                "TcpDataLink.\n"));
     return;
   }
 
   TcpDataLink_rch released_link;
 
   // Possible actions that will be taken to release the link.
   enum LinkAction { None, StopLink, ScheduleLinkRelease };
   LinkAction linkAction = None;
 
   // Scope for locking to protect the links (and pending_release) containers.
   GuardType guard(this->links_lock_);
 
   // Attempt to remove the TcpDataLink from our links_ map.
   PriorityKey key(
     tcp_link->transport_priority(),
     tcp_link->remote_address(),
     tcp_link->is_loopback(),
     tcp_link->is_active());
 
   VDBG_LVL((LM_DEBUG,
             "(%P|%t) TcpTransport::release_datalink link[%@] PriorityKey "
             "prio=%d, addr=%C, is_loopback=%d, is_active=%d\n",
             link,
             tcp_link->transport_priority(),
             LogAddr(tcp_link->remote_address()).c_str(),
             (int)tcp_link->is_loopback(),
             (int)tcp_link->is_active()), 2);
 
   if (this->links_.unbind(key, released_link) != 0) {
     //No op
   } else if (link->datalink_release_delay() > TimeDuration::zero_value) {
     link->set_scheduling_release(true);
 
     VDBG_LVL((LM_DEBUG,
               "(%P|%t) TcpTransport::release_datalink datalink_release_delay "
               "is %C\n",
               link->datalink_release_delay().str().c_str()), 4);
 
     // Atomic value update, safe to perform here.
     released_link->set_release_pending(true);
 
     switch (this->pending_release_links_.bind(key, released_link)) {
     case -1:
       ACE_ERROR((LM_ERROR,
                  "(%P|%t) ERROR: Unable to bind released TcpDataLink[%@] to "
                  "pending_release_links_ map: %p\n", released_link.in(), ACE_TEXT("bind")));
       linkAction = StopLink;
       break;
 
     case 1:
       ACE_ERROR((LM_ERROR,
                  "(%P|%t) ERROR: Unable to bind released TcpDataLink[%@] to "
                  "pending_release_links_ map: already bound\n", released_link.in()));
       linkAction = StopLink;
       break;
 
     case 0:
       linkAction = ScheduleLinkRelease;
       break;
 
     default:
       break;
     }
 
   } else { // datalink_release_delay_ is 0
     link->set_scheduling_release(true);
 
     linkAction = StopLink;
   }
 
   // Actions are executed outside of the lock scope.
   switch (linkAction) {
   case StopLink:
     link->schedule_stop(MonotonicTimePoint::now());
     break;
 
   case ScheduleLinkRelease:
     link->schedule_delayed_release();
     break;
 
   case None:
     break;
   }
 
   if (DCPS_debug_level > 9) {
     std::stringstream buffer;
     buffer << *link;
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) TcpTransport::release_datalink() - ")
                ACE_TEXT("link[%@] with priority %d released.\n%C"),
                link,
                link->transport_priority(),
                buffer.str().c_str()));
   }
 }

◆ shutdown_i()

void OpenDDS::DCPS::TcpTransport::shutdown_i ( )

privatevirtual

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

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 406 of file TcpTransport.cpp.

References acceptor_, connections_, connections_lock_, DBG_ENTRY_LVL, ACE_Hash_Map_Entry< EXT_ID, INT_ID >::int_id_, links_, links_lock_, ACE_Hash_Map_Iterator_Base_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::next(), OpenDDS::DCPS::TransportImpl::pending_connections_, OpenDDS::DCPS::TransportImpl::pending_connections_lock_, pending_release_links_, and ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind_all().

 {
   DBG_ENTRY_LVL("TcpTransport","shutdown_i",6);
 
   {
     GuardType guard(links_lock_);
 
     AddrLinkMap::ENTRY* entry;
 
     for (AddrLinkMap::ITERATOR itr(links_);
          itr.next(entry);
          itr.advance()) {
       entry->int_id_->pre_stop_i();
     }
   }
 
   // Don't accept any more connections.
   acceptor_->close();
   acceptor_->transport_shutdown();
 
   {
     {
       GuardType guard(connections_lock_);
 
       for (ConnectionMap::iterator it = connections_.begin(); it != connections_.end(); ++it) {
         it->second->shutdown();
       }
       connections_.clear();
     }
     {
       GuardType guard(pending_connections_lock_);
       pending_connections_.clear();
     }
   }
 
   // Disconnect all of our DataLinks, and clear our links_ collection.
   {
     GuardType guard(links_lock_);
 
     AddrLinkMap::ENTRY* entry;
 
     for (AddrLinkMap::ITERATOR itr(links_);
          itr.next(entry);
          itr.advance()) {
       entry->int_id_->transport_shutdown();
     }
 
     links_.unbind_all();
 
     for (AddrLinkMap::ITERATOR itr(pending_release_links_);
          itr.next(entry);
          itr.advance()) {
       entry->int_id_->transport_shutdown();
     }
 
     pending_release_links_.unbind_all();
   }
 
   // Tell our acceptor about this event so that it can drop its reference
   // it holds to this TcpTransport object (via smart-pointer).
   acceptor_->transport_shutdown();
 }

◆ stop_accepting_or_connecting()

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

privatevirtual

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

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 297 of file TcpTransport.cpp.

References LM_DEBUG, OpenDDS::DCPS::TransportImpl::pending_connections_, OpenDDS::DCPS::TransportImpl::pending_connections_lock_, and VDBG_LVL.

 {
   VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::stop_accepting_or_connecting "
             "stop connecting to remote: %C\n",
             LogGuid(remote_id).c_str()), 5);
 
   GuardType guard(pending_connections_lock_);
   typedef PendConnMap::iterator iter_t;
   const std::pair<iter_t, iter_t> range =
     pending_connections_.equal_range(client);
 
   for (iter_t iter = range.first; iter != range.second; ++iter) {
     iter->second->remove_on_start_callback(client, remote_id);
   }
 
   pending_connections_.erase(range.first, range.second);
 }

◆ transport_type()

virtual std::string OpenDDS::DCPS::TcpTransport::transport_type ( ) const

inlineprivatevirtual

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 86 of file TcpTransport.h.

86 { return "tcp"; }

◆ unbind_link()

void OpenDDS::DCPS::TcpTransport::unbind_link ( DataLink * link )

virtual

Remove any pending_release mappings.

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 761 of file TcpTransport.cpp.

References ACE_ERROR, OpenDDS::DCPS::DataLink::datalink_release_delay(), OpenDDS::DCPS::DataLink::is_active(), OpenDDS::DCPS::DataLink::is_loopback(), OpenDDS::DCPS::TimeDuration::is_zero(), links_lock_, LM_DEBUG, LM_ERROR, pending_release_links_, OpenDDS::DCPS::TcpDataLink::remote_address(), OpenDDS::DCPS::DataLink::transport_priority(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), and VDBG_LVL.

 {
   TcpDataLink* tcp_link = static_cast<TcpDataLink*>(link);
 
   if (tcp_link == 0) {
     // Really an assertion failure
     ACE_ERROR((LM_ERROR,
                "(%P|%t) TcpTransport::unbind_link INTERNAL ERROR - "
                "Failed to downcast DataLink to TcpDataLink.\n"));
     return;
   }
 
   // Attempt to remove the TcpDataLink from our links_ map.
   PriorityKey key(
     tcp_link->transport_priority(),
     tcp_link->remote_address(),
     tcp_link->is_loopback(),
     tcp_link->is_active());
 
   VDBG_LVL((LM_DEBUG,
             "(%P|%t) TcpTransport::unbind_link link %@ PriorityKey "
             "prio=%d, addr=%C, is_loopback=%d, is_active=%d\n",
             link,
             tcp_link->transport_priority(),
             LogAddr(tcp_link->remote_address()).c_str(),
             (int)tcp_link->is_loopback(),
             (int)tcp_link->is_active()), 2);
 
   GuardType guard(this->links_lock_);
 
   if (pending_release_links_.unbind(key) && !link->datalink_release_delay().is_zero()) {
     ACE_ERROR((LM_ERROR,
                "(%P|%t) TcpTransport::unbind_link INTERNAL ERROR - "
                "Failed to find link %@ tcp_link %@ PriorityKey "
                "prio=%d, addr=%C, is_loopback=%d, is_active=%d\n",
                link,
                tcp_link,
                tcp_link->transport_priority(),
                LogAddr(tcp_link->remote_address()).c_str(),
                (int)tcp_link->is_loopback(),
                (int)tcp_link->is_active()));
   }
 }

Friends And Related Function Documentation

◆ TcpConnection

friend class TcpConnection

friend

The TcpConnection is our friend. It tells us when it has been created (by our acceptor_), and is seeking the DataLink that should be (or will be) expecting the passive connection.

Definition at line 94 of file TcpTransport.h.

◆ TcpDataLink

friend class TcpDataLink

friend

Definition at line 95 of file TcpTransport.h.

Member Data Documentation

◆ acceptor_

unique_ptr<TcpAcceptor> OpenDDS::DCPS::TcpTransport::acceptor_

private

Used to accept passive connections on our local_address_.

Definition at line 144 of file TcpTransport.h.

Referenced by configure_i(), and shutdown_i().

◆ connections_

ConnectionMap OpenDDS::DCPS::TcpTransport::connections_

private

Map of passive connection objects that need to be paired with a DataLink.

Definition at line 163 of file TcpTransport.h.

Referenced by accept_datalink(), passive_connection(), and shutdown_i().

◆ connections_lock_

LockType OpenDDS::DCPS::TcpTransport::connections_lock_

private

This protects the connections_ and the pending_connections_ data members.

Definition at line 167 of file TcpTransport.h.

Referenced by accept_datalink(), passive_connection(), and shutdown_i().

◆ connector_

Connector OpenDDS::DCPS::TcpTransport::connector_

private

Open TcpConnections using non-blocking connect.

Definition at line 152 of file TcpTransport.h.

Referenced by OpenDDS::DCPS::TcpConnection::active_reconnect_i(), OpenDDS::DCPS::TcpConnection::close(), configure_i(), connect_datalink(), and OpenDDS::DCPS::TcpConnection::handle_timeout().

◆ last_link_

Atomic<size_t> OpenDDS::DCPS::TcpTransport::last_link_

private

Definition at line 168 of file TcpTransport.h.

Referenced by connect_tcp_datalink().

◆ links_

AddrLinkMap OpenDDS::DCPS::TcpTransport::links_

private

This is the map of connected DataLinks.

Definition at line 155 of file TcpTransport.h.

Referenced by accept_datalink(), async_connect_failed(), client_stop(), connect_datalink(), find_datalink_i(), fresh_link(), passive_connection(), release_datalink(), and shutdown_i().

◆ links_lock_

LockType OpenDDS::DCPS::TcpTransport::links_lock_

private

This lock is used to protect the links_ data member.

Definition at line 159 of file TcpTransport.h.

Referenced by accept_datalink(), async_connect_failed(), client_stop(), connect_datalink(), fresh_link(), passive_connection(), release_datalink(), shutdown_i(), and unbind_link().

◆ pending_release_links_

AddrLinkMap OpenDDS::DCPS::TcpTransport::pending_release_links_

private

Definition at line 156 of file TcpTransport.h.

Referenced by client_stop(), find_datalink_i(), release_datalink(), shutdown_i(), and unbind_link().

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

Classes

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ AddrLinkMap

◆ ConditionVariableType

◆ GuardType

◆ LockType

Constructor & Destructor Documentation

◆ TcpTransport()

◆ ~TcpTransport()

Member Function Documentation

◆ accept_datalink()

◆ async_connect_failed()

◆ blob_to_key()

◆ client_stop()

◆ config()

◆ configure_i()

◆ connect_datalink()

◆ connect_tcp_datalink()

◆ connection_info_i()

◆ find_datalink_i()

◆ fresh_link()

◆ OPENDDS_MAP() [1/2]

◆ OPENDDS_MAP() [2/2]

◆ passive_connection()

◆ release_datalink()

◆ shutdown_i()

◆ stop_accepting_or_connecting()

◆ transport_type()

◆ unbind_link()

Friends And Related Function Documentation

◆ TcpConnection

◆ TcpDataLink

Member Data Documentation

◆ acceptor_

◆ connections_

◆ connections_lock_

◆ connector_

◆ last_link_

◆ links_

◆ links_lock_

◆ pending_release_links_