#include <TcpConnection.h>

Inheritance diagram for OpenDDS::DCPS::TcpConnection:

[legend]

Collaboration diagram for OpenDDS::DCPS::TcpConnection:

[legend]

Public Types
enum	ReconnectState { INIT_STATE, LOST_STATE, RECONNECTED_STATE, ACTIVE_RECONNECTING_STATE, ACTIVE_WAITING_STATE, PASSIVE_WAITING_STATE, PASSIVE_TIMEOUT_CALLED_STATE }
	States are used during reconnecting. More...

Public Types inherited from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >
typedef ACE_SOCK_STREAM ::PEER_ADDR	addr_type

typedef ACE_SOCK_STREAM	stream_type

Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

Public Member Functions
	TcpConnection ()
	Passive side constructor (acceptor) More...

	TcpConnection (const ACE_INET_Addr &remote_address, Priority priority, const TcpInst_rch &config)
	Active side constructor (connector) More...

virtual	~TcpConnection ()

std::size_t &	id ()

int	active_open ()

int	active_reconnect_open ()

int	passive_open (void *)

void	disconnect ()

virtual int	open (void *arg)

TcpSendStrategy_rch	send_strategy ()

TcpReceiveStrategy_rch	receive_strategy ()

virtual int	handle_input (ACE_HANDLE)
	We pass this "event" along to the receive_strategy. More...

virtual int	handle_output (ACE_HANDLE)
	Handle back pressure when sending. More...

virtual int	close (u_long)

virtual int	handle_close (ACE_HANDLE, ACE_Reactor_Mask)

void	set_sock_options (const TcpInst_rch &tcp_config)

bool	is_connector () const

void	transfer (TcpConnection *connection)

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

void	set_datalink (const TcpDataLink_rch &link)

void	notify_lost_on_backpressure_timeout ()

ACE_INET_Addr	get_remote_address ()

void	relink_from_send (bool do_suspend)
	Reconnect initiated by send strategy. More...

void	relink_from_recv (bool do_suspend)
	Reconnect initiated by receive strategy. More...

void	tear_link ()

void	shutdown ()

TcpTransport_rch	impl ()

Priority &	transport_priority ()
	Access TRANSPORT_PRIORITY.value policy value if set. More...

Priority	transport_priority () const

virtual ACE_Event_Handler::Reference_Count	add_reference ()

virtual ACE_Event_Handler::Reference_Count	remove_reference ()

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

virtual void	_add_ref ()

virtual void	_remove_ref ()

long	ref_count () const

WeakObject *	_get_weak_object () const

Public Member Functions inherited from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >
	ACE_Svc_Handler (ACE_Thread_Manager thr_mgr=0, ACE_Message_Queue< ACE_NULL_SYNCH > mq=0, ACE_Reactor *reactor=ACE_Reactor::instance())

virtual	~ACE_Svc_Handler (void)

virtual int	idle (u_long flags=0)

virtual ACE_Recyclable_State	recycle_state (void) const

virtual int	recycle_state (ACE_Recyclable_State new_state)

virtual void	cleanup_hint (void **act_holder=0)

virtual int	init (int argc, ACE_TCHAR *argv[])

virtual int	fini (void)

virtual int	info (ACE_TCHAR **info_string, size_t length) const

virtual ACE_HANDLE	get_handle (void) const

virtual void	set_handle (ACE_HANDLE)

ACE_SOCK_STREAM &	peer (void) const

void *	operator new (size_t n)

void *	operator new (size_t n, void *p)

virtual void	destroy (void)

void	operator delete (void *)

void	operator delete (void , void )

void	shutdown (void)

void	dump (void) const

virtual void	recycler (ACE_Connection_Recycling_Strategy recycler, const void recycling_act)

virtual ACE_Connection_Recycling_Strategy *	recycler (void) const

virtual const void *	recycling_act (void) const

virtual int	recycle (void *=0)

Public Member Functions inherited from ACE_Task< ACE_NULL_SYNCH >
	ACE_Task (ACE_Thread_Manager thr_mgr=0, ACE_Message_Queue< ACE_SYNCH_USE, ACE_System_Time_Policy > mq=0)

virtual	~ACE_Task (void)

ACE_Message_Queue< ACE_SYNCH_USE, ACE_System_Time_Policy > *	msg_queue (void)

void	msg_queue (ACE_Message_Queue< ACE_SYNCH_USE, ACE_System_Time_Policy > *)

ACE_Time_Value_T< ACE_System_Time_Policy >	gettimeofday (void) const

void	set_time_policy (ACE_System_Time_Policy const &time_policy)

int	putq (ACE_Message_Block , ACE_Time_Value timeout=0)

int	getq (ACE_Message_Block &mb, ACE_Time_Value timeout=0)

int	ungetq (ACE_Message_Block , ACE_Time_Value timeout=0)

int	reply (ACE_Message_Block mb, ACE_Time_Value tv=0)

int	put_next (ACE_Message_Block msg, ACE_Time_Value timeout=0)

const ACE_TCHAR *	name (void) const

ACE_Task< ACE_SYNCH_USE, ACE_System_Time_Policy > *	next (void)

void	next (ACE_Task< ACE_SYNCH_USE, ACE_System_Time_Policy > *)

ACE_Task< ACE_SYNCH_USE, ACE_System_Time_Policy > *	sibling (void)

ACE_Module< ACE_SYNCH_USE, ACE_System_Time_Policy > *	module (void) const

int	flush (u_long flag=ACE_Task_Flags::ACE_FLUSHALL)

void	water_marks (ACE_IO_Cntl_Msg::ACE_IO_Cntl_Cmds, size_t)

void	dump (void) const

Public Member Functions inherited from ACE_Task_Base
	ACE_Task_Base (ACE_Thread_Manager *=0)

virtual	~ACE_Task_Base (void)

virtual int	module_closed (void)

virtual int	put (ACE_Message_Block , ACE_Time_Value =0)

virtual int	svc (void)

virtual int	activate (long flags=THR_NEW_LWP\|THR_JOINABLE\|THR_INHERIT_SCHED, int n_threads=1, int force_active=0, long priority=ACE_DEFAULT_THREAD_PRIORITY, int grp_id=-1, ACE_Task_Base task=0, ACE_hthread_t thread_handles[]=0, void stack[]=0, size_t stack_size[]=0, ACE_thread_t thread_ids[]=0, const char *thr_name[]=0)

virtual int	wait (void)

virtual int	suspend (void)

virtual int	resume (void)

int	grp_id (void) const

void	grp_id (int)

ACE_Thread_Manager *	thr_mgr (void) const

void	thr_mgr (ACE_Thread_Manager *)

int	is_reader (void) const

int	is_writer (void) const

size_t	thr_count (void) const

ACE_thread_t	last_thread (void) const

Public Member Functions inherited from ACE_Service_Object
	ACE_Service_Object (ACE_Reactor *=0)

virtual	~ACE_Service_Object (void)

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

virtual int	priority (void) const

virtual void	priority (int priority)

