OpenDDS::DCPS::TcpConnection Class Reference

#include <TcpConnection.h>

Inheritance diagram for OpenDDS::DCPS::TcpConnection:

[legend]

Collaboration diagram for OpenDDS::DCPS::TcpConnection:

[legend]

List of all members.

Public Types
enum	ReconnectState {   INIT_STATE, LOST_STATE, RECONNECTED_STATE, PASSIVE_WAITING_STATE,   PASSIVE_TIMEOUT_CALLED_STATE }
	States are used during reconnecting. More...
Public Member Functions
	TcpConnection ()
	Passive side constructor (acceptor).
	TcpConnection (const ACE_INET_Addr &remote_address, Priority priority, const TcpInst &config)
	Active side constructor (connector).
virtual	~TcpConnection ()
std::size_t &	id ()
int	active_open ()
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.
virtual int	handle_output (ACE_HANDLE)
	Handle back pressure when sending.
virtual int	close (u_long)
virtual int	handle_close (ACE_HANDLE, ACE_Reactor_Mask)
void	set_sock_options (const TcpInst *tcp_config)
int	reconnect (bool on_new_association=false)
bool	is_connector () const
bool	is_connected () const
	Return true if connection is connected.
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.
void	relink_from_recv (bool do_suspend)
	Reconnect initiated by receive strategy.
bool	tear_link ()
void	shutdown ()
Priority &	transport_priority ()
	Access TRANSPORT_PRIORITY.value policy value if set.
Priority	transport_priority () const
virtual ACE_Event_Handler::Reference_Count	add_reference ()
virtual ACE_Event_Handler::Reference_Count	remove_reference ()
Private Types
typedef ACE_SYNCH_MUTEX	LockType
typedef ACE_Guard< LockType >	GuardType
Private Member Functions
int	active_establishment (bool initiate_connect=true)
int	active_reconnect_i ()
int	passive_reconnect_i ()
int	active_reconnect_on_new_association ()
int	handle_setup_input (ACE_HANDLE h)
const std::string &	config_name () const
void	spawn_reconnect_thread ()
OPENDDS_STRING	reconnect_state_string () const
	Get name of the current reconnect state as a string.
Static Private Member Functions
static ACE_THR_FUNC_RETURN	reconnect_thread_fun (void *conn)
Private Attributes
LockType	reconnect_lock_
ACE_Atomic_Op< ACE_SYNCH_MUTEX, bool >	connected_
bool	is_connector_
	Flag indicate this connection object is the connector or acceptor.
ACE_INET_Addr	remote_address_
	Remote address.
ACE_INET_Addr	local_address_
	Local address.
const TcpInst *	tcp_config_
	The configuration used by this connection.
TcpDataLink_rch	link_
	Datalink object which is needed for connection lost callback.
int	passive_reconnect_timer_id_
ReconnectState	reconnect_state_
	The state indicates each step of the reconnecting.
ACE_Time_Value	last_reconnect_attempted_
	Last time the connection is re-established.
Priority	transport_priority_
	TRANSPORT_PRIORITY.value policy value.
bool	shutdown_
	shutdown flag
bool	passive_setup_
ACE_Message_Block	passive_setup_buffer_
TcpTransport *	transport_during_setup_
std::size_t	id_
	Small unique identifying value.
ACE_thread_t	reconnect_thread_

---

Detailed Description

Definition at line 36 of file TcpConnection.h.

---

Member Typedef Documentation

typedef ACE_Guard<LockType> OpenDDS::DCPS::TcpConnection::GuardType [private]

Definition at line 159 of file TcpConnection.h.

typedef ACE_SYNCH_MUTEX OpenDDS::DCPS::TcpConnection::LockType [private]

Definition at line 158 of file TcpConnection.h.

---

Member Enumeration Documentation

enum OpenDDS::DCPS::TcpConnection::ReconnectState

States are used during reconnecting.

Enumerator:

INIT_STATE
LOST_STATE
RECONNECTED_STATE
PASSIVE_WAITING_STATE
PASSIVE_TIMEOUT_CALLED_STATE

Definition at line 42 of file TcpConnection.h.

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

---

Constructor & Destructor Documentation

OpenDDS::DCPS::TcpConnection::TcpConnection

(

)

Passive side constructor (acceptor).

Definition at line 44 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().

  : connected_(false)
  , is_connector_(false)
  , tcp_config_(0)
  , passive_reconnect_timer_id_(-1)
  , reconnect_state_(INIT_STATE)
  , last_reconnect_attempted_(ACE_Time_Value::zero)
  , 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)
  , reconnect_thread_(0)
{
  DBG_ENTRY_LVL("TcpConnection","TcpConnection",6);

  this->reference_counting_policy().value(ACE_Event_Handler::Reference_Counting_Policy::ENABLED);
}

Here is the call graph for this function:

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

Active side constructor (connector).

Definition at line 64 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().

  : connected_(false)
  , is_connector_(true)
  , remote_address_(remote_address)
  , local_address_(config.local_address())
  , tcp_config_(&config)
  , passive_reconnect_timer_id_(-1)
  , reconnect_state_(INIT_STATE)
  , last_reconnect_attempted_(ACE_Time_Value::zero)
  , transport_priority_(priority)
  , shutdown_(false)
  , passive_setup_(false)
  , transport_during_setup_(0)
  , id_(0)
  , reconnect_thread_(0)
{
  DBG_ENTRY_LVL("TcpConnection","TcpConnection",6);
  this->reference_counting_policy().value(ACE_Event_Handler::Reference_Counting_Policy::ENABLED);

}

Here is the call graph for this function:

