#include <DataLinkSet.h>

Inheritance diagram for OpenDDS::DCPS::DataLinkSet:

Collaboration diagram for OpenDDS::DCPS::DataLinkSet:

[legend]

Public Types
typedef ACE_SYNCH_MUTEX	LockType

typedef ACE_Guard< LockType >	GuardType

Public Member Functions
	DataLinkSet ()

virtual	~DataLinkSet ()

int	insert_link (const DataLink_rch &link)

void	remove_link (const DataLink_rch &link)

void	send (DataSampleElement *sample)
	Send to each DataLink in the set. More...

void	send_control (DataSampleElement *sample)
	Send a control message that is wrapped in a DataSampleElement. More...

SendControlStatus	send_control (GUID_t pub_id, const TransportSendListener_rch &listener, const DataSampleHeader &header, Message_Block_Ptr msg)
	Send control message to each DataLink in the set. More...

void	send_response (GUID_t sub_id, const DataSampleHeader &header, Message_Block_Ptr response)

bool	remove_sample (const DataSampleElement *sample)

bool	remove_all_msgs (const GUID_t &pub_id)

void	send_start (DataLinkSet *link_set)

void	send_stop (GUID_t repoId)

DataLinkSet_rch	select_links (const GUID_t *remoteIds, const CORBA::ULong num_targets)

bool	empty ()

void	send_final_acks (const GUID_t &readerid)

typedef	OPENDDS_MAP (DataLinkIdType, DataLink_rch) MapType

LockType &	lock ()
	Accessors for external iteration. More...

MapType &	map ()

void	terminate_send_if_suspended ()

bool	is_leading (const GUID_t &writer_id, const GUID_t &reader_id) const

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

Private Member Functions
void	copy_map_to (MapType &target)
	lock and copy map for lock-free access More...

Private Attributes
MapType	map_
	Hash map for DataLinks. More...

LockType	lock_

SendResponseListener	send_response_listener_
	Listener for TransportSendControlElements created in send_response. More...

Additional Inherited Members
Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Detailed Description

Definition at line 30 of file DataLinkSet.h.

Member Typedef Documentation

◆ GuardType

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

Definition at line 78 of file DataLinkSet.h.

◆ LockType

typedef ACE_SYNCH_MUTEX OpenDDS::DCPS::DataLinkSet::LockType

Definition at line 77 of file DataLinkSet.h.

Constructor & Destructor Documentation

◆ DataLinkSet()

OpenDDS::DCPS::DataLinkSet::DataLinkSet ( )

Definition at line 30 of file DataLinkSet.cpp.

References DBG_ENTRY_LVL.

   : send_response_listener_("DataLinkSet")
 {
   DBG_ENTRY_LVL("DataLinkSet","DataLinkSet",6);
 }

◆ ~DataLinkSet()

OpenDDS::DCPS::DataLinkSet::~DataLinkSet ( )

virtual

Definition at line 36 of file DataLinkSet.cpp.

References DBG_ENTRY_LVL.

 {
   DBG_ENTRY_LVL("DataLinkSet","~DataLinkSet",6);
 }

Member Function Documentation

◆ copy_map_to()

ACE_INLINE void OpenDDS::DCPS::DataLinkSet::copy_map_to ( MapType & target )

private

lock and copy map for lock-free access

Definition at line 256 of file DataLinkSet.inl.

References ACE_INLINE, lock_, and map_.

Referenced by send_control().

 {
   target.clear();
 
   // Lock the existing map
   GuardType guard(this->lock_);
 
 
   // Copy to target
   for (MapType::iterator itr = map_.begin();
        itr != map_.end();
        ++itr) {
     target.insert(*itr);
   }
 }

◆ empty()

bool OpenDDS::DCPS::DataLinkSet::empty ( void )

Definition at line 87 of file DataLinkSet.cpp.

References lock_, and map_.

 {
   GuardType guard(this->lock_);
 
   return map_.empty();
 }

◆ insert_link()

int OpenDDS::DCPS::DataLinkSet::insert_link ( const DataLink_rch & link )

Definition at line 42 of file DataLinkSet.cpp.

References OpenDDS::DCPS::bind(), DBG_ENTRY_LVL, OpenDDS::DCPS::DataLink::id(), lock_, and map_.