virtual int	handle_exception (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 ACE_Shared_Object
	ACE_Shared_Object (void)

virtual	~ACE_Shared_Object (void)

Private Types
typedef ACE_SYNCH_MUTEX	LockType

typedef ACE_Guard< LockType >	GuardType

Private Member Functions
int	on_active_connection_established ()
	Handle the logic after an active connection has been established. More...

void	active_reconnect_i ()

void	passive_reconnect_i ()

void	notify_connection_lost ()

void	handle_stop_reconnecting ()

int	handle_setup_input (ACE_HANDLE h)

const std::string &	config_name () const

const char *	reconnect_state_string () const
	Get name of the current reconnect state as a string. More...

Private Attributes
LockType	reconnect_lock_
	Lock to synchronize state between reactor and non-reactor threads. More...

bool	is_connector_
	Flag indicate this connection object is the connector or acceptor. More...

ACE_INET_Addr	remote_address_
	Remote address. More...

ACE_INET_Addr	local_address_
	Local address. More...

WeakRcHandle< TcpInst >	tcp_config_
	The configuration used by this connection. More...

TcpDataLink_rch	link_
	Datalink object which is needed for connection lost callback. More...

TcpTransport_rch	impl_
	Impl object which is needed for connection objects and reconnect task. More...

ReconnectState	reconnect_state_
	The state indicates each step of the reconnecting. More...

Priority	transport_priority_
	TRANSPORT_PRIORITY.value policy value. More...

bool	shutdown_
	shutdown flag More...

bool	passive_setup_

ACE_Message_Block	passive_setup_buffer_

TcpTransport *	transport_during_setup_

std::size_t	id_
	Small unique identifying value. More...

int	conn_retry_counter_

Additional Inherited Members
Static Public Member Functions inherited from ACE_Task_Base
static ACE_THR_FUNC_RETURN	svc_run (void *)

static void	cleanup (void object, void params)

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_Task< ACE_NULL_SYNCH >
ACE_Message_Queue< ACE_SYNCH_USE, ACE_System_Time_Policy > *	msg_queue_

bool	delete_msg_queue_

ACE_Module< ACE_SYNCH_USE, ACE_System_Time_Policy > *	mod_

ACE_Task< ACE_SYNCH_USE, ACE_System_Time_Policy > *	next_

	ACE_ALLOC_HOOK_DECLARE

Public Attributes inherited from ACE_Service_Object
	ACE_ALLOC_HOOK_DECLARE

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 ACE_Event_Handler
typedef ACE_Atomic_Op< ACE_SYNCH_MUTEX, Reference_Count >	Atomic_Reference_Count

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

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

Protected Attributes inherited from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >
ACE_SOCK_STREAM	peer_

bool	dynamic_

bool	closing_

ACE_Connection_Recycling_Strategy *	recycler_

const void *	recycling_act_

Protected Attributes inherited from ACE_Task_Base
size_t	thr_count_

ACE_Thread_Manager *	thr_mgr_

u_long	flags_

int	grp_id_

ACE_thread_t	last_thread_id_

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Definition at line 37 of file TcpConnection.h.

Member Typedef Documentation

◆ GuardType

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

private

Definition at line 156 of file TcpConnection.h.

◆ LockType

typedef ACE_SYNCH_MUTEX OpenDDS::DCPS::TcpConnection::LockType

private

Definition at line 155 of file TcpConnection.h.

Member Enumeration Documentation

◆ ReconnectState

enum OpenDDS::DCPS::TcpConnection::ReconnectState

States are used during reconnecting.

Enumerator
INIT_STATE
LOST_STATE
RECONNECTED_STATE
ACTIVE_RECONNECTING_STATE
ACTIVE_WAITING_STATE
PASSIVE_WAITING_STATE
PASSIVE_TIMEOUT_CALLED_STATE

Definition at line 43 of file TcpConnection.h.

                       {
     INIT_STATE,
     LOST_STATE,
     RECONNECTED_STATE,
     ACTIVE_RECONNECTING_STATE,
     ACTIVE_WAITING_STATE,
     PASSIVE_WAITING_STATE,
     PASSIVE_TIMEOUT_CALLED_STATE
   };

Constructor & Destructor Documentation

◆ TcpConnection() [1/2]

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL OpenDDS::DCPS::TcpConnection::TcpConnection ( )

Passive side constructor (acceptor)

Definition at line 35 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, ACE_Event_Handler::Reference_Counting_Policy::ENABLED, ACE_Event_Handler::reference_counting_policy(), and ACE_Event_Handler::Reference_Counting_Policy::value().

   : is_connector_(false)
   , tcp_config_()
   , reconnect_state_(INIT_STATE)
   , transport_priority_(0)  // TRANSPORT_PRIORITY.value default value - 0.
   , shutdown_(false)
   , passive_setup_(false)
   , passive_setup_buffer_(sizeof(ACE_UINT32))
   , transport_during_setup_(0)
   , id_(0)
   , conn_retry_counter_(0)
 {
   DBG_ENTRY_LVL("TcpConnection","TcpConnection",6);
   this->reference_counting_policy().value(ACE_Event_Handler::Reference_Counting_Policy::ENABLED);
 }

◆ TcpConnection() [2/2]

OpenDDS::DCPS::TcpConnection::TcpConnection	(	const ACE_INET_Addr &	remote_address,
		Priority	priority,
		const TcpInst_rch &	config
	)

Active side constructor (connector)

Definition at line 51 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, ACE_Event_Handler::Reference_Counting_Policy::ENABLED, ACE_Event_Handler::reference_counting_policy(), and ACE_Event_Handler::Reference_Counting_Policy::value().

   : is_connector_(true)
   , remote_address_(remote_address)
   , local_address_(config->local_address())
   , tcp_config_(config)
   , reconnect_state_(INIT_STATE)
   , transport_priority_(priority)
   , shutdown_(false)
   , passive_setup_(false)
   , transport_during_setup_(0)
   , id_(0)
   , conn_retry_counter_(0)
 {
   DBG_ENTRY_LVL("TcpConnection","TcpConnection",6);
   this->reference_counting_policy().value(ACE_Event_Handler::Reference_Counting_Policy::ENABLED);
 }

◆ ~TcpConnection()

OpenDDS::DCPS::TcpConnection::~TcpConnection ( )

virtual

Definition at line 70 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, and shutdown().

 {
   DBG_ENTRY_LVL("TcpConnection","~TcpConnection",6);
   shutdown();
 }

Member Function Documentation

◆ active_open()

int OpenDDS::DCPS::TcpConnection::active_open ( )

Protocol setup (handshake) on the active side. The local address is sent to the remote (passive) side to identify ourselves to the remote side.

Definition at line 131 of file TcpConnection.cpp.

References OpenDDS::DCPS::TcpTransport::async_connect_failed(), OpenDDS::DCPS::TcpTransport::connect_tcp_datalink(), OpenDDS::DCPS::dynamic_rchandle_cast(), OpenDDS::DCPS::DataLink::impl(), link_, LM_DEBUG, local_address_, on_active_connection_established(), OpenDDS::DCPS::rchandle_from(), remote_address_, transport_priority_, VDBG, and VDBG_LVL.

Referenced by open().

 {
   VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::active_open.\n"), 2);
   VDBG((LM_DEBUG, "(%P|%t) DBG:   active_open(%C->%C)\n",
         LogAddr(local_address_).c_str(), LogAddr(remote_address_).c_str()));
 
   RcHandle<TcpTransport> transport = dynamic_rchandle_cast<TcpTransport>(link_->impl());
 
   if (transport) {
     if (on_active_connection_established() != -1 && transport->connect_tcp_datalink(*link_, rchandle_from(this)) != -1) {
       return 0;
     }
 
     const bool is_loop(local_address_ == remote_address_);
     const PriorityKey key(transport_priority_, remote_address_,
                       is_loop, true /* active */);
     transport->async_connect_failed(key);
   }
   return -1;
 }

◆ active_reconnect_i()

void OpenDDS::DCPS::TcpConnection::active_reconnect_i ( )

private

Definition at line 597 of file TcpConnection.cpp.

References ACE_DEBUG, ACE_ERROR, ACTIVE_RECONNECTING_STATE, ACTIVE_WAITING_STATE, ACE_Synch_Options::asynch, conn_retry_counter_, ACE_Connector< class, class >::connect(), OpenDDS::DCPS::TcpTransport::connector_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::DataLink::DISCONNECTED, OpenDDS::DCPS::dynamic_rchandle_cast(), EALREADY, EWOULDBLOCK, handle_stop_reconnecting(), OpenDDS::DCPS::DataLink::impl(), OpenDDS::DCPS::TcpDataLink::is_release_pending(), link_, LM_DEBUG, LM_ERROR, local_address_, ACE_Time_Value::msec(), OpenDDS::DCPS::DataLink::notify(), ACE_Event_Handler::reactor(), reconnect_lock_, reconnect_state_, reconnect_state_string(), remote_address_, ACE_Reactor::schedule_timer(), ACE_Time_Value::sec(), shutdown_, and tcp_config_.

Referenced by handle_close(), handle_timeout(), and impl().

 {
   DBG_ENTRY_LVL("TcpConnection","active_reconnect_i",6);
 
   if (this->link_->is_release_pending()) {
     return;
   }
 
   if (this->shutdown_) {
     return;
   }
 
   TcpInst_rch cfg = tcp_config_.lock();
   if (!cfg) {
     return;
   }
 
   if (this->conn_retry_counter_ < cfg->conn_retry_attempts_ ) {
     this->reconnect_state_ = ACTIVE_RECONNECTING_STATE;
     if (this->conn_retry_counter_ == 0)
       this->link_->notify(DataLink::DISCONNECTED);
 
     double retry_delay_msec = cfg->conn_retry_initial_delay_;
     retry_delay_msec *= std::pow(cfg->conn_retry_backoff_multiplier_, this->conn_retry_counter_);
 
     if (DCPS_debug_level >= 1) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::"
             "active_reconnect_i(%C->%C) reconnect_state = %C, conn_retry_counter_=%d, retry_delay_msec=%f\n",
             LogAddr(remote_address_).c_str(), LogAddr(local_address_).c_str(),
             reconnect_state_string(), this->conn_retry_counter_, retry_delay_msec));
     }
 
     ACE_Time_Value timeout;
     timeout.msec(static_cast<int>(retry_delay_msec));
 
     TcpConnection* pconn = this;
     int ret = -1;
     errno = ENODEV;
     {
       RcHandle<TcpTransport> transport = dynamic_rchandle_cast<TcpTransport>(link_->impl());
       if (transport) {
         ACE_Reverse_Lock<LockType> rev_lock(this->reconnect_lock_);
         ACE_Guard<ACE_Reverse_Lock<LockType> > guard(rev_lock);
         // We need to temporarily release the lock here because the connect could occasionally be synchronous
         // if the source and destination are on the same host. When the call become synchronous, active_reconnect_open()
         // would be called and try to acquired the lock in the same thread.
         ret = transport->connector_.connect(pconn, this->remote_address_,  ACE_Synch_Options::asynch);
       }
     }
 
     if (ret == -1 && errno != EWOULDBLOCK)
     {
       if (errno == EALREADY) {
         // This could happen on Windows, it may due to the close() on non-blocking socket needs more time to complete.
         // In this case, we just wait another second to initiate the connect again without incrementing the conn_retry_counter_.
         timeout.sec(1);
         --conn_retry_counter_;
         if (DCPS_debug_level >= 1) {
           ACE_DEBUG((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::"
                                "active_reconnect_i() socket operation is already in progress, wait another second to initiate the connect\n"));
         }
       } else {
         ACE_ERROR((LM_ERROR, "(%P|%t) TcpConnection::active_reconnect_i error %m.\n"));
       }
       this->reconnect_state_ = ACTIVE_WAITING_STATE;
     }
 
     this->reactor()->schedule_timer(this, 0, timeout);
     this->conn_retry_counter_ ++;
   } else {
     this->handle_stop_reconnecting();
   }
 }