OpenDDS::DCPS::TcpConnection::~TcpConnection

(

)

[virtual]

Definition at line 86 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, ACE_Thread_Manager::instance(), ACE_Thread_Manager::join(), reconnect_thread_, ACE_OS::thr_equal(), and ACE_OS::thr_self().

{
  DBG_ENTRY_LVL("TcpConnection","~TcpConnection",6);
  if (reconnect_thread_ &&
    // This is for Windows, where join doesn't check if the thread is the same
    // and the thread will hang itself if it tries to join itself.
    !ACE_OS::thr_equal(ACE_OS::thr_self(), reconnect_thread_)
  ) {
    ACE_Thread_Manager::instance()->join(reconnect_thread_);
  }
}

Here is the call graph for this function:

---

Member Function Documentation

int OpenDDS::DCPS::TcpConnection::active_establishment

(

bool

initiate_connect = true

)

[private]

Attempt an active connection establishment to the remote address. The local address is sent to the remote (passive) side to identify ourselves to the remote side. Note this method is not thread protected. The caller need acquire the reconnect_lock_ before calling this function.

Definition at line 438 of file TcpConnection.cpp.

References ACE_TEXT(), OpenDDS::DCPS::DirectPriorityMapper::codepoint(), ACE_SOCK_Connector::connect(), connected_, DBG_ENTRY_LVL, OpenDDS::DCPS::TcpInst::dump_to_str(), ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), OpenDDS::DCPS::TcpInst::get_public_address(), is_connector_, len, link_, LM_DEBUG, LM_ERROR, local_address_, ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >::peer(), remote_address_, send_n(), set_sock_options(), shutdown_, tcp_config_, transport_priority_, and VDBG.

Referenced by active_open(), active_reconnect_i(), and active_reconnect_on_new_association().

{
  DBG_ENTRY_LVL("TcpConnection","active_establishment",6);

  // Safety check - This should not happen since is_connector_ defaults to
  // true and the role in a connection connector is not changed when reconnecting.
  if (this->is_connector_ == false) {
    ACE_ERROR_RETURN((LM_ERROR,
                      "(%P|%t) ERROR: Failed to connect because it's previously an acceptor.\n"),
                     -1);
  }

  if (this->shutdown_)
    return -1;

  // Now use a connector object to establish the connection.
  ACE_SOCK_Connector connector;

  if (initiate_connect && connector.connect(this->peer(), remote_address_) != 0) {

    ACE_DEBUG((LM_DEBUG,
                      ACE_TEXT("(%P|%t) DBG: Failed to connect. this->shutdown_=%d %p\n%C"),
                      int(this->shutdown_), ACE_TEXT("connect"), this->tcp_config_->dump_to_str().c_str()));
                      return -1;

  } else {
    this->connected_ = true;
    const std::string remote_host = this->remote_address_.get_host_addr();
    VDBG((LM_DEBUG, "(%P|%t) DBG:   active_establishment(%C:%d->%C:%d)\n",
          this->local_address_.get_host_addr(), this->local_address_.get_port_number(),
          remote_host.c_str(), this->remote_address_.get_port_number()));
  }

  // Set the DiffServ codepoint according to the priority value.
  DirectPriorityMapper mapper(this->transport_priority_);
  this->link_->set_dscp_codepoint(mapper.codepoint(), this->peer());

  set_sock_options(tcp_config_);

  // 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.
  std::string address = tcp_config_->get_public_address();
  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) {
    // TBD later - Anything we are supposed to do to close the connection.
    ACE_ERROR_RETURN((LM_ERROR,
                      "(%P|%t) ERROR: Unable to send address string length to "
                      "the passive side to complete the active connection "
                      "establishment.\n"),
                     -1);
  }

  if (this->peer().send_n(address.c_str(), len)  == -1) {
    // TBD later - Anything we are supposed to do to close the connection.
    ACE_ERROR_RETURN((LM_ERROR,
                      "(%P|%t) ERROR: Unable to send our address to "
                      "the passive side to complete the active connection "
                      "establishment.\n"),
                     -1);
  }

  ACE_UINT32 npriority = htonl(this->transport_priority_);

  if (this->peer().send_n(&npriority, sizeof(ACE_UINT32)) == -1) {
    // TBD later - Anything we are supposed to do to close the connection.
    ACE_ERROR_RETURN((LM_ERROR,
                      "(%P|%t) ERROR: Unable to send publication priority to "
                      "the passive side to complete the active connection "
                      "establishment.\n"),
                     -1);
  }

  return 0;
}

Here is the call graph for this function:

Here is the caller graph for this function:

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 576 of file TcpConnection.cpp.

References active_establishment(), connected_, DBG_ENTRY_LVL, reconnect_lock_, shutdown_, and ACE_Atomic_Op< ACE_LOCK, TYPE >::value().

Referenced by open().

{
  DBG_ENTRY_LVL("TcpConnection","active_open",6);

  GuardType guard(reconnect_lock_);
  if (this->shutdown_)
    return -1;

  if (connected_.value()) {
    return 0;
  }

  return active_establishment(false /* !initiate_connect */);
}

Here is the call graph for this function:

Here is the caller graph for this function:

int OpenDDS::DCPS::TcpConnection::active_reconnect_i

(

)

[private]

Definition at line 668 of file TcpConnection.cpp.