Referenced by OpenDDS::DCPS::TransportClient::add_link(), OpenDDS::DCPS::TransportClient::send_control_to(), and OpenDDS::DCPS::TransportClient::send_response().

 {
   DBG_ENTRY_LVL("DataLinkSet","insert_link",6);
   GuardType guard(this->lock_);
   return OpenDDS::DCPS::bind(map_, link->id(), link);
 }

◆ is_leading()

bool OpenDDS::DCPS::DataLinkSet::is_leading	(	const GUID_t &	writer_id,
		const GUID_t &	reader_id
	)		const

Definition at line 107 of file DataLinkSet.cpp.

References lock_, and map_.

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

 {
   GuardType guard(this->lock_);
   for (MapType::const_iterator pos = map_.begin(), limit = map_.end(); pos != limit; ++pos) {
     if (pos->second->is_leading(writer_id, reader_id)) {
       return true;
     }
   }
 
   return false;
 }

◆ lock()

LockType& OpenDDS::DCPS::DataLinkSet::lock ( void )

inline

Accessors for external iteration.

Definition at line 84 of file DataLinkSet.h.

References lock_.

84 { return lock_; }

OpenDDS::DCPS::DataLinkSet::lock_

LockType lock_

Definition: DataLinkSet.h:100

◆ map()

MapType& OpenDDS::DCPS::DataLinkSet::map ( void )

inline

Definition at line 85 of file DataLinkSet.h.

85 { return map_; }

OpenDDS::DCPS::DataLinkSet::map_

MapType map_

Hash map for DataLinks.

Definition: DataLinkSet.h:96

◆ OPENDDS_MAP()

typedef OpenDDS::DCPS::DataLinkSet::OPENDDS_MAP	(	DataLinkIdType	,
		DataLink_rch
	)

◆ remove_all_msgs()

ACE_INLINE bool OpenDDS::DCPS::DataLinkSet::remove_all_msgs ( const GUID_t & pub_id )

Definition at line 185 of file DataLinkSet.inl.

References ACE_INLINE, DBG_ENTRY_LVL, lock_, and map_.

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

 {
   DBG_ENTRY_LVL("DataLinkSet", "remove_all_msgs", 6);
   MapType map_copy;
   {
     GuardType guard(lock_);
     map_copy = map_;
   }
   const MapType::iterator end = map_copy.end();
   for (MapType::iterator itr = map_copy.begin(); itr != end; ++itr) {
     itr->second->remove_all_msgs(pub_id);
   }
 
   return true;
 }

◆ remove_link()

void OpenDDS::DCPS::DataLinkSet::remove_link ( const DataLink_rch & link )

Definition at line 50 of file DataLinkSet.cpp.

References ACE_TEXT(), DBG_ENTRY_LVL, OpenDDS::DCPS::DataLink::id(), LM_DEBUG, lock_, map_, OpenDDS::DCPS::unbind(), and VDBG.

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

 {
   DBG_ENTRY_LVL("DataLinkSet", "remove_link", 6);
   GuardType guard1(this->lock_);
   if (unbind(map_, link->id()) != 0) {
     // Just report to the log that we tried.
     VDBG((LM_DEBUG,
           ACE_TEXT("(%P|%t) DataLinkSet::remove_links: ")
           ACE_TEXT("link_id %d not found in map.\n"),
           link->id()));
   }
 }

◆ remove_sample()

ACE_INLINE bool OpenDDS::DCPS::DataLinkSet::remove_sample ( const DataSampleElement * sample )

Definition at line 166 of file DataLinkSet.inl.

References ACE_INLINE, DBG_ENTRY_LVL, lock_, map_, and OpenDDS::DCPS::REMOVE_RELEASED.

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

 {
   DBG_ENTRY_LVL("DataLinkSet", "remove_sample", 6);
   MapType map_copy;
   {
     GuardType guard(lock_);
     map_copy = map_;
   }
   const MapType::iterator end = map_copy.end();
   for (MapType::iterator itr = map_copy.begin(); itr != end; ++itr) {
     if (itr->second->remove_sample(sample) == REMOVE_RELEASED) {
       return true;
     }
   }
 
   return false;
 }

◆ select_links()

OpenDDS::DCPS::DataLinkSet_rch OpenDDS::DCPS::DataLinkSet::select_links	(	const GUID_t *	remoteIds,
		const CORBA::ULong	num_targets
	)