◆ active_reconnect_open()

int OpenDDS::DCPS::TcpConnection::active_reconnect_open ( )

Definition at line 964 of file TcpConnection.cpp.

References ACE_DEBUG, ACE_ERROR_RETURN, ACE_TEXT(), config_name(), conn_retry_counter_, DBG_ENTRY_LVL, link_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::DataLink::notify(), on_active_connection_established(), OPENDDS_END_VERSIONED_NAMESPACE_DECL, ACE_Event_Handler::reactor(), ACE_Event_Handler::READ_MASK, reconnect_lock_, reconnect_state_, OpenDDS::DCPS::DataLink::RECONNECTED, RECONNECTED_STATE, ACE_Reactor::register_handler(), remote_address_, send_strategy(), and shutdown_.

Referenced by open().

 {
   DBG_ENTRY_LVL("TcpConnection","active_reconnect_open",6);
 
   ACE_DEBUG((LM_DEBUG, "(%P|%t) re-established connection on transport: %C to %C.\n",
     config_name().c_str(), LogAddr(remote_address_).c_str()));
 
   GuardType guard(reconnect_lock_);
 
   if (shutdown_) {
     return 0;
   }
 
   if (on_active_connection_established() == -1) {
     return -1;
   }
 
   int result = reactor()->register_handler(this, READ_MASK);
   if (result == -1) {
       ACE_ERROR_RETURN((LM_ERROR,
                         "(%P|%t) ERROR: OpenDDS::DCPS::TcpConnection::active_reconnect_open() can't register "
                         "with reactor %X %p\n", this, ACE_TEXT("register_handler")),
                         -1);
   }
 
   this->reconnect_state_ = RECONNECTED_STATE;
   this->link_->notify(DataLink::RECONNECTED);
   this->send_strategy()->resume_send();
   this->conn_retry_counter_ = 0;
 
   return 0;
 }

◆ add_reference()

ACE_Event_Handler::Reference_Count OpenDDS::DCPS::TcpConnection::add_reference ( void )

virtual

Reimplemented from ACE_Event_Handler.

Definition at line 924 of file TcpConnection.cpp.

References OpenDDS::DCPS::RcObject::_add_ref().

Referenced by impl().

 {
   RcObject::_add_ref();
   return 1;
 }

◆ close()

int OpenDDS::DCPS::TcpConnection::close ( u_long )

virtual

Reimplemented from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >.

Definition at line 339 of file TcpConnection.cpp.

References ACE_DEBUG, ACTIVE_RECONNECTING_STATE, ACTIVE_WAITING_STATE, ACE_Connector< class, class >::close(), config_name(), conn_retry_counter_, OpenDDS::DCPS::TcpTransport::connector_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, disconnect(), OpenDDS::DCPS::dynamic_rchandle_cast(), handle_stop_reconnecting(), OpenDDS::DCPS::DataLink::impl(), link_, LM_DEBUG, reconnect_state_, reconnect_state_string(), remote_address_, send_strategy(), and tcp_config_.

 {
   DBG_ENTRY_LVL("TcpConnection","close",6);
 
   ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::close, reconnect_state_=%C\n", reconnect_state_string()));
 
   TcpInst_rch cfg = tcp_config_.lock();
   if (cfg && reconnect_state_ == ACTIVE_RECONNECTING_STATE) {
     // This would be called when using ACE_Connector to initiate an async connect and
     // the network stack detects the destination is unreachable before timeout.
     if (DCPS_debug_level >= 1) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection()::close() on transport: %C to %C because of reconnect failure.\n",
         config_name().c_str(), LogAddr(remote_address_).c_str()));
     }
 
     if (conn_retry_counter_ >= cfg->conn_retry_attempts_) {
       handle_stop_reconnecting();
     } else {
       TcpTransport_rch transport = dynamic_rchandle_cast<TcpTransport>(link_->impl());
       if (transport) {
         transport->connector_.close();
         reconnect_state_ = ACTIVE_WAITING_STATE;
       }
     }
   } else {
     TcpSendStrategy_rch ss = send_strategy();
     if (ss) {
       ss->terminate_send();
     }
 
     disconnect();
   }
 
   return 0;
 }

◆ config_name()

const std::string & OpenDDS::DCPS::TcpConnection::config_name ( ) const

private

Definition at line 376 of file TcpConnection.cpp.

References tcp_config_.

Referenced by active_reconnect_open(), close(), handle_close(), handle_stop_reconnecting(), handle_timeout(), and impl().

 {
   static const std::string null_name("(couldn't get name)");
   TcpInst_rch cfg = tcp_config_.lock();
   return cfg ? cfg->name() : null_name;
 }

◆ disconnect()

void OpenDDS::DCPS::TcpConnection::disconnect ( void )

This will be called by the DataLink (that "owns" us) when the TcpTransport has been told to shutdown(), or when the DataLink finds itself no longer needed, and is "self-releasing".

Definition at line 105 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, OpenDDS::DCPS::TcpDataLink::drop_pending_request_acks(), link_, and ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >::peer().

Referenced by close(), handle_close(), and notify_lost_on_backpressure_timeout().

 {
   DBG_ENTRY_LVL("TcpConnection","disconnect",6);
 
   if (this->link_) {
     this->link_->drop_pending_request_acks();
   }
 
   this->peer().close();
 }

◆ get_remote_address()

ACE_INLINE ACE_INET_Addr OpenDDS::DCPS::TcpConnection::get_remote_address ( )

Definition at line 35 of file TcpConnection.inl.

References ACE_INLINE, and remote_address_.

 {
   return this->remote_address_;
 }

◆ handle_close()

int OpenDDS::DCPS::TcpConnection::handle_close	(	ACE_HANDLE	,
		ACE_Reactor_Mask
	)

virtual

Reimplemented from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >.

Definition at line 384 of file TcpConnection.cpp.

