OpenDDS::DCPS::UdpTransport Class Reference

#include <UdpTransport.h>

Inheritance diagram for OpenDDS::DCPS::UdpTransport:

[legend]Collaboration diagram for OpenDDS::DCPS::UdpTransport:

[legend]List of all members.

Public Member Functions
	UdpTransport (const TransportInst_rch &inst)
void	passive_connection (const ACE_INET_Addr &remote_address, ACE_Message_Block *data)
Protected Member Functions
virtual AcceptConnectResult	connect_datalink (const RemoteTransport &remote, const ConnectionAttribs &attribs, TransportClient *client)
virtual AcceptConnectResult	accept_datalink (const RemoteTransport &remote, const ConnectionAttribs &attribs, TransportClient *client)
virtual void	stop_accepting_or_connecting (TransportClient *client, const RepoId &remote_id)
virtual bool	configure_i (TransportInst *config)
virtual void	shutdown_i ()
virtual bool	connection_info_i (TransportLocator &info) const
ACE_INET_Addr	get_connection_addr (const TransportBLOB &data) const
virtual void	release_datalink (DataLink *link)
virtual std::string	transport_type () const
Private Types
typedef ACE_SYNCH_MUTEX	LockType
typedef ACE_Guard< LockType >	GuardType
typedef ACE_Condition< LockType >	ConditionType
typedef std::vector< DataLink::OnStartCallback >	Callbacks
Private Member Functions
UdpDataLink *	make_datalink (const ACE_INET_Addr &remote_address, Priority priority, bool active)
PriorityKey	blob_to_key (const TransportBLOB &remote, Priority priority, ACE_INET_Addr local_addr, bool active)
typedef	OPENDDS_MAP (PriorityKey, UdpDataLink_rch) UdpDataLinkMap
	Map of fully associated DataLinks for this transport. Protected.
typedef	OPENDDS_MAP (PriorityKey, Callbacks) PendConnMap
Private Attributes
RcHandle< UdpInst >	config_i_
LockType	client_links_lock_
	This lock is used to protect the client_links_ data member.
UdpDataLinkMap	client_links_
UdpDataLink_rch	server_link_
	The single datalink for the passive side. No locking required.
ACE_Recursive_Thread_Mutex	connections_lock_
std::set< PriorityKey >	server_link_keys_
PendConnMap	pending_connections_
std::set< PriorityKey >	pending_server_link_keys_

---

Detailed Description

Definition at line 25 of file UdpTransport.h.

---

Member Typedef Documentation

typedef std::vector<DataLink::OnStartCallback> OpenDDS::DCPS::UdpTransport::Callbacks [private]

Definition at line 90 of file UdpTransport.h.

typedef ACE_Condition<LockType> OpenDDS::DCPS::UdpTransport::ConditionType [private]

Definition at line 66 of file UdpTransport.h.

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

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 65 of file UdpTransport.h.

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

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 64 of file UdpTransport.h.

---

Constructor & Destructor Documentation

OpenDDS::DCPS::UdpTransport::UdpTransport

(

const TransportInst_rch &

inst

)

[explicit]

Definition at line 26 of file UdpTransport.cpp.

References OpenDDS::DCPS::TransportImpl::configure(), OpenDDS::DCPS::RcHandle< T >::in(), and OpenDDS::DCPS::RcHandle< T >::is_nil().

{
  if (!inst.is_nil()) {
    if (!configure(inst.in())) {
      throw Transport::UnableToCreate();
    }
  }
}

---

Member Function Documentation

TransportImpl::AcceptConnectResult OpenDDS::DCPS::UdpTransport::accept_datalink	(	const RemoteTransport &	remote,
		const ConnectionAttribs &	attribs,
		TransportClient *	client
	)			[protected, virtual]

Definition at line 114 of file UdpTransport.cpp.

References blob_to_key(), config_i_, connections_lock_, pending_connections_, pending_server_link_keys_, OpenDDS::DCPS::TransportClient::repo_id_, server_link_, server_link_keys_, and VDBG.