References ACE_TEXT(), active_establishment(), config_name(), OpenDDS::DCPS::TcpInst::conn_retry_attempts_, OpenDDS::DCPS::TcpInst::conn_retry_backoff_multiplier_, OpenDDS::DCPS::TcpInst::conn_retry_initial_delay_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, disconnect(), OpenDDS::DCPS::DataLink::DISCONNECTED, ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), ACE_Task< ACE_NULL_SYNCH >::gettimeofday(), ACE_OS::gettimeofday(), INIT_STATE, last_reconnect_attempted_, link_, LM_DEBUG, LM_ERROR, local_address_, OpenDDS::DCPS::DataLink::LOST, LOST_STATE, ACE_Event_Handler::READ_MASK, receive_strategy(), reconnect_delay(), reconnect_lock_, reconnect_state_, reconnect_state_string(), OpenDDS::DCPS::DataLink::RECONNECTED, RECONNECTED_STATE, remote_address_, send_strategy(), shutdown_, ACE_OS::sleep(), tcp_config_, and VDBG.

Referenced by reconnect().

{
  DBG_ENTRY_LVL("TcpConnection","active_reconnect_i",6);

  GuardType guard(this->reconnect_lock_);
  if (this->shutdown_)
    return -1;

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "active_reconnect_i(%C:%d->%C:%d) reconnect_state = %C\n",
        this->remote_address_.get_host_addr(), this->remote_address_.get_port_number(),
        this->local_address_.get_host_addr(), this->local_address_.get_port_number(),
        this->reconnect_state_string().c_str()));
  if (DCPS_debug_level >= 1) {
    ACE_DEBUG((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::"
          "active_reconnect_i(%C:%d->%C:%d) reconnect_state = %C\n",
          this->remote_address_.get_host_addr(), this->remote_address_.get_port_number(),
          this->local_address_.get_host_addr(), this->local_address_.get_port_number(),
          this->reconnect_state_string().c_str()));
  }
  // We need reset the state to INIT_STATE if we are previously reconnected.
  // This would allow re-establishing connection after the re-established
  // connection lost again.
  if (ACE_OS::gettimeofday() - this->last_reconnect_attempted_ > reconnect_delay
      && this->reconnect_state_ == RECONNECTED_STATE) {
    VDBG((LM_DEBUG, "(%P|%t) DBG:   "
          "We are in RECONNECTED_STATE and now flip reconnect state to INIT_STATE.\n"));
    this->reconnect_state_ = INIT_STATE;
  }

  if (this->reconnect_state_ == INIT_STATE) {
    // Suspend send once.
    TcpSendStrategy_rch send_strategy = this->send_strategy();
    if (send_strategy)
      send_strategy->suspend_send();
    this->reconnect_lock_.release();
    this->disconnect();
    this->reconnect_lock_.acquire();

    if (this->shutdown_)
      return -1;

    if (this->tcp_config_->conn_retry_attempts_ > 0) {
      this->link_->notify(DataLink::DISCONNECTED);
    }

    // else the conn_retry_attempts is 0 then we do not need this extra
    // notify_disconnected() since the user application should get the
    // notify_lost() without delay.

    double retry_delay_msec = this->tcp_config_->conn_retry_initial_delay_;
    int ret = -1;

    for (int i = 0; i < this->tcp_config_->conn_retry_attempts_; ++i) {
      ret = this->active_establishment();

      if (this->shutdown_)
        break;

      if (ret == -1) {
        ACE_Time_Value delay_tv(((int)retry_delay_msec)/1000,
                                ((int)retry_delay_msec)%1000*1000);
        ACE_OS::sleep(delay_tv);
        retry_delay_msec *= this->tcp_config_->conn_retry_backoff_multiplier_;

      } else {
        break;
      }
    }

    if (ret == -1) {
      if (this->tcp_config_->conn_retry_attempts_ > 0) {
        ACE_DEBUG((LM_DEBUG, "(%P|%t) we tried and failed to re-establish connection on transport: %C to %C:%d.\n",
                   this->config_name().c_str(),
                   this->remote_address_.get_host_addr(),
                   this->remote_address_.get_port_number()));

      } else {
        ACE_DEBUG((LM_DEBUG, "(%P|%t) we did not try to re-establish connection on transport: %C to %C:%d.\n",
                   this->config_name().c_str(),
                   this->remote_address_.get_host_addr(),
                   this->remote_address_.get_port_number()));
      }

      this->reconnect_state_ = LOST_STATE;
      this->link_->notify(DataLink::LOST);
      if (send_strategy)
        send_strategy->terminate_send();

    } else {
      ACE_DEBUG((LM_DEBUG, "(%P|%t) re-established connection on transport: %C to %C:%d.\n",
                 this->config_name().c_str(),
                 this->remote_address_.get_host_addr(),
                 this->remote_address_.get_port_number()));
      TcpReceiveStrategy_rch receive_strategy = this->receive_strategy();
      if (receive_strategy->get_reactor()->register_handler(this, ACE_Event_Handler::READ_MASK) == -1) {
        ACE_ERROR_RETURN((LM_ERROR,
                          "(%P|%t) ERROR: OpenDDS::DCPS::TcpConnection::active_reconnect_i() can't register "
                          "with reactor %X %p\n", this, ACE_TEXT("register_handler")),
                         -1);
      }
      this->reconnect_state_ = RECONNECTED_STATE;
      this->link_->notify(DataLink::RECONNECTED);
      send_strategy->resume_send();
    }

    this->last_reconnect_attempted_ = ACE_OS::gettimeofday();
  }

  return this->reconnect_state_ == LOST_STATE ? -1 : 0;
}

Here is the call graph for this function:

Here is the caller graph for this function:

int OpenDDS::DCPS::TcpConnection::active_reconnect_on_new_association

(

)

[private]

Definition at line 592 of file TcpConnection.cpp.

References active_establishment(), connected_, DBG_ENTRY_LVL, INIT_STATE, reconnect_lock_, reconnect_state_, send_strategy(), and shutdown_.

