OpenDDS::DCPS::DataLink Class Reference

#include <DataLink.h>

Inheritance diagram for OpenDDS::DCPS::DataLink:

Collaboration diagram for OpenDDS::DCPS::DataLink:

Classes
class	ImmediateStart
class	Interceptor
struct	LocalAssociationInfo

Public Types
enum	ConnectionNotice { DISCONNECTED, RECONNECTED, LOST }
typedef WeakRcHandle< TransportClient >	TransportClient_wrch
typedef std::pair< TransportClient_wrch, GUID_t >	OnStartCallback
Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

Public Member Functions
	DataLink (const TransportImpl_rch &impl, Priority priority, bool is_loopback, bool is_active)
	Only called by our TransportImpl object. More...
virtual	~DataLink ()
int	handle_exception (ACE_HANDLE)
	Reactor invokes this after being notified in schedule_stop or cancel_release. More...
void	schedule_stop (const MonotonicTimePoint &schedule_to_stop_at)
void	stop ()
	The stop method is used to stop the DataLink prior to shutdown. More...
void	resume_send ()
virtual int	make_reservation (const GUID_t &remote_subscription_id, const GUID_t &local_publication_id, const TransportSendListener_wrch &send_listener, bool reliable)
virtual int	make_reservation (const GUID_t &remote_publication_id, const GUID_t &local_subscription_id, const TransportReceiveListener_wrch &receive_listener, bool reliable)
void	release_reservations (GUID_t remote_id, GUID_t local_id, DataLinkSetMap &released_locals)
void	schedule_delayed_release ()
const TimeDuration &	datalink_release_delay () const
void	remove_listener (const GUID_t &local_id)
void	send_start ()
void	send (TransportQueueElement *element)
void	send_stop (GUID_t repoId)
virtual RemoveResult	remove_sample (const DataSampleElement *sample)
virtual void	remove_all_msgs (const GUID_t &pub_id)
int	data_received (ReceivedDataSample &sample, const GUID_t &readerId=GUID_UNKNOWN)
void	data_received_include (ReceivedDataSample &sample, const RepoIdSet &incl)
DataLinkIdType	id () const
	Obtain a unique identifier for this DataLink object. More...
void	transport_shutdown ()
void	notify (ConnectionNotice notice)
virtual void	pre_stop_i ()
void	release_resources ()
void	terminate_send ()
void	terminate_send_if_suspended ()
bool	is_target (const GUID_t &remote_id)
void	clear_associations ()
int	handle_timeout (const ACE_Time_Value &tv, const void *arg)
int	handle_close (ACE_HANDLE h, ACE_Reactor_Mask m)
void	set_dscp_codepoint (int cp, ACE_SOCK &socket)
Priority &	transport_priority ()
Priority	transport_priority () const
bool &	is_loopback ()
bool	is_loopback () const
bool &	is_active ()
bool	is_active () const
bool	cancel_release ()
ACE_Message_Block *	create_control (char submessage_id, DataSampleHeader &header, Message_Block_Ptr data)
SendControlStatus	send_control (const DataSampleHeader &header, Message_Block_Ptr data)
GUIDSeq *	target_intersection (const GUID_t &pub_id, const GUIDSeq &in, size_t &n_subs)
TransportImpl_rch	impl () const
void	default_listener (const TransportReceiveListener_wrch &trl)
TransportReceiveListener_wrch	default_listener () const
bool	add_on_start_callback (const TransportClient_wrch &client, const GUID_t &remote)
void	remove_on_start_callback (const TransportClient_wrch &client, const GUID_t &remote)
void	invoke_on_start_callbacks (bool success)
bool	invoke_on_start_callbacks (const GUID_t &local, const GUID_t &remote, bool success)
void	remove_startup_callbacks (const GUID_t &local, const GUID_t &remote)
void	set_scheduling_release (bool scheduling_release)
virtual void	send_final_acks (const GUID_t &readerid)
virtual WeakRcHandle< ICE::Endpoint >	get_ice_endpoint () const
virtual bool	is_leading (const GUID_t &, const GUID_t &) const
Public Member Functions inherited from OpenDDS::DCPS::RcEventHandler
	RcEventHandler ()
ACE_Event_Handler::Reference_Count	add_reference ()
ACE_Event_Handler::Reference_Count	remove_reference ()
Public Member Functions inherited from ACE_Event_Handler
virtual	~ACE_Event_Handler (void)
virtual ACE_HANDLE	get_handle (void) const
virtual void	set_handle (ACE_HANDLE)
virtual int	priority (void) const
virtual void	priority (int priority)
virtual int	handle_input (ACE_HANDLE fd=ACE_INVALID_HANDLE)
virtual int	handle_output (ACE_HANDLE fd=ACE_INVALID_HANDLE)
virtual int	handle_exit (ACE_Process *)
virtual int	handle_signal (int signum, siginfo_t *=0, ucontext_t *=0)
virtual int	resume_handler (void)
virtual int	handle_qos (ACE_HANDLE=ACE_INVALID_HANDLE)
virtual int	handle_group_qos (ACE_HANDLE=ACE_INVALID_HANDLE)
virtual void	reactor (ACE_Reactor *reactor)
virtual ACE_Reactor *	reactor (void) const
virtual ACE_Reactor_Timer_Interface *	reactor_timer_interface (void) const
Reference_Counting_Policy &	reference_counting_policy (void)
Public Member Functions inherited from OpenDDS::DCPS::RcObject
virtual	~RcObject ()
virtual void	_add_ref ()
virtual void	_remove_ref ()
long	ref_count () const
WeakObject *	_get_weak_object () const

Protected Types
typedef ACE_Guard< LockType >	GuardType
Protected Types inherited from ACE_Event_Handler
typedef ACE_Atomic_Op< ACE_SYNCH_MUTEX, Reference_Count >	Atomic_Reference_Count

Protected Member Functions
int	start (const TransportSendStrategy_rch &send_strategy, const TransportStrategy_rch &receive_strategy, bool invoke_all=true)
virtual void	stop_i ()
void	send_start_i ()
virtual void	send_i (TransportQueueElement *element, bool relink=true)
void	send_stop_i (GUID_t repoId)
GUIDSeq *	peer_ids (const GUID_t &local_id) const
void	network_change () const
void	replay_durable_data (const GUID_t &local_pub_id, const GUID_t &remote_sub_id) const
TransportSendStrategy_rch	get_send_strategy ()
typedef	OPENDDS_MAP_CMP (GUID_t, TransportClient_wrch, GUID_tKeyLessThan) RepoToClientMap
typedef	OPENDDS_MAP_CMP (GUID_t, RepoToClientMap, GUID_tKeyLessThan) OnStartCallbackMap
typedef	OPENDDS_MAP_CMP (GUID_t, RepoIdSet, GUID_tKeyLessThan) PendingOnStartsMap
Protected Member Functions inherited from ACE_Event_Handler
	ACE_Event_Handler (ACE_Reactor *=0, int priority=ACE_Event_Handler::LO_PRIORITY)
Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Static Protected Member Functions
static ACE_UINT64	get_next_datalink_id ()
	Used to provide unique Ids to all DataLink methods. More...

Protected Attributes
TransportStrategy_rch	receive_strategy_
	The transport receive strategy object for this DataLink. More...
TransportSendStrategy_rch	send_strategy_
	The transport send strategy object for this DataLink. More...
LockType	strategy_lock_
OnStartCallbackMap	on_start_callbacks_
PendingOnStartsMap	pending_on_starts_
TimeDuration	datalink_release_delay_
unique_ptr< MessageBlockAllocator >	mb_allocator_
unique_ptr< DataBlockAllocator >	db_allocator_
bool	is_loopback_
	Is remote attached to same transport ? More...
bool	is_active_
	Is pub or sub ? More...
bool	started_
SendResponseListener	send_response_listener_
	Listener for TransportSendControlElements created in send_control. More...
Interceptor	interceptor_
Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Private Types
typedef ACE_SYNCH_MUTEX	LockType

Private Member Functions
const char *	connection_notice_as_str (ConnectionNotice notice)
	Helper function to output the enum as a string to help debugging. More...
TransportSendListener_rch	send_listener_for (const GUID_t &pub_id) const
TransportReceiveListener_rch	recv_listener_for (const GUID_t &sub_id) const
void	prepare_release ()
virtual bool	handle_send_request_ack (TransportQueueElement *element)
virtual TransportQueueElement *	customize_queue_element (TransportQueueElement *element)
virtual void	release_remote_i (const GUID_t &)
virtual void	release_reservations_i (const GUID_t &, const GUID_t &)
void	data_received_i (ReceivedDataSample &sample, const GUID_t &readerId, const RepoIdSet &incl_excl, ReceiveListenerSet::ConstrainReceiveSet constrain)
void	notify_reactor ()
typedef	OPENDDS_MAP_CMP (GUID_t, TransportSendListener_wrch, GUID_tKeyLessThan) IdToSendListenerMap
	Map publication Id value to TransportSendListener. More...
typedef	OPENDDS_MAP_CMP (GUID_t, TransportReceiveListener_wrch, GUID_tKeyLessThan) IdToRecvListenerMap
	Map subscription Id value to TransportReceieveListener. More...
typedef	OPENDDS_MAP_CMP (GUID_t, ReceiveListenerSet_rch, GUID_tKeyLessThan) AssocByRemote
typedef	OPENDDS_MAP_CMP (GUID_t, LocalAssociationInfo, GUID_tKeyLessThan) AssocByLocal

Private Attributes
bool	stopped_
MonotonicTimePoint	scheduled_to_stop_at_
IdToSendListenerMap	send_listeners_
IdToRecvListenerMap	recv_listeners_
TransportReceiveListener_wrch	default_listener_
LockType	pub_sub_maps_lock_
AssocByRemote	assoc_by_remote_
AssocByLocal	assoc_by_local_
WeakRcHandle< TransportImpl >	impl_
	A weak rchandle to the TransportImpl that created this DataLink. More...
ACE_UINT64	id_
	The id for this DataLink. More...
unique_ptr< ThreadPerConnectionSendTask >	thr_per_con_send_task_
AssocByLocal	assoc_releasing_
Priority	transport_priority_
	TRANSPORT_PRIORITY value associated with the link. More...
bool	scheduling_release_

Friends
class	DataLinkCleanupTask
class	ThreadPerConnectionSendTask
OpenDDS_Dcps_Export std::ostream &	operator<< (std::ostream &str, const DataLink &value)
	Convenience function for diagnostic information. More...

Additional Inherited Members
Static Public Member Functions inherited from ACE_Event_Handler
static ACE_THR_FUNC_RETURN	read_adapter (void *event_handler)
static int	register_stdin_handler (ACE_Event_Handler *eh, ACE_Reactor *reactor, ACE_Thread_Manager *thr_mgr, int flags=THR_DETACHED)
static int	remove_stdin_handler (ACE_Reactor *reactor, ACE_Thread_Manager *thr_mgr)
Public Attributes inherited from ACE_Event_Handler
	LO_PRIORITY
	HI_PRIORITY
	NULL_MASK
	READ_MASK
	WRITE_MASK
	EXCEPT_MASK
	ACCEPT_MASK
	CONNECT_MASK
	TIMER_MASK
	QOS_MASK
	GROUP_QOS_MASK
	SIGNAL_MASK
	ALL_EVENTS_MASK
	RWE_MASK
	DONT_CALL
	ACE_EVENT_HANDLER_NOT_RESUMED
	ACE_REACTOR_RESUMES_HANDLER
	ACE_APPLICATION_RESUMES_HANDLER

Detailed Description

This class manages the reservations based on the associations between datareader and datawriter. It basically delegate the samples to send strategy for sending and deliver the samples received by receive strategy to the listener.

Notes about object ownership: 1) Own the send strategy object and receive strategy object. 2) Own ThreadPerConnectionSendTask object which is used when thread_per_connection is enabled.

Definition at line 73 of file DataLink.h.

Member Typedef Documentation

GuardType

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

protected

Definition at line 437 of file DataLink.h.

LockType

typedef ACE_SYNCH_MUTEX OpenDDS::DCPS::DataLink::LockType

private

Definition at line 377 of file DataLink.h.

OnStartCallback

typedef std::pair<TransportClient_wrch, GUID_t> OpenDDS::DCPS::DataLink::OnStartCallback

Definition at line 259 of file DataLink.h.

TransportClient_wrch

typedef WeakRcHandle<TransportClient> OpenDDS::DCPS::DataLink::TransportClient_wrch

Definition at line 258 of file DataLink.h.

Member Enumeration Documentation

ConnectionNotice

enum OpenDDS::DCPS::DataLink::ConnectionNotice

Enumerator
DISCONNECTED
RECONNECTED
LOST

Definition at line 80 of file DataLink.h.

                        {
    DISCONNECTED,
    RECONNECTED,
    LOST
  };

Constructor & Destructor Documentation

DataLink()

OpenDDS::DCPS::DataLink::DataLink	(	const TransportImpl_rch &	impl,
		Priority	priority,
		bool	is_loopback,
		bool	is_active
	)

Only called by our TransportImpl object.