Definition at line 64 of file DataLinkSet.cpp.

References OpenDDS::DCPS::bind(), DBG_ENTRY_LVL, lock_, and map_.

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

 {
   DBG_ENTRY_LVL("DataLinkSet","select_links",6);
 
   DataLinkSet_rch selected_links ( make_rch<DataLinkSet>() );
   GuardType guard(this->lock_);
   for (MapType::iterator itr = map_.begin();
        itr != map_.end();
        ++itr) {
     for (CORBA::ULong i = 0; i < num_targets; ++i) {
       if (itr->second->is_target(remoteIds[i])) {
         OpenDDS::DCPS::bind(selected_links->map_,
              itr->second->id(), itr->second);
         break;
       }
     }
   }
 
   return selected_links;
 }

◆ send()

ACE_INLINE void OpenDDS::DCPS::DataLinkSet::send ( DataSampleElement * sample )

Send to each DataLink in the set.

Definition at line 22 of file DataLinkSet.inl.

References ACE_INLINE, OpenDDS::DCPS::DataSampleHeader::add_cfentries(), OpenDDS::DCPS::CONTENT_FILTER_FLAG, OpenDDS::DCPS::TransportSendListener::data_dropped(), DBG_ENTRY_LVL, ACE_Message_Block::duplicate(), OpenDDS::DCPS::DataSampleElement::get_filter_per_link(), OpenDDS::DCPS::DataSampleElement::get_sample(), OpenDDS::DCPS::DataSampleElement::get_send_listener(), LM_DEBUG, lock_, map_, OpenDDS::DCPS::move(), OpenDDS::DCPS::TransportSendElement::msg(), OpenDDS::DCPS::DataSampleHeader::test_flag(), and VDBG_LVL.

 {
   DBG_ENTRY_LVL("DataLinkSet", "send", 6);
   VDBG_LVL((LM_DEBUG, "(%P|%t) DBG: DataLinkSet::send element %@.\n",
             sample), 5);
 
 #ifndef OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE
   const bool customHeader =
     DataSampleHeader::test_flag(CONTENT_FILTER_FLAG, sample->get_sample());
 #endif
 
   MapType map_copy;
   {
     GuardType guard(lock_);
     map_copy = map_;
   }
 
   if (map_copy.size()) {
     TransportSendElement* send_element = new TransportSendElement(static_cast<int>(map_copy.size()), sample);
     for (MapType::iterator itr = map_copy.begin(); itr != map_copy.end(); ++itr) {
 
 #ifndef OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE
       if (customHeader) {
         typedef std::map<DataLinkIdType, GUIDSeq_var>::iterator FilterIter;
         FilterIter fi = sample->get_filter_per_link().find(itr->first);
         GUIDSeq* guids = 0;
         if (fi != sample->get_filter_per_link().end()) {
           guids = fi->second.ptr();
         }
 
         VDBG_LVL((LM_DEBUG,
                   "(%P|%t) DBG: DataLink %@ filtering %d subscribers.\n",
                   itr->second.in(), guids ? guids->length() : 0), 5);
 
         Message_Block_Ptr mb (send_element->msg()->duplicate());
 
         DataSampleHeader::add_cfentries(guids, mb.get());
 
         TransportCustomizedElement* tce = new TransportCustomizedElement(send_element);
         tce->set_msg(move(mb)); // tce now owns ACE_Message_Block chain
 
         itr->second->send(tce);
 
       } else {
 #endif /* OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE */
 
         // Tell the DataLink to send it.
         itr->second->send(send_element);
 
 #ifndef OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE
       }
 #endif
     }
   } else if (sample->get_send_listener()) {
     sample->get_send_listener()->data_dropped(sample, true);
   }
 }

◆ send_control() [1/2]

ACE_INLINE void OpenDDS::DCPS::DataLinkSet::send_control ( DataSampleElement * sample )

Send a control message that is wrapped in a DataSampleElement.

Definition at line 81 of file DataLinkSet.inl.

References ACE_INLINE, DBG_ENTRY_LVL, LM_DEBUG, lock_, map_, and VDBG.