{
  ACE_Guard<ACE_Recursive_Thread_Mutex> guard(connections_lock_);
  //GuardType guard(connections_lock_);
  const PriorityKey key = blob_to_key(remote.blob_,
                                      attribs.priority_, config_i_->local_address(), false /* !active */);
  if (server_link_keys_.count(key)) {
    VDBG((LM_DEBUG, "(%P|%t) UdpTransport::accept_datalink found\n"));
    return AcceptConnectResult(UdpDataLink_rch(server_link_)._retn());
  }

  else if (pending_server_link_keys_.count(key)) {
    pending_server_link_keys_.erase(key);
    server_link_keys_.insert(key);
    VDBG((LM_DEBUG, "(%P|%t) UdpTransport::accept_datalink completed\n"));
    return AcceptConnectResult(UdpDataLink_rch(server_link_)._retn());
  } else {
    const DataLink::OnStartCallback callback(client, remote.repo_id_);
    pending_connections_[key].push_back(callback);
    VDBG((LM_DEBUG, "(%P|%t) UdpTransport::accept_datalink pending\n"));
    return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
  }
     return AcceptConnectResult();
}

PriorityKey OpenDDS::DCPS::UdpTransport::blob_to_key	(	const TransportBLOB &	remote,
		Priority	priority,
		ACE_INET_Addr	local_addr,
		bool	active
	)			[private]

Definition at line 251 of file UdpTransport.cpp.

References OpenDDS::DCPS::NetworkAddress::to_addr().

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

{
  NetworkAddress network_order_address;
  ACE_InputCDR cdr((const char*)remote.get_buffer(), remote.length());

  if ((cdr >> network_order_address) == 0) {
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) ERROR: UdpTransport::blob_to_key")
               ACE_TEXT(" failed to de-serialize the NetworkAddress\n")));
  }

  ACE_INET_Addr remote_address;
  network_order_address.to_addr(remote_address);
  const bool is_loopback = remote_address == local_addr;

  return PriorityKey(priority, remote_address, is_loopback, active);
}

bool OpenDDS::DCPS::UdpTransport::configure_i

(

TransportInst *

config

)

[protected, virtual]

Concrete subclass gets a shot at the config object. The subclass will likely downcast the TransportInst object to a subclass type that it expects/requires.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 166 of file UdpTransport.cpp.

References OpenDDS::DCPS::TransportImpl::config(), config_i_, OpenDDS::DCPS::TransportImpl::create_reactor_task(), make_datalink(), server_link_, and TheServiceParticipant.

{
  config_i_ = UdpInst_rch(dynamic_cast<UdpInst*>(config), false);

  if (config_i_ == 0) {
    ACE_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: ")
                      ACE_TEXT("UdpTransport::configure_i: ")
                      ACE_TEXT("invalid configuration!\n")),
                     false);
  }

  create_reactor_task();

  // Override with DCPSDefaultAddress.
  if (this->config_i_->local_address() == ACE_INET_Addr () &&
      !TheServiceParticipant->default_address ().empty ()) {
    this->config_i_->local_address(0, TheServiceParticipant->default_address ().c_str ());
  }

  // Our "server side" data link is created here, similar to the acceptor_
  // in the TcpTransport implementation.  This establishes a socket as an
  // endpoint that we can advertise to peers via connection_info_i().
  server_link_ = make_datalink(this->config_i_->local_address(), 0 /* priority */, false);
  return true;
}

TransportImpl::AcceptConnectResult OpenDDS::DCPS::UdpTransport::connect_datalink	(	const RemoteTransport &	remote,
		const ConnectionAttribs &	attribs,
		TransportClient *	client
	)			[protected, virtual]

Definition at line 77 of file UdpTransport.cpp.

References blob_to_key(), client_links_, client_links_lock_, config_i_, get_connection_addr(), make_datalink(), and VDBG.

