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
	INIT_STATE
	LOST_STATE
	RECONNECTED_STATE
	PASSIVE_WAITING_STATE
	PASSIVE_TIMEOUT_CALLED_STATE
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_rch &config)
	Active side constructor (connector).
virtual	~TcpConnection ()
std::size_t &	id ()
int	active_open ()
void	disconnect ()
virtual int	open (void *arg)
void	set_receive_strategy (TcpReceiveStrategy *receive_strategy)
void	remove_receive_strategy ()
void	set_send_strategy (TcpSendStrategy *send_strategy)
void	remove_send_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 (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 (TcpDataLink *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
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)
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.
TcpReceiveStrategy_rch	receive_strategy_
	Reference to the receiving strategy.
TcpSendStrategy_rch	send_strategy_
	Reference to the send strategy.
ACE_INET_Addr	remote_address_
	Remote address.
ACE_INET_Addr	local_address_
	Local address.
TcpInst_rch	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_
TcpReconnectTask	reconnect_task_
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_rch	transport_during_setup_
std::size_t	id_
	Small unique identifying value.

---

Detailed Description

Definition at line 33 of file TcpConnection.h.

---

Member Typedef Documentation

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

Definition at line 154 of file TcpConnection.h.

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

Definition at line 153 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 39 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 41 of file TcpConnection.cpp.

References DBG_ENTRY_LVL.

  : connected_(false)
  , is_connector_(false)
  , passive_reconnect_timer_id_(-1)
  , reconnect_task_(this)
  , 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))
  , id_(0)
{
  DBG_ENTRY_LVL("TcpConnection","TcpConnection",6);

  if (this->reconnect_task_.open()) {
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) ERROR: Reconnect task failed to open : %p\n"),
               ACE_TEXT("open")));
  }

}

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

Active side constructor (connector).

Definition at line 64 of file TcpConnection.cpp.

References DBG_ENTRY_LVL.

  : connected_(false)
  , is_connector_(true)
  , remote_address_(remote_address)
  , local_address_(config->local_address())
  , tcp_config_(config)
  , passive_reconnect_timer_id_(-1)
  , reconnect_task_(this)
  , reconnect_state_(INIT_STATE)
  , last_reconnect_attempted_(ACE_Time_Value::zero)
  , transport_priority_(priority)
  , shutdown_(false)
  , passive_setup_(false)
{
  DBG_ENTRY_LVL("TcpConnection","TcpConnection",6);

  // Open the reconnect task
  if (this->reconnect_task_.open()) {
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) ERROR: Reconnect task failed to open : %p\n"),
               ACE_TEXT("open")));
  }

}

OpenDDS::DCPS::TcpConnection::~TcpConnection

(

)

[virtual]

Definition at line 90 of file TcpConnection.cpp.

References OpenDDS::DCPS::QueueTaskBase< T >::close(), DBG_ENTRY_LVL, link_, reconnect_task_, and OpenDDS::DCPS::Transport_debug_level.

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

  // The Reconnect task belongs to the Connection object.
  // Cleanup before leaving the house.
  this->reconnect_task_.close(1);
  //this->reconnect_task_.wait ();

  if (!this->link_.is_nil()) {
    if (Transport_debug_level > 5) {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) TcpConnection::~TcpConnection: about to notify link[%@] connection deleted\n"),
                 this->link_.in()));
    }
    this->link_->notify_connection_deleted();
  }

}

---

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

References OpenDDS::DCPS::DirectPriorityMapper::codepoint(), connected_, DBG_ENTRY_LVL, OpenDDS::DCPS::RcHandle< T >::in(), link_, local_address_, remote_address_, set_sock_options(), tcp_config_, and VDBG.

Referenced by active_open().

{
  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_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: Failed to connect. %p\n%C"),
                      ACE_TEXT("connect"), this->tcp_config_->dump_to_str().c_str()),
                     -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());

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

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

References active_establishment(), connected_, DBG_ENTRY_LVL, and reconnect_lock_.

