#include <StaticDiscovery.h>

Inheritance diagram for OpenDDS::DCPS::StaticEndpointManager:

[legend]

Collaboration diagram for OpenDDS::DCPS::StaticEndpointManager:

[legend]

Classes
struct	DiscoveredPublication

struct	DiscoveredSubscription

struct	LocalEndpoint

struct	LocalPublication

struct	LocalSubscription

struct	TypeIdOrigSeqNumber

Public Member Functions
	StaticEndpointManager (const GUID_t &participant_id, ACE_Thread_Mutex &lock, const EndpointRegistry &registry, StaticParticipant &participant)

	~StaticEndpointManager ()

void	init_bit ()

virtual void	assign_publication_key (GUID_t &rid, const GUID_t &topicId, const DDS::DataWriterQos &qos)

virtual void	assign_subscription_key (GUID_t &rid, const GUID_t &topicId, const DDS::DataReaderQos &qos)

virtual bool	update_topic_qos (const GUID_t &, const DDS::TopicQos &)

virtual bool	update_publication_qos (const GUID_t &, const DDS::DataWriterQos &, const DDS::PublisherQos &)

virtual bool	update_subscription_qos (const GUID_t &, const DDS::DataReaderQos &, const DDS::SubscriberQos &)

virtual bool	update_subscription_params (const GUID_t &, const DDS::StringSeq &)

virtual bool	disassociate ()

virtual DDS::ReturnCode_t	add_publication_i (const GUID_t &, LocalPublication &)

virtual DDS::ReturnCode_t	remove_publication_i (const GUID_t &, LocalPublication &)

virtual DDS::ReturnCode_t	add_subscription_i (const GUID_t &, LocalSubscription &)

virtual DDS::ReturnCode_t	remove_subscription_i (const GUID_t &, LocalSubscription &)

virtual bool	is_expectant_opendds (const GUID_t &endpoint) const

virtual bool	shutting_down () const

virtual void	populate_transport_locator_sequence (TransportLocatorSeq *&, DiscoveredSubscriptionIter &, const GUID_t &)

virtual void	populate_transport_locator_sequence (TransportLocatorSeq *&, DiscoveredPublicationIter &, const GUID_t &)

virtual void	reader_exists (const GUID_t &readerid, const GUID_t &writerid)

virtual void	reader_does_not_exist (const GUID_t &readerid, const GUID_t &writerid)

virtual void	writer_exists (const GUID_t &writerid, const GUID_t &readerid)

virtual void	writer_does_not_exist (const GUID_t &writerid, const GUID_t &readerid)

void	cleanup_type_lookup_data (const GuidPrefix_t &prefix, const XTypes::TypeIdentifier &ti, bool secure)

PublicationBuiltinTopicDataDataReaderImpl *	pub_bit ()

SubscriptionBuiltinTopicDataDataReaderImpl *	sub_bit ()

void	type_lookup_init (ReactorInterceptor_rch reactor_interceptor)

void	type_lookup_fini ()

void	type_lookup_service (const XTypes::TypeLookupService_rch type_lookup_service)

void	purge_dead_topic (const String &topic_name)

void	ignore (const GUID_t &to_ignore)

bool	ignoring (const GUID_t &guid) const

bool	ignoring (const char *topic_name) const

TopicStatus	assert_topic (GUID_t &topicId, const char topicName, const char dataTypeName, const DDS::TopicQos &qos, bool hasDcpsKey, TopicCallbacks *topic_callbacks)

TopicStatus	find_topic (const char *topicName, CORBA::String_out dataTypeName, DDS::TopicQos_out qos, GUID_t &topicId)

TopicStatus	remove_topic (const GUID_t &topicId)

GUID_t	add_publication (const GUID_t &topicId, DataWriterCallbacks_rch publication, const DDS::DataWriterQos &qos, const TransportLocatorSeq &transInfo, const DDS::PublisherQos &publisherQos, const XTypes::TypeInformation &type_info)

void	remove_publication (const GUID_t &publicationId)

void	update_publication_locators (const GUID_t &publicationId, const TransportLocatorSeq &transInfo)

GUID_t	add_subscription (const GUID_t &topicId, DataReaderCallbacks_rch subscription, const DDS::DataReaderQos &qos, const TransportLocatorSeq &transInfo, const DDS::SubscriberQos &subscriberQos, const char filterClassName, const char filterExpr, const DDS::StringSeq &params, const XTypes::TypeInformation &type_info)

void	remove_subscription (const GUID_t &subscriptionId)

void	update_subscription_locators (const GUID_t &subscriptionId, const TransportLocatorSeq &transInfo)

void	match_endpoints (GUID_t repoId, const TopicDetails &td, bool remove=false)

void	remove_assoc (const GUID_t &remove_from, const GUID_t &removing)

virtual void	add_assoc_i (const GUID_t &, const LocalPublication &, const GUID_t &, const DiscoveredSubscription &)

virtual void	remove_assoc_i (const GUID_t &, const LocalPublication &, const GUID_t &)

virtual void	add_assoc_i (const GUID_t &, const LocalSubscription &, const GUID_t &, const DiscoveredPublication &)

virtual void	remove_assoc_i (const GUID_t &, const LocalSubscription &, const GUID_t &)

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_exception (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_timeout (const ACE_Time_Value &current_time, const void *act=0)

virtual int	handle_exit (ACE_Process *)

virtual int	handle_close (ACE_HANDLE handle, ACE_Reactor_Mask close_mask)

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

Public Member Functions inherited from OpenDDS::DCPS::DiscoveryListener
	DiscoveryListener ()

virtual	~DiscoveryListener ()

Protected Types
typedef DiscoveredSubscriptionMap::iterator	DiscoveredSubscriptionIter

typedef DiscoveredPublicationMap::iterator	DiscoveredPublicationIter

typedef LocalPublicationMap::iterator	LocalPublicationIter

typedef LocalPublicationMap::const_iterator	LocalPublicationCIter

typedef LocalSubscriptionMap::iterator	LocalSubscriptionIter

typedef LocalSubscriptionMap::const_iterator	LocalSubscriptionCIter

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

Protected Member Functions
typedef	OPENDDS_MAP_CMP (GUID_t, DiscoveredSubscription, GUID_tKeyLessThan) DiscoveredSubscriptionMap

typedef	OPENDDS_MAP_CMP (GUID_t, DiscoveredPublication, GUID_tKeyLessThan) DiscoveredPublicationMap

typedef	OPENDDS_MAP_CMP (GUID_t, LocalPublication, GUID_tKeyLessThan) LocalPublicationMap

typedef	OPENDDS_MAP_CMP (GUID_t, LocalSubscription, GUID_tKeyLessThan) LocalSubscriptionMap

typedef	OPENDDS_MAP_CMP (GUID_t, String, GUID_tKeyLessThan) TopicNameMap

virtual void	remove_from_bit_i (const DiscoveredPublication &)

virtual void	remove_from_bit_i (const DiscoveredSubscription &)

virtual DDS::ReturnCode_t	write_publication_data (const GUID_t &, LocalPublication &, const GUID_t &reader=GUID_UNKNOWN)

virtual DDS::ReturnCode_t	write_subscription_data (const GUID_t &, LocalSubscription &, const GUID_t &reader=GUID_UNKNOWN)

virtual bool	send_type_lookup_request (const XTypes::TypeIdentifierSeq &, const GUID_t &, bool, bool)

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 DDS::BuiltinTopicKey_t	get_key (const DiscoveredPublication &pub)

static DDS::BuiltinTopicKey_t	get_key (const DiscoveredSubscription &sub)

Private Types
typedef PmfSporadicTask< StaticEndpointManager >	StaticEndpointManagerSporadic

Private Member Functions
void	match (const GUID_t &writer, const GUID_t &reader)

void	need_minimal_and_or_complete_types (const XTypes::TypeInformation *type_info, bool &need_minimal, bool &need_complete) const

void	remove_expired_endpoints (const MonotonicTimePoint &)

void	match_continue (const GUID_t &writer, const GUID_t &reader)

void	remove_from_bit (const DiscoveredPublication &pub)

void	remove_from_bit (const DiscoveredSubscription &sub)

GUID_t	make_topic_guid ()

bool	has_dcps_key (const GUID_t &topicId) const

	OPENDDS_SET (OPENDDS_STRING) ignored_topics_

	OPENDDS_SET_CMP (GUID_t, GUID_tKeyLessThan) relay_only_readers_

typedef	OPENDDS_MAP (SequenceNumber, TypeIdOrigSeqNumber) OrigSeqNumberMap

Private Attributes
ACE_Thread_Mutex &	lock_

GUID_t	participant_id_

RepoIdSet	ignored_guids_

unsigned int	topic_counter_

LocalPublicationMap	local_publications_

LocalSubscriptionMap	local_subscriptions_

DiscoveredPublicationMap	discovered_publications_

DiscoveredSubscriptionMap	discovered_subscriptions_

TopicDetailsMap	topics_

TopicNameMap	topic_names_

const EndpointRegistry &	registry_

StaticParticipant &	participant_

XTypes::TypeLookupService_rch	type_lookup_service_

RcHandle< StaticEndpointManagerSporadic >	type_lookup_reply_deadline_processor_

TimeDuration	max_type_lookup_service_reply_period_

SequenceNumber	type_lookup_service_sequence_number_

OrigSeqNumberMap	orig_seq_numbers_

Additional Inherited Members
Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

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

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Definition at line 130 of file StaticDiscovery.h.

Member Typedef Documentation

◆ DiscoveredPublicationIter

typedef DiscoveredPublicationMap::iterator OpenDDS::DCPS::StaticEndpointManager::DiscoveredPublicationIter

protected

Definition at line 210 of file StaticDiscovery.h.

◆ DiscoveredSubscriptionIter

typedef DiscoveredSubscriptionMap::iterator OpenDDS::DCPS::StaticEndpointManager::DiscoveredSubscriptionIter

protected

Definition at line 172 of file StaticDiscovery.h.

◆ LocalPublicationCIter

typedef LocalPublicationMap::const_iterator OpenDDS::DCPS::StaticEndpointManager::LocalPublicationCIter

protected

Definition at line 247 of file StaticDiscovery.h.

◆ LocalPublicationIter

typedef LocalPublicationMap::iterator OpenDDS::DCPS::StaticEndpointManager::LocalPublicationIter

protected

Definition at line 246 of file StaticDiscovery.h.

◆ LocalSubscriptionCIter

typedef LocalSubscriptionMap::const_iterator OpenDDS::DCPS::StaticEndpointManager::LocalSubscriptionCIter

protected

Definition at line 252 of file StaticDiscovery.h.

◆ LocalSubscriptionIter

typedef LocalSubscriptionMap::iterator OpenDDS::DCPS::StaticEndpointManager::LocalSubscriptionIter

protected

Definition at line 251 of file StaticDiscovery.h.

◆ StaticEndpointManagerSporadic

typedef PmfSporadicTask<StaticEndpointManager> OpenDDS::DCPS::StaticEndpointManager::StaticEndpointManagerSporadic

private

Definition at line 459 of file StaticDiscovery.h.

Constructor & Destructor Documentation

◆ StaticEndpointManager()

OpenDDS::DCPS::StaticEndpointManager::StaticEndpointManager	(	const GUID_t &	participant_id,
		ACE_Thread_Mutex &	lock,
		const EndpointRegistry &	registry,
		StaticParticipant &	participant
	)

Definition at line 77 of file StaticDiscovery.cpp.

References TheServiceParticipant, and type_lookup_init().

   : lock_(lock)
   , participant_id_(participant_id)
   , topic_counter_(0)
   , registry_(registry)
 #ifndef DDS_HAS_MINIMUM_BIT
   , participant_(participant)
 #endif
   , max_type_lookup_service_reply_period_(0)
   , type_lookup_service_sequence_number_(0)
 {
 #ifdef DDS_HAS_MINIMUM_BIT
   ACE_UNUSED_ARG(participant);
 #endif
   type_lookup_init(TheServiceParticipant->interceptor());
 }

◆ ~StaticEndpointManager()

OpenDDS::DCPS::StaticEndpointManager::~StaticEndpointManager ( )

Definition at line 97 of file StaticDiscovery.cpp.

References type_lookup_fini().

 {
   type_lookup_fini();
 }

Member Function Documentation

◆ add_assoc_i() [1/2]

virtual void OpenDDS::DCPS::StaticEndpointManager::add_assoc_i	(	const GUID_t &	,
		const LocalPublication &	,
		const GUID_t &	,
		const DiscoveredSubscription &
	)

inlinevirtual

Definition at line 410 of file StaticDiscovery.h.

Referenced by match_continue().

411 {}

◆ add_assoc_i() [2/2]

virtual void OpenDDS::DCPS::StaticEndpointManager::add_assoc_i	(	const GUID_t &	,
		const LocalSubscription &	,
		const GUID_t &	,
		const DiscoveredPublication &
	)

inlinevirtual

Definition at line 414 of file StaticDiscovery.h.

415 {}

◆ add_publication()

GUID_t OpenDDS::DCPS::StaticEndpointManager::add_publication	(	const GUID_t &	topicId,
		DataWriterCallbacks_rch	publication,
		const DDS::DataWriterQos &	qos,
		const TransportLocatorSeq &	transInfo,
		const DDS::PublisherQos &	publisherQos,
		const XTypes::TypeInformation &	type_info
	)

Definition at line 760 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::TopicDetails::add_local_publication(), add_publication_i(), assign_publication_key(), OpenDDS::DCPS::DCPS_debug_level, LM_DEBUG, local_publications_, lock_, match_endpoints(), OPENDDS_STRING, participant_id_, OpenDDS::DCPS::StaticEndpointManager::LocalPublication::publication_, OpenDDS::DCPS::StaticEndpointManager::LocalPublication::publisher_qos_, OpenDDS::DCPS::StaticEndpointManager::LocalPublication::qos_, DDS::RETCODE_OK, OpenDDS::DCPS::StaticEndpointManager::LocalEndpoint::topic_id_, topic_names_, topics_, OpenDDS::DCPS::StaticEndpointManager::LocalEndpoint::trans_info_, OpenDDS::DCPS::StaticEndpointManager::LocalEndpoint::type_info_, and write_publication_data().

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, lock_, GUID_t());
 
   GUID_t rid = participant_id_;
   assign_publication_key(rid, topicId, qos);
   LocalPublication& pb = local_publications_[rid];
   pb.topic_id_ = topicId;
   pb.publication_ = publication;
   pb.qos_ = qos;
   pb.trans_info_ = transInfo;
   pb.publisher_qos_ = publisherQos;
   pb.type_info_ = type_info;
   const OPENDDS_STRING& topic_name = topic_names_[topicId];
 
   TopicDetails& td = topics_[topic_name];
   td.add_local_publication(rid);
 
   if (DDS::RETCODE_OK != add_publication_i(rid, pb)) {
     return GUID_t();
   }
 
   if (DDS::RETCODE_OK != write_publication_data(rid, pb)) {
     return GUID_t();
   }
 
   if (DCPS_debug_level > 3) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) StaticEndpointManager::add_publication - ")
                ACE_TEXT("calling match_endpoints\n")));
   }
   match_endpoints(rid, td);
 
   return rid;
 }