Referenced by reconnect().

{
  DBG_ENTRY_LVL("TcpConnection","active_reconnect_on_new_association",6);
  GuardType guard(this->reconnect_lock_);
  if (this->shutdown_)
    return -1;

  if (this->connected_ == true)
    return 0;

  else if (this->active_establishment() == 0) {
    this->reconnect_state_ = INIT_STATE;
    TcpSendStrategy_rch send_strategy = this->send_strategy();
    if (send_strategy)
      send_strategy->resume_send();
    return 0;
  }

  return -1;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ACE_Event_Handler::Reference_Count OpenDDS::DCPS::TcpConnection::add_reference

(

void

)

[virtual]

Reimplemented from ACE_Event_Handler.

Definition at line 999 of file TcpConnection.cpp.

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

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

Here is the call graph for this function:

int OpenDDS::DCPS::TcpConnection::close

(

u_long

)

[virtual]

Reimplemented from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >.

Definition at line 331 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, disconnect(), and send_strategy().

{
  DBG_ENTRY_LVL("TcpConnection","close",6);

  // TBD SOON - Find out exactly when close() is called.
  //            I have no clue when and who might call this.
  TcpSendStrategy_rch send_strategy = this->send_strategy();
  if (send_strategy)
    send_strategy->terminate_send();

  this->disconnect();

  return 0;
}

Here is the call graph for this function:

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

(

)

const [private]

Definition at line 347 of file TcpConnection.cpp.

References link_.

Referenced by active_reconnect_i(), handle_close(), handle_timeout(), and reconnect().

{
  return this->link_->impl().config().name();
}

Here is the caller graph for this function:

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 111 of file TcpConnection.cpp.

References connected_, DBG_ENTRY_LVL, ACE_Event_Handler::DONT_CALL, link_, ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >::peer(), ACE_Event_Handler::READ_MASK, and receive_strategy().

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

{
  DBG_ENTRY_LVL("TcpConnection","disconnect",6);
  this->connected_ = false;
  TcpReceiveStrategy_rch receive_strategy = this->receive_strategy();
  if (receive_strategy) {
    receive_strategy->get_reactor()->remove_handler(this,
                                                    ACE_Event_Handler::READ_MASK | ACE_Event_Handler::DONT_CALL);
  }

  if (this->link_) {
    this->link_->drop_pending_request_acks();
  }

  this->peer().close();

}

Here is the call graph for this function:

Here is the caller graph for this function:

ACE_INLINE ACE_INET_Addr OpenDDS::DCPS::TcpConnection::get_remote_address

(

)

Definition at line 46 of file TcpConnection.inl.

References remote_address_.

{
  return this->remote_address_;
}

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 353 of file TcpConnection.cpp.

References config_name(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, disconnect(), OpenDDS::DCPS::DataLink::DISCONNECTED, ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), link_, LM_DEBUG, receive_strategy(), remote_address_, send_strategy(), and spawn_reconnect_thread().

{
  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:%d.\n",
               this->config_name().c_str(),
               this->remote_address_.get_host_addr(),
               this->remote_address_.get_port_number()));
  }

  TcpReceiveStrategy_rch receive_strategy = this->receive_strategy();
  bool graceful = receive_strategy && receive_strategy->gracefully_disconnected();

  TcpSendStrategy_rch send_strategy = this->send_strategy();
  if (send_strategy) {
    if (graceful) {
      send_strategy->terminate_send();
    } else {
      send_strategy->suspend_send();
    }
  }

  this->disconnect();

  if (graceful) {
    this->link_->notify(DataLink::DISCONNECTED);
  } else {
    this->spawn_reconnect_thread();
  }

  return 0;
}

Here is the call graph for this function:

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 288 of file TcpConnection.cpp.

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

{
  DBG_ENTRY_LVL("TcpConnection","handle_input",6);

  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);
}

Here is the call graph for this function:

int OpenDDS::DCPS::TcpConnection::handle_output

(

ACE_HANDLE

)

[virtual]

Handle back pressure when sending.

Reimplemented from ACE_Event_Handler.

Definition at line 304 of file TcpConnection.cpp.

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

{
  DBG_ENTRY_LVL("TcpConnection","handle_output",6);
  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;
}

Here is the call graph for this function:

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 219 of file TcpConnection.cpp.

References ACE_Message_Block::base(), connected_, ACE_Event_Handler::DONT_CALL, ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), 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().

{
  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));
    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 NetworkAddress network_order_address(bufstr);
      network_order_address.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:%d->%C:%d, priority==%d, reconnect_state = %C\n", this,
            remote_address_.get_host_addr(), remote_address_.get_port_number(),
            local_address_.get_host_addr(), local_address_.get_port_number(),
            transport_priority_, reconnect_state_string().c_str()));

      // 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));
      connected_ = true;

      return 0;
    }
  }

  passive_setup_buffer_.rd_ptr(passive_setup_buffer_.base());

  return 0;
}

Here is the call graph for this function:

Here is the caller graph for this function:

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 783 of file TcpConnection.cpp.

References ACE_TEXT(), config_name(), DBG_ENTRY_LVL, ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), link_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::DataLink::LOST, LOST_STATE, PASSIVE_TIMEOUT_CALLED_STATE, PASSIVE_WAITING_STATE, reconnect_lock_, reconnect_state_, RECONNECTED_STATE, remote_address_, send_strategy(), and tear_link().