{
  UdpInst_rch tmp_config(this->config_i_.in(), false);
  if (this->is_shut_down() || this->config_i_.is_nil()) {
    return AcceptConnectResult(AcceptConnectResult::ACR_FAILED);
  }
  const ACE_INET_Addr remote_address = get_connection_addr(remote.blob_);
  const bool active = true;
  const PriorityKey key = blob_to_key(remote.blob_, attribs.priority_, tmp_config->local_address(), active);
  tmp_config = 0; //no longer need to hold on to local config
  GuardType guard(client_links_lock_);
  if (this->is_shut_down() || this->config_i_.is_nil()) {
    return AcceptConnectResult(AcceptConnectResult::ACR_FAILED);
  }

  const UdpDataLinkMap::iterator it(client_links_.find(key));
  if (it != client_links_.end()) {
    VDBG((LM_DEBUG, "(%P|%t) UdpTransport::connect_datalink found\n"));
    return AcceptConnectResult(UdpDataLink_rch(it->second)._retn());
  }

  // Create new DataLink for logical connection:
  UdpDataLink_rch link = make_datalink(remote_address,
                                       attribs.priority_,
                                       active);

  if (!link.is_nil()) {
    client_links_.insert(UdpDataLinkMap::value_type(key, link));
    VDBG((LM_DEBUG, "(%P|%t) UdpTransport::connect_datalink connected\n"));
  }

  return AcceptConnectResult(link._retn());
}

bool OpenDDS::DCPS::UdpTransport::connection_info_i

(

TransportLocator &

info

)

const [protected, virtual]

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

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 211 of file UdpTransport.cpp.

References config_i_.

{
  this->config_i_->populate_locator(info);
  return true;
}

ACE_INET_Addr OpenDDS::DCPS::UdpTransport::get_connection_addr

(

const TransportBLOB &

data

)

const [protected]

Definition at line 218 of file UdpTransport.cpp.

References OpenDDS::DCPS::NetworkAddress::to_addr().

Referenced by connect_datalink().

{
  ACE_INET_Addr local_address;
  NetworkAddress network_address;

  size_t len = data.length();
  const char* buffer = reinterpret_cast<const char*>(data.get_buffer());

  ACE_InputCDR cdr(buffer, len);
  cdr >> network_address;

  network_address.to_addr(local_address);

  return local_address;
}

UdpDataLink * OpenDDS::DCPS::UdpTransport::make_datalink	(	const ACE_INET_Addr &	remote_address,
		Priority	priority,
		bool	active
	)			[private]

Definition at line 36 of file UdpTransport.cpp.

References OpenDDS::DCPS::RcHandle< T >::_retn(), config_i_, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::RcHandle< T >::is_nil(), and OpenDDS::DCPS::TransportImpl::reactor_task().

Referenced by configure_i(), and connect_datalink().

{
  UdpDataLink_rch link;
  ACE_NEW_RETURN(link, UdpDataLink(this, priority, active), 0);

  if (link.is_nil()) {
    ACE_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: ")
                      ACE_TEXT("UdpTransport::make_datalink: ")
                      ACE_TEXT("failed to create DataLink!\n")),
                     0);
  }

  // Configure link with transport configuration and reactor task:
  TransportReactorTask_rch rtask = reactor_task();
  link->configure(config_i_.in(), rtask.in());

  // Assign send strategy:
  UdpSendStrategy* send_strategy;
  ACE_NEW_RETURN(send_strategy, UdpSendStrategy(link.in()), 0);
  link->send_strategy(send_strategy);

  // Assign receive strategy:
  UdpReceiveStrategy* recv_strategy;
  ACE_NEW_RETURN(recv_strategy, UdpReceiveStrategy(link.in()), 0);
  link->receive_strategy(recv_strategy);

  // Open logical connection:
  if (!link->open(remote_address)) {
    ACE_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: ")
                      ACE_TEXT("UdpTransport::make_datalink: ")
                      ACE_TEXT("failed to open DataLink!\n")),
                     0);
  }

  return link._retn();
}

typedef OpenDDS::DCPS::UdpTransport::OPENDDS_MAP	(	PriorityKey	,
		Callbacks
	)			[private]

typedef OpenDDS::DCPS::UdpTransport::OPENDDS_MAP	(	PriorityKey	,
		UdpDataLink_rch
	)			[private]

Map of fully associated DataLinks for this transport. Protected.