A DataLink object is always created by a TransportImpl object. Thus, the TransportImpl object passed-in here is the object that created this DataLink. The ability to specify a priority for individual links is included for construction so its value can be available for activating any threads.

Definition at line 42 of file DataLink.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::TransportImpl::config(), OpenDDS::DCPS::TransportInst::datalink_control_chunks_, datalink_release_delay(), OpenDDS::DCPS::TransportInst::datalink_release_delay_, datalink_release_delay_, db_allocator_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TransportInst::DEFAULT_DATALINK_CONTROL_CHUNKS, OpenDDS::DCPS::TransportInst::DEFAULT_DATALINK_RELEASE_DELAY, OpenDDS::DCPS::TimeDuration::from_msec(), get_next_datalink_id(), id_, LM_DEBUG, LM_ERROR, mb_allocator_, thr_per_con_send_task_, OpenDDS::DCPS::TransportInst::thread_per_connection_, and ThreadPerConnectionSendTask.

  : stopped_(false),
    impl_(impl),
    transport_priority_(priority),
    scheduling_release_(false),
    is_loopback_(is_loopback),
    is_active_(is_active),
    started_(false),
    send_response_listener_("DataLink"),
    interceptor_(impl->reactor(), impl->reactor_owner())
{
  DBG_ENTRY_LVL("DataLink", "DataLink", 6);

  id_ = DataLink::get_next_datalink_id();

  long datalink_release_delay = TransportInst::DEFAULT_DATALINK_RELEASE_DELAY;
  size_t control_chunks = TransportInst::DEFAULT_DATALINK_CONTROL_CHUNKS;

  TransportInst_rch cfg = impl->config();
  if (cfg) {
    datalink_release_delay = cfg->datalink_release_delay_;
    if (cfg->thread_per_connection_) {
      thr_per_con_send_task_.reset(new ThreadPerConnectionSendTask(this));

      if (thr_per_con_send_task_->open() == -1) {
        ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) DataLink::DataLink: ")
                   ACE_TEXT("failed to open ThreadPerConnectionSendTask\n")));

      } else if (DCPS_debug_level > 4) {
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) DataLink::DataLink - ")
                   ACE_TEXT("started new thread to send data with.\n")));
      }
    }
    control_chunks = cfg->datalink_control_chunks_;
  }

  // Initialize transport control sample allocators:
  datalink_release_delay_ = TimeDuration::from_msec(datalink_release_delay);

  this->mb_allocator_.reset(new MessageBlockAllocator(control_chunks));
  this->db_allocator_.reset(new DataBlockAllocator(control_chunks));
}

~DataLink()

OpenDDS::DCPS::DataLink::~DataLink ( )

virtual

Definition at line 88 of file DataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), assoc_by_local_, DBG_ENTRY_LVL, LM_WARNING, and thr_per_con_send_task_.

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

  if (!assoc_by_local_.empty()) {
    ACE_DEBUG((LM_WARNING,
               ACE_TEXT("(%P|%t) WARNING: DataLink[%@]::~DataLink() - ")
               ACE_TEXT("link still in use by %d entities when deleted!\n"),
               this, assoc_by_local_.size()));
  }

  if (this->thr_per_con_send_task_ != 0) {
    this->thr_per_con_send_task_->close(1);
  }
}

Member Function Documentation

add_on_start_callback()

bool OpenDDS::DCPS::DataLink::add_on_start_callback	(	const TransportClient_wrch &	client,
		const GUID_t &	remote
	)

Definition at line 111 of file DataLink.cpp.

References OpenDDS::DCPS::ReactorInterceptor::execute_or_enqueue(), OpenDDS::DCPS::GUID_UNKNOWN, interceptor_, OpenDDS::DCPS::RcHandle< T >::is_nil(), OpenDDS::DCPS::WeakRcHandle< T >::lock(), on_start_callbacks_, pending_on_starts_, ACE_Guard< ACE_LOCK >::release(), send_strategy_, started_, and strategy_lock_.

Referenced by OpenDDS::DCPS::TcpTransport::accept_datalink(), OpenDDS::DCPS::RtpsUdpDataLink::associated(), and OpenDDS::DCPS::TcpTransport::connect_datalink().

{
  const DataLink_rch link(this, inc_count());

  TransportClient_rch client_lock = client.lock();
  const GUID_t client_id = client_lock ? client_lock->get_guid() : GUID_UNKNOWN;

  GuardType guard(strategy_lock_);

  if (client_lock) {
    PendingOnStartsMap::iterator it = pending_on_starts_.find(remote);
    if (it != pending_on_starts_.end()) {
      RepoIdSet::iterator it2 = it->second.find(client_id);
      if (it2 != it->second.end()) {
        it->second.erase(it2);
        if (it->second.empty()) {
          pending_on_starts_.erase(it);
        }
        guard.release();
        interceptor_.execute_or_enqueue(make_rch<ImmediateStart>(link, client, remote));
      } else {
        on_start_callbacks_[remote][client_id] = client;
      }
    } else {
      on_start_callbacks_[remote][client_id] = client;
    }
  }

  if (started_ && !send_strategy_.is_nil()) {
    return false; // link already started
  }
  return true;
}

cancel_release()

bool OpenDDS::DCPS::DataLink::cancel_release

(

)

Definition at line 601 of file DataLink.cpp.

References ACE_DEBUG, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, LM_DEBUG, notify_reactor(), scheduled_to_stop_at_, scheduling_release_, set_scheduling_release(), stopped_, and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::zero_value.

Referenced by OpenDDS::DCPS::TcpTransport::find_datalink_i().

{
  DBG_ENTRY_LVL("DataLink", "cancel_release", 6);
  if (stopped_) {
    if (DCPS_debug_level > 0) {
      ACE_DEBUG((LM_DEBUG, "(%P|%t) DataLink::cancel_release - link[%@] already stopped_ cannot cancel release\n", this));
    }
    return false;
  }
  if (scheduling_release_) {
    if (DCPS_debug_level > 0) {
      ACE_DEBUG((LM_DEBUG, "(%P|%t) DataLink::cancel_release - link[%@] currently scheduling release, notify reactor of cancel\n", this));
    }
    this->set_scheduling_release(false);
    scheduled_to_stop_at_ = MonotonicTimePoint::zero_value;
    notify_reactor();
  }
  return true;
}

clear_associations()

void OpenDDS::DCPS::DataLink::clear_associations

(

)

This is called by DataLinkCleanupTask thread to remove the associations based on the snapshot in release_resources().

Definition at line 1059 of file DataLink.cpp.

References assoc_releasing_, recv_listener_for(), OpenDDS::DCPS::TransportReceiveListener::remove_associations(), send_listener_for(), and OpenDDS::DCPS::set_to_seq().

Referenced by OpenDDS::DCPS::TransportImpl::DoClear::handle_event().

{
  for (AssocByLocal::iterator iter = assoc_releasing_.begin();
       iter != assoc_releasing_.end(); ++iter) {
    TransportSendListener_rch tsl = send_listener_for(iter->first);
    if (tsl) {
      ReaderIdSeq sub_ids;
      set_to_seq(iter->second.associated_, sub_ids);
      tsl->remove_associations(sub_ids, false);
      continue;
    }
    TransportReceiveListener_rch trl = recv_listener_for(iter->first);
    if (trl) {
      WriterIdSeq pub_ids;
      set_to_seq(iter->second.associated_, pub_ids);
      trl->remove_associations(pub_ids, false);
    }
  }
  assoc_releasing_.clear();
}

connection_notice_as_str()

ACE_INLINE const char * OpenDDS::DCPS::DataLink::connection_notice_as_str ( ConnectionNotice  notice )

private

Helper function to output the enum as a string to help debugging.

Definition at line 267 of file DataLink.inl.

References ACE_INLINE.

Referenced by notify().

{
  static const char* NoticeStr[] = { "DISCONNECTED",
                                     "RECONNECTED",
                                     "LOST"
                                   };

  return NoticeStr [notice];
}

create_control()

ACE_Message_Block * OpenDDS::DCPS::DataLink::create_control	(	char	submessage_id,
		DataSampleHeader &	header,
		Message_Block_Ptr	data
	)

This allows a subclass to easily create a transport control sample to send via send_control.

Definition at line 628 of file DataLink.cpp.

References ACE_CDR_BYTE_ORDER, ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, ACE_DES_FREE, ACE_ERROR, ACE_NEW_MALLOC_RETURN, ACE_TEXT(), OpenDDS::DCPS::DataSampleHeader::byte_order_, db_allocator_, DBG_ENTRY_LVL, OpenDDS::DCPS::DataSampleHeader::get_max_serialized_size(), LM_ERROR, ACE_Time_Value::max_time, mb_allocator_, ACE_Message_Block::MB_DATA, OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::DataSampleHeader::message_length_, OpenDDS::DCPS::unique_ptr< T, Deleter >::release(), OpenDDS::DCPS::DataSampleHeader::submessage_id_, ACE_Message_Block::total_length(), OpenDDS::DCPS::TRANSPORT_CONTROL, and ACE_Time_Value::zero.

Referenced by OpenDDS::DCPS::MulticastSession::send_control(), and OpenDDS::DCPS::MulticastDataLink::syn_received_no_session().

{
  DBG_ENTRY_LVL("DataLink", "create_control", 6);

  header.byte_order_ = ACE_CDR_BYTE_ORDER;
  header.message_id_ = TRANSPORT_CONTROL;
  header.submessage_id_ = submessage_id;
  header.message_length_ = static_cast<ACE_UINT32>(data->total_length());

  ACE_Message_Block* message = 0;
  ACE_NEW_MALLOC_RETURN(message,
                        static_cast<ACE_Message_Block*>(
                          this->mb_allocator_->malloc(sizeof(ACE_Message_Block))),
                        ACE_Message_Block(header.get_max_serialized_size(),
                                          ACE_Message_Block::MB_DATA,
                                          data.release(),
                                          0,  // data
                                          0,  // allocator_strategy
                                          0,  // locking_strategy
                                          ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                                          ACE_Time_Value::zero,
                                          ACE_Time_Value::max_time,
                                          this->db_allocator_.get(),
                                          this->mb_allocator_.get()),
                        0);

  if (!(*message << header)) {
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) DataLink::create_control: ")
               ACE_TEXT("cannot put header in message\n")));
    ACE_DES_FREE(message, this->mb_allocator_->free, ACE_Message_Block);
    message = 0;
  }

  return message;
}

customize_queue_element()

virtual TransportQueueElement* OpenDDS::DCPS::DataLink::customize_queue_element ( TransportQueueElement *  element )

inlineprivatevirtual

Allow derived classes to provide an alternate "customized" queue element for this DataLink (not shared with other links in the DataLinkSet).

Reimplemented in OpenDDS::DCPS::RtpsUdpDataLink.

Definition at line 360 of file DataLink.h.

Referenced by send().

  {
    return element;
  }

data_received()

int OpenDDS::DCPS::DataLink::data_received	(	ReceivedDataSample &	sample,
		const GUID_t &	readerId = GUID_UNKNOWN
	)

This is called by our TransportReceiveStrategy object when it has received a complete data sample. This method will cause the appropriate TransportReceiveListener objects to be told that data_received(). If readerId is not GUID_UNKNOWN, only the TransportReceiveListener with that ID (if one exists) will receive the data.

This method will "deliver" the sample to all TransportReceiveListeners within this DataLink that are interested in the (remote) publisher id that sent the sample.

Definition at line 690 of file DataLink.cpp.

References data_received_i(), and OpenDDS::DCPS::ReceiveListenerSet::SET_EXCLUDED.