◆ add_publication_i()

DDS::ReturnCode_t OpenDDS::DCPS::StaticEndpointManager::add_publication_i	(	const GUID_t &	writerid,
		LocalPublication &	pub
	)

virtual

Definition at line 311 of file StaticDiscovery.cpp.

References OpenDDS::DCPS::EndpointRegistry::Writer::best_effort_readers, OpenDDS::DCPS::WeakRcHandle< T >::lock(), participant_id_, OpenDDS::DCPS::StaticEndpointManager::LocalPublication::publication_, OpenDDS::DCPS::EndpointRegistry::Reader::qos, OpenDDS::DCPS::EndpointRegistry::reader_map, registry_, OpenDDS::DCPS::EndpointRegistry::Writer::reliable_readers, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::EndpointRegistry::Reader::subscriber_qos, OpenDDS::DCPS::EndpointRegistry::Reader::trans_info, and OpenDDS::DCPS::EndpointRegistry::writer_map.

Referenced by add_publication().

 {
   /*
     Find all matching remote readers.
     If the reader is best effort, then associate immediately.
     If the reader is reliable (we are reliable by implication), register with the transport to receive notification that the remote reader is up.
     */
   EndpointRegistry::WriterMapType::const_iterator pos = registry_.writer_map.find(writerid);
   if (pos == registry_.writer_map.end()) {
     return DDS::RETCODE_ERROR;
   }
   const EndpointRegistry::Writer& writer = pos->second;
 
   for (RepoIdSet::const_iterator pos = writer.best_effort_readers.begin(), limit = writer.best_effort_readers.end();
        pos != limit;
        ++pos) {
     const GUID_t& readerid = *pos;
     const EndpointRegistry::Reader& reader = registry_.reader_map.find(readerid)->second;
 
     const ReaderAssociation ra =
       {reader.trans_info, TransportLocator(), 0, readerid, reader.subscriber_qos, reader.qos, "", "", 0, 0, {0, 0}};
     DataWriterCallbacks_rch pl = pub.publication_.lock();
     if (pl) {
       pl->add_association(writerid, ra, true);
     }
   }
 
   for (RepoIdSet::const_iterator pos = writer.reliable_readers.begin(), limit = writer.reliable_readers.end();
        pos != limit;
        ++pos) {
     const GUID_t& readerid = *pos;
     const EndpointRegistry::Reader& reader = registry_.reader_map.find(readerid)->second;
     DataWriterCallbacks_rch pl = pub.publication_.lock();
     if (pl) {
       pl->register_for_reader(participant_id_, writerid, readerid, reader.trans_info, this);
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ add_subscription()

GUID_t OpenDDS::DCPS::StaticEndpointManager::add_subscription	(	const GUID_t &	topicId,
		DataReaderCallbacks_rch	subscription,
		const DDS::DataReaderQos &	qos,
		const TransportLocatorSeq &	transInfo,
		const DDS::SubscriberQos &	subscriberQos,
		const char *	filterClassName,
		const char *	filterExpr,
		const DDS::StringSeq &	params,
		const XTypes::TypeInformation &	type_info
	)

Definition at line 839 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::TopicDetails::add_local_subscription(), add_subscription_i(), assign_subscription_key(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::ContentFilterProperty_t::expressionParameters, OpenDDS::DCPS::ContentFilterProperty_t::filterClassName, OpenDDS::DCPS::ContentFilterProperty_t::filterExpression, OpenDDS::DCPS::StaticEndpointManager::LocalSubscription::filterProperties, LM_DEBUG, local_subscriptions_, lock_, match_endpoints(), OPENDDS_STRING, participant_id_, OpenDDS::DCPS::StaticEndpointManager::LocalSubscription::qos_, DDS::RETCODE_OK, OpenDDS::DCPS::StaticEndpointManager::LocalSubscription::subscriber_qos_, OpenDDS::DCPS::StaticEndpointManager::LocalSubscription::subscription_, OpenDDS::DCPS::StaticEndpointManager::LocalEndpoint::topic_id_, topic_names_, topics_, OpenDDS::DCPS::StaticEndpointManager::LocalEndpoint::trans_info_, OpenDDS::DCPS::StaticEndpointManager::LocalEndpoint::type_info_, and write_subscription_data().

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, lock_, GUID_t());
 
   GUID_t rid = participant_id_;
   assign_subscription_key(rid, topicId, qos);
   LocalSubscription& sb = local_subscriptions_[rid];
   sb.topic_id_ = topicId;
   sb.subscription_ = subscription;
   sb.qos_ = qos;
   sb.trans_info_ = transInfo;
   sb.subscriber_qos_ = subscriberQos;
   sb.filterProperties.filterClassName = filterClassName;
   sb.filterProperties.filterExpression = filterExpr;
   sb.filterProperties.expressionParameters = params;
   sb.type_info_ = type_info;
   const OPENDDS_STRING& topic_name = topic_names_[topicId];
 
   TopicDetails& td = topics_[topic_name];
   td.add_local_subscription(rid);
 
   if (DDS::RETCODE_OK != add_subscription_i(rid, sb)) {
     return GUID_t();
   }
 
   if (DDS::RETCODE_OK != write_subscription_data(rid, sb)) {
     return GUID_t();
   }
 
   if (DCPS_debug_level > 3) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) StaticEndpointManager::add_subscription - ")
                ACE_TEXT("calling match_endpoints\n")));
   }
   match_endpoints(rid, td);
 
   return rid;
 }

◆ add_subscription_i()

DDS::ReturnCode_t OpenDDS::DCPS::StaticEndpointManager::add_subscription_i	(	const GUID_t &	readerid,
		LocalSubscription &	sub
	)

virtual

Definition at line 381 of file StaticDiscovery.cpp.

References OpenDDS::DCPS::EndpointRegistry::Reader::best_effort_writers, OpenDDS::DCPS::WeakRcHandle< T >::lock(), participant_id_, OpenDDS::DCPS::EndpointRegistry::Writer::publisher_qos, OpenDDS::DCPS::EndpointRegistry::Writer::qos, OpenDDS::DCPS::EndpointRegistry::reader_map, registry_, OpenDDS::DCPS::EndpointRegistry::Reader::reliable_writers, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::StaticEndpointManager::LocalSubscription::subscription_, OpenDDS::DCPS::EndpointRegistry::Writer::trans_info, and OpenDDS::DCPS::EndpointRegistry::writer_map.