References ACE_DEBUG, active_reconnect_i(), config_name(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, disconnect(), OpenDDS::DCPS::DataLink::DISCONNECTED, is_connector_, link_, LM_DEBUG, OpenDDS::DCPS::DataLink::notify(), passive_reconnect_i(), OpenDDS::DCPS::TcpDataLink::receive_strategy(), receive_strategy(), reconnect_lock_, reconnect_state_string(), remote_address_, OpenDDS::DCPS::TcpDataLink::send_strategy(), send_strategy(), and TheServiceParticipant.

 {
   ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());
 
   DBG_ENTRY_LVL("TcpConnection","handle_close",6);
 
   if (DCPS_debug_level >= 1) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::handle_close() called on transport: %C to %C , reconnect_state = %C.\n",
       config_name().c_str(), LogAddr(remote_address_).c_str(), reconnect_state_string()));
   }
 
   GuardType guard(reconnect_lock_);
   TcpDataLink_rch link = link_;
 
   if (!link) {
     if (DCPS_debug_level >= 1) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::handle_close() link is null.\n"));
     }
     return 0;
   }
 
   TcpReceiveStrategy_rch receive_strategy = link->receive_strategy();
   TcpSendStrategy_rch send_strategy = link->send_strategy();
 
   const bool graceful = receive_strategy && receive_strategy->gracefully_disconnected();
 
   if (send_strategy) {
     if (graceful) {
       send_strategy->terminate_send();
     } else {
       send_strategy->suspend_send();
     }
   }
 
   this->disconnect();
 
   if (graceful) {
     link->notify(DataLink::DISCONNECTED);
   } else if (this->is_connector_) {
     this->active_reconnect_i();
   } else {
     this->passive_reconnect_i();
   }
 
   return 0;
 }

◆ handle_input()

int OpenDDS::DCPS::TcpConnection::handle_input ( ACE_HANDLE fd )

virtual

We pass this "event" along to the receive_strategy.

Reimplemented from ACE_Event_Handler.

Definition at line 291 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, handle_setup_input(), passive_setup_, receive_strategy(), and TheServiceParticipant.

 {
   DBG_ENTRY_LVL("TcpConnection","handle_input",6);
 
   ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());
 
   if (passive_setup_) {
     return handle_setup_input(fd);
   }
   TcpReceiveStrategy_rch receive_strategy = this->receive_strategy();
   if (!receive_strategy) {
     return 0;
   }
 
   return receive_strategy->handle_dds_input(fd);
 }

◆ handle_output()

int OpenDDS::DCPS::TcpConnection::handle_output ( ACE_HANDLE )

virtual

Handle back pressure when sending.

Reimplemented from ACE_Event_Handler.

Definition at line 309 of file TcpConnection.cpp.

References ACE_DEBUG, ACE_TEXT(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, id_, LM_DEBUG, send_strategy(), TheServiceParticipant, and OpenDDS::DCPS::ThreadSynchWorker::WORK_OUTCOME_MORE_TO_DO.

 {
   DBG_ENTRY_LVL("TcpConnection","handle_output",6);
 
   ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());
 
   TcpSendStrategy_rch send_strategy = this->send_strategy();
   if (send_strategy) {
     if (DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TcpConnection::handle_output() [%d] - ")
                  ACE_TEXT("sending queued data.\n"),
                  id_));
     }
 
     // Process data to be sent from the queue.
     if (ThreadSynchWorker::WORK_OUTCOME_MORE_TO_DO
         != send_strategy->perform_work()) {
 
       // Stop handling output ready events when there is nothing to output.
       // N.B. This calls back into the reactor.  Is the reactor lock
       //      recursive?
       send_strategy->schedule_output();
     }
   }
 
   return 0;
 }

◆ handle_setup_input()

int OpenDDS::DCPS::TcpConnection::handle_setup_input ( ACE_HANDLE h )

private

During the connection setup phase, the passive side sets passive_setup_, redirecting handle_input() events here (there is no recv strategy yet).

Definition at line 221 of file TcpConnection.cpp.

References ACE_Message_Block::base(), OpenDDS::DCPS::DCPS_debug_level, ACE_Event_Handler::DONT_CALL, ETIME, ACE_Message_Block::length(), LM_DEBUG, local_address_, ACE_OS::memcpy(), OpenDDS::DCPS::TcpTransport::passive_connection(), passive_setup_, passive_setup_buffer_, ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >::peer(), OpenDDS::DCPS::rchandle_from(), ACE_Message_Block::rd_ptr(), ACE_Event_Handler::reactor(), ACE_Event_Handler::READ_MASK, reconnect_state_string(), remote_address_, ACE_Message_Block::reset(), ACE_Message_Block::size(), ACE_Message_Block::space(), transport_during_setup_, transport_priority_, VDBG, VDBG_LVL, ACE_Message_Block::wr_ptr(), and ACE_Time_Value::zero.

Referenced by handle_input(), and impl().

 {
   const ssize_t ret = peer().recv(passive_setup_buffer_.wr_ptr(),
                                   passive_setup_buffer_.space(),
                                   &ACE_Time_Value::zero);
 
   if (ret < 0 && errno == ETIME) {
     return 0;
   }
 
   VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::handle_setup_input %@ "
             "recv returned %b %m.\n", this, ret), 4);
 
   if (ret <= 0) {
     return -1;
   }
 
   passive_setup_buffer_.wr_ptr(ret);
   // Parse the setup message: <len><addr><prio>
   // len and prio are network order 32-bit ints
   // addr is a string of length len, including null
   ACE_UINT32 nlen = 0;
 
   if (passive_setup_buffer_.length() >= sizeof(nlen)) {
 
     ACE_OS::memcpy(&nlen, passive_setup_buffer_.rd_ptr(), sizeof(nlen));
     passive_setup_buffer_.rd_ptr(sizeof(nlen));
     const ACE_UINT32 hlen = ntohl(nlen);
     passive_setup_buffer_.size(hlen + 2 * sizeof(nlen));
 
     ACE_UINT32 nprio = 0;
 
     if (passive_setup_buffer_.length() >= hlen + sizeof(nprio)) {
 
       const std::string bufstr(passive_setup_buffer_.rd_ptr());
       const NetworkResource network_resource(bufstr);
       network_resource.to_addr(remote_address_);
 
       ACE_OS::memcpy(&nprio, passive_setup_buffer_.rd_ptr() + hlen, sizeof(nprio));
       transport_priority_ = ntohl(nprio);
 
       passive_setup_buffer_.reset();
       passive_setup_ = false;
 
       VDBG((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::handle_setup_input "
             "%@ %C->%C, priority==%d, reconnect_state = %C\n", this,
             LogAddr(remote_address_).c_str(), LogAddr(local_address_).c_str(),
             transport_priority_, reconnect_state_string()));
       if (DCPS_debug_level > 9) {
         network_resource.dump();
       }
 
       // remove from reactor, normal recv strategy setup will add us back
       if (reactor()->remove_handler(this, READ_MASK | DONT_CALL) == -1) {
         VDBG((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::handle_setup_input "
               "remove_handler failed %m.\n"));
       }
 
       transport_during_setup_->passive_connection(remote_address_, rchandle_from(this));
 
       return 0;
     }
   }
 
   passive_setup_buffer_.rd_ptr(passive_setup_buffer_.base());
 
   return 0;
 }

◆ handle_stop_reconnecting()

void OpenDDS::DCPS::TcpConnection::handle_stop_reconnecting ( )

private

Definition at line 685 of file TcpConnection.cpp.

References ACE_DEBUG, config_name(), LM_DEBUG, LOST_STATE, notify_connection_lost(), reconnect_state_, remote_address_, and tcp_config_.

Referenced by active_reconnect_i(), close(), and impl().

 {
   this->reconnect_state_ = LOST_STATE;
   notify_connection_lost();
   TcpInst_rch cfg = tcp_config_.lock();
   if (cfg && cfg->conn_retry_attempts_ > 0) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) we tried and failed to re-establish connection on transport: %C to %C.\n",
       config_name().c_str(), LogAddr(remote_address_).c_str()));
   } else {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) we did not try to re-establish connection on transport: %C to %C.\n",
       config_name().c_str(), LogAddr(remote_address_).c_str()));
   }
 }

◆ handle_timeout()

int OpenDDS::DCPS::TcpConnection::handle_timeout	(	const ACE_Time_Value &	tv,
		const void *	arg
	)