Referenced by OpenDDS::DCPS::ReliableSession::deliver_held_data(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::deliver_held_data(), OpenDDS::DCPS::ShmemReceiveStrategy::deliver_sample(), OpenDDS::DCPS::UdpReceiveStrategy::deliver_sample(), OpenDDS::DCPS::TcpReceiveStrategy::deliver_sample(), OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample_i(), OpenDDS::DCPS::ShmemDataLink::request_ack_received(), and OpenDDS::DCPS::MulticastDataLink::sample_received().

{
  data_received_i(sample, readerId, RepoIdSet(), ReceiveListenerSet::SET_EXCLUDED);
  return 0;
}

data_received_i()

void OpenDDS::DCPS::DataLink::data_received_i ( ReceivedDataSample &  sample, const GUID_t &  readerId, const RepoIdSet &  incl_excl, ReceiveListenerSet::ConstrainReceiveSet  constrain  )

private

Definition at line 704 of file DataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), assoc_by_remote_, OpenDDS::DCPS::DataSampleHeader::content_filter_, OpenDDS::DCPS::DataSampleHeader::content_filter_entries_, OpenDDS::DCPS::TransportReceiveListener::data_received(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, default_listener_, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::RcHandle< T >::is_nil(), LM_DEBUG, OpenDDS::DCPS::WeakRcHandle< T >::lock(), OPENDDS_STRING, pub_sub_maps_lock_, OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::ReceiveListenerSet::remove_all(), OpenDDS::DCPS::ReceiveListenerSet::SET_EXCLUDED, OpenDDS::DCPS::to_string(), and OpenDDS::DCPS::Transport_debug_level.

Referenced by data_received(), and data_received_include().

{
  DBG_ENTRY_LVL("DataLink", "data_received_i", 6);
  // Which remote publication sent this message?
  const GUID_t& publication_id = sample.header_.publication_id_;

  // Locate the set of TransportReceiveListeners associated with this
  // DataLink that are interested in hearing about any samples received
  // from the remote publisher_id.
  if (DCPS_debug_level > 9) {
    const GuidConverter converter(publication_id);
    const GuidConverter reader(readerId);
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) DataLink::data_received_i: ")
               ACE_TEXT("from publication %C received sample: %C to readerId %C (%C).\n"),
               OPENDDS_STRING(converter).c_str(),
               to_string(sample.header_).c_str(),
               OPENDDS_STRING(reader).c_str(),
               constrain == ReceiveListenerSet::SET_EXCLUDED ? "SET_EXCLUDED" : "SET_INCLUDED"));
  }

  if (Transport_debug_level > 9) {
    const GuidConverter converter(publication_id);
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) DataLink::data_received_i: ")
               ACE_TEXT("from publication %C received sample: %C.\n"),
               OPENDDS_STRING(converter).c_str(),
               to_string(sample.header_).c_str()));
  }

  ReceiveListenerSet_rch listener_set;
  TransportReceiveListener_rch listener;
  {
    GuardType guard(this->pub_sub_maps_lock_);
    AssocByRemote::iterator iter = assoc_by_remote_.find(publication_id);
    if (iter != assoc_by_remote_.end()) {
      listener_set = iter->second;
    } else {
      listener = this->default_listener_.lock();
    }
  }

  if (listener_set.is_nil()) {
    if (listener) {
      listener->data_received(sample);
    } else {
      // Nobody has any interest in this message.  Drop it on the floor.
      if (Transport_debug_level > 4) {
        const GuidConverter converter(publication_id);
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) DataLink::data_received_i: ")
                   ACE_TEXT(" discarding sample from publication %C due to no listeners.\n"),
                   OPENDDS_STRING(converter).c_str()));
      }
    }
    return;
  }

  if (readerId != GUID_UNKNOWN) {
    listener_set->data_received(sample, readerId);
    return;
  }

#ifndef OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE

  if (sample.header_.content_filter_
      && sample.header_.content_filter_entries_.length()) {
    ReceiveListenerSet subset(*listener_set.in());
    subset.remove_all(sample.header_.content_filter_entries_);
    subset.data_received(sample, incl_excl, constrain);

  } else {
#endif /* OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE */

    if (DCPS_debug_level > 9) {
      // Just get the set to do our dirty work by having it iterate over its
      // collection of TransportReceiveListeners, and invoke the data_received()
      // method on each one.
      OPENDDS_STRING included_ids;
      bool first = true;
      RepoIdSet::const_iterator iter = incl_excl.begin();
      while(iter != incl_excl.end()) {
        included_ids += (first ? "" : "\n") + OPENDDS_STRING(GuidConverter(*iter));
        first = false;
        ++iter;
      }
      ACE_DEBUG((LM_DEBUG, "(%P|%t) DataLink::data_received_i - normal data received to each subscription in listener_set %C ids:%C\n",
                 constrain == ReceiveListenerSet::SET_EXCLUDED ? "exclude" : "include", included_ids.c_str()));
    }
    listener_set->data_received(sample, incl_excl, constrain);
#ifndef OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE
  }

#endif /* OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE */
}

data_received_include()

void OpenDDS::DCPS::DataLink::data_received_include	(	ReceivedDataSample &	sample,
		const RepoIdSet &	incl
	)

Varation of data_received() that allows for excluding a subset of readers by specifying which readers specifically should receive. Any reader ID that does not appear in the include set will be skipped.

Definition at line 698 of file DataLink.cpp.

References data_received_i(), OpenDDS::DCPS::GUID_UNKNOWN, and OpenDDS::DCPS::ReceiveListenerSet::SET_INCLUDED.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::deliver_sample_i().

{
  data_received_i(sample, GUID_UNKNOWN, incl, ReceiveListenerSet::SET_INCLUDED);
}

datalink_release_delay()

ACE_INLINE const TimeDuration & OpenDDS::DCPS::DataLink::datalink_release_delay

(

)

const

Definition at line 62 of file DataLink.inl.

References ACE_INLINE, and datalink_release_delay_.

Referenced by DataLink(), OpenDDS::DCPS::TcpTransport::release_datalink(), and OpenDDS::DCPS::TcpTransport::unbind_link().

{
  return this->datalink_release_delay_;
}

default_listener() [1/2]

ACE_INLINE void OpenDDS::DCPS::DataLink::default_listener

(

const TransportReceiveListener_wrch &

trl

)

Definition at line 352 of file DataLink.inl.

References ACE_INLINE, default_listener_, and pub_sub_maps_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpTransport::configure_i().

{
  GuardType guard(this->pub_sub_maps_lock_);
  this->default_listener_ = trl;
}

default_listener() [2/2]

ACE_INLINE TransportReceiveListener_wrch OpenDDS::DCPS::DataLink::default_listener

(

)

const

Definition at line 360 of file DataLink.inl.

References ACE_INLINE, default_listener_, and pub_sub_maps_lock_.

{
  GuardType guard(this->pub_sub_maps_lock_);
  return this->default_listener_;
}

get_ice_endpoint()

virtual WeakRcHandle<ICE::Endpoint> OpenDDS::DCPS::DataLink::get_ice_endpoint ( ) const

inlinevirtual

Reimplemented in OpenDDS::DCPS::RtpsUdpDataLink.

Definition at line 287 of file DataLink.h.

{ return WeakRcHandle<ICE::Endpoint>(); }

get_next_datalink_id()

ACE_UINT64 OpenDDS::DCPS::DataLink::get_next_datalink_id ( )

staticprotected

Used to provide unique Ids to all DataLink methods.

Definition at line 805 of file DataLink.cpp.

References ACE_ERROR, ACE_TEXT(), id(), and LM_ERROR.

Referenced by DataLink().

{
  static ACE_UINT64 next_id = 0;
  static LockType lock;

  ACE_UINT64 id;
  {
    GuardType guard(lock);
    id = next_id++;

    if (0 == next_id) {
      ACE_ERROR((LM_ERROR,
                 ACE_TEXT("ERROR: DataLink::get_next_datalink_id: ")
                 ACE_TEXT("has rolled over and is reusing ids!\n")));
    }
  }

  return id;
}

get_send_strategy()

ACE_INLINE TransportSendStrategy_rch OpenDDS::DCPS::DataLink::get_send_strategy ( )

protected

Definition at line 374 of file DataLink.inl.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL, send_strategy_, and strategy_lock_.

Referenced by make_reservation(), remove_all_msgs(), remove_sample(), resume_send(), schedule_delayed_release(), send_i(), send_start_i(), send_stop_i(), terminate_send(), and terminate_send_if_suspended().

{
  GuardType guard(strategy_lock_);
  return send_strategy_;
}

handle_close()

int OpenDDS::DCPS::DataLink::handle_close ( ACE_HANDLE  h, ACE_Reactor_Mask  m  )

virtual

Reimplemented from ACE_Event_Handler.

Definition at line 1101 of file DataLink.cpp.

References handle_timeout(), TheServiceParticipant, ACE_Event_Handler::TIMER_MASK, and ACE_Time_Value::zero.

{
  ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());

  if (h == ACE_INVALID_HANDLE && m == TIMER_MASK) {
    // Reactor is shutting down with this timer still pending.
    // Take the same cleanup actions as if the timeout had expired.
    handle_timeout(ACE_Time_Value::zero, 0);
  }

  return 0;
}

handle_exception()

int OpenDDS::DCPS::DataLink::handle_exception ( ACE_HANDLE  )

virtual

Reactor invokes this after being notified in schedule_stop or cancel_release.

Reimplemented from ACE_Event_Handler.

Definition at line 270 of file DataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), ACE_Reactor_Timer_Interface::cancel_timer(), OpenDDS::DCPS::DCPS_debug_level, handle_timeout(), impl(), impl_, OpenDDS::DCPS::TimePoint_T< AceClock >::is_zero(), LM_DEBUG, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), ACE_Event_Handler::reactor(), ACE_Reactor_Timer_Interface::schedule_timer(), scheduled_to_stop_at_, scheduling_release_, stop(), TheServiceParticipant, OpenDDS::DCPS::TransportImpl::timer(), OpenDDS::DCPS::TimeDuration::value(), and ACE_Time_Value::zero.

{
  ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());

  const MonotonicTimePoint now = MonotonicTimePoint::now();
  if (scheduled_to_stop_at_.is_zero()) {
    if (DCPS_debug_level > 0) {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) DataLink::handle_exception() - not scheduling or stopping\n")));
    }
    TransportImpl_rch impl = impl_.lock();
    if (impl) {
      ACE_Reactor_Timer_Interface* reactor = impl->timer();
      if (reactor && reactor->cancel_timer(this) > 0) {
        if (DCPS_debug_level > 0) {
          ACE_DEBUG((LM_DEBUG,
                     ACE_TEXT("(%P|%t) DataLink::handle_exception() - cancelled future release timer\n")));
        }
      }
    }
    return 0;
  } else if (scheduled_to_stop_at_ <= now) {
    if (this->scheduling_release_) {
      if (DCPS_debug_level > 0) {
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) DataLink::handle_exception() - delay already elapsed so handle_timeout now\n")));
      }
      this->handle_timeout(ACE_Time_Value::zero, 0);
      return 0;
    }
    if (DCPS_debug_level > 0) {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) DataLink::handle_exception() - stopping now\n")));
    }
    this->stop();
    return 0;
  } else /* SCHEDULE TO STOP IN THE FUTURE*/ {
    if (DCPS_debug_level > 0) {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) DataLink::handle_exception() - (delay) scheduling timer for future release\n")));
    }
    TransportImpl_rch impl = impl_.lock();
    if (impl) {
      ACE_Reactor_Timer_Interface* reactor = impl->timer();
      const TimeDuration future_release_time = scheduled_to_stop_at_ - now;
      reactor->schedule_timer(this, 0, future_release_time.value());
    }
  }
  return 0;
}

handle_send_request_ack()

bool OpenDDS::DCPS::DataLink::handle_send_request_ack ( TransportQueueElement *  element )

privatevirtual

Reimplemented in OpenDDS::DCPS::TcpDataLink.

Definition at line 1191 of file DataLink.cpp.

References OpenDDS::DCPS::TransportQueueElement::data_delivered().

Referenced by send().

{
  element->data_delivered();
  return true;
}

handle_timeout()

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

virtual

Reimplemented from ACE_Event_Handler.

Definition at line 1081 of file DataLink.cpp.

References assoc_by_local_, assoc_by_remote_, impl(), impl_, OpenDDS::DCPS::TimePoint_T< AceClock >::is_zero(), LM_DEBUG, scheduled_to_stop_at_, stop(), TheServiceParticipant, OpenDDS::DCPS::TransportImpl::unbind_link(), and VDBG_LVL.

Referenced by handle_close(), and handle_exception().

{
  ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());

  if (!scheduled_to_stop_at_.is_zero()) {
    VDBG_LVL((LM_DEBUG, "(%P|%t) DataLink::handle_timeout called\n"), 4);
    {
      TransportImpl_rch impl = impl_.lock();
      if (impl) {
        impl->unbind_link(this);
      }
    }
    if (assoc_by_remote_.empty() && assoc_by_local_.empty()) {
      this->stop();
    }
  }
  return 0;
}

id()

ACE_INLINE DataLinkIdType OpenDDS::DCPS::DataLink::id

(

void

)

const

Obtain a unique identifier for this DataLink object.

Definition at line 205 of file DataLink.inl.

References ACE_INLINE, DBG_ENTRY_LVL, and id_.

Referenced by get_next_datalink_id(), OpenDDS::DCPS::DataLinkSet::insert_link(), and OpenDDS::DCPS::DataLinkSet::remove_link().

{
  DBG_ENTRY_LVL("DataLink","id",6);
  return id_;
}

impl()

TransportImpl_rch OpenDDS::DCPS::DataLink::impl

(

void

)

const

Definition at line 105 of file DataLink.cpp.

References impl_.

Referenced by OpenDDS::DCPS::TcpConnection::active_open(), OpenDDS::DCPS::TcpConnection::active_reconnect_i(), OpenDDS::DCPS::DataReaderImpl::add_link(), OpenDDS::DCPS::TcpConnection::close(), OpenDDS::DCPS::RtpsUdpDataLink::config(), OpenDDS::DCPS::UdpDataLink::control_received(), OpenDDS::DCPS::RtpsUdpDataLink::get_ice_endpoint(), handle_exception(), OpenDDS::DCPS::TcpConnection::handle_timeout(), handle_timeout(), notify_reactor(), OpenDDS::DCPS::UdpDataLink::open(), OpenDDS::DCPS::TcpDataLink::reconnect(), release_reservations(), release_resources(), OpenDDS::DCPS::TcpConnection::set_datalink(), OpenDDS::DCPS::MulticastDataLink::transport(), OpenDDS::DCPS::ShmemDataLink::transport(), OpenDDS::DCPS::RtpsUdpDataLink::transport(), and transport_shutdown().