Referenced by add_subscription().

 {
   /*
     Find all matching remote writers.
     If we (the reader) is best effort, then associate immediately.
     If we (the reader) are reliable, then register with the transport to receive notification that the remote writer is up.
     */
   EndpointRegistry::ReaderMapType::const_iterator pos = registry_.reader_map.find(readerid);
   if (pos == registry_.reader_map.end()) {
     return DDS::RETCODE_ERROR;
   }
   const EndpointRegistry::Reader& reader = pos->second;
 
   for (RepoIdSet::const_iterator pos = reader.best_effort_writers.begin(), limit = reader.best_effort_writers.end();
        pos != limit;
        ++pos) {
     const GUID_t& writerid = *pos;
     const EndpointRegistry::Writer& writer = registry_.writer_map.find(writerid)->second;
 
     DDS::OctetSeq type_info;
     const WriterAssociation wa = {
       writer.trans_info, TransportLocator(), 0, writerid, writer.publisher_qos, writer.qos, type_info, {0, 0}
     };
     DataReaderCallbacks_rch sl = sub.subscription_.lock();
     if (sl) {
       sl->add_association(readerid, wa, false);
     }
   }
 
   for (RepoIdSet::const_iterator pos = reader.reliable_writers.begin(), limit = reader.reliable_writers.end();
        pos != limit;
        ++pos) {
     const GUID_t& writerid = *pos;
     const EndpointRegistry::Writer& writer = registry_.writer_map.find(writerid)->second;
     DataReaderCallbacks_rch sl = sub.subscription_.lock();
     if (sl) {
       sl->register_for_writer(participant_id_, readerid, writerid, writer.trans_info, this);
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ assert_topic()

TopicStatus OpenDDS::DCPS::StaticEndpointManager::assert_topic	(	GUID_t &	topicId,
		const char *	topicName,
		const char *	dataTypeName,
		const DDS::TopicQos &	qos,
		bool	hasDcpsKey,
		TopicCallbacks *	topic_callbacks
	)

Definition at line 698 of file StaticDiscovery.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::CONFLICTING_TYPENAME, OpenDDS::DCPS::CREATED, OpenDDS::DCPS::FOUND, OpenDDS::DCPS::TopicDetails::init(), OpenDDS::DCPS::INTERNAL_ERROR, lock_, make_topic_guid(), OpenDDS::DCPS::TopicDetails::set_local(), topic_names_, and topics_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, lock_, INTERNAL_ERROR);
   TopicDetailsMap::iterator iter = topics_.find(topicName);
   if (iter != topics_.end()) {
     if (iter->second.local_is_set() && iter->second.local_data_type_name() != dataTypeName) {
       return CONFLICTING_TYPENAME;
     }
     topicId = iter->second.topic_id();
     iter->second.set_local(dataTypeName, qos, hasDcpsKey, topic_callbacks);
     return FOUND;
   }
 
   TopicDetails& td = topics_[topicName];
   topicId = make_topic_guid();
   td.init(topicName, topicId);
   topic_names_[topicId] = topicName;
   td.set_local(dataTypeName, qos, hasDcpsKey, topic_callbacks);
 
   return CREATED;
 }

◆ assign_publication_key()

void OpenDDS::DCPS::StaticEndpointManager::assign_publication_key	(	GUID_t &	rid,
		const GUID_t &	topicId,
		const DDS::DataWriterQos &	qos
	)

virtual

Definition at line 197 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::EntityId_t::entityKey, OpenDDS::DCPS::EntityId_t::entityKind, OpenDDS::DCPS::ENTITYKIND_USER_WRITER_WITH_KEY, LM_ERROR, LM_INFO, LM_WARNING, registry_, DDS::DataWriterQos::user_data, DDS::UserDataQosPolicy::value, and OpenDDS::DCPS::EndpointRegistry::writer_map.

Referenced by add_publication().

 {
   if (qos.user_data.value.length() != BYTES_IN_ENTITY) {
     ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::assign_publication_key: no user data to identify writer\n")));
     return;
   }
 
   rid.entityId.entityKey[0] = qos.user_data.value[0];
   rid.entityId.entityKey[1] = qos.user_data.value[1];
   rid.entityId.entityKey[2] = qos.user_data.value[2];
   rid.entityId.entityKind = ENTITYKIND_USER_WRITER_WITH_KEY;
 
   if (DCPS_debug_level > 8) {
     ACE_DEBUG((LM_INFO, "(%P|%t) looking up writer ID %C\n",
                LogGuid(rid).c_str()));
   }
 
   EndpointRegistry::WriterMapType::const_iterator pos = registry_.writer_map.find(rid);
   if (pos == registry_.writer_map.end()) {
     ACE_DEBUG((LM_WARNING, ACE_TEXT("(%P|%t) WARNING: StaticEndpointManager::assign_publication_key: unknown writer: %C\n"), LogGuid(rid).c_str()));
     return;
   }
 
   DDS::DataWriterQos qos2(qos);
   // Qos in registry will not have the user data so overwrite.
   qos2.user_data = pos->second.qos.user_data;
 
   DDS::DataWriterQos qos3(pos->second.qos);
 
   if (qos2 != qos3) {
     ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::assign_publication_key: dynamic and static QoS differ\n")));
   }
 }

◆ assign_subscription_key()

void OpenDDS::DCPS::StaticEndpointManager::assign_subscription_key	(	GUID_t &	rid,
		const GUID_t &	topicId,
		const DDS::DataReaderQos &	qos
	)

virtual

Definition at line 233 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::EntityId_t::entityKey, OpenDDS::DCPS::EntityId_t::entityKind, OpenDDS::DCPS::ENTITYKIND_USER_READER_WITH_KEY, LM_ERROR, LM_WARNING, OpenDDS::DCPS::EndpointRegistry::reader_map, registry_, DDS::DataReaderQos::user_data, and DDS::UserDataQosPolicy::value.

Referenced by add_subscription().

 {
   if (qos.user_data.value.length() != BYTES_IN_ENTITY) {
     ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::assign_subscription_key: no user data to identify reader\n")));
     return;
   }
 
   rid.entityId.entityKey[0] = qos.user_data.value[0];
   rid.entityId.entityKey[1] = qos.user_data.value[1];
   rid.entityId.entityKey[2] = qos.user_data.value[2];
   rid.entityId.entityKind = ENTITYKIND_USER_READER_WITH_KEY;
 
   EndpointRegistry::ReaderMapType::const_iterator pos = registry_.reader_map.find(rid);
   if (pos == registry_.reader_map.end()) {
     ACE_DEBUG((LM_WARNING, ACE_TEXT("(%P|%t) WARNING: StaticEndpointManager::assign_subscription_key: unknown reader: %C\n"), LogGuid(rid).c_str()));
     return;
   }
 
   DDS::DataReaderQos qos2(qos);
   // Qos in registry will not have the user data so overwrite.
   qos2.user_data = pos->second.qos.user_data;
 
   DDS::DataReaderQos qos3(pos->second.qos);
 
   if (qos2 != qos3) {
     ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::assign_subscription_key: dynamic and static QoS differ\n")));
   }
 }

◆ cleanup_type_lookup_data()

void OpenDDS::DCPS::StaticEndpointManager::cleanup_type_lookup_data	(	const GuidPrefix_t &	prefix,
		const XTypes::TypeIdentifier &	ti,
		bool	secure
	)

Definition at line 557 of file StaticDiscovery.cpp.

Referenced by remove_expired_endpoints().

 {
   // Do nothing.
 }

◆ disassociate()

bool OpenDDS::DCPS::StaticEndpointManager::disassociate ( )

virtual

Definition at line 303 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_TEXT(), and LM_NOTICE.

 {
   ACE_DEBUG((LM_NOTICE, ACE_TEXT("(%P|%t) StaticEndpointManager::disassociate TODO\n")));
   // TODO
   return false;
 }

◆ find_topic()

TopicStatus OpenDDS::DCPS::StaticEndpointManager::find_topic	(	const char *	topicName,
		CORBA::String_out	dataTypeName,
		DDS::TopicQos_out	qos,
		GUID_t &	topicId
	)

Definition at line 723 of file StaticDiscovery.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::FOUND, OpenDDS::DCPS::INTERNAL_ERROR, OpenDDS::DCPS::TopicDetails::local_data_type_name(), OpenDDS::DCPS::TopicDetails::local_qos(), lock_, OpenDDS::DCPS::NOT_FOUND, OpenDDS::DCPS::TopicDetails::topic_id(), Update::TopicQos, and topics_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, lock_, INTERNAL_ERROR);
   TopicDetailsMap::const_iterator iter = topics_.find(topicName);
   if (iter == topics_.end()) {
     return NOT_FOUND;
   }
 
   const TopicDetails& td = iter->second;
 
   dataTypeName = td.local_data_type_name().c_str();
   qos = new DDS::TopicQos(td.local_qos());
   topicId = td.topic_id();
   return FOUND;
 }

◆ get_key() [1/2]

static DDS::BuiltinTopicKey_t OpenDDS::DCPS::StaticEndpointManager::get_key ( const DiscoveredPublication & pub )

inlinestaticprotected

Definition at line 256 of file StaticDiscovery.h.

References OpenDDS::DCPS::DiscoveredWriterData::ddsPublicationData, and OpenDDS::DCPS::StaticEndpointManager::DiscoveredPublication::writer_data_.

   {
     return pub.writer_data_.ddsPublicationData.key;
   }

◆ get_key() [2/2]

static DDS::BuiltinTopicKey_t OpenDDS::DCPS::StaticEndpointManager::get_key ( const DiscoveredSubscription & sub )

inlinestaticprotected

Definition at line 260 of file StaticDiscovery.h.

References OpenDDS::DCPS::DiscoveredReaderData::ddsSubscriptionData, and OpenDDS::DCPS::StaticEndpointManager::DiscoveredSubscription::reader_data_.

   {
     return sub.reader_data_.ddsSubscriptionData.key;
   }

◆ has_dcps_key()

bool OpenDDS::DCPS::StaticEndpointManager::has_dcps_key ( const GUID_t & topicId ) const

private

Definition at line 1487 of file StaticDiscovery.cpp.

References OPENDDS_MAP_CMP(), OPENDDS_STRING, topic_names_, and topics_.

 {
   typedef OPENDDS_MAP_CMP(GUID_t, OPENDDS_STRING, GUID_tKeyLessThan) TNMap;
   TNMap::const_iterator tn = topic_names_.find(topicId);
   if (tn == topic_names_.end()) return false;
 
   TopicDetailsMap::const_iterator td = topics_.find(tn->second);
   if (td == topics_.end()) return false;
 
   return td->second.has_dcps_key();
 }

◆ ignore()

void OpenDDS::DCPS::StaticEndpointManager::ignore ( const GUID_t & to_ignore )

Definition at line 618 of file StaticDiscovery.cpp.

References OpenDDS::DCPS::TopicDetails::discovered_publications(), discovered_publications_, OpenDDS::DCPS::TopicDetails::discovered_subscriptions(), discovered_subscriptions_, ignored_guids_, OpenDDS::DCPS::TopicDetails::is_dead(), match_endpoints(), OPENDDS_MAP_CMP(), OPENDDS_STRING, purge_dead_topic(), OpenDDS::DCPS::TopicDetails::remove_discovered_publication(), OpenDDS::DCPS::TopicDetails::remove_discovered_subscription(), remove_from_bit(), shutting_down(), topic_names_, and topics_.

 {
   // Locked prior to call from Spdp.
   ignored_guids_.insert(to_ignore);
   {
     const DiscoveredPublicationIter iter = discovered_publications_.find(to_ignore);
     if (iter != discovered_publications_.end()) {
       // clean up tracking info
       const String topic_name = iter->second.get_topic_name();
       TopicDetails& td = topics_[topic_name];
       td.remove_discovered_publication(to_ignore);
       remove_from_bit(iter->second);
       discovered_publications_.erase(iter);
       // break associations
       match_endpoints(to_ignore, td, true /*remove*/);
       if (td.is_dead()) {
         purge_dead_topic(topic_name);
       }
       return;
     }
   }
   {
     const DiscoveredSubscriptionIter iter =
       discovered_subscriptions_.find(to_ignore);
     if (iter != discovered_subscriptions_.end()) {
       // clean up tracking info
       const String topic_name = iter->second.get_topic_name();
       TopicDetails& td = topics_[topic_name];
       td.remove_discovered_publication(to_ignore);
       remove_from_bit(iter->second);
       discovered_subscriptions_.erase(iter);
       // break associations
       match_endpoints(to_ignore, td, true /*remove*/);
       if (td.is_dead()) {
         purge_dead_topic(topic_name);
       }
       return;
     }
   }
   {
     const OPENDDS_MAP_CMP(GUID_t, OPENDDS_STRING, GUID_tKeyLessThan)::iterator
       iter = topic_names_.find(to_ignore);
     if (iter != topic_names_.end()) {
       ignored_topics_.insert(iter->second);
       // Remove all publications and subscriptions on this topic
       TopicDetails& td = topics_[iter->second];
       {
         const RepoIdSet ids = td.discovered_publications();
         for (RepoIdSet::const_iterator ep = ids.begin(); ep!= ids.end(); ++ep) {
           match_endpoints(*ep, td, true /*remove*/);
           td.remove_discovered_publication(*ep);
           // TODO: Do we need to remove from discovered_subscriptions?
           if (shutting_down()) { return; }
         }
       }
       {
         const RepoIdSet ids = td.discovered_subscriptions();
         for (RepoIdSet::const_iterator ep = ids.begin(); ep!= ids.end(); ++ep) {
           match_endpoints(*ep, td, true /*remove*/);
           td.remove_discovered_subscription(*ep);
           // TODO: Do we need to remove from discovered_publications?
           if (shutting_down()) { return; }
         }
       }
       if (td.is_dead()) {
         purge_dead_topic(iter->second);
       }
     }
   }
 }

◆ ignoring() [1/2]

bool OpenDDS::DCPS::StaticEndpointManager::ignoring ( const GUID_t & guid ) const

Definition at line 689 of file StaticDiscovery.cpp.

References ignored_guids_.

 {
   return ignored_guids_.count(guid);
 }

◆ ignoring() [2/2]

bool OpenDDS::DCPS::StaticEndpointManager::ignoring ( const char * topic_name ) const

Definition at line 693 of file StaticDiscovery.cpp.

 {
   return ignored_topics_.count(topic_name);
 }

◆ init_bit()

void OpenDDS::DCPS::StaticEndpointManager::init_bit ( )