virtual

A timer is scheduled on acceptor side to check if a new connection is accepted after the connection is lost.

Reimplemented from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >.

Definition at line 702 of file TcpConnection.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), active_reconnect_i(), ACTIVE_RECONNECTING_STATE, ACTIVE_WAITING_STATE, OpenDDS::DCPS::TcpTransport::async_connect_failed(), ACE_Connector< class, class >::cancel(), config_name(), OpenDDS::DCPS::TcpTransport::connector_, DBG_ENTRY_LVL, OpenDDS::DCPS::dynamic_rchandle_cast(), OpenDDS::DCPS::DataLink::impl(), INIT_STATE, link_, LM_DEBUG, LM_ERROR, local_address_, LOST_STATE, notify_connection_lost(), PASSIVE_TIMEOUT_CALLED_STATE, PASSIVE_WAITING_STATE, reconnect_lock_, reconnect_state_, reconnect_state_string(), RECONNECTED_STATE, remote_address_, tear_link(), TheServiceParticipant, and transport_priority_.

 {
   ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());
 
   DBG_ENTRY_LVL("TcpConnection","handle_timeout",6);
   ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::handle_timeout, this->reconnect_state_ = %C\n", reconnect_state_string()));
   GuardType guard(this->reconnect_lock_);
 
   switch (this->reconnect_state_) {
   case PASSIVE_WAITING_STATE: {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::handle_timeout, we tried and failed to re-establish connection on transport: %C to %C.\n",
       config_name().c_str(), LogAddr(remote_address_).c_str()));
 
     this->reconnect_state_ = PASSIVE_TIMEOUT_CALLED_STATE;
     // We stay in PASSIVE_TIMEOUT_CALLED_STATE indicates there is no new connection.
     // Now we need declare the connection is lost.
     this->notify_connection_lost();
     this->reconnect_state_ = LOST_STATE;
     this->tear_link();
   }
   break;
 
   case RECONNECTED_STATE:
     // reconnected successfully.
     ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::handle_timeout, re-established connection on transport: %C to %C.\n",
       config_name().c_str(), LogAddr(remote_address_).c_str()));
     break;
 
   case INIT_STATE: {
     // couldn't initialize connection successfully.
     ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::handle_timeout, failed connection initialization due to timeout.: %C to %C.\n",
       config_name().c_str(), LogAddr(remote_address_).c_str()));
 
     // build key and remove from service
     const bool is_loop(local_address_ == remote_address_);
     const PriorityKey key(transport_priority_, remote_address_,
                           is_loop, true /* active */);
 
     RcHandle<TcpTransport> transport = dynamic_rchandle_cast<TcpTransport>(link_->impl());
     if (transport) {
       transport->async_connect_failed(key);
     }
     break;
   }
   case ACTIVE_RECONNECTING_STATE: {
     // we get the timeout before the network stack reports the destination is unreachable
     // cancel the async connect operation and retry it.
     {
       RcHandle<TcpTransport> transport = dynamic_rchandle_cast<TcpTransport>(link_->impl());
       if (transport) {
         transport->connector_.cancel(this);
       }
     }
     this->active_reconnect_i();
     break;
   }
   case ACTIVE_WAITING_STATE:
     this->active_reconnect_i();
     break;
   case LOST_STATE:
     break;
   default :
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TcpConnection::handle_timeout, ")
                ACE_TEXT(" unknown state or it should not be in state = %d\n"),
                reconnect_state_));
     break;
   }
 
   return 0;
 }

◆ id()

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL ACE_INLINE std::size_t & OpenDDS::DCPS::TcpConnection::id ( void )

Definition at line 22 of file TcpConnection.inl.

References ACE_INLINE, and id_.

 {
   return id_;
 }

◆ impl()

TcpTransport_rch OpenDDS::DCPS::TcpConnection::impl ( void )

inline

Definition at line 127 of file TcpConnection.h.

References active_reconnect_i(), add_reference(), config_name(), handle_setup_input(), handle_stop_reconnecting(), impl_, notify_connection_lost(), on_active_connection_established(), OPENDDS_POOL_ALLOCATION_FWD, passive_reconnect_i(), remove_reference(), and transport_priority().

127 { return impl_; }

OpenDDS::DCPS::TcpConnection::impl_

TcpTransport_rch impl_

Impl object which is needed for connection objects and reconnect task.

Definition: TcpConnection.h:177

◆ is_connector()

ACE_INLINE bool OpenDDS::DCPS::TcpConnection::is_connector ( ) const

Return true if the object represents the connector side, otherwise it's the acceptor side. The acceptor/connector role is not changed when re-establishing the connection.

Definition at line 29 of file TcpConnection.inl.

References ACE_INLINE, and is_connector_.

 {
   return this->is_connector_;
 }

◆ notify_connection_lost()

void OpenDDS::DCPS::TcpConnection::notify_connection_lost ( )

private

Definition at line 672 of file TcpConnection.cpp.

References OpenDDS::DCPS::TcpDataLink::drop_pending_request_acks(), link_, OpenDDS::DCPS::DataLink::LOST, OpenDDS::DCPS::DataLink::notify(), OpenDDS::DCPS::TcpDataLink::send_strategy(), and send_strategy().

Referenced by handle_stop_reconnecting(), handle_timeout(), impl(), and notify_lost_on_backpressure_timeout().

 {
   if (link_) {
     link_->drop_pending_request_acks();
     link_->notify(DataLink::LOST);
     TcpSendStrategy_rch send_strategy = link_->send_strategy();
     if (send_strategy) {
       send_strategy->terminate_send();
     }
   }
 }

◆ notify_lost_on_backpressure_timeout()

void OpenDDS::DCPS::TcpConnection::notify_lost_on_backpressure_timeout ( )

This function is called when the backpressure occurs and timed out after "max_output_pause_period". The lost connection notification should be sent and the connection needs be closed since we declared it as a "lost" connection.

Definition at line 857 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, disconnect(), INIT_STATE, LOST_STATE, notify_connection_lost(), reconnect_lock_, and reconnect_state_.

 {
   DBG_ENTRY_LVL("TcpConnection","notify_lost_on_backpressure_timeout",6);
   bool notify_lost = false;
   {
     GuardType guard(this->reconnect_lock_);
 
     if (this->reconnect_state_ == INIT_STATE) {
       this->reconnect_state_ = LOST_STATE;
       notify_lost = true;
 
     }
   }
 
   if (notify_lost) {
     this->disconnect();
     this->notify_connection_lost();
   }
 
 }

◆ on_active_connection_established()

int OpenDDS::DCPS::TcpConnection::on_active_connection_established ( )

private

Handle the logic after an active connection has been established.

Definition at line 482 of file TcpConnection.cpp.

References ACE_DEBUG, ACE_ERROR, OpenDDS::DCPS::DirectPriorityMapper::codepoint(), OpenDDS::DCPS::DCPS_debug_level, link_, LM_DEBUG, LM_NOTICE, LM_WARNING, OpenDDS::DCPS::log_level, OpenDDS::DCPS::LogLevel::Notice, ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >::peer(), send_n(), OpenDDS::DCPS::DataLink::set_dscp_codepoint(), set_sock_options(), tcp_config_, and transport_priority_.