{
  return impl_.lock();
}

invoke_on_start_callbacks() [1/2]

void OpenDDS::DCPS::DataLink::invoke_on_start_callbacks

(

bool

success

)

Definition at line 194 of file DataLink.cpp.

References OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::WeakRcHandle< T >::lock(), on_start_callbacks_, ACE_Guard< ACE_LOCK >::release(), and strategy_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::associated(), OpenDDS::DCPS::TcpTransport::async_connect_failed(), OpenDDS::DCPS::TcpTransport::connect_datalink(), OpenDDS::DCPS::TcpDataLink::do_association_actions(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_i(), OpenDDS::DCPS::TcpDataLink::request_ack_received(), OpenDDS::DCPS::ShmemDataLink::request_ack_received(), start(), and OpenDDS::DCPS::MulticastSession::synack_received().

{
  const DataLink_rch link(success ? this : 0, inc_count());

  while (true) {
    GuardType guard(strategy_lock_);

    if (on_start_callbacks_.empty()) {
      break;
    }

    GUID_t remote = GUID_UNKNOWN;
    TransportClient_wrch client;
    OnStartCallbackMap::iterator it = on_start_callbacks_.begin();
    if (it != on_start_callbacks_.end()) {
      remote = it->first;
      RepoToClientMap::iterator it2 = it->second.begin();
      if (it2 != it->second.end()) {
        client = it2->second;
        it->second.erase(it2);
        if (it->second.empty()) {
          on_start_callbacks_.erase(it);
        }
      }
    }

    guard.release();
    if (success) {
      TransportClient_rch client_lock = client.lock();
      if (client_lock) {
        client_lock->use_datalink(remote, link);
      }
    }
  }
}

invoke_on_start_callbacks() [2/2]

bool OpenDDS::DCPS::DataLink::invoke_on_start_callbacks	(	const GUID_t &	local,
		const GUID_t &	remote,
		bool	success
	)

Definition at line 230 of file DataLink.cpp.

References OpenDDS::DCPS::WeakRcHandle< T >::lock(), on_start_callbacks_, pending_on_starts_, and strategy_lock_.

{
  const DataLink_rch link(success ? this : 0, inc_count());

  TransportClient_wrch client;
  bool made_callback = false;

  {
    GuardType guard(strategy_lock_);

    OnStartCallbackMap::iterator it = on_start_callbacks_.find(remote);
    if (it != on_start_callbacks_.end()) {
      RepoToClientMap::iterator it2 = it->second.find(local);
      if (it2 != it->second.end()) {
        client = it2->second;
        it->second.erase(it2);
        if (it->second.empty()) {
          on_start_callbacks_.erase(it);
        }
      } else {
        pending_on_starts_[remote].insert(local);
      }
    } else {
      pending_on_starts_[remote].insert(local);
    }
  }

  if (success) {
    TransportClient_rch client_lock = client.lock();
    if (client_lock) {
      client_lock->use_datalink(remote, link);
      made_callback = true;
    }
  }

  return made_callback;
}

is_active() [1/2]

ACE_INLINE bool & OpenDDS::DCPS::DataLink::is_active

(

)

Definition at line 49 of file DataLink.inl.

References ACE_INLINE, and is_active_.

Referenced by OpenDDS::DCPS::MulticastDataLink::check_header(), OpenDDS::DCPS::TcpTransport::release_datalink(), OpenDDS::DCPS::MulticastDataLink::sample_received(), and OpenDDS::DCPS::TcpTransport::unbind_link().

{
  return this->is_active_;
}

is_active() [2/2]

ACE_INLINE bool OpenDDS::DCPS::DataLink::is_active

(

)

const

Definition at line 56 of file DataLink.inl.

References ACE_INLINE, and is_active_.

{
  return this->is_active_;
}

is_leading()

virtual bool OpenDDS::DCPS::DataLink::is_leading ( const GUID_t &  , const GUID_t &    ) const

inlinevirtual

Reimplemented in OpenDDS::DCPS::RtpsUdpDataLink.

Definition at line 289 of file DataLink.h.

                                                 { return false; }

is_loopback() [1/2]

ACE_INLINE bool & OpenDDS::DCPS::DataLink::is_loopback

(

void

)

Definition at line 35 of file DataLink.inl.

References ACE_INLINE, and is_loopback_.

Referenced by OpenDDS::DCPS::TcpTransport::release_datalink(), and OpenDDS::DCPS::TcpTransport::unbind_link().

{
  return this->is_loopback_;
}

is_loopback() [2/2]

ACE_INLINE bool OpenDDS::DCPS::DataLink::is_loopback

(

void

)

const

Definition at line 42 of file DataLink.inl.

References ACE_INLINE, and is_loopback_.

{
  return this->is_loopback_;
}

is_target()

bool OpenDDS::DCPS::DataLink::is_target

(

const GUID_t &

remote_id

)

This is called on publisher side to see if this link communicates with the provided sub or by the subscriber side to see if this link communicates with the provided pub

Definition at line 1013 of file DataLink.cpp.

References assoc_by_remote_, and pub_sub_maps_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpReceiveStrategy::reassemble_i().

{
  GuardType guard(this->pub_sub_maps_lock_);
  return assoc_by_remote_.count(remote_id);
}

make_reservation() [1/2]

int OpenDDS::DCPS::DataLink::make_reservation ( const GUID_t &  remote_subscription_id, const GUID_t &  local_publication_id, const TransportSendListener_wrch &  send_listener, bool  reliable  )

virtual

Only called by our TransportImpl object.

Return Codes: 0 means successful reservation made. -1 means failure.

Reimplemented in OpenDDS::DCPS::TcpDataLink.

Definition at line 398 of file DataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), assoc_by_local_, assoc_by_remote_, OpenDDS::DCPS::DataLink::LocalAssociationInfo::associated_, OpenDDS::DCPS::LogGuid::c_str(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, get_send_strategy(), OpenDDS::DCPS::RcHandle< T >::is_nil(), OpenDDS::DCPS::TransportSendStrategy::link_released(), LM_DEBUG, pub_sub_maps_lock_, OpenDDS::DCPS::DataLink::LocalAssociationInfo::reliable_, and send_listeners_.

Referenced by OpenDDS::DCPS::TransportClient::add_link(), OpenDDS::DCPS::ShmemDataLink::make_reservation(), OpenDDS::DCPS::TcpDataLink::make_reservation(), OpenDDS::DCPS::MulticastDataLink::make_reservation(), and OpenDDS::DCPS::RtpsUdpDataLink::make_reservation().

{
  DBG_ENTRY_LVL("DataLink", "make_reservation", 6);

  if (DCPS_debug_level > 9) {
    LogGuid local_log(local_publication_id), remote_log(remote_subscription_id);
    ACE_DEBUG((LM_DEBUG,
        ACE_TEXT("(%P|%t) DataLink::make_reservation() - ")
        ACE_TEXT("creating association local publication %C ")
        ACE_TEXT("<--> with remote subscription %C.\n"),
        local_log .c_str(),
        remote_log.c_str()));
  }

  TransportSendStrategy_rch strategy = get_send_strategy();

  if (strategy) {
    strategy->link_released(false);
  }

  {
    GuardType guard(pub_sub_maps_lock_);

    LocalAssociationInfo& info = assoc_by_local_[local_publication_id];
    info.reliable_ = reliable;
    info.associated_.insert(remote_subscription_id);
    ReceiveListenerSet_rch& rls = assoc_by_remote_[remote_subscription_id];

    if (rls.is_nil())
      rls = make_rch<ReceiveListenerSet>();
    rls->insert(local_publication_id, TransportReceiveListener_rch());

    send_listeners_.insert(std::make_pair(local_publication_id, send_listener));
  }
  return 0;
}

make_reservation() [2/2]

int OpenDDS::DCPS::DataLink::make_reservation ( const GUID_t &  remote_publication_id, const GUID_t &  local_subscription_id, const TransportReceiveListener_wrch &  receive_listener, bool  reliable  )

virtual

Only called by our TransportImpl object.

Return Codes: 0 means successful reservation made. -1 means failure.

Reimplemented in OpenDDS::DCPS::RtpsUdpDataLink, OpenDDS::DCPS::MulticastDataLink, OpenDDS::DCPS::TcpDataLink, and OpenDDS::DCPS::ShmemDataLink.

Definition at line 439 of file DataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), assoc_by_local_, assoc_by_remote_, OpenDDS::DCPS::DataLink::LocalAssociationInfo::associated_, OpenDDS::DCPS::LogGuid::c_str(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, get_send_strategy(), OpenDDS::DCPS::RcHandle< T >::is_nil(), OpenDDS::DCPS::TransportSendStrategy::link_released(), LM_DEBUG, pub_sub_maps_lock_, recv_listeners_, and OpenDDS::DCPS::DataLink::LocalAssociationInfo::reliable_.

{
  DBG_ENTRY_LVL("DataLink", "make_reservation", 6);

  if (DCPS_debug_level > 9) {
    LogGuid local(local_subscription_id), remote(remote_publication_id);
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) DataLink::make_reservation() - ")
               ACE_TEXT("creating association local subscription %C ")
               ACE_TEXT("<--> with remote publication %C.\n"),
               local.c_str(), remote.c_str()));
  }

  TransportSendStrategy_rch strategy = get_send_strategy();

  if (strategy) {
    strategy->link_released(false);
  }

  {
    GuardType guard(pub_sub_maps_lock_);

    LocalAssociationInfo& info = assoc_by_local_[local_subscription_id];
    info.reliable_ = reliable;
    info.associated_.insert(remote_publication_id);
    ReceiveListenerSet_rch& rls = assoc_by_remote_[remote_publication_id];

    if (rls.is_nil())
      rls = make_rch<ReceiveListenerSet>();
    rls->insert(local_subscription_id, receive_listener);

    recv_listeners_.insert(std::make_pair(local_subscription_id,
                                          receive_listener));
  }
  return 0;
}

network_change()

void OpenDDS::DCPS::DataLink::network_change ( ) const

protected

Definition at line 1212 of file DataLink.cpp.

References pub_sub_maps_lock_, recv_listeners_, send_listeners_, and OpenDDS::DCPS::TransportReceiveListener::transport_discovery_change().

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::on_data_available().

{
  IdToSendListenerMap send_listeners;
  IdToRecvListenerMap recv_listeners;
  {
    GuardType guard(pub_sub_maps_lock_);
    send_listeners = send_listeners_;
    recv_listeners = recv_listeners_;
  }
  for (IdToSendListenerMap::const_iterator itr = send_listeners.begin();
       itr != send_listeners.end(); ++itr) {
    TransportSendListener_rch tsl = itr->second.lock();
    if (tsl) {
      tsl->transport_discovery_change();
    }
  }

  for (IdToRecvListenerMap::const_iterator itr = recv_listeners.begin();
       itr != recv_listeners.end(); ++itr) {
    TransportReceiveListener_rch trl = itr->second.lock();
    if (trl) {
      trl->transport_discovery_change();
    }
  }
}

notify()

void OpenDDS::DCPS::DataLink::notify

(

ConnectionNotice

notice

)

Notify the datawriters and datareaders that the connection is disconnected, lost, or reconnected. The datareader/datawriter will notify the corresponding listener.

Definition at line 848 of file DataLink.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), assoc_by_local_, connection_notice_as_str(), DBG_ENTRY_LVL, DISCONNECTED, LM_DEBUG, LM_ERROR, LOST, OpenDDS::DCPS::TransportReceiveListener::notify_subscription_disconnected(), OpenDDS::DCPS::TransportReceiveListener::notify_subscription_lost(), OpenDDS::DCPS::TransportReceiveListener::notify_subscription_reconnected(), OPENDDS_STRING, pub_sub_maps_lock_, RECONNECTED, recv_listeners_, send_listeners_, OpenDDS::DCPS::set_to_seq(), OpenDDS::DCPS::Transport_debug_level, and VDBG.

Referenced by OpenDDS::DCPS::TcpConnection::active_reconnect_i(), OpenDDS::DCPS::TcpConnection::active_reconnect_open(), OpenDDS::DCPS::TcpConnection::handle_close(), OpenDDS::DCPS::TcpConnection::notify_connection_lost(), OpenDDS::DCPS::TcpConnection::passive_reconnect_i(), and OpenDDS::DCPS::TcpConnection::transfer().