Definition at line 102 of file StaticDiscovery.cpp.

 {
   // Discover all remote publications and subscriptions.
 
   for (EndpointRegistry::WriterMapType::const_iterator pos = registry_.writer_map.begin(),
          limit = registry_.writer_map.end();
        pos != limit;
        ++pos) {
     const GUID_t& remoteid = pos->first;
     const EndpointRegistry::Writer& writer = pos->second;
 
     if (!equal_guid_prefixes(participant_id_, remoteid)) {
       const DDS::BuiltinTopicKey_t key = guid_to_bit_key(remoteid);
 
       // pos represents a remote.
       // Populate data.
       DDS::PublicationBuiltinTopicData data = DDS::PublicationBuiltinTopicData();
 
       data.key = key;
       OPENDDS_STRING topic_name = writer.topic_name;
       data.topic_name = topic_name.c_str();
       const EndpointRegistry::Topic& topic = registry_.topic_map.find(topic_name)->second;
       data.type_name = topic.type_name.c_str();
       data.durability = writer.qos.durability;
       data.durability_service = writer.qos.durability_service;
       data.deadline = writer.qos.deadline;
       data.latency_budget = writer.qos.latency_budget;
       data.liveliness = writer.qos.liveliness;
       data.reliability = writer.qos.reliability;
       data.lifespan = writer.qos.lifespan;
       data.user_data = writer.qos.user_data;
       data.ownership = writer.qos.ownership;
       data.ownership_strength = writer.qos.ownership_strength;
       data.destination_order = writer.qos.destination_order;
       data.presentation = writer.publisher_qos.presentation;
       data.partition = writer.publisher_qos.partition;
       // If the TopicQos becomes available, this can be populated.
       //data.topic_data = topic_details.qos_.topic_data;
       data.group_data = writer.publisher_qos.group_data;
       data.representation = writer.qos.representation;
 
 #ifndef DDS_HAS_MINIMUM_BIT
       OpenDDS::DCPS::PublicationBuiltinTopicDataDataReaderImpl* bit = pub_bit();
       if (bit) { // bit may be null if the DomainParticipant is shutting down
         bit->store_synthetic_data(data, DDS::NEW_VIEW_STATE);
       }
 #endif /* DDS_HAS_MINIMUM_BIT */
     }
   }
 
   for (EndpointRegistry::ReaderMapType::const_iterator pos = registry_.reader_map.begin(),
          limit = registry_.reader_map.end();
        pos != limit;
        ++pos) {
     const GUID_t& remoteid = pos->first;
     const EndpointRegistry::Reader& reader = pos->second;
 
     if (!equal_guid_prefixes(participant_id_, remoteid)) {
       const DDS::BuiltinTopicKey_t key = guid_to_bit_key(remoteid);
 
       // pos represents a remote.
       // Populate data.
       DDS::SubscriptionBuiltinTopicData data = DDS::SubscriptionBuiltinTopicData();
 
       data.key = key;
       OPENDDS_STRING topic_name = reader.topic_name;
       data.topic_name = topic_name.c_str();
       const EndpointRegistry::Topic& topic = registry_.topic_map.find(topic_name)->second;
       data.type_name = topic.type_name.c_str();
       data.durability = reader.qos.durability;
       data.deadline = reader.qos.deadline;
       data.latency_budget = reader.qos.latency_budget;
       data.liveliness = reader.qos.liveliness;
       data.reliability = reader.qos.reliability;
       data.ownership = reader.qos.ownership;
       data.destination_order = reader.qos.destination_order;
       data.user_data = reader.qos.user_data;
       data.time_based_filter = reader.qos.time_based_filter;
       data.presentation = reader.subscriber_qos.presentation;
       data.partition = reader.subscriber_qos.partition;
       // // If the TopicQos becomes available, this can be populated.
       //data.topic_data = topic_details.qos_.topic_data;
       data.group_data = reader.subscriber_qos.group_data;
       data.representation = reader.qos.representation;
 
 #ifndef DDS_HAS_MINIMUM_BIT
       OpenDDS::DCPS::SubscriptionBuiltinTopicDataDataReaderImpl* bit = sub_bit();
       if (bit) { // bit may be null if the DomainParticipant is shutting down
         bit->store_synthetic_data(data, DDS::NEW_VIEW_STATE);
       }
 #endif /* DDS_HAS_MINIMUM_BIT */
     }
   }
 }

◆ is_expectant_opendds()

bool OpenDDS::DCPS::StaticEndpointManager::is_expectant_opendds ( const GUID_t & endpoint ) const

virtual

Definition at line 453 of file StaticDiscovery.cpp.

 {
   // We can't propagate associated writers via SEDP announcments if we're
   // using static discovery, so nobody ought to be "expecting" them
   return false;
 }

◆ make_topic_guid()

GUID_t OpenDDS::DCPS::StaticEndpointManager::make_topic_guid ( )

private

Definition at line 1469 of file StaticDiscovery.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::assign(), OpenDDS::DCPS::EntityId_t::entityKey, OpenDDS::DCPS::EntityId_t::entityKind, OpenDDS::DCPS::ENTITYKIND_OPENDDS_TOPIC, LM_ERROR, OpenDDS::DCPS::make_id(), participant_id_, and topic_counter_.

Referenced by assert_topic().

 {
   EntityId_t entity_id;
   assign(entity_id.entityKey, topic_counter_);
   ++topic_counter_;
   entity_id.entityKind = ENTITYKIND_OPENDDS_TOPIC;
 
   if (topic_counter_ == 0x1000000) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::make_topic_guid: ")
                ACE_TEXT("Exceeded Maximum number of topic entity keys!")
                ACE_TEXT("Next key will be a duplicate!\n")));
     topic_counter_ = 0;
   }
 
   return make_id(participant_id_, entity_id);
 }

◆ match()

void OpenDDS::DCPS::StaticEndpointManager::match	(	const GUID_t &	writer,
		const GUID_t &	reader
	)

private

Definition at line 1028 of file StaticDiscovery.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DCPS_debug_level, LM_DEBUG, and match_continue().

Referenced by match_endpoints().

 {
   if (DCPS_debug_level >= 4) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) StaticEndpointManager::match: w: %C r: %C\n",
       LogGuid(writer).c_str(), LogGuid(reader).c_str()));
   }
 
   match_continue(writer, reader);
 }

◆ match_continue()

void OpenDDS::DCPS::StaticEndpointManager::match_continue	(	const GUID_t &	writer,
		const GUID_t &	reader
	)

private

Definition at line 1060 of file StaticDiscovery.cpp.