void OpenDDS::DCPS::UdpTransport::passive_connection	(	const ACE_INET_Addr &	remote_address,
		ACE_Message_Block *	data
	)

Definition at line 273 of file UdpTransport.cpp.

References blob_to_key(), config_i_, connections_lock_, OpenDDS::DCPS::find(), pending_connections_, pending_server_link_keys_, OpenDDS::DCPS::Serializer::read_octet_array(), server_link_, server_link_keys_, and VDBG.

{
  CORBA::ULong octet_size =
    static_cast<CORBA::ULong>(data->length() - sizeof(Priority));
  Priority priority;
  Serializer serializer(data);
  serializer >> priority;
  TransportBLOB blob(octet_size);
  blob.length(octet_size);
  serializer.read_octet_array(blob.get_buffer(), octet_size);

  // Send an ack so that the active side can return from
  // connect_datalink_i().  This is just a single byte of
  // arbitrary data, the remote side is not yet using the
  // framework (TransportHeader, DataSampleHeader,
  // ReceiveStrategy).
  const char ack_data = 23;
  server_link_->socket().send(&ack_data, 1, remote_address);

  const PriorityKey key = blob_to_key(blob, priority, config_i_->local_address(), false /* passive */);

  ACE_Guard<ACE_Recursive_Thread_Mutex> guard(connections_lock_);

  //GuardType guard(connections_lock_);
  const PendConnMap::iterator pend = pending_connections_.find(key);

  if (pend != pending_connections_.end()) {

     //don't hold connections_lock_ while calling use_datalink
     //guard.release();

    VDBG((LM_DEBUG, "(%P|%t) UdpTransport::passive_connection completing\n"));

    const DataLink_rch link = static_rchandle_cast<DataLink>(server_link_);

    //Insert key now to make sure when releasing guard to call use_datalink
    //if an accept_datalink obtains lock first it will see that it can proceed
    //with using the link and do its own use_datalink call.
    server_link_keys_.insert(key);

    //create a copy of the size of callback vector so that if use_datalink_i -> stop_accepting_or_connecting
    //finds that callbacks vector is empty and deletes pending connection & its callback vector for loop can
    //still exit the loop without checking the size of invalid memory
    //size_t num_callbacks = pend->second.size();

    //Create a copy of the vector of callbacks to process, making sure that each is
    //still present in the actual pending_connections_ before calling use_datalink
    Callbacks tmp(pend->second);
    for (size_t i = 0; i < tmp.size(); ++i) {
      const PendConnMap::iterator pend = pending_connections_.find(key);
      if (pend != pending_connections_.end()) {
        const Callbacks::iterator tmp_iter = find(pend->second.begin(),
                                                  pend->second.end(),
                                                  tmp.at(i));
        if (tmp_iter != pend->second.end()) {
          TransportClient* pend_client = tmp.at(i).first;
          RepoId remote_repo = tmp.at(i).second;
          guard.release();
          pend_client->use_datalink(remote_repo, link);
          guard.acquire();
        }
      }
    }
  } else {
    // still hold guard(connections_lock_) at this point so
    // pending_server_link_keys_ is protected for insert

    VDBG((LM_DEBUG, "(%P|%t) UdpTransport::passive_connection pending\n"));
    // accept_datalink() will complete the connection.
    pending_server_link_keys_.insert(key);
  }
}

void OpenDDS::DCPS::UdpTransport::release_datalink

(

DataLink *

link

)

[protected, virtual]

The DataLink itself calls this method when it thinks it is no longer used for any associations. This occurs during a "remove associations" operation being performed by some TransportClient that uses this TransportImpl. The TransportClient is known to have acquired our reservation_lock_, so there won't be any reserve_datalink() calls being made from any other threads while we perform this release.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 235 of file UdpTransport.cpp.

References client_links_, client_links_lock_, and OpenDDS::DCPS::DataLink::stop().

{
  GuardType guard(client_links_lock_);
  for (UdpDataLinkMap::iterator it(client_links_.begin());
       it != client_links_.end(); ++it) {
    // We are guaranteed to have exactly one matching DataLink
    // in the map; release any resources held and return.
    if (link == static_cast<DataLink*>(it->second.in())) {
      link->stop();
      client_links_.erase(it);
      return;
    }
  }
}