Referenced by open().

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

  GuardType guard(reconnect_lock_);

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

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

int OpenDDS::DCPS::TcpConnection::active_reconnect_i

(

)

[private]

Definition at line 680 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, disconnect(), OpenDDS::DCPS::DataLink::DISCONNECTED, INIT_STATE, last_reconnect_attempted_, link_, local_address_, OpenDDS::DCPS::DataLink::LOST, LOST_STATE, reconnect_delay(), reconnect_state_, OpenDDS::DCPS::DataLink::RECONNECTED, RECONNECTED_STATE, remote_address_, send_strategy_, tcp_config_, and VDBG.

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

  GuardType guard(this->reconnect_lock_);

  VDBG((LM_DEBUG, "(%P|%t) DBG:   "
        "active_reconnect_i(%C:%d->%C:%d) reconnect_state = %d\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_));
  if (DCPS_debug_level >= 1) {
    ACE_DEBUG((LM_DEBUG, "(%P|%t) DBG:   TcpConnection::"
          "active_reconnect_i(%C:%d->%C:%d) reconnect_state = %d\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_));
  }
  // 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.
    this->send_strategy_->suspend_send();

    this->disconnect();

    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->link_->get_transport_impl()->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->link_->get_transport_impl()->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);
      this->send_strategy_->terminate_send();

    } else {
      ACE_DEBUG((LM_DEBUG, "(%P|%t) re-established connection on transport: %C to %C:%d.\n",
                 this->link_->get_transport_impl()->config()->name().c_str(),
                 this->remote_address_.get_host_addr(),
                 this->remote_address_.get_port_number()));
      if (this->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);
      this->send_strategy_->resume_send();
    }

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

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

int OpenDDS::DCPS::TcpConnection::active_reconnect_on_new_association

(

)

[private]

Definition at line 607 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, and INIT_STATE.

Referenced by reconnect().

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

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

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

  return -1;
}

int OpenDDS::DCPS::TcpConnection::close

(

u_long

)

[virtual]

Definition at line 355 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, and disconnect().

{
  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.

  if (!this->send_strategy_.is_nil())
    this->send_strategy_->terminate_send();

  this->disconnect();

  return 0;
}

void OpenDDS::DCPS::TcpConnection::disconnect

(

)

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_, and DBG_ENTRY_LVL.

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

{
  DBG_ENTRY_LVL("TcpConnection","disconnect",6);
  this->connected_ = false;

  if (!this->receive_strategy_.is_nil()) {
    this->receive_strategy_->get_reactor()->remove_handler(this,
                                                           ACE_Event_Handler::READ_MASK | ACE_Event_Handler::DONT_CALL);
  }

  this->peer().close();

}

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

(

)

Definition at line 60 of file TcpConnection.inl.

References remote_address_.

{
  return this->remote_address_;
}

int OpenDDS::DCPS::TcpConnection::handle_close	(	ACE_HANDLE	,
		ACE_Reactor_Mask
	)			[virtual]

Definition at line 371 of file TcpConnection.cpp.

References OpenDDS::DCPS::QueueTaskBase< T >::add(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, disconnect(), OpenDDS::DCPS::DataLink::DISCONNECTED, OpenDDS::DCPS::DO_RECONNECT, OpenDDS::DCPS::RcHandle< T >::is_nil(), link_, receive_strategy_, reconnect_task_, remote_address_, and send_strategy_.

{
  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->link_->get_transport_impl()->config()->name().c_str(),
               this->remote_address_.get_host_addr(),
               this->remote_address_.get_port_number()));
  }

  bool graceful = !this->receive_strategy_.is_nil() && this->receive_strategy_->gracefully_disconnected();

  if (!this->send_strategy_.is_nil()) {
    if (graceful) {
      this->send_strategy_->terminate_send();
    } else {
      this->send_strategy_->suspend_send();
    }
  }

  this->disconnect();

  if (graceful) {
    this->link_->notify(DataLink::DISCONNECTED);
  } else {
    ReconnectOpType op = DO_RECONNECT;
    this->reconnect_task_.add(op);
  }

  return 0;
}