Referenced by active_open(), active_reconnect_open(), and impl().

 {
 
   // Set the DiffServ codepoint according to the priority value.
   DirectPriorityMapper mapper(this->transport_priority_);
   this->link_->set_dscp_codepoint(mapper.codepoint(), this->peer());
 
   TcpInst_rch cfg = tcp_config_.lock();
   if (!cfg) {
     if (log_level >= LogLevel::Notice) {
       ACE_ERROR((LM_NOTICE, "((%P|%t)) NOTICE: TcpConnection::on_active_connection_established() - Invalid Transport Instance.\n"));
     }
     return -1;
   }
   set_sock_options(cfg);
 
   // In order to complete the connection establishment from the active
   // side, we need to tell the remote side about our public address.
   // It will use that as an "identifier" of sorts.  To the other
   // (passive) side, our local_address that we send here will be known
   // as the remote_address.
   const std::string address = cfg->get_public_address();
 
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) TcpConnection::on_active_connection_established: "
                "Sending public address <%C> to remote side\n",
                address.c_str()));
   }
 
   ACE_UINT32 len = static_cast<ACE_UINT32>(address.length()) + 1;
 
   ACE_UINT32 nlen = htonl(len);
 
   if (this->peer().send_n(&nlen, sizeof(ACE_UINT32)) == -1) {
     if (DCPS_debug_level >= 2) {
       ACE_DEBUG((LM_WARNING,
                  "(%P|%t) WARNING: TcpConnection::on_active_connection_established: "
                  "Unable to send address string length to "
                  "the passive side to complete the active connection "
                  "establishment.\n"));
     }
     return -1;
   }
 
   if (this->peer().send_n(address.c_str(), len)  == -1) {
     if (DCPS_debug_level >= 2) {
       ACE_DEBUG((LM_WARNING,
                  "(%P|%t) WARNING: TcpConnection::on_active_connection_established: "
                  "Unable to send our address to "
                  "the passive side to complete the active connection "
                  "establishment.\n"));
     }
     return -1;
   }
 
   ACE_UINT32 npriority = htonl(this->transport_priority_);
 
   if (this->peer().send_n(&npriority, sizeof(ACE_UINT32)) == -1) {
     if (DCPS_debug_level >= 2) {
       ACE_DEBUG((LM_WARNING,
                  "(%P|%t) WARNING: TcpConnection::on_active_connection_established: "
                  "Unable to send publication priority to "
                  "the passive side to complete the active connection "
                  "establishment.\n"));
     }
     return -1;
   }
 
   return 0;
 }

◆ open()

int OpenDDS::DCPS::TcpConnection::open ( void * arg )

virtual

Reimplemented from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >.

Definition at line 117 of file TcpConnection.cpp.

References active_open(), active_reconnect_open(), ACTIVE_RECONNECTING_STATE, DBG_ENTRY_LVL, is_connector_, passive_open(), and reconnect_state_.

 {
   DBG_ENTRY_LVL("TcpConnection","open",6);
 
   if (is_connector_) {
     if (this->reconnect_state_ == ACTIVE_RECONNECTING_STATE) {
       return active_reconnect_open();
     }
     return active_open();
   }
   return passive_open(arg);
 }

◆ passive_open()

int OpenDDS::DCPS::TcpConnection::passive_open ( void * arg )

Definition at line 153 of file TcpConnection.cpp.

References ACE_ERROR, ACE_ERROR_RETURN, ACE_TEXT(), ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >::get_handle(), LM_DEBUG, LM_ERROR, LM_NOTICE, local_address_, OpenDDS::DCPS::log_level, OpenDDS::DCPS::LogLevel::Notice, passive_setup_, passive_setup_buffer_, ACE_Event_Handler::reactor(), ACE_Event_Handler::READ_MASK, set_sock_options(), ACE_Message_Block::size(), tcp_config_, OpenDDS::DCPS::TcpAcceptor::transport(), transport_during_setup_, and VDBG_LVL.

Referenced by open().

 {
   VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::passive_open.\n"), 2);
   // The passed-in arg is really the acceptor object that created this
   // TcpConnection object, and is also the caller of this open()
   // method.  We need to cast the arg to the TcpAcceptor* type.
   TcpAcceptor* acceptor = static_cast<TcpAcceptor*>(arg);
 
   if (acceptor == 0) {
     // The cast failed.
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: TcpConnection::open() - ")
                       ACE_TEXT("failed to cast void* arg to ")
                       ACE_TEXT("TcpAcceptor* type.\n")),
                      -1);
   }
 
   TcpConnection_rch self(this, keep_count());
 
   // Now we need to ask the TcpAcceptor object to provide us with
   // a pointer to the TcpTransport object that "owns" the acceptor.
   RcHandle<TcpTransport> transport = acceptor->transport();
 
   if (!transport) {
     // The acceptor gave us a nil transport (smart) pointer.
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: TcpConnection::open() - ")
                       ACE_TEXT("acceptor's transport is nil.\n")),
                      -1);
   }
 
   // Keep a "copy" of the reference to TcpInst object
   // for ourselves.
   TcpInst_rch cfg = transport->config();
   if (!cfg) {
     if (log_level >= LogLevel::Notice) {
       ACE_ERROR((LM_NOTICE, "((%P|%t)) NOTICE: TcpConnection::open() - Invalid Transport Instance.\n"));
     }
     return -1;
   }
   tcp_config_ = cfg;
   local_address_ = cfg->local_address();
 
   set_sock_options(cfg);
 
   // We expect that the active side of the connection (the remote side
   // in this case) will supply its listening ACE_INET_Addr as the first
   // message it sends to the socket.  This is a one-way connection
   // establishment protocol message.
   passive_setup_ = true;
   transport_during_setup_ = transport.get();
   passive_setup_buffer_.size(sizeof(ACE_UINT32));
 
   if (reactor()->register_handler(this, READ_MASK) == -1) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: TcpConnection::open() - ")
                       ACE_TEXT("unable to register with the reactor.%p\n"),
                       ACE_TEXT("register_handler")),
                      -1);
   }
 
   VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::open passive handle=%d.\n",
             static_cast<int>(intptr_t(get_handle()))), 2);
 
   return 0;
 }

◆ passive_reconnect_i()

void OpenDDS::DCPS::TcpConnection::passive_reconnect_i ( )

private

Definition at line 558 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, OpenDDS::DCPS::DataLink::DISCONNECTED, OpenDDS::DCPS::TimeDuration::from_msec(), INIT_STATE, link_, OpenDDS::DCPS::DataLink::notify(), PASSIVE_WAITING_STATE, ACE_Event_Handler::reactor(), reconnect_state_, ACE_Reactor::schedule_timer(), shutdown_, tcp_config_, OpenDDS::DCPS::TimeDuration::value(), and OpenDDS::DCPS::TimeDuration::zero_value.

Referenced by handle_close(), and impl().

 {
   DBG_ENTRY_LVL("TcpConnection","passive_reconnect_i",6);
 
   if (this->shutdown_) {
     return;
   }
 
   TcpInst_rch cfg = tcp_config_.lock();
   if (!cfg) {
     return;
   }
 
   if (this->reconnect_state_ == INIT_STATE) {
     // Mark the connection lost since the recv/send just failed.
     // this->connected_ = false;
 
     if (cfg->passive_reconnect_duration_ == 0)
       return;
 
     this->reconnect_state_ = PASSIVE_WAITING_STATE;
     this->link_->notify(DataLink::DISCONNECTED);
 
     TimeDuration delay = TimeDuration::from_msec(cfg->passive_reconnect_duration_);
     this->reactor()->schedule_timer(this, 0, delay.value(), TimeDuration::zero_value.value());
   }
 }

◆ receive_strategy()

OpenDDS::DCPS::TcpReceiveStrategy_rch OpenDDS::DCPS::TcpConnection::receive_strategy ( )

Definition at line 99 of file TcpConnection.cpp.

References link_, and OpenDDS::DCPS::TcpDataLink::receive_strategy().

Referenced by handle_close(), and handle_input().

 {
   return this->link_->receive_strategy();
 }

◆ reconnect_state_string()

const char * OpenDDS::DCPS::TcpConnection::reconnect_state_string ( ) const

private

Get name of the current reconnect state as a string.

Definition at line 937 of file TcpConnection.cpp.

References ACE_ERROR, ACE_TEXT(), ACTIVE_RECONNECTING_STATE, ACTIVE_WAITING_STATE, INIT_STATE, LM_ERROR, LOST_STATE, PASSIVE_TIMEOUT_CALLED_STATE, PASSIVE_WAITING_STATE, reconnect_state_, and RECONNECTED_STATE.

Referenced by active_reconnect_i(), close(), handle_close(), handle_setup_input(), and handle_timeout().

 {
   switch (reconnect_state_) {
   case INIT_STATE:
     return "INIT_STATE";
   case LOST_STATE:
     return "LOST_STATE";
   case RECONNECTED_STATE:
     return "RECONNECTED_STATE";
   case ACTIVE_RECONNECTING_STATE:
     return "ACTIVE_RECONNECTING_STATE";
   case ACTIVE_WAITING_STATE:
     return "ACTIVE_WAITING_STATE";
   case PASSIVE_WAITING_STATE:
     return "PASSIVE_WAITING_STATE";
   case PASSIVE_TIMEOUT_CALLED_STATE:
     return "PASSIVE_TIMEOUT_CALLED_STATE";
   default:
     ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: TcpConnection::reconnect_state_string: ")
       ACE_TEXT("%d is either invalid or not recognized.\n"),
       reconnect_state_));
     return "Invalid reconnect state";
   }
 }