void OpenDDS::DCPS::UdpTransport::shutdown_i

(

)

[protected, virtual]

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

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 194 of file UdpTransport.cpp.

References client_links_, client_links_lock_, config_i_, and server_link_.

{
  // Shutdown reserved datalinks and release configuration:
  GuardType guard(client_links_lock_);
  for (UdpDataLinkMap::iterator it(client_links_.begin());
       it != client_links_.end(); ++it) {
    it->second->transport_shutdown();
  }
  client_links_.clear();

  server_link_->transport_shutdown();
  server_link_ = 0;

  config_i_ = 0;
}

void OpenDDS::DCPS::UdpTransport::stop_accepting_or_connecting	(	TransportClient *	client,
		const RepoId &	remote_id
	)			[protected, virtual]

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

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 142 of file UdpTransport.cpp.

References connections_lock_, pending_connections_, and VDBG.

{
  VDBG((LM_DEBUG, "(%P|%t) UdpTransport::stop_accepting_or_connecting\n"));

  //GuardType guard(connections_lock_);
  ACE_Guard<ACE_Recursive_Thread_Mutex> guard(connections_lock_);

  for (PendConnMap::iterator it = pending_connections_.begin();
       it != pending_connections_.end(); ++it) {
    for (size_t i = 0; i < it->second.size(); ++i) {
      if (it->second[i].first == client && it->second[i].second == remote_id) {
        it->second.erase(it->second.begin() + i);
        break;
      }
    }
    if (it->second.empty()) {
      pending_connections_.erase(it);
      return;
    }
  }
}

virtual std::string OpenDDS::DCPS::UdpTransport::transport_type

(

)

const [inline, protected, virtual]

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 53 of file UdpTransport.h.

{ return "udp"; }

---

Member Data Documentation

UdpDataLinkMap OpenDDS::DCPS::UdpTransport::client_links_ [private]

Definition at line 74 of file UdpTransport.h.

Referenced by connect_datalink(), release_datalink(), and shutdown_i().

LockType OpenDDS::DCPS::UdpTransport::client_links_lock_ [private]

This lock is used to protect the client_links_ data member.

Definition at line 69 of file UdpTransport.h.

Referenced by connect_datalink(), release_datalink(), and shutdown_i().

RcHandle<UdpInst> OpenDDS::DCPS::UdpTransport::config_i_ [private]

Definition at line 62 of file UdpTransport.h.

Referenced by accept_datalink(), configure_i(), connect_datalink(), connection_info_i(), make_datalink(), passive_connection(), and shutdown_i().

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::UdpTransport::connections_lock_ [private]

This protects the pending_connections_, pending_server_link_keys_, and server_link_keys_ data members.

Definition at line 82 of file UdpTransport.h.

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

PendConnMap OpenDDS::DCPS::UdpTransport::pending_connections_ [private]

Locked by connections_lock_. Tracks expected connections that we have learned about in accept_datalink() but have not yet performed the handshake.

Definition at line 95 of file UdpTransport.h.

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

std::set<PriorityKey> OpenDDS::DCPS::UdpTransport::pending_server_link_keys_ [private]

Locked by connections_lock_. These are passive-side PriorityKeys that have finished handshaking, but have not been processed by accept_datalink()

Definition at line 100 of file UdpTransport.h.

Referenced by accept_datalink(), and passive_connection().

UdpDataLink_rch OpenDDS::DCPS::UdpTransport::server_link_ [private]

The single datalink for the passive side. No locking required.

Definition at line 77 of file UdpTransport.h.

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

std::set<PriorityKey> OpenDDS::DCPS::UdpTransport::server_link_keys_ [private]

Locked by connections_lock_. These are passive-side PriorityKeys that have been fully associated (processed by accept_datalink() and finished handshaking). They are ready for use and reuse via server_link_.

Definition at line 88 of file UdpTransport.h.

Referenced by accept_datalink(), and passive_connection().

---

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

• UdpTransport.h
• UdpTransport.cpp

---