Referenced by match().

 {
   if (DCPS_debug_level >= 4) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) StaticEndpointManager::match_continue: w: %C r: %C\n",
       LogGuid(writer).c_str(), LogGuid(reader).c_str()));
   }
 
   // 0. For discovered endpoints, we'll have the QoS info in the form of the
   // publication or subscription BIT data which doesn't use the same structures
   // for QoS.  In those cases we can copy the individual QoS policies to temp
   // QoS structs:
   DDS::DataWriterQos tempDwQos;
   DDS::PublisherQos tempPubQos;
   DDS::DataReaderQos tempDrQos;
   DDS::SubscriberQos tempSubQos;
   ContentFilterProperty_t tempCfp;
 
   DiscoveredPublicationIter dpi = discovered_publications_.find(writer);
   DiscoveredSubscriptionIter dsi = discovered_subscriptions_.find(reader);
   if (dpi != discovered_publications_.end() && dsi != discovered_subscriptions_.end()) {
     // This is a discovered/discovered match, nothing for us to do
     return;
   }
 
   // 1. Collect details about the writer, which may be local or discovered
   const DDS::DataWriterQos* dwQos = 0;
   const DDS::PublisherQos* pubQos = 0;
   TransportLocatorSeq* wTls = 0;
   ACE_CDR::ULong wTransportContext = 0;
   XTypes::TypeInformation* writer_type_info = 0;
   OPENDDS_STRING topic_name;
   MonotonicTime_t writer_participant_discovered_at;
 
   const LocalPublicationIter lpi = local_publications_.find(writer);
   bool writer_local = false, already_matched = false;
   if (lpi != local_publications_.end()) {
     writer_local = true;
     dwQos = &lpi->second.qos_;
     pubQos = &lpi->second.publisher_qos_;
     wTls = &lpi->second.trans_info_;
     wTransportContext = lpi->second.transport_context_;
     already_matched = lpi->second.matched_endpoints_.count(reader);
     writer_type_info = &lpi->second.type_info_;
     topic_name = topic_names_[lpi->second.topic_id_];
     writer_participant_discovered_at = lpi->second.participant_discovered_at_;
   } else if (dpi != discovered_publications_.end()) {
     wTls = &dpi->second.writer_data_.writerProxy.allLocators;
     wTransportContext = dpi->second.transport_context_;
     writer_type_info = &dpi->second.type_info_;
     topic_name = dpi->second.get_topic_name();
     writer_participant_discovered_at = dpi->second.participant_discovered_at_;
 
     const DDS::PublicationBuiltinTopicData& bit =
       dpi->second.writer_data_.ddsPublicationData;
     tempDwQos.durability = bit.durability;
     tempDwQos.durability_service = bit.durability_service;
     tempDwQos.deadline = bit.deadline;
     tempDwQos.latency_budget = bit.latency_budget;
     tempDwQos.liveliness = bit.liveliness;
     tempDwQos.reliability = bit.reliability;
     tempDwQos.destination_order = bit.destination_order;
     tempDwQos.history = TheServiceParticipant->initial_HistoryQosPolicy();
     tempDwQos.resource_limits =
       TheServiceParticipant->initial_ResourceLimitsQosPolicy();
     tempDwQos.transport_priority =
       TheServiceParticipant->initial_TransportPriorityQosPolicy();
     tempDwQos.lifespan = bit.lifespan;
     tempDwQos.user_data = bit.user_data;
     tempDwQos.ownership = bit.ownership;
     tempDwQos.ownership_strength = bit.ownership_strength;
     tempDwQos.writer_data_lifecycle =
       TheServiceParticipant->initial_WriterDataLifecycleQosPolicy();
     tempDwQos.representation = bit.representation;
     dwQos = &tempDwQos;
 
     tempPubQos.presentation = bit.presentation;
     tempPubQos.partition = bit.partition;
     tempPubQos.group_data = bit.group_data;
     tempPubQos.entity_factory =
       TheServiceParticipant->initial_EntityFactoryQosPolicy();
     pubQos = &tempPubQos;
 
     populate_transport_locator_sequence(wTls, dpi, writer);
   } else {
     return; // Possible and ok, since lock is released
   }
 
   // 2. Collect details about the reader, which may be local or discovered
   const DDS::DataReaderQos* drQos = 0;
   const DDS::SubscriberQos* subQos = 0;
   TransportLocatorSeq* rTls = 0;
   ACE_CDR::ULong rTransportContext = 0;
   const ContentFilterProperty_t* cfProp = 0;
   XTypes::TypeInformation* reader_type_info = 0;
   MonotonicTime_t reader_participant_discovered_at;
 
   const LocalSubscriptionIter lsi = local_subscriptions_.find(reader);
   bool reader_local = false;
   if (lsi != local_subscriptions_.end()) {
     reader_local = true;
     drQos = &lsi->second.qos_;
     subQos = &lsi->second.subscriber_qos_;
     rTls = &lsi->second.trans_info_;
     rTransportContext = lsi->second.transport_context_;
     reader_type_info = &lsi->second.type_info_;
     if (lsi->second.filterProperties.filterExpression[0] != 0) {
       tempCfp.filterExpression = lsi->second.filterProperties.filterExpression;
       tempCfp.expressionParameters = lsi->second.filterProperties.expressionParameters;
     }
     cfProp = &tempCfp;
     if (!already_matched) {
       already_matched = lsi->second.matched_endpoints_.count(writer);
     }
     reader_participant_discovered_at = lsi->second.participant_discovered_at_;
   } else if (dsi != discovered_subscriptions_.end()) {
     rTls = &dsi->second.reader_data_.readerProxy.allLocators;
 
     populate_transport_locator_sequence(rTls, dsi, reader);
     rTransportContext = dsi->second.transport_context_;
 
     const DDS::SubscriptionBuiltinTopicData& bit =
       dsi->second.reader_data_.ddsSubscriptionData;
     tempDrQos.durability = bit.durability;
     tempDrQos.deadline = bit.deadline;
     tempDrQos.latency_budget = bit.latency_budget;
     tempDrQos.liveliness = bit.liveliness;
     tempDrQos.reliability = bit.reliability;
     tempDrQos.destination_order = bit.destination_order;
     tempDrQos.history = TheServiceParticipant->initial_HistoryQosPolicy();
     tempDrQos.resource_limits =
       TheServiceParticipant->initial_ResourceLimitsQosPolicy();
     tempDrQos.user_data = bit.user_data;
     tempDrQos.ownership = bit.ownership;
     tempDrQos.time_based_filter = bit.time_based_filter;
     tempDrQos.reader_data_lifecycle =
       TheServiceParticipant->initial_ReaderDataLifecycleQosPolicy();
     tempDrQos.representation = bit.representation;
     tempDrQos.type_consistency = bit.type_consistency;
     drQos = &tempDrQos;
 
     tempSubQos.presentation = bit.presentation;
     tempSubQos.partition = bit.partition;
     tempSubQos.group_data = bit.group_data;
     tempSubQos.entity_factory =
       TheServiceParticipant->initial_EntityFactoryQosPolicy();
     subQos = &tempSubQos;
 
     cfProp = &dsi->second.reader_data_.contentFilterProperty;
     reader_type_info = &dsi->second.type_info_;
     reader_participant_discovered_at = dsi->second.participant_discovered_at_;
   } else {
     return; // Possible and ok, since lock is released
   }
 
   // 3. Perform type consistency check (XTypes 1.3, Section 7.6.3.4.2)
   bool consistent = false;
 
   TopicDetailsMap::iterator td_iter = topics_.find(topic_name);
   if (td_iter == topics_.end()) {
     ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) StaticEndpointManager::match_continue - ERROR ")
               ACE_TEXT("Didn't find topic for consistency check\n")));
     return;
   } else {
     const XTypes::TypeIdentifier& writer_type_id = writer_type_info->minimal.typeid_with_size.type_id;
     const XTypes::TypeIdentifier& reader_type_id = reader_type_info->minimal.typeid_with_size.type_id;
     if (writer_type_id.kind() != XTypes::TK_NONE && reader_type_id.kind() != XTypes::TK_NONE) {
       if (!writer_local || !reader_local) {
         Encoding::Kind encoding_kind;
         if (tempDwQos.representation.value.length() > 0 &&
             repr_to_encoding_kind(tempDwQos.representation.value[0], encoding_kind) &&
             encoding_kind == Encoding::KIND_XCDR1) {
           const XTypes::TypeFlag extensibility_mask = XTypes::IS_APPENDABLE;
           if (type_lookup_service_->extensibility(extensibility_mask, writer_type_id)) {
             if (DCPS_debug_level) {
               ACE_DEBUG((LM_WARNING, ACE_TEXT("(%P|%t) WARNING: ")
                 ACE_TEXT("StaticEndpointManager::match_continue: ")
                 ACE_TEXT("Encountered unsupported combination of XCDR1 encoding and appendable extensibility\n")));
             }
           }
         }
       }
 
       XTypes::TypeConsistencyAttributes type_consistency;
       type_consistency.ignore_sequence_bounds = drQos->type_consistency.ignore_sequence_bounds;
       type_consistency.ignore_string_bounds = drQos->type_consistency.ignore_string_bounds;
       type_consistency.ignore_member_names = drQos->type_consistency.ignore_member_names;
       type_consistency.prevent_type_widening = drQos->type_consistency.prevent_type_widening;
       XTypes::TypeAssignability ta(type_lookup_service_, type_consistency);
 
       if (drQos->type_consistency.kind == DDS::ALLOW_TYPE_COERCION) {
         consistent = ta.assignable(reader_type_id, writer_type_id);
       } else {
         // The two types must be equivalent for DISALLOW_TYPE_COERCION
         consistent = reader_type_id == writer_type_id;
       }
     } else {
       if (drQos->type_consistency.force_type_validation) {
         // Cannot do type validation since not both TypeObjects are available
         consistent = false;
       } else {
         // Fall back to matching type names
         OPENDDS_STRING writer_type_name;
         OPENDDS_STRING reader_type_name;
         if (writer_local) {
           writer_type_name = td_iter->second.local_data_type_name();
         } else {
           writer_type_name = dpi->second.get_type_name();
         }
         if (reader_local) {
           reader_type_name = td_iter->second.local_data_type_name();
         } else {
           reader_type_name = dsi->second.get_type_name();
         }
         consistent = writer_type_name == reader_type_name;
       }
     }
 
     if (!consistent) {
       td_iter->second.increment_inconsistent();
       if (DCPS::DCPS_debug_level) {
         ACE_DEBUG((LM_WARNING,
                   ACE_TEXT("(%P|%t) StaticEndpointManager::match_continue - WARNING ")
                   ACE_TEXT("Data types of topic %C does not match (inconsistent)\n"),
                   topic_name.c_str()));
       }
       return;
     }
   }
 
   // Need to release lock, below, for callbacks into DCPS which could
   // call into Spdp/Sedp.  Note that this doesn't unlock, it just constructs
   // an ACE object which will be used below for unlocking.
   ACE_Reverse_Lock<ACE_Thread_Mutex> rev_lock(lock_);
 
   // 4. Check transport and QoS compatibility
 
   // Copy entries from local publication and local subscription maps
   // prior to releasing lock
   DataWriterCallbacks_wrch dwr;
   DataReaderCallbacks_wrch drr;
   if (writer_local) {
     dwr = lpi->second.publication_;
     OPENDDS_ASSERT(lpi->second.publication_);
     OPENDDS_ASSERT(dwr);
   }
   if (reader_local) {
     drr = lsi->second.subscription_;
     OPENDDS_ASSERT(lsi->second.subscription_);
     OPENDDS_ASSERT(drr);
   }
 
   IncompatibleQosStatus writerStatus = {0, 0, 0, DDS::QosPolicyCountSeq()};
   IncompatibleQosStatus readerStatus = {0, 0, 0, DDS::QosPolicyCountSeq()};
 
   if (compatibleQOS(&writerStatus, &readerStatus, *wTls, *rTls,
       dwQos, drQos, pubQos, subQos)) {
 
     bool call_writer = false, call_reader = false;
 
     if (writer_local) {
       call_writer = lpi->second.matched_endpoints_.insert(reader).second;
       dwr = lpi->second.publication_;
       if (!reader_local) {
         dsi->second.matched_endpoints_.insert(writer);
       }
     }
     if (reader_local) {
       call_reader = lsi->second.matched_endpoints_.insert(writer).second;
       drr = lsi->second.subscription_;
       if (!writer_local) {
         dpi->second.matched_endpoints_.insert(reader);
       }
     }
 
     if (writer_local && !reader_local) {
       add_assoc_i(writer, lpi->second, reader, dsi->second);
     }
     if (reader_local && !writer_local) {
       add_assoc_i(reader, lsi->second, writer, dpi->second);
     }
 
     if (!call_writer && !call_reader) {
       return; // nothing more to do
     }
 
     // Copy reader and writer association data prior to releasing lock
     DDS::OctetSeq octet_seq_type_info_reader;
     XTypes::serialize_type_info(*reader_type_info, octet_seq_type_info_reader);
     const ReaderAssociation ra = {
       *rTls, TransportLocator(), rTransportContext, reader, *subQos, *drQos,
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
       cfProp->filterClassName, cfProp->filterExpression,
 #else
       "", "",
 #endif
       cfProp->expressionParameters,
       octet_seq_type_info_reader,
       reader_participant_discovered_at
     };
 
     DDS::OctetSeq octet_seq_type_info_writer;
     XTypes::serialize_type_info(*writer_type_info, octet_seq_type_info_writer);
     const WriterAssociation wa = {
       *wTls, TransportLocator(), wTransportContext, writer, *pubQos, *dwQos,
       octet_seq_type_info_writer,
       writer_participant_discovered_at
     };
 
     ACE_GUARD(ACE_Reverse_Lock<ACE_Thread_Mutex>, rg, rev_lock);
     static const bool writer_active = true;
 
     if (call_writer) {
       if (DCPS_debug_level > 3) {
         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) StaticEndpointManager::match_continue - ")
           ACE_TEXT("adding writer %C association for reader %C\n"), LogGuid(writer).c_str(),
           LogGuid(reader).c_str()));
       }
       DataWriterCallbacks_rch dwr_lock = dwr.lock();
       if (dwr_lock) {
         if (call_reader) {
           DataReaderCallbacks_rch drr_lock = drr.lock();
           if (drr_lock) {
             DcpsUpcalls thr(drr_lock, reader, wa, !writer_active, dwr_lock);
             thr.activate();
             dwr_lock->add_association(writer, ra, writer_active);
             thr.writer_done();
           }
         } else {
           dwr_lock->add_association(writer, ra, writer_active);
         }
       }
     } else if (call_reader) {
       if (DCPS_debug_level > 3) {
         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) StaticEndpointManager::match_continue - ")
           ACE_TEXT("adding reader %C association for writer %C\n"),
           LogGuid(reader).c_str(), LogGuid(writer).c_str()));
       }
       DataReaderCallbacks_rch drr_lock = drr.lock();
       if (drr_lock) {
         drr_lock->add_association(reader, wa, !writer_active);
       }
     }
 
   } else if (already_matched) { // break an existing associtaion
     if (writer_local) {
       lpi->second.matched_endpoints_.erase(reader);
       lpi->second.remote_expectant_opendds_associations_.erase(reader);
       if (dsi != discovered_subscriptions_.end()) {
         dsi->second.matched_endpoints_.erase(writer);
       }
     }
     if (reader_local) {
       lsi->second.matched_endpoints_.erase(writer);
       lsi->second.remote_expectant_opendds_associations_.erase(writer);
       if (dpi != discovered_publications_.end()) {
         dpi->second.matched_endpoints_.erase(reader);
       }
     }
     if (writer_local && !reader_local) {
       remove_assoc_i(writer, lpi->second, reader);
     }
     if (reader_local && !writer_local) {
       remove_assoc_i(reader, lsi->second, writer);
     }
     ACE_GUARD(ACE_Reverse_Lock<ACE_Thread_Mutex>, rg, rev_lock);
     if (writer_local) {
       ReaderIdSeq reader_seq(1);
       reader_seq.length(1);
       reader_seq[0] = reader;
       DataWriterCallbacks_rch dwr_lock = dwr.lock();
       if (dwr_lock) {
         dwr_lock->remove_associations(reader_seq, false /*notify_lost*/);
       }
     }
     if (reader_local) {
       WriterIdSeq writer_seq(1);
       writer_seq.length(1);
       writer_seq[0] = writer;
       DataReaderCallbacks_rch drr_lock = drr.lock();
       if (drr_lock) {
         drr_lock->remove_associations(writer_seq, false /*notify_lost*/);
       }
     }
   } else { // something was incompatible
     ACE_GUARD(ACE_Reverse_Lock< ACE_Thread_Mutex>, rg, rev_lock);
     if (writer_local && writerStatus.count_since_last_send) {
       if (DCPS_debug_level > 3) {
         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) StaticEndpointManager::match - ")
                    ACE_TEXT("writer incompatible\n")));
       }
       DataWriterCallbacks_rch dwr_lock = dwr.lock();
       if (dwr_lock) {
         dwr_lock->update_incompatible_qos(writerStatus);
       }
     }
     if (reader_local && readerStatus.count_since_last_send) {
       if (DCPS_debug_level > 3) {
         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) StaticEndpointManager::match - ")
                    ACE_TEXT("reader incompatible\n")));
       }
       DataReaderCallbacks_rch drr_lock = drr.lock();
       if (drr_lock) {
         drr_lock->update_incompatible_qos(readerStatus);
       }
     }
   }
 }

◆ match_endpoints()

void OpenDDS::DCPS::StaticEndpointManager::match_endpoints	(	GUID_t	repoId,
		const TopicDetails &	td,
		bool	remove = `false`
	)

Definition at line 926 of file StaticDiscovery.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TopicDetails::discovered_publications(), OpenDDS::DCPS::TopicDetails::discovered_subscriptions(), OpenDDS::DCPS::equal_guid_prefixes(), OpenDDS::DCPS::GuidConverter::isReader(), LM_DEBUG, OpenDDS::DCPS::TopicDetails::local_publications(), OpenDDS::DCPS::TopicDetails::local_subscriptions(), match(), participant_id_, and remove_assoc().