int OpenDDS::DCPS::TcpConnection::handle_input

(

ACE_HANDLE

)

[virtual]

We pass this "event" along to the receive_strategy.

Definition at line 312 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, handle_setup_input(), OpenDDS::DCPS::RcHandle< T >::is_nil(), passive_setup_, and receive_strategy_.

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

  if (passive_setup_) {
    return handle_setup_input(fd);
  }

  if (receive_strategy_.is_nil()) {
    return 0;
  }

  return receive_strategy_->handle_dds_input(fd);
}

int OpenDDS::DCPS::TcpConnection::handle_output

(

ACE_HANDLE

)

[virtual]

Handle back pressure when sending.

Definition at line 328 of file TcpConnection.cpp.

References DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, id_, send_strategy_, and OpenDDS::DCPS::ThreadSynchWorker::WORK_OUTCOME_MORE_TO_DO.

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

  if (!this->send_strategy_.is_nil()) {
    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;
}

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

References connected_, local_address_, passive_setup_, passive_setup_buffer_, reconnect_state_, remote_address_, transport_during_setup_, transport_priority_, VDBG, and VDBG_LVL.

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 = %d\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_));

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

      const TcpConnection_rch self(this, false);

      transport_during_setup_->passive_connection(remote_address_, self);
      transport_during_setup_ = 0;
      connected_ = true;

      return 0;
    }
  }

  passive_setup_buffer_.rd_ptr(passive_setup_buffer_.base());

  return 0;
}

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

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

Definition at line 785 of file TcpConnection.cpp.

References OpenDDS::DCPS::RcObject< ACE_SYNCH_MUTEX >::_remove_ref(), DBG_ENTRY_LVL, link_, OpenDDS::DCPS::DataLink::LOST, LOST_STATE, PASSIVE_TIMEOUT_CALLED_STATE, PASSIVE_WAITING_STATE, reconnect_state_, RECONNECTED_STATE, remote_address_, 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->link_->get_transport_impl()->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.
    if (!this->send_strategy_.is_nil())
      this->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->link_->get_transport_impl()->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"), this->reconnect_state_));
    break;
  }

  // Take back the "copy" we gave to reactor when we schedule the timer.
  this->_remove_ref();

  return 0;
}

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

(

)

Definition at line 18 of file TcpConnection.inl.

References id_.

{
  return id_;
}

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

(

)

const

Return true if connection is connected.

Definition at line 46 of file TcpConnection.inl.

References connected_.

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

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

References DBG_ENTRY_LVL, disconnect(), INIT_STATE, link_, OpenDDS::DCPS::DataLink::LOST, LOST_STATE, 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;

      this->disconnect();
    }
  }

  if (notify_lost) {
    this->link_->notify(DataLink::LOST);
    this->send_strategy_->terminate_send();
  }

}

int OpenDDS::DCPS::TcpConnection::open

(

void *

arg

)

[virtual]

Definition at line 150 of file TcpConnection.cpp.

References OpenDDS::DCPS::RcObject< T >::_add_ref(), active_open(), DBG_ENTRY_LVL, OpenDDS::DCPS::TcpAcceptor::get_configuration(), OpenDDS::DCPS::RcHandle< T >::in(), is_connector_, OpenDDS::DCPS::RcHandle< T >::is_nil(), link_, local_address_, passive_setup_, passive_setup_buffer_, remote_address_, set_sock_options(), 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().
    const TcpConnection_rch self(this);
    const TcpTransport_rch transport = link_->get_transport_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_, self);

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

  // Now we need to ask the TcpAcceptor object to provide us with
  // a pointer to the TcpTransport object that "owns" the acceptor.
  TcpTransport_rch transport = acceptor->transport();

  if (transport.is_nil()) {
    // 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);
  }

  TcpInst* tcp_config = acceptor->get_configuration();

  // Keep a "copy" of the reference to TcpInst object
  // for ourselves.
  tcp_config->_add_ref();
  tcp_config_ = tcp_config;
  local_address_ = tcp_config_->local_address();

  set_sock_options(tcp_config_.in());

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