{
  DBG_ENTRY_LVL("DataLink", "notify", 6);

  VDBG((LM_DEBUG,
        ACE_TEXT("(%P|%t) DataLink::notify: this(%X) notify %C\n"),
        this,
        connection_notice_as_str(notice)));

  GuardType guard(this->pub_sub_maps_lock_);

  // Notify the datawriters
  // the lost publications due to a connection problem.
  for (IdToSendListenerMap::iterator itr = send_listeners_.begin();
       itr != send_listeners_.end(); ++itr) {

    TransportSendListener_rch tsl = itr->second.lock();

    if (tsl) {
      if (Transport_debug_level > 0) {
        GuidConverter converter(itr->first);
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) DataLink::notify: ")
                   ACE_TEXT("notify pub %C %C.\n"),
                   OPENDDS_STRING(converter).c_str(),
                   connection_notice_as_str(notice)));
      }
      AssocByLocal::iterator local_it = assoc_by_local_.find(itr->first);
      if (local_it == assoc_by_local_.end()) {
        if (Transport_debug_level) {
          GuidConverter converter(itr->first);
          ACE_DEBUG((LM_DEBUG,
                     ACE_TEXT("(%P|%t) DataLink::notify: ")
                     ACE_TEXT("try to notify pub %C %C - no associations to notify.\n"),
                     OPENDDS_STRING(converter).c_str(),
                     connection_notice_as_str(notice)));
        }
        break;
      }
      const RepoIdSet& rids = local_it->second.associated_;

      ReaderIdSeq subids;
      set_to_seq(rids, subids);

      switch (notice) {
      case DISCONNECTED:
        tsl->notify_publication_disconnected(subids);
        break;

      case RECONNECTED:
        tsl->notify_publication_reconnected(subids);
        break;

      case LOST:
        tsl->notify_publication_lost(subids);
        break;

      default:
        ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) ERROR: DataLink::notify: ")
                   ACE_TEXT("unknown notice to TransportSendListener\n")));
        break;
      }

    } else {
      if (Transport_debug_level > 0) {
        GuidConverter converter(itr->first);
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) DataLink::notify: ")
                   ACE_TEXT("not notify pub %C %C\n"),
                   OPENDDS_STRING(converter).c_str(),
                   connection_notice_as_str(notice)));
      }
    }
  }

  // Notify the datareaders registered with TransportImpl
  // the lost subscriptions due to a connection problem.
  for (IdToRecvListenerMap::iterator itr = recv_listeners_.begin();
       itr != recv_listeners_.end(); ++itr) {

    TransportReceiveListener_rch trl = itr->second.lock();

    if (trl) {
      if (Transport_debug_level > 0) {
        GuidConverter converter(itr->first);
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) DataLink::notify: ")
                   ACE_TEXT("notify sub %C %C.\n"),
                   OPENDDS_STRING(converter).c_str(),
                   connection_notice_as_str(notice)));
      }
      AssocByLocal::iterator local_it = assoc_by_local_.find(itr->first);
      if (local_it == assoc_by_local_.end()) {
        if (Transport_debug_level) {
          GuidConverter converter(itr->first);
          ACE_DEBUG((LM_DEBUG,
                     ACE_TEXT("(%P|%t) DataLink::notify: ")
                     ACE_TEXT("try to notify sub %C %C - no associations to notify.\n"),
                     OPENDDS_STRING(converter).c_str(),
                     connection_notice_as_str(notice)));
        }
        break;
      }
      const RepoIdSet& rids = local_it->second.associated_;

      WriterIdSeq pubids;
      set_to_seq(rids, pubids);

      switch (notice) {
      case DISCONNECTED:
        trl->notify_subscription_disconnected(pubids);
        break;

      case RECONNECTED:
        trl->notify_subscription_reconnected(pubids);
        break;

      case LOST:
        trl->notify_subscription_lost(pubids);
        break;

      default:
        ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) ERROR: DataLink::notify: ")
                   ACE_TEXT("unknown notice to datareader.\n")));
        break;
      }

    } else {
      if (Transport_debug_level > 0) {
        GuidConverter converter(itr->first);
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) DataLink::notify: ")
                   ACE_TEXT("not notify sub %C subscription lost.\n"),
                   OPENDDS_STRING(converter).c_str()));
      }

    }
  }
}

notify_reactor()

void OpenDDS::DCPS::DataLink::notify_reactor ( void  )

private

Definition at line 340 of file DataLink.cpp.

References impl(), impl_, ACE_Reactor::notify(), ACE_Event_Handler::reactor(), and OpenDDS::DCPS::TransportImpl::reactor_task().

Referenced by cancel_release(), and schedule_stop().

{
  TransportImpl_rch impl = impl_.lock();
  if (impl) {
    ReactorTask_rch rt(impl->reactor_task());
    if (rt) {
      ACE_Reactor* reactor = rt->get_reactor();
      if (reactor) {
        reactor->notify(this);
      }
    }
  }
}

OPENDDS_MAP_CMP() [1/7]

typedef OpenDDS::DCPS::DataLink::OPENDDS_MAP_CMP ( GUID_t  , TransportSendListener_wrch  , GUID_tKeyLessThan    )

private

Map publication Id value to TransportSendListener.

OPENDDS_MAP_CMP() [2/7]

typedef OpenDDS::DCPS::DataLink::OPENDDS_MAP_CMP ( GUID_t  , TransportReceiveListener_wrch  , GUID_tKeyLessThan    )

private

Map subscription Id value to TransportReceieveListener.

OPENDDS_MAP_CMP() [3/7]

typedef OpenDDS::DCPS::DataLink::OPENDDS_MAP_CMP ( GUID_t  , ReceiveListenerSet_rch  , GUID_tKeyLessThan    )

private

OPENDDS_MAP_CMP() [4/7]

typedef OpenDDS::DCPS::DataLink::OPENDDS_MAP_CMP ( GUID_t  , LocalAssociationInfo  , GUID_tKeyLessThan    )

private

OPENDDS_MAP_CMP() [5/7]

typedef OpenDDS::DCPS::DataLink::OPENDDS_MAP_CMP ( GUID_t  , TransportClient_wrch  , GUID_tKeyLessThan    )

protected

OPENDDS_MAP_CMP() [6/7]

typedef OpenDDS::DCPS::DataLink::OPENDDS_MAP_CMP ( GUID_t  , RepoToClientMap  , GUID_tKeyLessThan    )

protected

OPENDDS_MAP_CMP() [7/7]

typedef OpenDDS::DCPS::DataLink::OPENDDS_MAP_CMP ( GUID_t  , RepoIdSet  , GUID_tKeyLessThan    )

protected

peer_ids()

GUIDSeq * OpenDDS::DCPS::DataLink::peer_ids ( const GUID_t &  local_id ) const

protected

For a given local GUID_t (publication or subscription), return the list of remote peer GUID_ts (subscriptions or publications) that this link knows about due to make_reservation().

Definition at line 490 of file DataLink.cpp.

References assoc_by_local_, pub_sub_maps_lock_, and OpenDDS::DCPS::set_to_seq().

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element(), and OpenDDS::DCPS::RtpsUdpDataLink::get_addresses_i().

{
  GuardType guard(pub_sub_maps_lock_);

  const AssocByLocal::const_iterator iter = assoc_by_local_.find(local_id);

  if (iter == assoc_by_local_.end())
    return 0;

  GUIDSeq_var result = new GUIDSeq;
  set_to_seq(iter->second.associated_, static_cast<GUIDSeq&>(result));
  return result._retn();
}

pre_stop_i()

void OpenDDS::DCPS::DataLink::pre_stop_i ( )

virtual

Called before release the datalink or before shutdown to let the concrete DataLink to do anything necessary.

Reimplemented in OpenDDS::DCPS::RtpsUdpDataLink, and OpenDDS::DCPS::TcpDataLink.

Definition at line 993 of file DataLink.cpp.

References thr_per_con_send_task_.

Referenced by OpenDDS::DCPS::TcpDataLink::pre_stop_i(), OpenDDS::DCPS::RtpsUdpDataLink::pre_stop_i(), and stop().

{
  if (this->thr_per_con_send_task_ != 0) {
    this->thr_per_con_send_task_->close(1);
  }
}

prepare_release()

void OpenDDS::DCPS::DataLink::prepare_release ( )

private

Save current sub and pub association maps for releasing and create empty maps for new associations.

Definition at line 1045 of file DataLink.cpp.

References ACE_ERROR, ACE_TEXT(), assoc_by_local_, assoc_releasing_, LM_ERROR, and pub_sub_maps_lock_.

Referenced by release_resources().

{
  GuardType guard(this->pub_sub_maps_lock_);

  if (!assoc_releasing_.empty()) {
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) DataLink::prepare_release: ")
               ACE_TEXT("already prepared for release.\n")));
    return;
  }

  assoc_releasing_ = assoc_by_local_;
}

recv_listener_for()

ACE_INLINE TransportReceiveListener_rch OpenDDS::DCPS::DataLink::recv_listener_for ( const GUID_t &  sub_id ) const

private

Definition at line 338 of file DataLink.inl.

References ACE_INLINE, and recv_listeners_.

Referenced by clear_associations().

{
  // sub_map_ (and recv_listeners_) are already locked when entering this
  // private method.
  IdToRecvListenerMap::const_iterator found =
    this->recv_listeners_.find(sub_id);
  if (found == this->recv_listeners_.end()) {
    return TransportReceiveListener_rch();
  }
  return found->second.lock();
}

release_remote_i()

virtual void OpenDDS::DCPS::DataLink::release_remote_i ( const GUID_t &  )

inlineprivatevirtual

Reimplemented in OpenDDS::DCPS::MulticastDataLink.

Definition at line 366 of file DataLink.h.

Referenced by release_reservations().

{}

release_reservations()

void OpenDDS::DCPS::DataLink::release_reservations	(	GUID_t	remote_id,
		GUID_t	local_id,
		DataLinkSetMap &	released_locals
	)

This will release reservations that were made by one of the make_reservation() methods. All we know is that the supplied GUID_t is considered to be a remote id. It could be a remote subscriber or a remote publisher.

This gets invoked when a TransportClient::remove_associations() call has been made. Because this DataLink can be shared amongst different TransportClient objects, and different threads could be "managing" the different TransportClient objects, we need to make sure that this release_reservations() works in conjunction with a simultaneous call (in another thread) to one of this DataLink's make_reservation() methods.

Definition at line 512 of file DataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), assoc_by_local_, assoc_by_remote_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, impl(), impl_, OpenDDS::DCPS::RcHandle< T >::is_nil(), LM_DEBUG, OPENDDS_STRING, pub_sub_maps_lock_, OpenDDS::DCPS::rchandle_from(), ACE_Guard< ACE_LOCK >::release(), OpenDDS::DCPS::TransportImpl::release_datalink(), release_remote_i(), release_reservations_i(), remove_startup_callbacks(), stopped_, and VDBG_LVL.

Referenced by OpenDDS::DCPS::TransportClient::disassociate().

{
  DBG_ENTRY_LVL("DataLink", "release_reservations", 6);

  if (DCPS_debug_level > 9) {
    GuidConverter local(local_id);
    GuidConverter remote(remote_id);
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) DataLink::release_reservations() - ")
               ACE_TEXT("releasing association local: %C ")
               ACE_TEXT("<--> with remote %C.\n"),
               OPENDDS_STRING(local).c_str(),
               OPENDDS_STRING(remote).c_str()));
  }

  remove_startup_callbacks(local_id, remote_id);

  //let the specific class release its reservations
  //done this way to prevent deadlock of holding pub_sub_maps_lock_
  //then obtaining a specific class lock in release_reservations_i
  //which reverses lock ordering of the active send logic of needing
  //the specific class lock before obtaining the over arching DataLink
  //pub_sub_maps_lock_
  this->release_reservations_i(remote_id, local_id);

  bool release_remote_required = false;
  {
    GuardType guard(this->pub_sub_maps_lock_);

    if (this->stopped_) return;

    ReceiveListenerSet_rch& rls = assoc_by_remote_[remote_id];
    if (rls->size() == 1) {
      assoc_by_remote_.erase(remote_id);
      release_remote_required = true;
    } else {
      rls->remove(local_id);
    }
    RepoIdSet& ris = assoc_by_local_[local_id].associated_;
    if (ris.size() == 1) {
      DataLinkSet_rch& links = released_locals[local_id];
      if (links.is_nil()) {
        links = make_rch<DataLinkSet>();
      }
      links->insert_link(rchandle_from(this));
      assoc_by_local_.erase(local_id);
    } else {
      ris.erase(remote_id);
    }

    if (assoc_by_local_.empty()) {
      VDBG_LVL((LM_DEBUG,
                ACE_TEXT("(%P|%t) DataLink::release_reservations: ")
                ACE_TEXT("release_datalink due to no remaining pubs or subs.\n")), 5);

      guard.release();
      TransportImpl_rch impl = impl_.lock();
      if (impl) {
        impl->release_datalink(this);
      }
    }
  }
  if (release_remote_required) {
    release_remote_i(remote_id);
  }
}

release_reservations_i()

virtual void OpenDDS::DCPS::DataLink::release_reservations_i ( const GUID_t &  , const GUID_t &    )

inlineprivatevirtual

Reimplemented in OpenDDS::DCPS::RtpsUdpDataLink, and OpenDDS::DCPS::MulticastDataLink.

Definition at line 367 of file DataLink.h.

Referenced by release_reservations().

                                                       {}

release_resources()

void OpenDDS::DCPS::DataLink::release_resources

(

)

Definition at line 1001 of file DataLink.cpp.

References DBG_ENTRY_LVL, impl(), impl_, prepare_release(), and OpenDDS::DCPS::TransportImpl::release_link_resources().

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

{
  DBG_ENTRY_LVL("DataLink", "release_resources", 6);

  this->prepare_release();
  TransportImpl_rch impl = impl_.lock();
  if (impl) {
    impl->release_link_resources(this);
  }
}

remove_all_msgs()