Referenced by add_publication(), add_subscription(), ignore(), remove_publication(), and remove_subscription().

 {
   if (DCPS_debug_level >= 4) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) StaticEndpointManager::match_endpoints %C%C\n",
       remove ? "remove " : "", LogGuid(repoId).c_str()));
   }
 
   const bool reader = GuidConverter(repoId).isReader();
   // Copy the endpoint set - lock can be released in match()
   RepoIdSet local_endpoints;
   RepoIdSet discovered_endpoints;
   if (reader) {
     local_endpoints = td.local_publications();
     discovered_endpoints = td.discovered_publications();
   } else {
     local_endpoints = td.local_subscriptions();
     discovered_endpoints = td.discovered_subscriptions();
   }
 
   const bool is_remote = !equal_guid_prefixes(repoId, participant_id_);
   if (is_remote && local_endpoints.empty()) {
     // Nothing to match.
     return;
   }
 
   for (RepoIdSet::const_iterator iter = local_endpoints.begin();
        iter != local_endpoints.end(); ++iter) {
     // check to make sure it's a Reader/Writer or Writer/Reader match
     if (GuidConverter(*iter).isReader() != reader) {
       if (remove) {
         remove_assoc(*iter, repoId);
       } else {
         match(reader ? *iter : repoId, reader ? repoId : *iter);
       }
     }
   }
 
   // Remote/remote matches are a waste of time
   if (is_remote) {
     return;
   }
 
   for (RepoIdSet::const_iterator iter = discovered_endpoints.begin();
        iter != discovered_endpoints.end(); ++iter) {
     // check to make sure it's a Reader/Writer or Writer/Reader match
     if (GuidConverter(*iter).isReader() != reader) {
       if (remove) {
         remove_assoc(*iter, repoId);
       } else {
         match(reader ? *iter : repoId, reader ? repoId : *iter);
       }
     }
   }
 }

◆ need_minimal_and_or_complete_types()

void OpenDDS::DCPS::StaticEndpointManager::need_minimal_and_or_complete_types	(	const XTypes::TypeInformation *	type_info,
		bool &	need_minimal,
		bool &	need_complete
	)		const

private

◆ OPENDDS_MAP()

typedef OpenDDS::DCPS::StaticEndpointManager::OPENDDS_MAP	(	SequenceNumber	,
		TypeIdOrigSeqNumber
	)

private

◆ OPENDDS_MAP_CMP() [1/5]

typedef OpenDDS::DCPS::StaticEndpointManager::OPENDDS_MAP_CMP	(	GUID_t	,
		DiscoveredSubscription	,
		GUID_tKeyLessThan
	)

protected

Referenced by has_dcps_key(), and ignore().

◆ OPENDDS_MAP_CMP() [2/5]

typedef OpenDDS::DCPS::StaticEndpointManager::OPENDDS_MAP_CMP	(	GUID_t	,
		DiscoveredPublication	,
		GUID_tKeyLessThan
	)

protected

◆ OPENDDS_MAP_CMP() [3/5]

typedef OpenDDS::DCPS::StaticEndpointManager::OPENDDS_MAP_CMP	(	GUID_t	,
		LocalPublication	,
		GUID_tKeyLessThan
	)

protected

◆ OPENDDS_MAP_CMP() [4/5]

typedef OpenDDS::DCPS::StaticEndpointManager::OPENDDS_MAP_CMP	(	GUID_t	,
		LocalSubscription	,
		GUID_tKeyLessThan
	)

protected

◆ OPENDDS_MAP_CMP() [5/5]

typedef OpenDDS::DCPS::StaticEndpointManager::OPENDDS_MAP_CMP	(	GUID_t	,
		String	,
		GUID_tKeyLessThan
	)

protected

◆ OPENDDS_SET()

OpenDDS::DCPS::StaticEndpointManager::OPENDDS_SET ( OPENDDS_STRING )

private

◆ OPENDDS_SET_CMP()

OpenDDS::DCPS::StaticEndpointManager::OPENDDS_SET_CMP	(	GUID_t	,
		GUID_tKeyLessThan
	)

private

◆ populate_transport_locator_sequence() [1/2]

void OpenDDS::DCPS::StaticEndpointManager::populate_transport_locator_sequence	(	TransportLocatorSeq *&	,
		DiscoveredSubscriptionIter &	,
		const GUID_t &
	)

virtual

Definition at line 469 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_TEXT(), and LM_NOTICE.

Referenced by match_continue().

 {
   ACE_DEBUG((LM_NOTICE, ACE_TEXT("(%P|%t) StaticEndpointManager::populate_transport_locator_sequence TODO\n")));
   // TODO
 }

◆ populate_transport_locator_sequence() [2/2]

void OpenDDS::DCPS::StaticEndpointManager::populate_transport_locator_sequence	(	TransportLocatorSeq *&	,
		DiscoveredPublicationIter &	,
		const GUID_t &
	)

virtual

Definition at line 478 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_TEXT(), and LM_NOTICE.

 {
   ACE_DEBUG((LM_NOTICE, ACE_TEXT("(%P|%t) StaticEndpointManager::populate_transport_locator_sequence TODO\n")));
   // TODO
 }

◆ pub_bit()

OpenDDS::DCPS::PublicationBuiltinTopicDataDataReaderImpl * OpenDDS::DCPS::StaticEndpointManager::pub_bit ( )

Definition at line 566 of file StaticDiscovery.cpp.

References OpenDDS::DCPS::StaticParticipant::bit_subscriber(), OpenDDS::DCPS::BUILT_IN_PUBLICATION_TOPIC, and participant_.

Referenced by init_bit().

 {
   DDS::Subscriber_var sub = participant_.bit_subscriber();
   if (!sub.in())
     return 0;
 
   DDS::DataReader_var d = sub->lookup_datareader(BUILT_IN_PUBLICATION_TOPIC);
   return dynamic_cast<OpenDDS::DCPS::PublicationBuiltinTopicDataDataReaderImpl*>(d.in());
 }

◆ purge_dead_topic()

void OpenDDS::DCPS::StaticEndpointManager::purge_dead_topic ( const String & topic_name )

Definition at line 611 of file StaticDiscovery.cpp.

References topic_names_, and topics_.

Referenced by ignore(), and remove_topic().

 {
   TopicDetailsMap::iterator top_it = topics_.find(topic_name);
   topic_names_.erase(top_it->second.topic_id());
   topics_.erase(top_it);
 }

◆ reader_does_not_exist()

void OpenDDS::DCPS::StaticEndpointManager::reader_does_not_exist	(	const GUID_t &	readerid,
		const GUID_t &	writerid
	)

virtual

Implements OpenDDS::DCPS::DiscoveryListener.

Definition at line 505 of file StaticDiscovery.cpp.

References ACE_GUARD, local_publications_, lock_, OpenDDS::DCPS::EndpointRegistry::reader_map, and registry_.

 {
   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
   LocalPublicationMap::const_iterator lp_pos = local_publications_.find(writerid);
   EndpointRegistry::ReaderMapType::const_iterator reader_pos = registry_.reader_map.find(readerid);
   if (lp_pos != local_publications_.end() &&
       reader_pos != registry_.reader_map.end()) {
     DataWriterCallbacks_rch dwr = lp_pos->second.publication_.lock();
     if (dwr) {
       ReaderIdSeq ids;
       ids.length(1);
       ids[0] = readerid;
       dwr->remove_associations(ids, true);
     }
   }
 }

◆ reader_exists()

void OpenDDS::DCPS::StaticEndpointManager::reader_exists	(	const GUID_t &	readerid,
		const GUID_t &	writerid
	)

virtual

Implements OpenDDS::DCPS::DiscoveryListener.

Definition at line 487 of file StaticDiscovery.cpp.

References ACE_GUARD, local_publications_, lock_, OpenDDS::DCPS::EndpointRegistry::reader_map, and registry_.

 {
   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
   LocalPublicationMap::const_iterator lp_pos = local_publications_.find(writerid);
   EndpointRegistry::ReaderMapType::const_iterator reader_pos = registry_.reader_map.find(readerid);
   if (lp_pos != local_publications_.end() &&
       reader_pos != registry_.reader_map.end()) {
     DataWriterCallbacks_rch dwr = lp_pos->second.publication_.lock();
     if (dwr) {
       const ReaderAssociation ra =
         {reader_pos->second.trans_info, TransportLocator(), 0, readerid, reader_pos->second.subscriber_qos, reader_pos->second.qos,
          "", "", DDS::StringSeq(), DDS::OctetSeq(), {0, 0}};
       dwr->add_association(writerid, ra, true);
     }
   }
 }

◆ remove_assoc()

void OpenDDS::DCPS::StaticEndpointManager::remove_assoc	(	const GUID_t &	remove_from,
		const GUID_t &	removing
	)

Definition at line 982 of file StaticDiscovery.cpp.

References discovered_publications_, discovered_subscriptions_, local_publications_, local_subscriptions_, remove_assoc_i(), and write_subscription_data().

Referenced by match_endpoints().

 {
   if (GuidConverter(remove_from).isReader()) {
     const LocalSubscriptionIter lsi = local_subscriptions_.find(remove_from);
     if (lsi != local_subscriptions_.end()) {
       lsi->second.matched_endpoints_.erase(removing);
       const DiscoveredPublicationIter dpi = discovered_publications_.find(removing);
       if (dpi != discovered_publications_.end()) {
         dpi->second.matched_endpoints_.erase(remove_from);
       }
       WriterIdSeq writer_seq(1);
       writer_seq.length(1);
       writer_seq[0] = removing;
       const size_t count = lsi->second.remote_expectant_opendds_associations_.erase(removing);
       DataReaderCallbacks_rch drr = lsi->second.subscription_.lock();
       if (drr) {
         drr->remove_associations(writer_seq, false /*notify_lost*/);
       }
       remove_assoc_i(remove_from, lsi->second, removing);
       // Update writer
       if (count) {
         write_subscription_data(remove_from, lsi->second);
       }
     }
 
   } else {
     const LocalPublicationIter lpi = local_publications_.find(remove_from);
     if (lpi != local_publications_.end()) {
       lpi->second.matched_endpoints_.erase(removing);
       const DiscoveredSubscriptionIter dsi = discovered_subscriptions_.find(removing);
       if (dsi != discovered_subscriptions_.end()) {
         dsi->second.matched_endpoints_.erase(remove_from);
       }
       ReaderIdSeq reader_seq(1);
       reader_seq.length(1);
       reader_seq[0] = removing;
       lpi->second.remote_expectant_opendds_associations_.erase(removing);
       DataWriterCallbacks_rch dwr = lpi->second.publication_.lock();
       if (dwr) {
         dwr->remove_associations(reader_seq, false /*notify_lost*/);
       }
       remove_assoc_i(remove_from, lpi->second, removing);
     }
   }
 }

◆ remove_assoc_i() [1/2]

virtual void OpenDDS::DCPS::StaticEndpointManager::remove_assoc_i	(	const GUID_t &	,
		const LocalPublication &	,
		const GUID_t &
	)

inlinevirtual

Definition at line 412 of file StaticDiscovery.h.

Referenced by match_continue(), and remove_assoc().

413 {}

◆ remove_assoc_i() [2/2]

virtual void OpenDDS::DCPS::StaticEndpointManager::remove_assoc_i	(	const GUID_t &	,
		const LocalSubscription &	,
		const GUID_t &
	)

inlinevirtual

Definition at line 416 of file StaticDiscovery.h.

417 {}

◆ remove_expired_endpoints()

void OpenDDS::DCPS::StaticEndpointManager::remove_expired_endpoints ( const MonotonicTimePoint & )

private

Definition at line 1038 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_GUARD, OpenDDS::DCPS::LogGuid::c_str(), cleanup_type_lookup_data(), OpenDDS::DCPS::DCPS_debug_level, LM_DEBUG, lock_, max_type_lookup_service_reply_period_, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), and orig_seq_numbers_.

Referenced by type_lookup_init().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
   const MonotonicTimePoint now = MonotonicTimePoint::now();
 
   // Clean up internal data used by getTypeDependencies
   for (OrigSeqNumberMap::iterator it = orig_seq_numbers_.begin(); it != orig_seq_numbers_.end();) {
     if (now - it->second.time_started >= max_type_lookup_service_reply_period_) {
       if (DCPS_debug_level >= 4) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) StaticEndpointManager::remove_expired_endpoints: "
           "clean up type lookup data for %C\n",
           LogGuid(it->second.participant).c_str()));
       }
       cleanup_type_lookup_data(it->second.participant, it->second.type_id, it->second.secure);
       orig_seq_numbers_.erase(it++);
     } else {
       ++it;
     }
   }
 }

◆ remove_from_bit() [1/2]

void OpenDDS::DCPS::StaticEndpointManager::remove_from_bit ( const DiscoveredPublication & pub )

inlineprivate

Definition at line 427 of file StaticDiscovery.h.

Referenced by ignore().

   {
     remove_from_bit_i(pub);
   }

◆ remove_from_bit() [2/2]

void OpenDDS::DCPS::StaticEndpointManager::remove_from_bit ( const DiscoveredSubscription & sub )