{
  DBG_ENTRY_LVL("TcpConnection","handle_timeout",6);

  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:%d.\n",
               this->config_name().c_str(),
               this->remote_address_.get_host_addr(),
               this->remote_address_.get_port_number()));

    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->link_->notify(DataLink::LOST);

    // The handle_timeout may be called after the connection is re-established
    // and the send strategy of this old connection is reset to nil.
    TcpSendStrategy_rch send_strategy = this->send_strategy();
    if (send_strategy)
      send_strategy->terminate_send();

    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:%d.\n",
               this->config_name().c_str(),
               this->remote_address_.get_host_addr(),
               this->remote_address_.get_port_number()));
    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;
}

Here is the call graph for this function:

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL ACE_INLINE std::size_t & OpenDDS::DCPS::TcpConnection::id

(

void

)

Definition at line 20 of file TcpConnection.inl.

References id_.

{
  return id_;
}

ACE_INLINE bool OpenDDS::DCPS::TcpConnection::is_connected

(

void

)

const

Return true if connection is connected.

Definition at line 33 of file TcpConnection.inl.

References connected_, and ACE_Atomic_Op< ACE_LOCK, TYPE >::value().

{
  return this->connected_.value();
}

Here is the call graph for this function:

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 27 of file TcpConnection.inl.

References is_connector_.

{
  return this->is_connector_;
}

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 927 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, disconnect(), INIT_STATE, link_, OpenDDS::DCPS::DataLink::LOST, LOST_STATE, reconnect_lock_, reconnect_state_, and send_strategy().

{
  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->link_->notify(DataLink::LOST);

    TcpSendStrategy_rch send_strategy = this->send_strategy();
    if (send_strategy)
      send_strategy->terminate_send();
  }

}

Here is the call graph for this function:

int OpenDDS::DCPS::TcpConnection::open

(

void *

arg

)

[virtual]

Reimplemented from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >.

Definition at line 130 of file TcpConnection.cpp.

References ACE_TEXT(), active_open(), OpenDDS::DCPS::TcpTransport::async_connect_failed(), OpenDDS::DCPS::TcpTransport::connect_tcp_datalink(), DBG_ENTRY_LVL, OpenDDS::DCPS::TcpAcceptor::get_configuration(), ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >::get_handle(), is_connector_, link_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::TcpInst::local_address(), local_address_, passive_setup_, passive_setup_buffer_, OpenDDS::DCPS::rchandle_from(), ACE_Event_Handler::reactor(), ACE_Event_Handler::READ_MASK, remote_address_, set_sock_options(), ACE_Message_Block::size(), tcp_config_, OpenDDS::DCPS::TcpAcceptor::transport(), transport_during_setup_, transport_priority_, and VDBG_LVL.

{
  DBG_ENTRY_LVL("TcpConnection","open",6);

  if (is_connector_) {

    VDBG_LVL((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::open active.\n"), 2);
    // Take over the refcount from TcpTransport::connect_datalink().

    TcpTransport& transport = static_cast<TcpTransport&>(link_->impl());

    const bool is_loop(local_address_ == remote_address_);
    const PriorityKey key(transport_priority_, remote_address_,
                          is_loop, false /* !active */);

    int active_open_ = active_open();

    int connect_tcp_datalink_ = transport.connect_tcp_datalink(*link_, rchandle_from(this));

    if (active_open_ == -1 || connect_tcp_datalink_ == -1) {
      // if (active_open() == -1 ||
      //       transport->connect_tcp_datalink(link_, self) == -1) {

      transport.async_connect_failed(key);

      return -1;
    }

    return 0;
  }

  // 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.
  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.
  tcp_config_ =  &acceptor->get_configuration();
  local_address_ = tcp_config_->local_address();

  set_sock_options(tcp_config_);

  // 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;
  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;
}

Here is the call graph for this function:

int OpenDDS::DCPS::TcpConnection::passive_reconnect_i

(

)

[private]

Definition at line 617 of file TcpConnection.cpp.

References ACE_TEXT(), connected_, DBG_ENTRY_LVL, OpenDDS::DCPS::DataLink::DISCONNECTED, INIT_STATE, link_, LM_ERROR, OpenDDS::DCPS::TcpInst::passive_reconnect_duration_, passive_reconnect_timer_id_, PASSIVE_WAITING_STATE, receive_strategy(), reconnect_lock_, reconnect_state_, shutdown_, and tcp_config_.

Referenced by reconnect().

{
  DBG_ENTRY_LVL("TcpConnection","passive_reconnect_i",6);
  GuardType guard(this->reconnect_lock_);
  if (this->shutdown_)
    return -1;

  // The passive_reconnect_timer_id_ is used as flag to allow the timer scheduled just once.
  if (this->reconnect_state_ == INIT_STATE) {
    // Mark the connection lost since the recv/send just failed.
    this->connected_ = false;

    if (this->tcp_config_->passive_reconnect_duration_ == 0)
      return -1;

    ACE_Time_Value timeout(this->tcp_config_->passive_reconnect_duration_/1000,
                           this->tcp_config_->passive_reconnect_duration_%1000 * 1000);
    this->reconnect_state_ = PASSIVE_WAITING_STATE;
    this->link_->notify(DataLink::DISCONNECTED);

    // It is possible that the passive reconnect is called after the new connection
    // is accepted and the receive_strategy of this old connection is reset to nil.
    TcpReceiveStrategy_rch receive_strategy = this->receive_strategy();
    if (this->receive_strategy()) {

      // Give a copy to reactor.
      this->passive_reconnect_timer_id_ = receive_strategy->get_reactor()->schedule_timer(this, 0, timeout);

      if (this->passive_reconnect_timer_id_ == -1) {
        ACE_ERROR_RETURN((LM_ERROR,
                          ACE_TEXT("(%P|%t) ERROR: TcpConnection::passive_reconnect_i")
                          ACE_TEXT(", %p.\n"), ACE_TEXT("schedule_timer")),
                         -1);
      }
    }
  }

  return 0;
}

Here is the call graph for this function:

Here is the caller graph for this function:

OpenDDS::DCPS::TcpReceiveStrategy_rch OpenDDS::DCPS::TcpConnection::receive_strategy

(

)

Definition at line 105 of file TcpConnection.cpp.

References link_.

Referenced by active_reconnect_i(), disconnect(), handle_close(), handle_input(), passive_reconnect_i(), and transfer().

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

Here is the caller graph for this function:

int OpenDDS::DCPS::TcpConnection::reconnect

(

bool

on_new_association = false

)

This function is called to re-establish the connection. If this object is the connector side of the connection then it tries to reconnect to the remote, if it's the acceptor side of the connection then it schedules a timer to check if it passively accepted a connection from remote. The on_new_association true indicates this is called when the connection is previous lost and new association is added. The connector side needs to try to actively reconnect to remote.

Definition at line 528 of file TcpConnection.cpp.

References active_reconnect_i(), active_reconnect_on_new_association(), config_name(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), is_connector_, link_, LM_DEBUG, passive_reconnect_i(), and remote_address_.

Referenced by OpenDDS::DCPS::TcpReconnectTask::execute().

{
  DBG_ENTRY_LVL("TcpConnection","reconnect",6);
  if (DCPS_debug_level >= 1) {
    ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::reconnect initiated on transport: %C to %C:%d.\n",
               this->config_name().c_str(),
               this->remote_address_.get_host_addr(),
               this->remote_address_.get_port_number()));
  }

  if (on_new_association) {
    if (DCPS_debug_level >= 1) {
      ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::reconnect on new association\n"));
    }
    return this->active_reconnect_on_new_association();
  }

  // If on_new_association is false, it's called by the reconnect task.
  // We need make sure if the link release is pending. If does, do
  // not try to reconnect.
  else if (!this->link_->is_release_pending()) {
    if (DCPS_debug_level >= 1) {
      ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::reconnect release not currently pending\n"));
    }
    // Try to reconnect if it's connector previously.
    if (this->is_connector_ && this->active_reconnect_i() == -1) {
      if (DCPS_debug_level >= 1) {
        ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::reconnect is connector but active_reconnect_i failed\n"));
      }
      return -1;
    }

    // Schedule a timer to see if a incoming connection is accepted when timeout.
    else if (!this->is_connector_ && this->passive_reconnect_i() == -1) {
      if (DCPS_debug_level >= 1) {
        ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::reconnect is acceptor but passive_reconnect_i failed\n"));
      }
      return -1;
    }

  }
  if (DCPS_debug_level >= 1) {
    ACE_DEBUG((LM_DEBUG, "(%P|%t) TcpConnection::reconnect returning 0\n"));
  }
  return 0;
}