ACE_INLINE void OpenDDS::DCPS::DataLink::remove_all_msgs ( const GUID_t &  pub_id )

virtual

Reimplemented in OpenDDS::DCPS::RtpsUdpDataLink.

Definition at line 190 of file DataLink.inl.

References ACE_INLINE, DBG_ENTRY_LVL, get_send_strategy(), OpenDDS::DCPS::RcHandle< T >::is_nil(), and OpenDDS::DCPS::TransportSendStrategy::remove_all_msgs().

{
  DBG_ENTRY_LVL("DataLink","remove_all_msgs",6);

  // This one is easy.  Simply delegate to our TransportSendStrategy
  // data member.

  TransportSendStrategy_rch strategy = get_send_strategy();

  if (!strategy.is_nil()) {
    strategy->remove_all_msgs(pub_id);
  }
}

remove_listener()

ACE_INLINE void OpenDDS::DCPS::DataLink::remove_listener

(

const GUID_t &

local_id

)

Either send or receive listener for this local_id should be removed from internal DataLink structures so it no longer receives events.

Definition at line 289 of file DataLink.inl.

References ACE_DEBUG, ACE_INLINE, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), LM_DEBUG, pub_sub_maps_lock_, recv_listeners_, send_listeners_, and OpenDDS::DCPS::Transport_debug_level.

Referenced by OpenDDS::DCPS::TransportClient::disassociate().

{
  GuardType guard(pub_sub_maps_lock_);
  {
    IdToSendListenerMap::iterator pos = send_listeners_.find(local_id);
    if (pos != send_listeners_.end()) {
      send_listeners_.erase(pos);
      if (Transport_debug_level > 5) {
        LogGuid logger(local_id);
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) DataLink::remove_listener: ")
                   ACE_TEXT("removed %C from send_listeners\n"),
                   logger.c_str()));
      }
      return;
    }
  }
  {
    IdToRecvListenerMap::iterator pos = recv_listeners_.find(local_id);
    if (pos != recv_listeners_.end()) {
      recv_listeners_.erase(pos);
      if (Transport_debug_level > 5) {
        LogGuid logger(local_id);
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) DataLink::remove_listener: ")
                   ACE_TEXT("removed %C from recv_listeners\n"),
                   logger.c_str()));
      }
      return;
    }
  }
}

remove_on_start_callback()

void OpenDDS::DCPS::DataLink::remove_on_start_callback	(	const TransportClient_wrch &	client,
		const GUID_t &	remote
	)

Definition at line 173 of file DataLink.cpp.

References OpenDDS::DCPS::WeakRcHandle< T >::lock(), on_start_callbacks_, and strategy_lock_.

{
  TransportClient_rch client_lock = client.lock();
  if (client_lock) {
    const GUID_t id = client_lock->get_guid();

    GuardType guard(strategy_lock_);
    OnStartCallbackMap::iterator it = on_start_callbacks_.find(remote);
    if (it != on_start_callbacks_.end()) {
      RepoToClientMap::iterator it2 = it->second.find(id);
      if (it2 != it->second.end()) {
        it->second.erase(it2);
        if (it->second.empty()) {
          on_start_callbacks_.erase(it);
        }
      }
    }
  }
}

remove_sample()

ACE_INLINE RemoveResult OpenDDS::DCPS::DataLink::remove_sample ( const DataSampleElement *  sample )

virtual

This method is essentially an "undo_send()" method. It's goal is to remove all traces of the sample from this DataLink (if the sample is even known to the DataLink).

Reimplemented in OpenDDS::DCPS::RtpsUdpDataLink.

Definition at line 167 of file DataLink.inl.

References ACE_INLINE, DBG_ENTRY_LVL, get_send_strategy(), OpenDDS::DCPS::RcHandle< T >::is_nil(), LM_DEBUG, OpenDDS::DCPS::REMOVE_FOUND, OpenDDS::DCPS::REMOVE_NOT_FOUND, OpenDDS::DCPS::REMOVE_RELEASED, OpenDDS::DCPS::TransportSendStrategy::remove_sample(), thr_per_con_send_task_, and VDBG.

{
  DBG_ENTRY_LVL("DataLink", "remove_sample", 6);

  if (this->thr_per_con_send_task_ != 0) {
    const RemoveResult rr = this->thr_per_con_send_task_->remove_sample(sample);
    if (rr == REMOVE_RELEASED || rr == REMOVE_FOUND) {
      VDBG((LM_DEBUG, "(%P|%t) DBG:   "
            "Removed sample from ThreadPerConnection queue.\n"));
      return rr;
    }
  }

  TransportSendStrategy_rch strategy = get_send_strategy();

  if (!strategy.is_nil()) {
    return strategy->remove_sample(sample);
  }

  return REMOVE_NOT_FOUND;
}

remove_startup_callbacks()

void OpenDDS::DCPS::DataLink::remove_startup_callbacks	(	const GUID_t &	local,
		const GUID_t &	remote
	)

Definition at line 146 of file DataLink.cpp.

References on_start_callbacks_, pending_on_starts_, and strategy_lock_.

Referenced by release_reservations().

{
  GuardType guard(strategy_lock_);

  OnStartCallbackMap::iterator oit = on_start_callbacks_.find(remote);
  if (oit != on_start_callbacks_.end()) {
    RepoToClientMap::iterator oit2 = oit->second.find(local);
    if (oit2 != oit->second.end()) {
      oit->second.erase(oit2);
      if (oit->second.empty()) {
        on_start_callbacks_.erase(oit);
      }
    }
  }
  PendingOnStartsMap::iterator pit = pending_on_starts_.find(remote);
  if (pit != pending_on_starts_.end()) {
    RepoIdSet::iterator pit2 = pit->second.find(local);
    if (pit2 != pit->second.end()) {
      pit->second.erase(pit2);
      if (pit->second.empty()) {
        pending_on_starts_.erase(pit);
      }
    }
  }
}

replay_durable_data()

void OpenDDS::DCPS::DataLink::replay_durable_data ( const GUID_t &  local_pub_id, const GUID_t &  remote_sub_id  ) const

protected

Definition at line 1239 of file DataLink.cpp.

References send_listener_for().

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::ReplayDurableData::handle_event().

{
  GuidConverter local(local_pub_id);
  GuidConverter remote(remote_sub_id);
  TransportSendListener_rch send_listener = send_listener_for(local_pub_id);
  if (send_listener) {
    send_listener->replay_durable_data_for(remote_sub_id);
  }
}

resume_send()

void OpenDDS::DCPS::DataLink::resume_send

(

)

The resume_send is used in the case of reconnection on the subscriber's side.

Definition at line 388 of file DataLink.cpp.

References get_send_strategy(), OpenDDS::DCPS::TransportSendStrategy::isDirectMode(), and OpenDDS::DCPS::TransportSendStrategy::resume_send().

{
  TransportSendStrategy_rch strategy = get_send_strategy();

  if (strategy && strategy->isDirectMode()) {
    strategy->resume_send();
  }
}

schedule_delayed_release()

void OpenDDS::DCPS::DataLink::schedule_delayed_release

(

)

Definition at line 581 of file DataLink.cpp.

References OpenDDS::DCPS::TransportSendStrategy::clear(), datalink_release_delay_, DBG_ENTRY_LVL, get_send_strategy(), LM_DEBUG, OpenDDS::DCPS::TransportSendStrategy::MODE_DIRECT, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), schedule_stop(), and VDBG.

Referenced by OpenDDS::DCPS::TcpTransport::release_datalink().

{
  DBG_ENTRY_LVL("DataLink", "schedule_delayed_release", 6);

  VDBG((LM_DEBUG, "(%P|%t) DataLink[%@]::schedule_delayed_release\n", this));

  // The samples have to be removed at this point, otherwise the samples
  // can not be delivered when new association is added and still use
  // this connection/datalink.
  TransportSendStrategy_rch strategy = get_send_strategy();

  if (strategy) {
    strategy->clear(TransportSendStrategy::MODE_DIRECT);
  }

  const MonotonicTimePoint future_release_time(MonotonicTimePoint::now() + datalink_release_delay_);
  schedule_stop(future_release_time);
}

schedule_stop()

void OpenDDS::DCPS::DataLink::schedule_stop

(

const MonotonicTimePoint &

schedule_to_stop_at

)

Allows DataLink::stop to be done on the reactor thread so that this thread avoids possibly deadlocking trying to access reactor to stop strategies or schedule timers

Definition at line 325 of file DataLink.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TimePoint_T< AceClock >::is_zero(), LM_DEBUG, notify_reactor(), scheduled_to_stop_at_, and stopped_.

Referenced by OpenDDS::DCPS::TcpTransport::release_datalink(), and schedule_delayed_release().

{
  if (!stopped_ && scheduled_to_stop_at_.is_zero()) {
    this->scheduled_to_stop_at_ = schedule_to_stop_at;
    notify_reactor();
    // reactor will invoke our DataLink::handle_exception()
  } else {
    if (DCPS_debug_level > 0) {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) DataLink::schedule_stop() - Already stopped or already scheduled for stop\n")));
    }
  }
}

send()

ACE_INLINE void OpenDDS::DCPS::DataLink::send

(

TransportQueueElement *

element

)

Definition at line 94 of file DataLink.inl.

References ACE_INLINE, customize_queue_element(), OpenDDS::DCPS::TransportQueueElement::data_dropped(), DBG_ENTRY_LVL, handle_send_request_ack(), OpenDDS::DCPS::TransportQueueElement::is_request_ack(), OpenDDS::DCPS::SEND, send_i(), and thr_per_con_send_task_.

Referenced by send_control().

{
  DBG_ENTRY_LVL("DataLink","send",6);

  if (element->is_request_ack() && handle_send_request_ack(element)) {
    return;
  }

  element = this->customize_queue_element(element);
  if (!element) {
    return;
  }

  if (this->thr_per_con_send_task_ != 0) {
    if (this->thr_per_con_send_task_->add_request(SEND, element) == -1) {
      element->data_dropped(true);
    }

  } else {
    this->send_i(element);

  }
}

send_control()

SendControlStatus OpenDDS::DCPS::DataLink::send_control	(	const DataSampleHeader &	header,
		Message_Block_Ptr	data
	)

This allows a subclass to send transport control samples over this DataLink. This is useful for sending transport-specific control messages between one or more endpoints under this DataLink's control.

Definition at line 668 of file DataLink.cpp.

References DBG_ENTRY_LVL, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::move(), OpenDDS::DCPS::DataSampleHeader::publication_id_, send(), OpenDDS::DCPS::SEND_CONTROL_OK, send_response_listener_, send_start(), send_stop(), and OpenDDS::DCPS::SendResponseListener::track_message().

Referenced by OpenDDS::DCPS::MulticastSession::send_control(), and OpenDDS::DCPS::MulticastDataLink::syn_received_no_session().

{
  DBG_ENTRY_LVL("DataLink", "send_control", 6);

  TransportSendControlElement* const elem = new TransportSendControlElement(1, // initial_count
                                       GUID_UNKNOWN, &send_response_listener_,
                                       header, move(message));

  send_response_listener_.track_message();

  GUID_t senderId(header.publication_id_);
  send_start();
  send(elem);
  send_stop(senderId);

  return SEND_CONTROL_OK;
}

send_final_acks()

ACE_INLINE void OpenDDS::DCPS::DataLink::send_final_acks ( const GUID_t &  readerid )

virtual

Definition at line 368 of file DataLink.inl.

References ACE_INLINE.

{
}

send_i()

ACE_INLINE void OpenDDS::DCPS::DataLink::send_i ( TransportQueueElement *  element, bool  relink = true  )

protectedvirtual

Reimplemented in OpenDDS::DCPS::TcpDataLink.

Definition at line 119 of file DataLink.inl.

References ACE_INLINE, OpenDDS::DCPS::TransportQueueElement::data_dropped(), DBG_ENTRY_LVL, get_send_strategy(), and OpenDDS::DCPS::TransportSendStrategy::send().

Referenced by OpenDDS::DCPS::ThreadPerConnectionSendTask::execute(), send(), OpenDDS::DCPS::ShmemDataLink::send_association_msg(), and OpenDDS::DCPS::TcpDataLink::send_i().

{
  DBG_ENTRY_LVL("DataLink","send_i",6);
  // This one is easy.  Simply delegate to our TransportSendStrategy
  // data member.

  TransportSendStrategy_rch strategy = get_send_strategy();

  if (strategy) {
    strategy->send(element, relink);
  } else {
    element->data_dropped(true);
  }
}

send_listener_for()

ACE_INLINE TransportSendListener_rch OpenDDS::DCPS::DataLink::send_listener_for ( const GUID_t &  pub_id ) const

private

Definition at line 324 of file DataLink.inl.

References ACE_INLINE, and send_listeners_.

Referenced by clear_associations(), and replay_durable_data().

{
  // pub_map_ (and send_listeners_) are already locked when entering this
  // private method.
  IdToSendListenerMap::const_iterator found =
    this->send_listeners_.find(pub_id);
  if (found == this->send_listeners_.end()) {
    return TransportSendListener_rch();
  }
  return found->second.lock();
}

send_start()

ACE_INLINE void OpenDDS::DCPS::DataLink::send_start