inlineprivate

Definition at line 432 of file StaticDiscovery.h.

   {
     remove_from_bit_i(sub);
   }

◆ remove_from_bit_i() [1/2]

virtual void OpenDDS::DCPS::StaticEndpointManager::remove_from_bit_i ( const DiscoveredPublication & )

inlineprotectedvirtual

Definition at line 265 of file StaticDiscovery.h.

265 { }

◆ remove_from_bit_i() [2/2]

virtual void OpenDDS::DCPS::StaticEndpointManager::remove_from_bit_i ( const DiscoveredSubscription & )

inlineprotectedvirtual

Definition at line 266 of file StaticDiscovery.h.

266 { }

◆ remove_publication()

void OpenDDS::DCPS::StaticEndpointManager::remove_publication ( const GUID_t & publicationId )

Definition at line 801 of file StaticDiscovery.cpp.

References ACE_ERROR, ACE_GUARD, ACE_TEXT(), LM_ERROR, local_publications_, lock_, match_endpoints(), OPENDDS_STRING, remove_publication_i(), DDS::RETCODE_OK, topic_names_, and topics_.

 {
   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
   LocalPublicationIter iter = local_publications_.find(publicationId);
   if (iter != local_publications_.end()) {
     if (DDS::RETCODE_OK == remove_publication_i(publicationId, iter->second)) {
       OPENDDS_STRING topic_name = topic_names_[iter->second.topic_id_];
       local_publications_.erase(publicationId);
       TopicDetailsMap::iterator top_it = topics_.find(topic_name);
       if (top_it != topics_.end()) {
         match_endpoints(publicationId, top_it->second, true /*remove*/);
         top_it->second.remove_local_publication(publicationId);
         // Local, no need to check for dead topic.
       }
     } else {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::remove_publication - ")
                  ACE_TEXT("Failed to publish dispose msg\n")));
     }
   }
 }

◆ remove_publication_i()

DDS::ReturnCode_t OpenDDS::DCPS::StaticEndpointManager::remove_publication_i	(	const GUID_t &	writerid,
		LocalPublication &	pub
	)

virtual

Definition at line 354 of file StaticDiscovery.cpp.

References OpenDDS::DCPS::WeakRcHandle< T >::lock(), participant_id_, OpenDDS::DCPS::StaticEndpointManager::LocalPublication::publication_, registry_, OpenDDS::DCPS::EndpointRegistry::Writer::reliable_readers, DDS::RETCODE_ERROR, DDS::RETCODE_OK, and OpenDDS::DCPS::EndpointRegistry::writer_map.

Referenced by remove_publication().

 {
   EndpointRegistry::WriterMapType::const_iterator pos = registry_.writer_map.find(writerid);
   if (pos == registry_.writer_map.end()) {
     return DDS::RETCODE_ERROR;
   }
 
   const EndpointRegistry::Writer& writer = pos->second;
 
   ReaderIdSeq ids;
   ids.length((CORBA::ULong)writer.reliable_readers.size());
   CORBA::ULong idx = 0;
   for (RepoIdSet::const_iterator pos = writer.reliable_readers.begin(), limit = writer.reliable_readers.end();
         pos != limit;
         ++pos, ++idx) {
     const GUID_t& readerid = *pos;
     ids[idx] = readerid;
     DataWriterCallbacks_rch pl = pub.publication_.lock();
     if (pl) {
       pl->unregister_for_reader(participant_id_, writerid, readerid);
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ remove_subscription()

void OpenDDS::DCPS::StaticEndpointManager::remove_subscription ( const GUID_t & subscriptionId )

Definition at line 886 of file StaticDiscovery.cpp.

References ACE_ERROR, ACE_GUARD, ACE_TEXT(), LM_ERROR, local_subscriptions_, lock_, match_endpoints(), OPENDDS_STRING, remove_subscription_i(), DDS::RETCODE_OK, topic_names_, and topics_.

 {
   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
   LocalSubscriptionIter iter = local_subscriptions_.find(subscriptionId);
   if (iter != local_subscriptions_.end()) {
     if (DDS::RETCODE_OK == remove_subscription_i(subscriptionId, iter->second)) {
       OPENDDS_STRING topic_name = topic_names_[iter->second.topic_id_];
       local_subscriptions_.erase(subscriptionId);
       TopicDetailsMap::iterator top_it = topics_.find(topic_name);
       if (top_it != topics_.end()) {
         match_endpoints(subscriptionId, top_it->second, true /*remove*/);
         top_it->second.remove_local_subscription(subscriptionId);
         // Local, no need to check for dead topic.
       }
     } else {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::remove_subscription - ")
                  ACE_TEXT("Failed to publish dispose msg\n")));
     }
   }
 }

◆ remove_subscription_i()

DDS::ReturnCode_t OpenDDS::DCPS::StaticEndpointManager::remove_subscription_i	(	const GUID_t &	readerid,
		LocalSubscription &	sub
	)

virtual

Definition at line 425 of file StaticDiscovery.cpp.

References OpenDDS::DCPS::WeakRcHandle< T >::lock(), participant_id_, OpenDDS::DCPS::EndpointRegistry::reader_map, registry_, OpenDDS::DCPS::EndpointRegistry::Reader::reliable_writers, DDS::RETCODE_ERROR, DDS::RETCODE_OK, and OpenDDS::DCPS::StaticEndpointManager::LocalSubscription::subscription_.

Referenced by remove_subscription().

 {
   EndpointRegistry::ReaderMapType::const_iterator pos = registry_.reader_map.find(readerid);
   if (pos == registry_.reader_map.end()) {
     return DDS::RETCODE_ERROR;
   }
 
   const EndpointRegistry::Reader& reader = pos->second;
 
   WriterIdSeq ids;
   ids.length((CORBA::ULong)reader.reliable_writers.size());
   CORBA::ULong idx = 0;
   for (RepoIdSet::const_iterator pos = reader.reliable_writers.begin(), limit = reader.reliable_writers.end();
         pos != limit;
         ++pos, ++idx) {
     const GUID_t& writerid = *pos;
     ids[idx] = writerid;
     DataReaderCallbacks_rch sl = sub.subscription_.lock();
     if (sl) {
       sl->unregister_for_writer(participant_id_, readerid, writerid);
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ remove_topic()

TopicStatus OpenDDS::DCPS::StaticEndpointManager::remove_topic ( const GUID_t & topicId )

Definition at line 743 of file StaticDiscovery.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::INTERNAL_ERROR, OpenDDS::DCPS::TopicDetails::is_dead(), lock_, name, OpenDDS::DCPS::NOT_FOUND, purge_dead_topic(), OpenDDS::DCPS::REMOVED, topic_names_, topics_, and OpenDDS::DCPS::TopicDetails::unset_local().

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, lock_, INTERNAL_ERROR);
   TopicNameMap::iterator name_iter = topic_names_.find(topicId);
   if (name_iter == topic_names_.end()) {
     return NOT_FOUND;
   }
   const String& name = name_iter->second;
   TopicDetails& td = topics_[name];
   td.unset_local();
   if (td.is_dead()) {
     purge_dead_topic(name);
   }
 
   return REMOVED;
 }

◆ send_type_lookup_request()

virtual bool OpenDDS::DCPS::StaticEndpointManager::send_type_lookup_request	(	const XTypes::TypeIdentifierSeq &	,
		const GUID_t &	,
		bool	,
		bool
	)

inlineprotectedvirtual

Definition at line 284 of file StaticDiscovery.h.

   {
     return true;
   }

◆ shutting_down()

bool OpenDDS::DCPS::StaticEndpointManager::shutting_down ( void ) const

virtual

Definition at line 461 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_TEXT(), and LM_NOTICE.

Referenced by ignore().

 {
   ACE_DEBUG((LM_NOTICE, ACE_TEXT("(%P|%t) StaticEndpointManager::shutting_down TODO\n")));
   // TODO
   return false;
 }

◆ sub_bit()

OpenDDS::DCPS::SubscriptionBuiltinTopicDataDataReaderImpl * OpenDDS::DCPS::StaticEndpointManager::sub_bit ( )

Definition at line 577 of file StaticDiscovery.cpp.

References OpenDDS::DCPS::StaticParticipant::bit_subscriber(), OpenDDS::DCPS::BUILT_IN_SUBSCRIPTION_TOPIC, and participant_.

Referenced by init_bit().

 {
   DDS::Subscriber_var sub = participant_.bit_subscriber();
   if (!sub.in())
     return 0;
 
   DDS::DataReader_var d = sub->lookup_datareader(BUILT_IN_SUBSCRIPTION_TOPIC);
   return dynamic_cast<OpenDDS::DCPS::SubscriptionBuiltinTopicDataDataReaderImpl*>(d.in());
 }

◆ type_lookup_fini()

void OpenDDS::DCPS::StaticEndpointManager::type_lookup_fini ( )

Definition at line 597 of file StaticDiscovery.cpp.

References type_lookup_reply_deadline_processor_.

Referenced by ~StaticEndpointManager().

 {
   if (type_lookup_reply_deadline_processor_) {
     type_lookup_reply_deadline_processor_->cancel();
     type_lookup_reply_deadline_processor_.reset();
   }
 }

◆ type_lookup_init()

void OpenDDS::DCPS::StaticEndpointManager::type_lookup_init ( ReactorInterceptor_rch reactor_interceptor )

Definition at line 588 of file StaticDiscovery.cpp.

References OpenDDS::DCPS::rchandle_from(), remove_expired_endpoints(), TheServiceParticipant, and type_lookup_reply_deadline_processor_.

Referenced by StaticEndpointManager().

 {
   if (!type_lookup_reply_deadline_processor_) {
     type_lookup_reply_deadline_processor_ =
       DCPS::make_rch<StaticEndpointManagerSporadic>(TheServiceParticipant->time_source(), reactor_interceptor,
                                                     rchandle_from(this), &StaticEndpointManager::remove_expired_endpoints);
   }
 }

◆ type_lookup_service()

void OpenDDS::DCPS::StaticEndpointManager::type_lookup_service ( const XTypes::TypeLookupService_rch type_lookup_service )

Definition at line 605 of file StaticDiscovery.cpp.

References type_lookup_service_.

 {
   type_lookup_service_ = type_lookup_service;
 }

◆ update_publication_locators()

void OpenDDS::DCPS::StaticEndpointManager::update_publication_locators	(	const GUID_t &	publicationId,
		const TransportLocatorSeq &	transInfo
	)

Definition at line 823 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, LM_INFO, local_publications_, lock_, and write_publication_data().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
   LocalPublicationIter iter = local_publications_.find(publicationId);
   if (iter != local_publications_.end()) {
     if (DCPS_debug_level > 3) {
       ACE_DEBUG((LM_INFO,
         ACE_TEXT("(%P|%t) StaticEndpointManager::update_publication_locators - updating locators for %C\n"),
         LogGuid(publicationId).c_str()));
     }
     iter->second.trans_info_ = transInfo;
     write_publication_data(publicationId, iter->second);
   }
 }

◆ update_publication_qos()

bool OpenDDS::DCPS::StaticEndpointManager::update_publication_qos	(	const GUID_t &	,
		const DDS::DataWriterQos &	,
		const DDS::PublisherQos &
	)

virtual

Definition at line 274 of file StaticDiscovery.cpp.

References ACE_ERROR, ACE_TEXT(), and LM_ERROR.

 {
   ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::update_publication_qos - ")
              ACE_TEXT("Not allowed\n")));
   return false;
 }

◆ update_subscription_locators()

void OpenDDS::DCPS::StaticEndpointManager::update_subscription_locators	(	const GUID_t &	subscriptionId,
		const TransportLocatorSeq &	transInfo
	)

Definition at line 908 of file StaticDiscovery.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, LM_INFO, local_subscriptions_, lock_, and write_subscription_data().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
   LocalSubscriptionIter iter = local_subscriptions_.find(subscriptionId);
   if (iter != local_subscriptions_.end()) {
     if (DCPS_debug_level > 3) {
       ACE_DEBUG((LM_INFO,
         ACE_TEXT("(%P|%t) StaticEndpointManager::update_subscription_locators updating locators for %C\n"),
         LogGuid(subscriptionId).c_str()));
     }
     iter->second.trans_info_ = transInfo;
     write_subscription_data(subscriptionId, iter->second);
   }
 }