◆ relink_from_recv()

void OpenDDS::DCPS::TcpConnection::relink_from_recv ( bool do_suspend )

Reconnect initiated by receive strategy.

This is called by TcpReceiveStrategy when a disconnect is detected. It simply suspends any sends and lets the handle_close() handle the reconnect logic.

Definition at line 896 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, and send_strategy().

 {
   DBG_ENTRY_LVL("TcpConnection","relink_from_recv",6);
   TcpSendStrategy_rch send_strategy = this->send_strategy();
   if (do_suspend && send_strategy)
     send_strategy->suspend_send();
 }

◆ relink_from_send()

void OpenDDS::DCPS::TcpConnection::relink_from_send ( bool do_suspend )

Reconnect initiated by send strategy.

This is called by TcpSendStrategy when a send fails and a reconnect should be initiated. This method suspends any sends and kicks the reconnect thread into action.

Definition at line 883 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, and send_strategy().

 {
   DBG_ENTRY_LVL("TcpConnection","relink_from_send",6);
 
   TcpSendStrategy_rch send_strategy = this->send_strategy();
   if (do_suspend && send_strategy)
     send_strategy->suspend_send();
 }

◆ remove_reference()

ACE_Event_Handler::Reference_Count OpenDDS::DCPS::TcpConnection::remove_reference ( void )

virtual

Reimplemented from ACE_Event_Handler.

Definition at line 931 of file TcpConnection.cpp.

References OpenDDS::DCPS::RcObject::_remove_ref().

Referenced by impl().

 {
   RcObject::_remove_ref();
   return 1;
 }

◆ send_strategy()

OpenDDS::DCPS::TcpSendStrategy_rch OpenDDS::DCPS::TcpConnection::send_strategy ( )

Definition at line 93 of file TcpConnection.cpp.

References link_, and OpenDDS::DCPS::TcpDataLink::send_strategy().

Referenced by active_reconnect_open(), close(), handle_close(), handle_output(), notify_connection_lost(), relink_from_recv(), and relink_from_send().

 {
   return this->link_->send_strategy();
 }

◆ set_datalink()

void OpenDDS::DCPS::TcpConnection::set_datalink ( const TcpDataLink_rch & link )

Cache the reference to the datalink object for lost connection callbacks.

Definition at line 77 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, OpenDDS::DCPS::dynamic_rchandle_cast(), OpenDDS::DCPS::DataLink::impl(), impl_, link_, reconnect_lock_, and OpenDDS::DCPS::RcHandle< T >::reset().

 {
   DBG_ENTRY_LVL("TcpConnection","set_datalink",6);
 
   GuardType guard(reconnect_lock_);
 
   link_ = link;
   if (link_) {
     impl_ = dynamic_rchandle_cast<TcpTransport>(link_->impl());
   } else {
     impl_.reset();
   }
 }

◆ set_sock_options()

void OpenDDS::DCPS::TcpConnection::set_sock_options ( const TcpInst_rch & tcp_config )

Definition at line 432 of file TcpConnection.cpp.

References ACE_DEFAULT_MAX_SOCKET_BUFSIZ, ACE_ERROR, ENOTSUP, IPPROTO_TCP, LM_ERROR, ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >::peer(), SO_RCVBUF, SO_SNDBUF, SOL_SOCKET, and TCP_NODELAY.

Referenced by on_active_connection_established(), and passive_open().

 {
 #if defined (ACE_DEFAULT_MAX_SOCKET_BUFSIZ)
   int snd_size = ACE_DEFAULT_MAX_SOCKET_BUFSIZ;
   int rcv_size = ACE_DEFAULT_MAX_SOCKET_BUFSIZ;
   //ACE_SOCK_Stream sock = ACE_static_cast(ACE_SOCK_Stream, this->peer() );
 #  if !defined (ACE_LACKS_SOCKET_BUFSIZ)
 
   // A little screwy double negative logic: disabling nagle involves
   // enabling TCP_NODELAY
   int opt = (tcp_config->enable_nagle_algorithm_ == false);
 
   if (this->peer().set_option(IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt)) == -1) {
     ACE_ERROR((LM_ERROR, "Failed to set TCP_NODELAY\n"));
   }
 
   if (this->peer().set_option(SOL_SOCKET,
                               SO_SNDBUF,
                               (void *) &snd_size,
                               sizeof(snd_size)) == -1
       && errno != ENOTSUP) {
     ACE_ERROR((LM_ERROR,
                "(%P|%t) TcpConnection failed to set the send buffer size to %d errno %m\n",
                snd_size));
     return;
   }
 
   if (this->peer().set_option(SOL_SOCKET,
                               SO_RCVBUF,
                               (void *) &rcv_size,
                               sizeof(int)) == -1
       && errno != ENOTSUP) {
     ACE_ERROR((LM_ERROR,
                "(%P|%t) TcpConnection failed to set the receive buffer size to %d errno %m\n",
                rcv_size));
     return;
   }
 
 #  else
   ACE_UNUSED_ARG(tcp_config);
   ACE_UNUSED_ARG(snd_size);
   ACE_UNUSED_ARG(rcv_size);
 #  endif /* !ACE_LACKS_SOCKET_BUFSIZ */
 
 #else
   ACE_UNUSED_ARG(tcp_config);
 #endif /* !ACE_DEFAULT_MAX_SOCKET_BUFSIZ */
 }

◆ shutdown()

void OpenDDS::DCPS::TcpConnection::shutdown ( void )

Definition at line 913 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, reconnect_lock_, ACE_Svc_Handler< PEER_STREAM, SYNCH_TRAITS >::shutdown(), and shutdown_.

Referenced by ~TcpConnection().

 {
   DBG_ENTRY_LVL("TcpConnection", "shutdown", 6);
 
   GuardType guard(reconnect_lock_);
   shutdown_ = true;
 
   ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::shutdown();
 }

◆ tear_link()

void OpenDDS::DCPS::TcpConnection::tear_link ( )

Called by the reconnect task to inform us that the link & any associated data can be torn down. This call is done with no DCPS/transport locks held.

Definition at line 905 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, link_, and OpenDDS::DCPS::DataLink::release_resources().

Referenced by handle_timeout().

 {
   DBG_ENTRY_LVL("TcpConnection","tear_link",6);
 
   return link_->release_resources();
 }

◆ transfer()

void OpenDDS::DCPS::TcpConnection::transfer ( TcpConnection * connection )

This object would be "old" connection object and the provided is the new connection object. The "old" connection object will copy its states to to the "new" connection object. This is called by the TcpDataLink when a new connection is accepted (with a new TcpConnection object). We need make the state in "new" connection object consistent with the "old" connection object.

Definition at line 782 of file TcpConnection.cpp.