Here is the call graph for this function:

Here is the caller graph for this function:

OPENDDS_STRING OpenDDS::DCPS::TcpConnection::reconnect_state_string

(

)

const [private]

Get name of the current reconnect state as a string.

Definition at line 1059 of file TcpConnection.cpp.

References ACE_TEXT(), INIT_STATE, LM_ERROR, LOST_STATE, OPENDDS_STRING, PASSIVE_TIMEOUT_CALLED_STATE, PASSIVE_WAITING_STATE, reconnect_state_, RECONNECTED_STATE, and OpenDDS::DCPS::to_dds_string().

Referenced by active_reconnect_i(), and handle_setup_input().

{
  switch (reconnect_state_) {
  case INIT_STATE:
    return "INIT_STATE";
  case LOST_STATE:
    return "LOST_STATE";
  case RECONNECTED_STATE:
    return "RECONENCTED_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(
      "OpenDDS::DCPS::TcpConnection::reconnect_state_string(): "
      "%d is either completely invalid or at least not defined in this function.\n"),
      reconnect_state_
    ));
    return OPENDDS_STRING("(Unknown Reconnect State: ")
      + to_dds_string(reconnect_state_) + ")";
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

ACE_THR_FUNC_RETURN OpenDDS::DCPS::TcpConnection::reconnect_thread_fun

(

void *

conn

)

[static, private]

Definition at line 1035 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, SIG_SETMASK, ACE_OS::sigfillset(), sigset_t, and ACE_OS::thr_sigsetmask().

Referenced by spawn_reconnect_thread().

{
  DBG_ENTRY_LVL("TcpConnection","reconnect_thread_fun",6);

  // Ignore all signals to avoid
  //     ERROR: <something descriptive> Interrupted system call
  // The main thread will handle signals.
  sigset_t set;
  ACE_OS::sigfillset(&set);
  ACE_OS::thr_sigsetmask(SIG_SETMASK, &set, NULL);

  // Make sure the associated transport_config outlives the connection object.
  RcHandle<TransportInst> transport_config;
  TcpConnection_rch connection(static_cast<TcpConnection*>(arg), keep_count());
  transport_config = RcHandle<TransportInst>(&connection->link_->impl().config(), keep_count());

  if (connection->reconnect() == -1) {
    connection->tear_link();
  }

  return 0;
}

Here is the call graph for this function:

Here is the caller graph for this function:

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 972 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();
}

Here is the call graph for this function:

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 957 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, send_strategy(), and spawn_reconnect_thread().

{
  DBG_ENTRY_LVL("TcpConnection","relink_from_send",6);

  TcpSendStrategy_rch send_strategy = this->send_strategy();
  if (do_suspend && send_strategy)
    send_strategy->suspend_send();

  this->spawn_reconnect_thread();
}