(

)

Called by the TransportClient objects that reference this DataLink. Used by the TransportClient to send a sample, or to send a control message. These functions either give the request to the PerThreadConnectionSendTask when thread_per_connection configuration is true or just simply delegate to the send strategy.

Definition at line 68 of file DataLink.inl.

References ACE_INLINE, DBG_ENTRY_LVL, OpenDDS::DCPS::SEND_START, send_start_i(), and thr_per_con_send_task_.

Referenced by send_control().

{
  DBG_ENTRY_LVL("DataLink","send_start",6);

  if (this->thr_per_con_send_task_ != 0) {
    this->thr_per_con_send_task_->add_request(SEND_START);

  } else
    this->send_start_i();
}

send_start_i()

ACE_INLINE void OpenDDS::DCPS::DataLink::send_start_i ( )

protected

The implementation of the functions that accomplish the sample or control message delivery. They just simply delegate to the send strategy.

Definition at line 80 of file DataLink.inl.

References ACE_INLINE, DBG_ENTRY_LVL, get_send_strategy(), OpenDDS::DCPS::RcHandle< T >::is_nil(), and OpenDDS::DCPS::TransportSendStrategy::send_start().

Referenced by OpenDDS::DCPS::ThreadPerConnectionSendTask::execute(), and send_start().

{
  DBG_ENTRY_LVL("DataLink","send_start_i",6);
  // This one is easy.  Simply delegate to our TransportSendStrategy
  // data member.

  TransportSendStrategy_rch strategy = get_send_strategy();

  if (!strategy.is_nil()) {
    strategy->send_start();
  }
}

send_stop()

ACE_INLINE void OpenDDS::DCPS::DataLink::send_stop

(

GUID_t

repoId

)

Definition at line 135 of file DataLink.inl.

References ACE_INLINE, DBG_ENTRY_LVL, OpenDDS::DCPS::SEND_STOP, send_stop_i(), and thr_per_con_send_task_.

Referenced by send_control().

{
  DBG_ENTRY_LVL("DataLink","send_stop",6);

  if (this->thr_per_con_send_task_ != 0) {
    this->thr_per_con_send_task_->add_request(SEND_STOP);

  } else
    this->send_stop_i(repoId);
}

send_stop_i()

ACE_INLINE void OpenDDS::DCPS::DataLink::send_stop_i ( GUID_t  repoId )

protected

Definition at line 147 of file DataLink.inl.

References ACE_INLINE, DBG_ENTRY_LVL, get_send_strategy(), OpenDDS::DCPS::RcHandle< T >::is_nil(), and OpenDDS::DCPS::TransportSendStrategy::send_stop().

Referenced by OpenDDS::DCPS::ThreadPerConnectionSendTask::execute(), send_stop(), and OpenDDS::DCPS::TcpDataLink::send_stop_i().

{
  DBG_ENTRY_LVL("DataLink","send_stop_i",6);
  // This one is easy.  Simply delegate to our TransportSendStrategy
  // data member.

  TransportSendStrategy_rch strategy = get_send_strategy();

  if (!strategy.is_nil()) {
    strategy->send_stop(repoId);
  }
}

set_dscp_codepoint()

void OpenDDS::DCPS::DataLink::set_dscp_codepoint	(	int	cp,
		ACE_SOCK &	socket
	)

The following IPV6 code was lifted in spirit from the RTCORBA implementation of setting the DiffServ codepoint.

Definition at line 1115 of file DataLink.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, ENOTSUP, ACE_SOCK::get_local_addr(), ACE_Addr::get_type(), IPPROTO_IP, LM_DEBUG, LM_ERROR, and ACE_SOCK::set_option().

Referenced by OpenDDS::DCPS::TcpConnection::on_active_connection_established(), and OpenDDS::DCPS::UdpDataLink::open().

{
  /**
   * The following IPV6 code was lifted in spirit from the RTCORBA
   * implementation of setting the DiffServ codepoint.
   */
  int result = 0;

  // Shift the code point up to bits, so that we only use the DS field
  int tos = cp << 2;

  const char* which = "IPV4 TOS";
#if defined (ACE_HAS_IPV6)
  ACE_INET_Addr local_address;

  if (socket.get_local_addr(local_address) == -1) {
    return;

  } else if (local_address.get_type() == AF_INET6)
#if !defined (IPV6_TCLASS)
  {
    if (DCPS_debug_level > 0) {
      ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: DataLink::set_dscp_codepoint() - ")
                 ACE_TEXT("IPV6 TCLASS not supported yet, not setting codepoint %d.\n"),
                 cp));
    }

    return;
  }

#else /* IPV6_TCLASS */
  {
    which = "IPV6 TCLASS";
    result = socket.set_option(
               IPPROTO_IPV6,
               IPV6_TCLASS,
               &tos,
               sizeof(tos));

  } else // This is a bit tricky and might be hard to follow...

#endif /* IPV6_TCLASS */
#endif /* ACE_HAS_IPV6 */

#ifdef IP_TOS
  result = socket.set_option(
             IPPROTO_IP,
             IP_TOS,
             &tos,
             sizeof(tos));

  if ((result == -1) && (errno != ENOTSUP)
#ifdef WSAEINVAL
      && (errno != WSAEINVAL)
#endif /* WSAINVAL */
     ) {
#endif /* IP_TOS */
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) DataLink::set_dscp_codepoint() - ")
               ACE_TEXT("failed to set the %C codepoint to %d: %m, ")
               ACE_TEXT("try running as superuser.\n"),
               which,
               cp));
#ifdef IP_TOS
  } else if (DCPS_debug_level > 4) {
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) DataLink::set_dscp_codepoint() - ")
               ACE_TEXT("set %C codepoint to %d.\n"),
               which,
               cp));
  }
#endif /* IP_TOS */
}

set_scheduling_release()

ACE_INLINE void OpenDDS::DCPS::DataLink::set_scheduling_release

(

bool

scheduling_release

)

Definition at line 161 of file DataLink.inl.

References ACE_INLINE, and scheduling_release_.

Referenced by cancel_release(), OpenDDS::DCPS::TcpTransport::release_datalink(), and transport_shutdown().

{
  this->scheduling_release_ = scheduling_release;
}

start()

ACE_INLINE int OpenDDS::DCPS::DataLink::start ( const TransportSendStrategy_rch &  send_strategy, const TransportStrategy_rch &  receive_strategy, bool  invoke_all = true  )

protected

This is how the subclass "announces" to this DataLink base class that this DataLink has now been "connected" and should start the supplied strategy objects. This start method is also going to keep a "copy" of the references to the strategy objects. Also note that it is acceptable to pass-in a NULL (0) TransportReceiveStrategy*, but it is assumed that the TransportSendStrategy* argument is not NULL.

If the start() method fails to start either strategy, then a -1 is returned. Otherwise, a 0 is returned. In the failure case, if one of the strategy objects was started successfully, then it will be stopped before the start() method returns -1.

Definition at line 212 of file DataLink.inl.

References ACE_INLINE, DBG_ENTRY_LVL, invoke_on_start_callbacks(), OpenDDS::DCPS::RcHandle< T >::is_nil(), receive_strategy_, send_strategy_, OpenDDS::DCPS::TransportStrategy::start(), OpenDDS::DCPS::TransportSendStrategy::start(), started_, OpenDDS::DCPS::TransportSendStrategy::stop(), and strategy_lock_.