int OpenDDS::DCPS::TcpConnection::passive_reconnect_i

(

)

[private]

Definition at line 628 of file TcpConnection.cpp.

References OpenDDS::DCPS::RcObject< ACE_SYNCH_MUTEX >::_add_ref(), OpenDDS::DCPS::RcObject< ACE_SYNCH_MUTEX >::_remove_ref(), connected_, DBG_ENTRY_LVL, OpenDDS::DCPS::DataLink::DISCONNECTED, INIT_STATE, link_, passive_reconnect_timer_id_, PASSIVE_WAITING_STATE, and reconnect_state_.

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

  // 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.
    if (!this->receive_strategy_.is_nil()) {
      TcpReceiveStrategy* rs
        = dynamic_cast <TcpReceiveStrategy*>(this->receive_strategy_.in());

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

      if (this->passive_reconnect_timer_id_ == -1) {
        this->_remove_ref();
        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;
}

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

References active_reconnect_on_new_association(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, and remote_address_.

{
  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->link_->get_transport_impl()->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;
}

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

References DBG_ENTRY_LVL.

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

  if (do_suspend && !this->send_strategy_.is_nil())
    this->send_strategy_->suspend_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 956 of file TcpConnection.cpp.

References OpenDDS::DCPS::QueueTaskBase< T >::add(), DBG_ENTRY_LVL, OpenDDS::DCPS::DO_RECONNECT, and reconnect_task_.

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

  if (do_suspend && !this->send_strategy_.is_nil())
    this->send_strategy_->suspend_send();

  ReconnectOpType op = DO_RECONNECT;
  this->reconnect_task_.add(op);
}

ACE_INLINE void OpenDDS::DCPS::TcpConnection::remove_receive_strategy

(

)

Definition at line 24 of file TcpConnection.inl.

References DBG_ENTRY_LVL, and receive_strategy_.

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

  this->receive_strategy_ = 0;
}

ACE_INLINE void OpenDDS::DCPS::TcpConnection::remove_send_strategy

(

)

Definition at line 32 of file TcpConnection.inl.

References DBG_ENTRY_LVL, and send_strategy_.

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

  this->send_strategy_ = 0;
}

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

(

TcpDataLink *

link

)

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

Definition at line 52 of file TcpConnection.inl.

References OpenDDS::DCPS::RcObject< T >::_add_ref(), and link_.

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

void OpenDDS::DCPS::TcpConnection::set_receive_strategy

(

TcpReceiveStrategy *

receive_strategy

)

Definition at line 129 of file TcpConnection.cpp.

References OpenDDS::DCPS::RcObject< T >::_add_ref(), and DBG_ENTRY_LVL.

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

  // Make a "copy" for ourselves
  receive_strategy->_add_ref();
  this->receive_strategy_ = receive_strategy;
}

void OpenDDS::DCPS::TcpConnection::set_send_strategy

(

TcpSendStrategy *

send_strategy

)

Give a "copy" of the TcpSendStrategy object to this connection object.

Definition at line 140 of file TcpConnection.cpp.

References OpenDDS::DCPS::RcObject< T >::_add_ref(), and DBG_ENTRY_LVL.

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

  // Make a "copy" for ourselves
  send_strategy->_add_ref();
  this->send_strategy_ = send_strategy;
}

void OpenDDS::DCPS::TcpConnection::set_sock_options

(

TcpInst *

tcp_config

)

Definition at line 405 of file TcpConnection.cpp.

References OpenDDS::DCPS::TcpInst::enable_nagle_algorithm_.

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

void OpenDDS::DCPS::TcpConnection::shutdown

(

)

Definition at line 988 of file TcpConnection.cpp.

References OpenDDS::DCPS::QueueTaskBase< T >::close(), DBG_ENTRY_LVL, reconnect_task_, and shutdown_.