Here is the call graph for this function:

ACE_Event_Handler::Reference_Count OpenDDS::DCPS::TcpConnection::remove_reference

(

void

)

[virtual]

Reimplemented from ACE_Event_Handler.

Definition at line 1006 of file TcpConnection.cpp.

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

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

Here is the call graph for this function:

OpenDDS::DCPS::TcpSendStrategy_rch OpenDDS::DCPS::TcpConnection::send_strategy

(

)

Definition at line 99 of file TcpConnection.cpp.

References link_.

Referenced by active_reconnect_i(), active_reconnect_on_new_association(), close(), handle_close(), handle_output(), handle_timeout(), notify_lost_on_backpressure_timeout(), relink_from_recv(), and relink_from_send().

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

Here is the caller graph for this function:

ACE_INLINE void OpenDDS::DCPS::TcpConnection::set_datalink

(

const TcpDataLink_rch &

link

)

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

Definition at line 39 of file TcpConnection.inl.

References link_.

{
  // Keep a "copy" of the reference to the data link for ourselves.
  this->link_ = link;
}

void OpenDDS::DCPS::TcpConnection::set_sock_options

(

const TcpInst *

tcp_config

)

Definition at line 388 of file TcpConnection.cpp.

References OpenDDS::DCPS::TcpInst::enable_nagle_algorithm_, LM_ERROR, and ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >::peer().

Referenced by active_establishment(), and 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 */
}

Here is the call graph for this function:

Here is the caller graph for this function:

void OpenDDS::DCPS::TcpConnection::shutdown

(

void

)

Reimplemented from ACE_Svc_Handler< ACE_SOCK_STREAM, ACE_NULL_SYNCH >.

Definition at line 989 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, reconnect_lock_, and shutdown_.

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

}

void OpenDDS::DCPS::TcpConnection::spawn_reconnect_thread

(

)

[private]

Definition at line 1013 of file TcpConnection.cpp.

References OpenDDS::DCPS::RcObject::_add_ref(), OpenDDS::DCPS::RcObject::_remove_ref(), DBG_ENTRY_LVL, ACE_Thread_Manager::instance(), link_, reconnect_lock_, reconnect_thread_, reconnect_thread_fun(), and shutdown_.

Referenced by handle_close(), and relink_from_send().

{
  DBG_ENTRY_LVL("TcpConnection","spawn_reconnect_thread",6);
  GuardType guard(this->reconnect_lock_);
  if (!shutdown_) {
    // Make sure the associated transport_config outlives the connection object.
    TransportInst& transport_config = this->link_->impl().config();
    transport_config._add_ref();
    // add the reference count to be picked up from the new thread
    this->_add_ref();
    if (ACE_Thread_Manager::instance()->spawn(&reconnect_thread_fun,
                                              this,
                                              THR_NEW_LWP|THR_JOINABLE|THR_INHERIT_SCHED,
                                              &reconnect_thread_) == -1){
      // we need to decrement the reference count when thread creation fails.
      this->_remove_ref();
      transport_config._remove_ref();
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool 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 981 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, and link_.

Referenced by OpenDDS::DCPS::TcpReconnectTask::execute(), and handle_timeout().

{
  DBG_ENTRY_LVL("TcpConnection","tear_link",6);

  return this->link_->release_resources();
}

Here is the caller graph for this function:

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 841 of file TcpConnection.cpp.

References ACE_TEXT(), DBG_ENTRY_LVL, ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), INIT_STATE, ACE_Thread_Manager::instance(), is_connector_, ACE_Thread_Manager::join(), link_, LM_DEBUG, LM_ERROR, local_address_, LOST_STATE, passive_reconnect_timer_id_, PASSIVE_TIMEOUT_CALLED_STATE, PASSIVE_WAITING_STATE, receive_strategy(), reconnect_lock_, reconnect_state_, reconnect_thread_, OpenDDS::DCPS::DataLink::RECONNECTED, RECONNECTED_STATE, remote_address_, tcp_config_, and VDBG.

{
  DBG_ENTRY_LVL("TcpConnection","transfer",6);

  GuardType guard(this->reconnect_lock_);

  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: {
    // Cancel the timer since we got new connection.
    TcpReceiveStrategy_rch receive_strategy = this->receive_strategy();
    if (receive_strategy->get_reactor()->cancel_timer(this) == -1) {
      ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: TcpConnection::transfer, ")
                 ACE_TEXT(" %p. \n"), ACE_TEXT("cancel_timer")));

    } else
      passive_reconnect_timer_id_ = -1;

    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")));
  }

  if (reconnect_thread_) {
      ACE_Thread_Manager::instance()->join(reconnect_thread_);
      reconnect_thread_ = 0;
  }
  // connection->receive_strategy_ = this->receive_strategy_;
  // connection->send_strategy_ = this->send_strategy_;
  connection->remote_address_ = this->remote_address_;
  connection->local_address_ = this->local_address_;
  connection->tcp_config_ = this->tcp_config_;
  connection->link_ = this->link_;

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "transfer(%C:%d->%C:%d) passive reconnected. new con %@   "
        " old con %@ \n",
        this->remote_address_.get_host_addr(), this->remote_address_.get_port_number(),
        this->local_address_.get_host_addr(), this->local_address_.get_port_number(),
        connection, this));

  if (notify_reconnect) {
    this->reconnect_state_ = RECONNECTED_STATE;
    this->link_->notify(DataLink::RECONNECTED);
  }

}

Here is the call graph for this function:

ACE_INLINE OpenDDS::DCPS::Priority OpenDDS::DCPS::TcpConnection::transport_priority