Referenced by OpenDDS::DCPS::TransportClient::send_control(), and OpenDDS::DCPS::TransportClient::send_control_to().

 {
   DBG_ENTRY_LVL("DataLinkSet", "send_control", 6);
   VDBG((LM_DEBUG, "(%P|%t) DBG: DataLinkSet::send_control %@.\n", sample));
   MapType map_copy;
   {
     GuardType guard(lock_);
     map_copy = map_;
   }
 
   TransportSendControlElement* send_element =
     new TransportSendControlElement(static_cast<int>(map_copy.size()), sample);
 
   for (MapType::iterator itr = map_copy.begin(); itr != map_copy.end(); ++itr) {
     itr->second->send(send_element);
   }
 }

◆ send_control() [2/2]

ACE_INLINE OpenDDS::DCPS::SendControlStatus OpenDDS::DCPS::DataLinkSet::send_control	(	GUID_t	pub_id,
		const TransportSendListener_rch &	listener,
		const DataSampleHeader &	header,
		Message_Block_Ptr	msg
	)

Send control message to each DataLink in the set.

Definition at line 100 of file DataLinkSet.inl.

References ACE_DEBUG, ACE_INLINE, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), copy_map_to(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, header, OpenDDS::DCPS::RcHandle< T >::in(), LM_DEBUG, OpenDDS::DCPS::move(), and OpenDDS::DCPS::SEND_CONTROL_OK.

 {
   DBG_ENTRY_LVL("DataLinkSet","send_control",6);
   //Optimized - use cached allocator.
 
   MapType dup_map;
   copy_map_to(dup_map);
 
   if (dup_map.empty()) {
     // similar to the "no links" case in TransportClient::send()
     if (DCPS_debug_level > 4) {
       const LogGuid logger(pub_id);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) DataLinkSet::send_control: ")
                  ACE_TEXT("no links for publication %C, ")
                  ACE_TEXT("not sending control message.\n"),
                  logger.c_str()));
     }
     listener->control_delivered(msg);
     return SEND_CONTROL_OK;
   }
 
   TransportSendControlElement* const send_element =
     new TransportSendControlElement(static_cast<int>(dup_map.size()), pub_id,
                                        listener.in(), header, move(msg));
 
   for (MapType::iterator itr = dup_map.begin();
        itr != dup_map.end();
        ++itr) {
     itr->second->send_start();
     itr->second->send(send_element);
     itr->second->send_stop(pub_id);
   }
 
   return SEND_CONTROL_OK;
 }

◆ send_final_acks()

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

Definition at line 273 of file DataLinkSet.inl.

References lock_, and map_.

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

 {
   MapType map_copy;
   {
     GuardType guard(lock_);
     map_copy = map_;
     map_.clear();
   }
 
   for (MapType::iterator itr = map_copy.begin(); itr != map_copy.end(); ++itr) {
     itr->second->send_final_acks(readerid);
   }
 }

◆ send_response()

ACE_INLINE void OpenDDS::DCPS::DataLinkSet::send_response	(	GUID_t	sub_id,
		const DataSampleHeader &	header,
		Message_Block_Ptr	response
	)

Definition at line 141 of file DataLinkSet.inl.

References ACE_INLINE, DBG_ENTRY_LVL, lock_, map_, OpenDDS::DCPS::move(), send_response_listener_, and OpenDDS::DCPS::SendResponseListener::track_message().

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

 {
   DBG_ENTRY_LVL("DataLinkSet","send_response",6);
   GuardType guard(this->lock_);
 
   TransportSendControlElement* const send_element =
     new TransportSendControlElement(static_cast<int>(map_.size()), pub_id,
                                        &send_response_listener_, header,
                                        move(response));
   if (!send_element) return;
   send_response_listener_.track_message();
 
   for (MapType::iterator itr = map_.begin();
        itr != map_.end();
        ++itr) {
     itr->second->send_start();
     itr->second->send(send_element);
     itr->second->send_stop(pub_id);
   }
 }

◆ send_start()

ACE_INLINE void OpenDDS::DCPS::DataLinkSet::send_start ( DataLinkSet * link_set )

Calls send_start() on the links in link_set and also adds the links from link_set to *this.

Definition at line 202 of file DataLinkSet.inl.