◆ update_subscription_params()

bool OpenDDS::DCPS::StaticEndpointManager::update_subscription_params	(	const GUID_t &	,
		const DDS::StringSeq &
	)

virtual

Definition at line 294 of file StaticDiscovery.cpp.

References ACE_ERROR, ACE_TEXT(), and LM_ERROR.

 {
   ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::update_subscription_qos - ")
              ACE_TEXT("Not allowed\n")));
   return false;
 }

◆ update_subscription_qos()

bool OpenDDS::DCPS::StaticEndpointManager::update_subscription_qos	(	const GUID_t &	,
		const DDS::DataReaderQos &	,
		const DDS::SubscriberQos &
	)

virtual

Definition at line 284 of file StaticDiscovery.cpp.

References ACE_ERROR, ACE_TEXT(), and LM_ERROR.

 {
   ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::update_subscription_qos - ")
              ACE_TEXT("Not allowed\n")));
   return false;
 }

◆ update_topic_qos()

bool OpenDDS::DCPS::StaticEndpointManager::update_topic_qos	(	const GUID_t &	,
		const DDS::TopicQos &
	)

virtual

Definition at line 265 of file StaticDiscovery.cpp.

References ACE_ERROR, ACE_TEXT(), and LM_ERROR.

 {
   ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: StaticEndpointManager::update_topic_qos - ")
              ACE_TEXT("Not allowed\n")));
   return false;
 }

◆ write_publication_data()

virtual DDS::ReturnCode_t OpenDDS::DCPS::StaticEndpointManager::write_publication_data	(	const GUID_t &	,
		LocalPublication &	,
		const GUID_t &	reader = `GUID_UNKNOWN`
	)

inlineprotectedvirtual

Definition at line 268 of file StaticDiscovery.h.

References DDS::RETCODE_OK.

Referenced by add_publication(), and update_publication_locators().

   {
     ACE_UNUSED_ARG(reader);
     return DDS::RETCODE_OK;
   }

◆ write_subscription_data()

virtual DDS::ReturnCode_t OpenDDS::DCPS::StaticEndpointManager::write_subscription_data	(	const GUID_t &	,
		LocalSubscription &	,
		const GUID_t &	reader = `GUID_UNKNOWN`
	)

inlineprotectedvirtual

Definition at line 276 of file StaticDiscovery.h.

References DDS::RETCODE_OK.

Referenced by add_subscription(), remove_assoc(), and update_subscription_locators().

   {
     ACE_UNUSED_ARG(reader);
     return DDS::RETCODE_OK;
   }

◆ writer_does_not_exist()

void OpenDDS::DCPS::StaticEndpointManager::writer_does_not_exist	(	const GUID_t &	writerid,
		const GUID_t &	readerid
	)

virtual

Implements OpenDDS::DCPS::DiscoveryListener.

Definition at line 540 of file StaticDiscovery.cpp.

References ACE_GUARD, local_subscriptions_, lock_, registry_, and OpenDDS::DCPS::EndpointRegistry::writer_map.

 {
   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
   LocalSubscriptionMap::const_iterator ls_pos = local_subscriptions_.find(readerid);
   EndpointRegistry::WriterMapType::const_iterator writer_pos = registry_.writer_map.find(writerid);
   if (ls_pos != local_subscriptions_.end() &&
       writer_pos != registry_.writer_map.end()) {
     DataReaderCallbacks_rch drr = ls_pos->second.subscription_.lock();
     if (drr) {
       WriterIdSeq ids;
       ids.length(1);
       ids[0] = writerid;
       drr->remove_associations(ids, true);
     }
   }
 }

◆ writer_exists()

void OpenDDS::DCPS::StaticEndpointManager::writer_exists	(	const GUID_t &	writerid,
		const GUID_t &	readerid
	)

virtual

Implements OpenDDS::DCPS::DiscoveryListener.

Definition at line 523 of file StaticDiscovery.cpp.

References ACE_GUARD, local_subscriptions_, lock_, registry_, and OpenDDS::DCPS::EndpointRegistry::writer_map.

 {
   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
   LocalSubscriptionMap::const_iterator ls_pos = local_subscriptions_.find(readerid);
   EndpointRegistry::WriterMapType::const_iterator writer_pos = registry_.writer_map.find(writerid);
   if (ls_pos != local_subscriptions_.end() &&
       writer_pos != registry_.writer_map.end()) {
     DataReaderCallbacks_rch drr = ls_pos->second.subscription_.lock();
     if (drr) {
       const WriterAssociation wa =
         {writer_pos->second.trans_info, TransportLocator(), 0, writerid, writer_pos->second.publisher_qos, writer_pos->second.qos, DDS::OctetSeq(), {0,0}};
       drr->add_association(readerid, wa, false);
     }
   }
 }

Member Data Documentation

◆ discovered_publications_

DiscoveredPublicationMap OpenDDS::DCPS::StaticEndpointManager::discovered_publications_

private

Definition at line 447 of file StaticDiscovery.h.

Referenced by ignore(), match_continue(), and remove_assoc().

◆ discovered_subscriptions_

DiscoveredSubscriptionMap OpenDDS::DCPS::StaticEndpointManager::discovered_subscriptions_

private

Definition at line 448 of file StaticDiscovery.h.

Referenced by ignore(), match_continue(), and remove_assoc().

◆ ignored_guids_

RepoIdSet OpenDDS::DCPS::StaticEndpointManager::ignored_guids_

private

Definition at line 443 of file StaticDiscovery.h.

Referenced by ignore(), and ignoring().

◆ local_publications_

LocalPublicationMap OpenDDS::DCPS::StaticEndpointManager::local_publications_

private

Definition at line 445 of file StaticDiscovery.h.

Referenced by add_publication(), match_continue(), reader_does_not_exist(), reader_exists(), remove_assoc(), remove_publication(), and update_publication_locators().

◆ local_subscriptions_

LocalSubscriptionMap OpenDDS::DCPS::StaticEndpointManager::local_subscriptions_

private

Definition at line 446 of file StaticDiscovery.h.

Referenced by add_subscription(), match_continue(), remove_assoc(), remove_subscription(), update_subscription_locators(), writer_does_not_exist(), and writer_exists().

◆ lock_

ACE_Thread_Mutex& OpenDDS::DCPS::StaticEndpointManager::lock_

private

Definition at line 441 of file StaticDiscovery.h.

Referenced by add_publication(), add_subscription(), assert_topic(), find_topic(), match_continue(), reader_does_not_exist(), reader_exists(), remove_expired_endpoints(), remove_publication(), remove_subscription(), remove_topic(), update_publication_locators(), update_subscription_locators(), writer_does_not_exist(), and writer_exists().

◆ max_type_lookup_service_reply_period_

TimeDuration OpenDDS::DCPS::StaticEndpointManager::max_type_lookup_service_reply_period_

private

Definition at line 461 of file StaticDiscovery.h.

Referenced by remove_expired_endpoints().

◆ orig_seq_numbers_

OrigSeqNumberMap OpenDDS::DCPS::StaticEndpointManager::orig_seq_numbers_

private

Definition at line 476 of file StaticDiscovery.h.

Referenced by remove_expired_endpoints().

◆ participant_

StaticParticipant& OpenDDS::DCPS::StaticEndpointManager::participant_

private

Definition at line 455 of file StaticDiscovery.h.

Referenced by pub_bit(), and sub_bit().

◆ participant_id_

GUID_t OpenDDS::DCPS::StaticEndpointManager::participant_id_

private

Definition at line 442 of file StaticDiscovery.h.

Referenced by add_publication(), add_publication_i(), add_subscription(), add_subscription_i(), init_bit(), make_topic_guid(), match_endpoints(), remove_publication_i(), and remove_subscription_i().

◆ registry_

const EndpointRegistry& OpenDDS::DCPS::StaticEndpointManager::registry_

private

Definition at line 453 of file StaticDiscovery.h.

Referenced by add_publication_i(), add_subscription_i(), assign_publication_key(), assign_subscription_key(), init_bit(), reader_does_not_exist(), reader_exists(), remove_publication_i(), remove_subscription_i(), writer_does_not_exist(), and writer_exists().

◆ topic_counter_

unsigned int OpenDDS::DCPS::StaticEndpointManager::topic_counter_

private

Definition at line 444 of file StaticDiscovery.h.

Referenced by make_topic_guid().

◆ topic_names_

TopicNameMap OpenDDS::DCPS::StaticEndpointManager::topic_names_

private

Definition at line 450 of file StaticDiscovery.h.

Referenced by add_publication(), add_subscription(), assert_topic(), has_dcps_key(), ignore(), match_continue(), purge_dead_topic(), remove_publication(), remove_subscription(), and remove_topic().

◆ topics_

TopicDetailsMap OpenDDS::DCPS::StaticEndpointManager::topics_

private

Definition at line 449 of file StaticDiscovery.h.

Referenced by add_publication(), add_subscription(), assert_topic(), find_topic(), has_dcps_key(), ignore(), match_continue(), purge_dead_topic(), remove_publication(), remove_subscription(), and remove_topic().

◆ type_lookup_reply_deadline_processor_

RcHandle<StaticEndpointManagerSporadic> OpenDDS::DCPS::StaticEndpointManager::type_lookup_reply_deadline_processor_

private

Definition at line 460 of file StaticDiscovery.h.

Referenced by type_lookup_fini(), and type_lookup_init().

◆ type_lookup_service_

XTypes::TypeLookupService_rch OpenDDS::DCPS::StaticEndpointManager::type_lookup_service_

private

Definition at line 458 of file StaticDiscovery.h.

Referenced by match_continue(), and type_lookup_service().

◆ type_lookup_service_sequence_number_

SequenceNumber OpenDDS::DCPS::StaticEndpointManager::type_lookup_service_sequence_number_

private

Definition at line 462 of file StaticDiscovery.h.

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

Classes

Public Member Functions

Protected Types

Protected Member Functions

Static Protected Member Functions

Private Types

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ DiscoveredPublicationIter

◆ DiscoveredSubscriptionIter

◆ LocalPublicationCIter

◆ LocalPublicationIter

◆ LocalSubscriptionCIter

◆ LocalSubscriptionIter

◆ StaticEndpointManagerSporadic

Constructor & Destructor Documentation

◆ StaticEndpointManager()

◆ ~StaticEndpointManager()

Member Function Documentation

◆ add_assoc_i() [1/2]

◆ add_assoc_i() [2/2]

◆ add_publication()

◆ add_publication_i()

◆ add_subscription()

◆ add_subscription_i()

◆ assert_topic()

◆ assign_publication_key()

◆ assign_subscription_key()

◆ cleanup_type_lookup_data()

◆ disassociate()

◆ find_topic()

◆ get_key() [1/2]

◆ get_key() [2/2]

◆ has_dcps_key()

◆ ignore()

◆ ignoring() [1/2]

◆ ignoring() [2/2]

◆ init_bit()

◆ is_expectant_opendds()

◆ make_topic_guid()

◆ match()

◆ match_continue()

◆ match_endpoints()

◆ need_minimal_and_or_complete_types()

◆ OPENDDS_MAP()

◆ OPENDDS_MAP_CMP() [1/5]

◆ OPENDDS_MAP_CMP() [2/5]

◆ OPENDDS_MAP_CMP() [3/5]

◆ OPENDDS_MAP_CMP() [4/5]

◆ OPENDDS_MAP_CMP() [5/5]

◆ OPENDDS_SET()

◆ OPENDDS_SET_CMP()

◆ populate_transport_locator_sequence() [1/2]

◆ populate_transport_locator_sequence() [2/2]

◆ pub_bit()

◆ purge_dead_topic()

◆ reader_does_not_exist()

◆ reader_exists()

◆ remove_assoc()

◆ remove_assoc_i() [1/2]

◆ remove_assoc_i() [2/2]

◆ remove_expired_endpoints()

◆ remove_from_bit() [1/2]

◆ remove_from_bit() [2/2]

◆ remove_from_bit_i() [1/2]

◆ remove_from_bit_i() [2/2]

◆ remove_publication()

◆ remove_publication_i()

◆ remove_subscription()

◆ remove_subscription_i()

◆ remove_topic()

◆ send_type_lookup_request()

◆ shutting_down()

◆ sub_bit()

◆ type_lookup_fini()

◆ type_lookup_init()

◆ type_lookup_service()