(

)

const

Definition at line 60 of file TcpConnection.inl.

References transport_priority_.

{
  return this->transport_priority_;
}

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

(

)

Access TRANSPORT_PRIORITY.value policy value if set.

Definition at line 53 of file TcpConnection.inl.

References transport_priority_.

{
  return this->transport_priority_;
}

---

Member Data Documentation

ACE_Atomic_Op<ACE_SYNCH_MUTEX, bool> OpenDDS::DCPS::TcpConnection::connected_ [private]

Flag indicates if connected or disconnected. It's set to true when actively connecting or passively accepting succeeds and set to false whenever the peer stream is closed.

Definition at line 168 of file TcpConnection.h.

Referenced by active_establishment(), active_open(), active_reconnect_on_new_association(), disconnect(), handle_setup_input(), is_connected(), and passive_reconnect_i().

std::size_t OpenDDS::DCPS::TcpConnection::id_ [private]

Small unique identifying value.

Definition at line 208 of file TcpConnection.h.

Referenced by handle_output(), and id().

bool OpenDDS::DCPS::TcpConnection::is_connector_ [private]

Flag indicate this connection object is the connector or acceptor.

Definition at line 171 of file TcpConnection.h.

Referenced by active_establishment(), is_connector(), open(), reconnect(), and transfer().

ACE_Time_Value OpenDDS::DCPS::TcpConnection::last_reconnect_attempted_ [private]

Last time the connection is re-established.

Definition at line 195 of file TcpConnection.h.

Referenced by active_reconnect_i().

TcpDataLink_rch OpenDDS::DCPS::TcpConnection::link_ [private]

Datalink object which is needed for connection lost callback.

Definition at line 183 of file TcpConnection.h.

Referenced by active_establishment(), active_reconnect_i(), config_name(), disconnect(), handle_close(), handle_timeout(), notify_lost_on_backpressure_timeout(), open(), passive_reconnect_i(), receive_strategy(), reconnect(), send_strategy(), set_datalink(), spawn_reconnect_thread(), tear_link(), and transfer().

ACE_INET_Addr OpenDDS::DCPS::TcpConnection::local_address_ [private]

Local address.

Definition at line 177 of file TcpConnection.h.

Referenced by active_establishment(), active_reconnect_i(), handle_setup_input(), open(), and transfer().

int OpenDDS::DCPS::TcpConnection::passive_reconnect_timer_id_ [private]

The id of the scheduled timer. The timer is scheduled to check if the connection is re-established during the passive_reconnect_duration_. This id controls that the timer is just scheduled once when there are multiple threads detect the lost connection.

Definition at line 189 of file TcpConnection.h.

Referenced by passive_reconnect_i(), and transfer().

bool OpenDDS::DCPS::TcpConnection::passive_setup_ [private]

Definition at line 203 of file TcpConnection.h.

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

ACE_Message_Block OpenDDS::DCPS::TcpConnection::passive_setup_buffer_ [private]

Definition at line 204 of file TcpConnection.h.

Referenced by handle_setup_input(), and open().

LockType OpenDDS::DCPS::TcpConnection::reconnect_lock_ [private]

Lock to avoid the reconnect() called multiple times when both send() and recv() fail.

Definition at line 163 of file TcpConnection.h.

Referenced by active_open(), active_reconnect_i(), active_reconnect_on_new_association(), handle_timeout(), notify_lost_on_backpressure_timeout(), passive_reconnect_i(), shutdown(), spawn_reconnect_thread(), and transfer().

ReconnectState OpenDDS::DCPS::TcpConnection::reconnect_state_ [private]

The state indicates each step of the reconnecting.

Definition at line 192 of file TcpConnection.h.

Referenced by active_reconnect_i(), active_reconnect_on_new_association(), handle_timeout(), notify_lost_on_backpressure_timeout(), passive_reconnect_i(), reconnect_state_string(), and transfer().

ACE_thread_t OpenDDS::DCPS::TcpConnection::reconnect_thread_ [private]

Definition at line 209 of file TcpConnection.h.

Referenced by spawn_reconnect_thread(), transfer(), and ~TcpConnection().

ACE_INET_Addr OpenDDS::DCPS::TcpConnection::remote_address_ [private]

Remote address.

Definition at line 174 of file TcpConnection.h.

Referenced by active_establishment(), active_reconnect_i(), get_remote_address(), handle_close(), handle_setup_input(), handle_timeout(), open(), reconnect(), and transfer().

bool OpenDDS::DCPS::TcpConnection::shutdown_ [private]

shutdown flag

Definition at line 201 of file TcpConnection.h.

Referenced by active_establishment(), active_open(), active_reconnect_i(), active_reconnect_on_new_association(), passive_reconnect_i(), shutdown(), and spawn_reconnect_thread().

const TcpInst* OpenDDS::DCPS::TcpConnection::tcp_config_ [private]

The configuration used by this connection.

Definition at line 180 of file TcpConnection.h.

Referenced by active_establishment(), active_reconnect_i(), open(), passive_reconnect_i(), and transfer().

TcpTransport* OpenDDS::DCPS::TcpConnection::transport_during_setup_ [private]

Definition at line 205 of file TcpConnection.h.

Referenced by handle_setup_input(), and open().

Priority OpenDDS::DCPS::TcpConnection::transport_priority_ [private]

TRANSPORT_PRIORITY.value policy value.

Definition at line 198 of file TcpConnection.h.

Referenced by active_establishment(), handle_setup_input(), open(), and transport_priority().

---

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

• TcpConnection.h
• TcpConnection.cpp
• TcpConnection.inl