References ACE_ERROR, ACE_INLINE, OpenDDS::DCPS::bind(), DBG_ENTRY_LVL, LM_ERROR, lock_, map_, and OpenDDS::DCPS::OPENDDS_VECTOR().

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

 {
   DBG_ENTRY_LVL("DataLinkSet","send_start",6);
 
   typedef OPENDDS_VECTOR(DataLink_rch) SendStartVec;
   SendStartVec send_start_vec;
 
   {
     GuardType guard1(lock_);
     GuardType guard2(in->lock_);
     for (MapType::iterator itr = in->map_.begin(); itr != in->map_.end(); ++itr) {
       // Attempt to add the current DataLink to this set.
       int result = OpenDDS::DCPS::bind(map_, itr->first, itr->second);
 
       if (result == 0) {
         // We successfully added the current DataLink to this set,
         // meaning that it wasn't already a member.  We should tell
         // the DataLink about the send_start() event.
         send_start_vec.push_back(itr->second);
 
       } else if (result == -1) {
         ACE_ERROR((LM_ERROR,
                    "(%P|%t) ERROR: Failed to bind data link into set.\n"));
       }
 
       // Note that there is a possibility that the result == 1, which
       // means that the DataLink already exists in our map_->  We skip
       // all of these cases.
     }
   }
 
   for (SendStartVec::iterator it = send_start_vec.begin(); it != send_start_vec.end(); ++it) {
     (*it)->send_start();
   }
 }

◆ send_stop()

ACE_INLINE void OpenDDS::DCPS::DataLinkSet::send_stop ( GUID_t repoId )

Calls send_stop() on the links with ID repoId and then clears the set.

Definition at line 239 of file DataLinkSet.inl.

References ACE_INLINE, DBG_ENTRY_LVL, lock_, and map_.

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

 {
   DBG_ENTRY_LVL("DataLinkSet","send_stop",6);
   // Iterate over our map_ and tell each DataLink about the send_stop() event.
   MapType map_copy;
   {
     GuardType guard(lock_);
     map_copy = map_;
     map_.clear();
   }
 
   for (MapType::iterator itr = map_copy.begin(); itr != map_copy.end(); ++itr) {
     itr->second->send_stop(repoId);
   }
 }

◆ terminate_send_if_suspended()

void OpenDDS::DCPS::DataLinkSet::terminate_send_if_suspended ( )

Definition at line 94 of file DataLinkSet.cpp.

References lock_, and map_.

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

 {
   MapType map_copy;
   {
     GuardType guard(lock_);
     map_copy = map_;
   }
   for (MapType::iterator itr = map_copy.begin();
       itr != map_copy.end(); ++itr) {
         itr->second->terminate_send_if_suspended();
   }
 }

Member Data Documentation

◆ lock_

LockType OpenDDS::DCPS::DataLinkSet::lock_

mutableprivate

This lock will protect critical sections of code that play a role in the sending of data.

Definition at line 100 of file DataLinkSet.h.

Referenced by copy_map_to(), empty(), insert_link(), is_leading(), remove_all_msgs(), remove_link(), remove_sample(), select_links(), send(), send_control(), send_final_acks(), send_response(), send_start(), send_stop(), and terminate_send_if_suspended().

◆ map_

MapType OpenDDS::DCPS::DataLinkSet::map_

private

Hash map for DataLinks.

Definition at line 96 of file DataLinkSet.h.

Referenced by copy_map_to(), empty(), insert_link(), is_leading(), remove_all_msgs(), remove_link(), remove_sample(), select_links(), send(), send_control(), send_final_acks(), send_response(), send_start(), send_stop(), and terminate_send_if_suspended().

◆ send_response_listener_

SendResponseListener OpenDDS::DCPS::DataLinkSet::send_response_listener_

private

Listener for TransportSendControlElements created in send_response.

Definition at line 103 of file DataLinkSet.h.

Referenced by send_response().

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

Public Types

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ GuardType

◆ LockType

Constructor & Destructor Documentation

◆ DataLinkSet()

◆ ~DataLinkSet()

Member Function Documentation

◆ copy_map_to()

◆ empty()

◆ insert_link()

◆ is_leading()

◆ lock()

◆ map()

◆ OPENDDS_MAP()

◆ remove_all_msgs()

◆ remove_link()

◆ remove_sample()

◆ select_links()

◆ send()

◆ send_control() [1/2]

◆ send_control() [2/2]

◆ send_final_acks()

◆ send_response()

◆ send_start()

◆ send_stop()

◆ terminate_send_if_suspended()

Member Data Documentation

◆ lock_

◆ map_

◆ send_response_listener_