{
  DBG_ENTRY_LVL("TcpConnection","shutdown",6);
  this->shutdown_ = true;

  this->reconnect_task_.close(1);

}

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

References DBG_ENTRY_LVL, and link_.

Referenced by handle_timeout().

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

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

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

References OpenDDS::DCPS::RcObject< ACE_SYNCH_MUTEX >::_remove_ref(), OpenDDS::DCPS::QueueTaskBase< T >::close(), DBG_ENTRY_LVL, OpenDDS::DCPS::TcpReceiveStrategy::get_reactor(), OpenDDS::DCPS::RcHandle< T >::in(), INIT_STATE, is_connector_, link_, local_address_, LOST_STATE, passive_reconnect_timer_id_, PASSIVE_TIMEOUT_CALLED_STATE, PASSIVE_WAITING_STATE, receive_strategy_, reconnect_state_, reconnect_task_, OpenDDS::DCPS::DataLink::RECONNECTED, RECONNECTED_STATE, remote_address_, send_strategy_, 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: {
    TcpReceiveStrategy* rs
      = dynamic_cast <TcpReceiveStrategy*>(this->receive_strategy_.in());

    // Cancel the timer since we got new connection.
    if (rs->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;

    this->_remove_ref();
    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=%d \n"), this->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")));
  }

  this->reconnect_task_.close(1);
  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);
  }

}

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

(

)

const

Definition at line 74 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 67 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 disconneted. It's set to true when actively connecting or passively acepting succeeds and set to false whenever the peer stream is closed.

Definition at line 163 of file TcpConnection.h.

Referenced by active_establishment(), active_open(), 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 215 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 166 of file TcpConnection.h.

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

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

Last time the connection is re-established.

Definition at line 202 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 184 of file TcpConnection.h.

Referenced by active_establishment(), active_reconnect_i(), handle_close(), handle_timeout(), notify_lost_on_backpressure_timeout(), open(), passive_reconnect_i(), set_datalink(), tear_link(), transfer(), and ~TcpConnection().

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

Local address.

Definition at line 178 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 190 of file TcpConnection.h.

Referenced by passive_reconnect_i(), and transfer().

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

Definition at line 210 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 211 of file TcpConnection.h.

Referenced by handle_setup_input(), and open().

TcpReceiveStrategy_rch OpenDDS::DCPS::TcpConnection::receive_strategy_ [private]

Reference to the receiving strategy.

Definition at line 169 of file TcpConnection.h.

Referenced by handle_close(), handle_input(), remove_receive_strategy(), and transfer().

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

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

Definition at line 158 of file TcpConnection.h.

Referenced by active_open().

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

The state indicates each step of the reconnecting.

Definition at line 199 of file TcpConnection.h.

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

TcpReconnectTask OpenDDS::DCPS::TcpConnection::reconnect_task_ [private]

The task to do the reconnecting. TODO: We might need reuse the PerConnectionSynch thread to do the reconnecting or create the reconnect task when we need reconnect.

Definition at line 196 of file TcpConnection.h.

Referenced by handle_close(), relink_from_send(), shutdown(), transfer(), and ~TcpConnection().

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

Remote address.

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

TcpSendStrategy_rch OpenDDS::DCPS::TcpConnection::send_strategy_ [private]

Reference to the send strategy.

Definition at line 172 of file TcpConnection.h.

Referenced by active_reconnect_i(), handle_close(), handle_output(), notify_lost_on_backpressure_timeout(), remove_send_strategy(), and transfer().

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

shutdown flag

Definition at line 208 of file TcpConnection.h.

Referenced by shutdown().

TcpInst_rch OpenDDS::DCPS::TcpConnection::tcp_config_ [private]

The configuration used by this connection.

Definition at line 181 of file TcpConnection.h.

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

TcpTransport_rch OpenDDS::DCPS::TcpConnection::transport_during_setup_ [private]

Definition at line 212 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 205 of file TcpConnection.h.

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

---

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

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

---