References ACE_ERROR, ACE_TEXT(), DBG_ENTRY_LVL, impl_, INIT_STATE, is_connector_, link_, LM_DEBUG, LM_ERROR, local_address_, LOST_STATE, OpenDDS::DCPS::DataLink::notify(), PASSIVE_TIMEOUT_CALLED_STATE, PASSIVE_WAITING_STATE, reconnect_lock_, reconnect_state_, OpenDDS::DCPS::DataLink::RECONNECTED, RECONNECTED_STATE, remote_address_, shutdown_, tcp_config_, and VDBG.

 {
   DBG_ENTRY_LVL("TcpConnection","transfer",6);
   GuardType guard(reconnect_lock_);
 
   if (shutdown_) {
     return;
   }
 
   bool notify_reconnect = false;
 
   switch (this->reconnect_state_) {
   case INIT_STATE:
     // We have not detected the lost connection and the peer is faster than us and
     // re-established the connection. so do not notify reconnected.
     break;
 
   case LOST_STATE:
 
     // The reconnect timed out.
   case PASSIVE_TIMEOUT_CALLED_STATE:
     // TODO: If the handle_timeout is called before the old connection
     // transfer its state to new connection then should we disconnect
     // the new connection or keep it alive ?
     // I think we should keep the connection, the user will get a
     // lost connection notification and then a reconnected notification.
     notify_reconnect = true;
     break;
 
   case PASSIVE_WAITING_STATE: {
     // we just let the timer expires by itself. When the timer
     // expires, it already transitions to the RECONNECTED_STATE,
     // and do nothing in handle_timeout(). We don't need to delete
     // the timer explicitly.
     notify_reconnect = true;
   }
   break;
 
   default :
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TcpConnection::transfer, ")
                ACE_TEXT(" unknown state or it should not be in state=%i\n"),
                reconnect_state_));
     break;
   }
 
   // Verify if this acceptor side.
   if (this->is_connector_ || connection->is_connector_) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TcpConnection::transfer, ")
                ACE_TEXT(" should NOT be called by the connector side\n")));
   }
 
   connection->remote_address_ = this->remote_address_;
   connection->local_address_ = this->local_address_;
   connection->tcp_config_ = this->tcp_config_;
   connection->link_ = this->link_;
   connection->impl_ = this->impl_;
 
   VDBG((LM_DEBUG, "(%P|%t) DBG:   transfer(%C->%C) passive reconnected. new con %@   old con %@\n",
         LogAddr(remote_address_).c_str(), LogAddr(local_address_).c_str(),
         connection, this));
 
   if (notify_reconnect) {
     this->reconnect_state_ = RECONNECTED_STATE;
     this->link_->notify(DataLink::RECONNECTED);
   }
 
 }

◆ transport_priority() [1/2]

ACE_INLINE OpenDDS::DCPS::Priority & OpenDDS::DCPS::TcpConnection::transport_priority ( )

Access TRANSPORT_PRIORITY.value policy value if set.

Definition at line 42 of file TcpConnection.inl.

References ACE_INLINE, and transport_priority_.

Referenced by impl().

 {
   return this->transport_priority_;
 }

◆ transport_priority() [2/2]

ACE_INLINE OpenDDS::DCPS::Priority OpenDDS::DCPS::TcpConnection::transport_priority ( ) const

Definition at line 49 of file TcpConnection.inl.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL, and transport_priority_.

 {
   return this->transport_priority_;
 }

Member Data Documentation

◆ conn_retry_counter_

int OpenDDS::DCPS::TcpConnection::conn_retry_counter_

private

Definition at line 194 of file TcpConnection.h.

Referenced by active_reconnect_i(), active_reconnect_open(), and close().

◆ id_

std::size_t OpenDDS::DCPS::TcpConnection::id_

private

Small unique identifying value.

Definition at line 193 of file TcpConnection.h.

Referenced by handle_output(), and id().

◆ impl_

TcpTransport_rch OpenDDS::DCPS::TcpConnection::impl_

private

Impl object which is needed for connection objects and reconnect task.

Definition at line 177 of file TcpConnection.h.

Referenced by impl(), set_datalink(), and transfer().

◆ is_connector_

bool OpenDDS::DCPS::TcpConnection::is_connector_

private

Flag indicate this connection object is the connector or acceptor.

Definition at line 162 of file TcpConnection.h.

Referenced by handle_close(), is_connector(), open(), and transfer().

◆ link_

TcpDataLink_rch OpenDDS::DCPS::TcpConnection::link_

private

Datalink object which is needed for connection lost callback.

Definition at line 174 of file TcpConnection.h.

Referenced by active_open(), active_reconnect_i(), active_reconnect_open(), close(), disconnect(), handle_close(), handle_timeout(), notify_connection_lost(), on_active_connection_established(), passive_reconnect_i(), receive_strategy(), send_strategy(), set_datalink(), tear_link(), and transfer().

◆ local_address_

ACE_INET_Addr OpenDDS::DCPS::TcpConnection::local_address_

private

Local address.

Definition at line 168 of file TcpConnection.h.

Referenced by active_open(), active_reconnect_i(), handle_setup_input(), handle_timeout(), passive_open(), and transfer().

◆ passive_setup_

bool OpenDDS::DCPS::TcpConnection::passive_setup_

private

Definition at line 188 of file TcpConnection.h.

Referenced by handle_input(), handle_setup_input(), and passive_open().

◆ passive_setup_buffer_

ACE_Message_Block OpenDDS::DCPS::TcpConnection::passive_setup_buffer_

private

Definition at line 189 of file TcpConnection.h.

Referenced by handle_setup_input(), and passive_open().

◆ reconnect_lock_

LockType OpenDDS::DCPS::TcpConnection::reconnect_lock_

private

Lock to synchronize state between reactor and non-reactor threads.

Definition at line 159 of file TcpConnection.h.

Referenced by active_reconnect_i(), active_reconnect_open(), handle_close(), handle_timeout(), notify_lost_on_backpressure_timeout(), set_datalink(), shutdown(), and transfer().

◆ reconnect_state_

ReconnectState OpenDDS::DCPS::TcpConnection::reconnect_state_

private

The state indicates each step of the reconnecting.

Definition at line 180 of file TcpConnection.h.

Referenced by active_reconnect_i(), active_reconnect_open(), close(), handle_stop_reconnecting(), handle_timeout(), notify_lost_on_backpressure_timeout(), open(), passive_reconnect_i(), reconnect_state_string(), and transfer().

◆ remote_address_

ACE_INET_Addr OpenDDS::DCPS::TcpConnection::remote_address_

private

Remote address.

Definition at line 165 of file TcpConnection.h.

Referenced by active_open(), active_reconnect_i(), active_reconnect_open(), close(), get_remote_address(), handle_close(), handle_setup_input(), handle_stop_reconnecting(), handle_timeout(), and transfer().

◆ shutdown_

bool OpenDDS::DCPS::TcpConnection::shutdown_

private

shutdown flag

Definition at line 186 of file TcpConnection.h.

Referenced by active_reconnect_i(), active_reconnect_open(), passive_reconnect_i(), shutdown(), and transfer().

◆ tcp_config_

WeakRcHandle<TcpInst> OpenDDS::DCPS::TcpConnection::tcp_config_

private

The configuration used by this connection.

Definition at line 171 of file TcpConnection.h.

Referenced by active_reconnect_i(), close(), config_name(), handle_stop_reconnecting(), on_active_connection_established(), passive_open(), passive_reconnect_i(), and transfer().

◆ transport_during_setup_

TcpTransport* OpenDDS::DCPS::TcpConnection::transport_during_setup_

private

Definition at line 190 of file TcpConnection.h.

Referenced by handle_setup_input(), and passive_open().

◆ transport_priority_

Priority OpenDDS::DCPS::TcpConnection::transport_priority_

private

TRANSPORT_PRIORITY.value policy value.

Definition at line 183 of file TcpConnection.h.

Referenced by active_open(), handle_setup_input(), handle_timeout(), on_active_connection_established(), and transport_priority().

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

Public Types

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ GuardType

◆ LockType

Member Enumeration Documentation

◆ ReconnectState

Constructor & Destructor Documentation

◆ TcpConnection() [1/2]

◆ TcpConnection() [2/2]

◆ ~TcpConnection()

Member Function Documentation

◆ active_open()

◆ active_reconnect_i()

◆ active_reconnect_open()

◆ add_reference()

◆ close()

◆ config_name()

◆ disconnect()

◆ get_remote_address()

◆ handle_close()

◆ handle_input()

◆ handle_output()

◆ handle_setup_input()

◆ handle_stop_reconnecting()

◆ handle_timeout()

◆ id()

◆ impl()

◆ is_connector()

◆ notify_connection_lost()

◆ notify_lost_on_backpressure_timeout()

◆ on_active_connection_established()

◆ open()

◆ passive_open()

◆ passive_reconnect_i()

◆ receive_strategy()

◆ reconnect_state_string()

◆ relink_from_recv()

◆ relink_from_send()

◆ remove_reference()

◆ send_strategy()

◆ set_datalink()

◆ set_sock_options()

◆ shutdown()

◆ tear_link()

◆ transfer()

◆ transport_priority() [1/2]

◆ transport_priority() [2/2]

Member Data Documentation

◆ conn_retry_counter_

◆ id_

◆ impl_

◆ is_connector_

◆ link_

◆ local_address_

◆ passive_setup_

◆ passive_setup_buffer_

◆ reconnect_lock_

◆ reconnect_state_

◆ remote_address_

◆ shutdown_

◆ tcp_config_

◆ transport_during_setup_

◆ transport_priority_