Referenced by OpenDDS::DCPS::TcpDataLink::connect(), OpenDDS::DCPS::MulticastDataLink::join(), OpenDDS::DCPS::UdpDataLink::open(), OpenDDS::DCPS::ShmemDataLink::open(), OpenDDS::DCPS::RtpsUdpDataLink::open(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_gap_i().

{
  DBG_ENTRY_LVL("DataLink","start",6);

  // We assume that the send_strategy is not NULL, but the receive_strategy
  // is allowed to be NULL.

  // Attempt to start the strategies, and if there is a start() failure,
  // make sure to stop() any strategy that was already start()'ed.
  if (send_strategy->start() != 0) {
    // Failed to start the TransportSendStrategy.
    invoke_on_start_callbacks(false);
    return -1;
  }

  if ((!receive_strategy.is_nil()) && (receive_strategy->start() != 0)) {
    // Failed to start the TransportReceiveStrategy.

    // Remember to stop() the TransportSendStrategy since we did start it,
    // and now need to "undo" that action.
    send_strategy->stop();
    invoke_on_start_callbacks(false);
    return -1;
  }

  // We started both strategy objects.  Save them to data members since
  // we will now take ownership of them.
  {
    GuardType guard(this->strategy_lock_);

    this->send_strategy_    = send_strategy;
    this->receive_strategy_ = receive_strategy;
  }
  if (invoke_all) {
    invoke_on_start_callbacks(true);
  }
  {
    //catch any associations added during initial invoke_on_start_callbacks
    //only after first use_datalink has resolved does datalink's state truly
    //change to started, thus can't let pending associations proceed normally yet
    GuardType guard(this->strategy_lock_);
    this->started_ = true;
  }
  //Now state transitioned to started so no new on_start_callbacks will be added
  //so resolve any added during transition to started.
  if (invoke_all) {
    invoke_on_start_callbacks(true);
  }
  return 0;
}

stop()

void OpenDDS::DCPS::DataLink::stop

(

void

)

The stop method is used to stop the DataLink prior to shutdown.

Definition at line 355 of file DataLink.cpp.

References OpenDDS::DCPS::RcHandle< T >::is_nil(), pre_stop_i(), receive_strategy_, OpenDDS::DCPS::RcHandle< T >::reset(), scheduled_to_stop_at_, send_strategy_, OpenDDS::DCPS::TransportStrategy::stop(), OpenDDS::DCPS::TransportSendStrategy::stop(), stop_i(), stopped_, strategy_lock_, and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::zero_value.

Referenced by handle_exception(), handle_timeout(), OpenDDS::DCPS::UdpTransport::release_datalink(), OpenDDS::DCPS::ShmemTransport::release_datalink(), and transport_shutdown().

{
  pre_stop_i();

  TransportSendStrategy_rch send_strategy;
  TransportStrategy_rch recv_strategy;

  {
    GuardType guard(strategy_lock_);

    if (stopped_) return;

    send_strategy = send_strategy_;
    send_strategy_.reset();

    recv_strategy = receive_strategy_;
    receive_strategy_.reset();
  }

  if (!send_strategy.is_nil()) {
    send_strategy->stop();
  }

  if (!recv_strategy.is_nil()) {
    recv_strategy->stop();
  }

  stop_i();
  stopped_ = true;
  scheduled_to_stop_at_ = MonotonicTimePoint::zero_value;
}

stop_i()

void OpenDDS::DCPS::DataLink::stop_i ( )

protectedvirtual

This announces the "stop" event to our subclass. The "stop" event will occur when this DataLink is handling a release_reservations() call and determines that it has just released all of the remaining reservations on this DataLink. The "stop" event will also occur when the TransportImpl is being shutdown() - we call stop_i() from our transport_shutdown() method to handle this case.

Reimplemented in OpenDDS::DCPS::RtpsUdpDataLink, OpenDDS::DCPS::MulticastDataLink, OpenDDS::DCPS::TcpDataLink, OpenDDS::DCPS::ShmemDataLink, and OpenDDS::DCPS::UdpDataLink.

Definition at line 622 of file DataLink.cpp.

References DBG_ENTRY_LVL.

Referenced by stop().

{
  DBG_ENTRY_LVL("DataLink", "stop_i", 6);
}

target_intersection()

GUIDSeq * OpenDDS::DCPS::DataLink::target_intersection	(	const GUID_t &	pub_id,
		const GUIDSeq &	in,
		size_t &	n_subs
	)

For a given publication "pub_id", store the total number of corresponding subscriptions in "n_subs" and given a set of subscriptions (the "in" sequence), return the subset of the input set "in" which are targets of this DataLink (see is_target()).

Definition at line 1020 of file DataLink.cpp.

References assoc_by_local_, pub_sub_maps_lock_, and OpenDDS::DCPS::push_back().

{
  GUIDSeq_var res;
  GuardType guard(this->pub_sub_maps_lock_);
  AssocByLocal::const_iterator iter = assoc_by_local_.find(pub_id);

  if (iter != assoc_by_local_.end()) {
    n_subs = iter->second.associated_.size();
    const CORBA::ULong len = in.length();

    for (CORBA::ULong i(0); i < len; ++i) {
      if (iter->second.associated_.count(in[i])) {
        if (res.ptr() == 0) {
          res = new GUIDSeq;
        }

        push_back(res.inout(), in[i]);
      }
    }
  }

  return res._retn();
}

terminate_send()

ACE_INLINE void OpenDDS::DCPS::DataLink::terminate_send

(

)

Definition at line 279 of file DataLink.inl.

References ACE_INLINE, get_send_strategy(), and OpenDDS::DCPS::TransportSendStrategy::terminate_send().

{
  TransportSendStrategy_rch strategy = get_send_strategy();
  if (strategy) {
    strategy->terminate_send(false);
  }
}

terminate_send_if_suspended()

void OpenDDS::DCPS::DataLink::terminate_send_if_suspended

(

)

Definition at line 1275 of file DataLink.cpp.

References get_send_strategy(), OPENDDS_END_VERSIONED_NAMESPACE_DECL, and OpenDDS::DCPS::TransportSendStrategy::terminate_send_if_suspended().

{
  TransportSendStrategy_rch strategy = get_send_strategy();

  if (strategy) {
    strategy->terminate_send_if_suspended();
  }
}

transport_priority() [1/2]

ACE_INLINE Priority & OpenDDS::DCPS::DataLink::transport_priority

(

)

Accessors for the TRANSPORT_PRIORITY value associated with this link.

Definition at line 21 of file DataLink.inl.

References ACE_INLINE, and transport_priority_.

Referenced by OpenDDS::DCPS::TcpTransport::connect_datalink(), OpenDDS::DCPS::TcpTransport::connect_tcp_datalink(), OpenDDS::DCPS::UdpDataLink::open(), OpenDDS::DCPS::ThreadPerConnectionSendTask::open(), OpenDDS::DCPS::TcpTransport::release_datalink(), and OpenDDS::DCPS::TcpTransport::unbind_link().

{
  return this->transport_priority_;
}

transport_priority() [2/2]

ACE_INLINE Priority OpenDDS::DCPS::DataLink::transport_priority

(

)

const

Definition at line 28 of file DataLink.inl.

References ACE_INLINE, and transport_priority_.

{
  return this->transport_priority_;
}

transport_shutdown()

void OpenDDS::DCPS::DataLink::transport_shutdown

(

)

Our TransportImpl will inform us if it is being shutdown() by calling this method.

Definition at line 826 of file DataLink.cpp.

References ACE_Reactor_Timer_Interface::cancel_timer(), DBG_ENTRY_LVL, impl(), impl_, ACE_Event_Handler::reactor(), scheduled_to_stop_at_, set_scheduling_release(), stop(), OpenDDS::DCPS::TransportImpl::timer(), and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::zero_value.

Referenced by OpenDDS::DCPS::UdpTransport::shutdown_i(), and OpenDDS::DCPS::RtpsUdpTransport::shutdown_i().

{
  DBG_ENTRY_LVL("DataLink", "transport_shutdown", 6);

  //this->cancel_release();
  this->set_scheduling_release(false);
  scheduled_to_stop_at_ = MonotonicTimePoint::zero_value;

  {
    TransportImpl_rch impl = impl_.lock();
    if (impl) {
      ACE_Reactor_Timer_Interface* reactor = impl->timer();
      reactor->cancel_timer(this);
    }
  }
  this->stop();
  // this->send_listeners_.clear();
  // this->recv_listeners_.clear();
  // Drop our reference to the TransportImpl object
}

Friends And Related Function Documentation

DataLinkCleanupTask

friend class DataLinkCleanupTask

friend

Definition at line 76 of file DataLink.h.

operator<<

OpenDDS_Dcps_Export std::ostream& operator<< ( std::ostream &  str, const DataLink &  value  )

friend

Convenience function for diagnostic information.

Definition at line 1251 of file DataLink.cpp.

{
  str << "   There are " << value.assoc_by_local_.size()
      << " local entities currently using this link";

  if (!value.assoc_by_local_.empty()) {
    str << " comprising following associations:";
  }
  str << std::endl;

  typedef DataLink::AssocByLocal::const_iterator assoc_iter_t;
  const DataLink::AssocByLocal& abl = value.assoc_by_local_;
  for (assoc_iter_t ait = abl.begin(); ait != abl.end(); ++ait) {
    const RepoIdSet& set = ait->second.associated_;
    for (RepoIdSet::const_iterator rit = set.begin(); rit != set.end(); ++rit) {
      str << GuidConverter(ait->first) << " --> "
          << GuidConverter(*rit) << "   " << std::endl;
    }
  }
  return str;
}

ThreadPerConnectionSendTask

friend class ThreadPerConnectionSendTask

friend

Definition at line 326 of file DataLink.h.

Referenced by DataLink().

Member Data Documentation

assoc_by_local_

AssocByLocal OpenDDS::DCPS::DataLink::assoc_by_local_

private

Definition at line 414 of file DataLink.h.

Referenced by handle_timeout(), make_reservation(), notify(), OpenDDS::DCPS::operator<<(), peer_ids(), prepare_release(), release_reservations(), target_intersection(), and ~DataLink().

assoc_by_remote_

AssocByRemote OpenDDS::DCPS::DataLink::assoc_by_remote_

private

Definition at line 406 of file DataLink.h.

Referenced by data_received_i(), handle_timeout(), is_target(), make_reservation(), and release_reservations().

assoc_releasing_

AssocByLocal OpenDDS::DCPS::DataLink::assoc_releasing_

private

Definition at line 428 of file DataLink.h.

Referenced by clear_associations(), and prepare_release().

datalink_release_delay_

TimeDuration OpenDDS::DCPS::DataLink::datalink_release_delay_

protected

Configurable delay in milliseconds that the datalink should be released after all associations are removed.

Definition at line 453 of file DataLink.h.

Referenced by DataLink(), datalink_release_delay(), and schedule_delayed_release().

db_allocator_

unique_ptr<DataBlockAllocator> OpenDDS::DCPS::DataLink::db_allocator_

protected

Definition at line 458 of file DataLink.h.

Referenced by create_control(), and DataLink().

default_listener_

TransportReceiveListener_wrch OpenDDS::DCPS::DataLink::default_listener_

private

If default_listener_ is not null and this DataLink receives a sample from a publication GUID that's not in pub_map_, it will call data_received() on the default_listener_.

Definition at line 401 of file DataLink.h.

Referenced by data_received_i(), and default_listener().

id_

ACE_UINT64 OpenDDS::DCPS::DataLink::id_

private

The id for this DataLink.

Definition at line 420 of file DataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::add_gap_submsg_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::customize_queue_element_helper(), DataLink(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::end_historic_samples_i(), id(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_all_msgs(), and OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::request_ack_i().

impl_

WeakRcHandle<TransportImpl> OpenDDS::DCPS::DataLink::impl_

private

A weak rchandle to the TransportImpl that created this DataLink.

Definition at line 417 of file DataLink.h.

Referenced by handle_exception(), handle_timeout(), impl(), notify_reactor(), release_reservations(), release_resources(), and transport_shutdown().

interceptor_

Interceptor OpenDDS::DCPS::DataLink::interceptor_

protected

Definition at line 469 of file DataLink.h.

Referenced by add_on_start_callback().

is_active_

bool OpenDDS::DCPS::DataLink::is_active_

protected

Is pub or sub ?

Definition at line 463 of file DataLink.h.

Referenced by is_active(), and OpenDDS::DCPS::TcpDataLink::reuse_existing_connection().

is_loopback_

bool OpenDDS::DCPS::DataLink::is_loopback_

protected

Is remote attached to same transport ?

Definition at line 461 of file DataLink.h.

Referenced by is_loopback(), and OpenDDS::DCPS::UdpDataLink::open().

mb_allocator_

unique_ptr<MessageBlockAllocator> OpenDDS::DCPS::DataLink::mb_allocator_

protected

Allocators for data and message blocks used by transport control samples when send_control is called.

Definition at line 457 of file DataLink.h.

Referenced by create_control(), and DataLink().

on_start_callbacks_

OnStartCallbackMap OpenDDS::DCPS::DataLink::on_start_callbacks_

protected

Definition at line 447 of file DataLink.h.

Referenced by add_on_start_callback(), OpenDDS::DCPS::TcpDataLink::do_association_actions(), invoke_on_start_callbacks(), remove_on_start_callback(), and remove_startup_callbacks().

pending_on_starts_

PendingOnStartsMap OpenDDS::DCPS::DataLink::pending_on_starts_

protected

Definition at line 449 of file DataLink.h.

Referenced by add_on_start_callback(), invoke_on_start_callbacks(), and remove_startup_callbacks().

pub_sub_maps_lock_

LockType OpenDDS::DCPS::DataLink::pub_sub_maps_lock_

mutableprivate

Definition at line 403 of file DataLink.h.

Referenced by data_received_i(), default_listener(), is_target(), make_reservation(), network_change(), notify(), peer_ids(), prepare_release(), release_reservations(), remove_listener(), and target_intersection().

receive_strategy_

TransportStrategy_rch OpenDDS::DCPS::DataLink::receive_strategy_

protected

The transport receive strategy object for this DataLink.

Definition at line 324 of file DataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::open(), OpenDDS::DCPS::TcpDataLink::receive_strategy(), OpenDDS::DCPS::RtpsUdpDataLink::receive_strategy(), OpenDDS::DCPS::TcpDataLink::reconnect(), OpenDDS::DCPS::TcpDataLink::reuse_existing_connection(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsUdpDataLink(), start(), and stop().

recv_listeners_

IdToRecvListenerMap OpenDDS::DCPS::DataLink::recv_listeners_

private

Definition at line 396 of file DataLink.h.

Referenced by make_reservation(), network_change(), notify(), recv_listener_for(), and remove_listener().

scheduled_to_stop_at_

MonotonicTimePoint OpenDDS::DCPS::DataLink::scheduled_to_stop_at_

private

Definition at line 388 of file DataLink.h.

Referenced by cancel_release(), handle_exception(), handle_timeout(), schedule_stop(), stop(), and transport_shutdown().

scheduling_release_

bool OpenDDS::DCPS::DataLink::scheduling_release_

private

Definition at line 433 of file DataLink.h.

Referenced by cancel_release(), handle_exception(), and set_scheduling_release().

send_listeners_

IdToSendListenerMap OpenDDS::DCPS::DataLink::send_listeners_

private

Definition at line 392 of file DataLink.h.

Referenced by make_reservation(), network_change(), notify(), remove_listener(), and send_listener_for().

send_response_listener_

SendResponseListener OpenDDS::DCPS::DataLink::send_response_listener_

protected

Listener for TransportSendControlElements created in send_control.

Definition at line 467 of file DataLink.h.

Referenced by send_control().

send_strategy_

TransportSendStrategy_rch OpenDDS::DCPS::DataLink::send_strategy_

protected

The transport send strategy object for this DataLink.

Definition at line 440 of file DataLink.h.

Referenced by add_on_start_callback(), OpenDDS::DCPS::RtpsUdpDataLink::client_stop(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::customize_queue_element_helper(), OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element_non_reliable_i(), OpenDDS::DCPS::TcpDataLink::do_association_actions(), get_send_strategy(), OpenDDS::DCPS::RtpsUdpDataLink::open(), OpenDDS::DCPS::TcpDataLink::reconnect(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_all_msgs(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_sample(), OpenDDS::DCPS::TcpDataLink::reuse_existing_connection(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsUdpDataLink(), OpenDDS::DCPS::TcpDataLink::send_graceful_disconnect_message(), OpenDDS::DCPS::TcpDataLink::send_strategy(), OpenDDS::DCPS::RtpsUdpDataLink::send_strategy(), start(), and stop().

started_

bool OpenDDS::DCPS::DataLink::started_

protected

Definition at line 464 of file DataLink.h.

Referenced by add_on_start_callback(), and start().

stopped_

bool OpenDDS::DCPS::DataLink::stopped_

private

A boolean indicating if the DataLink has been stopped. This value is protected by the strategy_lock_.

Definition at line 387 of file DataLink.h.

Referenced by cancel_release(), release_reservations(), schedule_stop(), and stop().

strategy_lock_

LockType OpenDDS::DCPS::DataLink::strategy_lock_

protected

Definition at line 441 of file DataLink.h.

Referenced by add_on_start_callback(), OpenDDS::DCPS::RtpsUdpDataLink::associated(), OpenDDS::DCPS::RtpsUdpDataLink::client_stop(), OpenDDS::DCPS::TcpDataLink::connect(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::customize_queue_element_helper(), OpenDDS::DCPS::RtpsUdpDataLink::customize_queue_element_non_reliable_i(), OpenDDS::DCPS::TcpDataLink::do_association_actions(), get_send_strategy(), invoke_on_start_callbacks(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::pre_stop_helper(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_data_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_gap_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_frag_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_i(), OpenDDS::DCPS::TcpDataLink::receive_strategy(), OpenDDS::DCPS::RtpsUdpDataLink::received(), OpenDDS::DCPS::TcpDataLink::reconnect(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_all_msgs(), remove_on_start_callback(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_sample(), remove_startup_callbacks(), OpenDDS::DCPS::TcpDataLink::send_strategy(), start(), and stop().

thr_per_con_send_task_

unique_ptr<ThreadPerConnectionSendTask> OpenDDS::DCPS::DataLink::thr_per_con_send_task_

private

The task used to do the sending. This ThreadPerConnectionSendTask object is created when the thread_per_connection configuration is true. It only dedicate to this datalink.

Definition at line 425 of file DataLink.h.

Referenced by DataLink(), pre_stop_i(), remove_sample(), send(), send_start(), send_stop(), and ~DataLink().

transport_priority_

Priority OpenDDS::DCPS::DataLink::transport_priority_

private

TRANSPORT_PRIORITY value associated with the link.

Definition at line 431 of file DataLink.h.

Referenced by transport_priority().

---

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

• DataLink.h
• DataLink.cpp
• DataLink.inl