Implementation of the DCPSInfo. More...

#include <DCPSInfo_i.h>

Inheritance diagram for TAO_DDS_DCPSInfo_i:

Collaboration diagram for TAO_DDS_DCPSInfo_i:

[legend]

Classes
struct	BIT_Cleanup_Handler

Public Member Functions
	TAO_DDS_DCPSInfo_i (CORBA::ORB_ptr orb, bool reincarnate, ShutdownInterface *shutdown, const TAO_DDS_DCPSFederationId &federation)

virtual	~TAO_DDS_DCPSInfo_i ()

virtual int	handle_timeout (const ACE_Time_Value &now, const void *arg)

virtual CORBA::Boolean	attach_participant (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId)

virtual OpenDDS::DCPS::TopicStatus	assert_topic (OpenDDS::DCPS::GUID_t_out topicId, DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const char topicName, const char dataTypeName, const DDS::TopicQos &qos, bool hasDcpsKey)

bool	add_topic (const OpenDDS::DCPS::GUID_t &topicId, DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const char topicName, const char dataTypeName, const DDS::TopicQos &qos)
	Add a previously existing topic to the repository. More...

virtual OpenDDS::DCPS::TopicStatus	find_topic (DDS::DomainId_t domainId, const char *topicName, CORBA::String_out dataTypeName, DDS::TopicQos_out qos, OpenDDS::DCPS::GUID_t_out topicId)

virtual OpenDDS::DCPS::TopicStatus	remove_topic (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &topicId)

virtual OpenDDS::DCPS::GUID_t	add_publication (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &topicId, OpenDDS::DCPS::DataWriterRemote_ptr publication, const DDS::DataWriterQos &qos, const OpenDDS::DCPS::TransportLocatorSeq &transInfo, const DDS::PublisherQos &publisherQos, const DDS::OctetSeq &serializedTypeInfo)

bool	add_publication (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &topicId, const OpenDDS::DCPS::GUID_t &pubId, const char *pub_str, const DDS::DataWriterQos &qos, const OpenDDS::DCPS::TransportLocatorSeq &transInfo, ACE_CDR::ULong transportContext, const DDS::PublisherQos &publisherQos, const DDS::OctetSeq &serializedTypeInfo, bool associate=false)
	Add a previously existing publication to the repository. More...

virtual void	remove_publication (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &publicationId)

virtual OpenDDS::DCPS::GUID_t	add_subscription (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &topicId, OpenDDS::DCPS::DataReaderRemote_ptr subscription, const DDS::DataReaderQos &qos, const OpenDDS::DCPS::TransportLocatorSeq &transInfo, const DDS::SubscriberQos &subscriberQos, const char filterClassName, const char filterExpression, const DDS::StringSeq &exprParams, const DDS::OctetSeq &serializedTypeInfo)

bool	add_subscription (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &topicId, const OpenDDS::DCPS::GUID_t &subId, const char sub_str, const DDS::DataReaderQos &qos, const OpenDDS::DCPS::TransportLocatorSeq &transInfo, ACE_CDR::ULong transportContext, const DDS::SubscriberQos &subscriberQos, const char filterClassName, const char *filterExpression, const DDS::StringSeq &exprParams, const DDS::OctetSeq &serializedTypeInfo, bool associate=false)
	Add a previously existing subscription to the repository. More...

virtual void	remove_subscription (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &subscriptionId)

virtual OpenDDS::DCPS::AddDomainStatus	add_domain_participant (DDS::DomainId_t domain, const DDS::DomainParticipantQos &qos)

bool	add_domain_participant (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const DDS::DomainParticipantQos &qos)
	Add a previously existing participant to the repository. More...

virtual void	remove_domain_participant (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId)

bool	remove_by_owner (DDS::DomainId_t domain, long owner)

virtual void	disassociate_participant (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &local_id, const OpenDDS::DCPS::GUID_t &remote_id)

virtual void	disassociate_subscription (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &local_id, const OpenDDS::DCPS::GUID_t &remote_id)

virtual void	disassociate_publication (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &local_id, const OpenDDS::DCPS::GUID_t &remote_id)

virtual void	ignore_domain_participant (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &myParticipantId, const OpenDDS::DCPS::GUID_t &ignoreId)

virtual void	ignore_topic (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &myParticipantId, const OpenDDS::DCPS::GUID_t &ignoreId)

virtual void	ignore_subscription (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &myParticipantId, const OpenDDS::DCPS::GUID_t &ignoreId)

virtual void	ignore_publication (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &myParticipantId, const OpenDDS::DCPS::GUID_t &ignoreId)

virtual CORBA::Boolean	update_publication_qos (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &partId, const OpenDDS::DCPS::GUID_t &dwId, const DDS::DataWriterQos &qos, const DDS::PublisherQos &publisherQos)

void	update_publication_qos (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &partId, const OpenDDS::DCPS::GUID_t &dwId, const DDS::DataWriterQos &qos)
	Entry for federation updates of DataWriterQos values. More...

void	update_publication_qos (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &partId, const OpenDDS::DCPS::GUID_t &dwId, const DDS::PublisherQos &qos)
	Entry for federation updates of PublisherQos values. More...

virtual CORBA::Boolean	update_subscription_qos (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &partId, const OpenDDS::DCPS::GUID_t &drId, const DDS::DataReaderQos &qos, const DDS::SubscriberQos &subscriberQos)

void	update_subscription_qos (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &partId, const OpenDDS::DCPS::GUID_t &drId, const DDS::DataReaderQos &qos)
	Entry for federation updates of DataReaderQos values. More...

void	update_subscription_qos (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &partId, const OpenDDS::DCPS::GUID_t &drId, const DDS::SubscriberQos &qos)
	Entry for federation updates of SubscriberQos values. More...

virtual ::CORBA::Boolean	update_subscription_params (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const OpenDDS::DCPS::GUID_t &subscriptionId, const DDS::StringSeq &params)

virtual CORBA::Boolean	update_topic_qos (const OpenDDS::DCPS::GUID_t &topicId, DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, const DDS::TopicQos &qos)

virtual CORBA::Boolean	update_domain_participant_qos (DDS::DomainId_t domain, const OpenDDS::DCPS::GUID_t &participantId, const DDS::DomainParticipantQos &qos)

virtual void	shutdown ()
	Cause the entire repository to exit. More...

virtual char *	dump_to_string ()
	Dump the Repos state to string. More...

bool	changeOwnership (DDS::DomainId_t domainId, const OpenDDS::DCPS::GUID_t &participantId, long sender, long owner)
	assert new ownership for a participant and its contained entities. More...

int	init_transport (int listen_address_given, const char *listen_str)

bool	receive_image (const Update::UImage &image)

void	add (Update::Updater *updater)
	Add an additional Updater interface. More...

DCPS_IR_Domain *	domain (DDS::DomainId_t domain)
	Convert a domain Id into a reference to a DCPS_IR_Domain object. More...

const DCPS_IR_Domain_Map &	domains () const
	Expose a readable reference of the domain map. More...

CORBA::ORB_ptr	orb ()
	Expose the ORB. More...

bool	init_persistence ()

bool	init_reassociation (const ACE_Time_Value &delay)

bool	init_dispatchChecking (const ACE_Time_Value &delay)

void	finalize ()
	Cleanup state for shutdown. More...

void	cleanup_all_built_in_topics ()

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_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

virtual Reference_Count	add_reference (void)

virtual Reference_Count	remove_reference (void)

Reference_Counting_Policy &	reference_counting_policy (void)

Private Attributes
DCPS_IR_Domain_Map	domains_

CORBA::ORB_var	orb_

CORBA::ORB_var	dispatchingOrb_

const TAO_DDS_DCPSFederationId &	federation_

OpenDDS::DCPS::RepoIdGenerator	participantIdGenerator_

Update::Manager *	um_

bool	reincarnate_

ShutdownInterface *	shutdown_
	Interface to effect shutdown of the process. More...

ACE_Recursive_Thread_Mutex	lock_

long	reassociate_timer_id_

long	dispatch_check_timer_id_

bool	in_cleanup_all_built_in_topics_

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 Types inherited from ACE_Event_Handler
typedef ACE_Atomic_Op< ACE_SYNCH_MUTEX, Reference_Count >	Atomic_Reference_Count

Protected Member Functions inherited from ACE_Event_Handler
	ACE_Event_Handler (ACE_Reactor *=0, int priority=ACE_Event_Handler::LO_PRIORITY)

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Implementation of the DCPSInfo.

This is the Information Repository object. Clients of the system will use the CORBA reference of this object.

Definition at line 53 of file DCPSInfo_i.h.

Constructor & Destructor Documentation

◆ TAO_DDS_DCPSInfo_i()

TAO_DDS_DCPSInfo_i::TAO_DDS_DCPSInfo_i	(	CORBA::ORB_ptr	orb,
		bool	reincarnate,
		ShutdownInterface *	shutdown,
		const TAO_DDS_DCPSFederationId &	federation
	)

Definition at line 42 of file DCPSInfo_i.cpp.

References _duplicate(), dispatchingOrb_, CORBA::ORB_init(), and TheServiceParticipant.

   : orb_(CORBA::ORB::_duplicate(orb))
   , federation_(federation)
   , participantIdGenerator_(federation.id())
   , um_(0)
   , reincarnate_(reincarnate)
   , shutdown_(shutdown)
   , reassociate_timer_id_(-1)
   , dispatch_check_timer_id_(-1)
 #ifndef DDS_HAS_MINIMUM_BIT
   , in_cleanup_all_built_in_topics_(false)
 #endif
 {
   if (!TheServiceParticipant->use_bidir_giop()) {
     int argc = 0;
     char** no_argv = 0;
     dispatchingOrb_ = CORBA::ORB_init(argc, no_argv, "dispatchingOnly");
   }
 }

◆ ~TAO_DDS_DCPSInfo_i()

TAO_DDS_DCPSInfo_i::~TAO_DDS_DCPSInfo_i ( )

virtual

Definition at line 65 of file DCPSInfo_i.cpp.

66 {

67 }

Member Function Documentation

◆ add()

void TAO_DDS_DCPSInfo_i::add ( Update::Updater * updater )

Add an additional Updater interface.

Definition at line 2403 of file DCPSInfo_i.cpp.

References Update::Manager::add(), and um_.

 {
   if (this->um_) {
     this->um_->add(updater);
   }
 }

◆ add_domain_participant() [1/2]

OpenDDS::DCPS::AddDomainStatus TAO_DDS_DCPSInfo_i::add_domain_participant	(	DDS::DomainId_t	domain,
		const DDS::DomainParticipantQos &	qos
	)

virtual

Definition at line 1010 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domain(), OpenDDS::DCPS::AddDomainStatus::federated, federation_, OpenDDS::DCPS::RcHandle< T >::get(), DCPS_IR_Domain::get_next_participant_id(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::AddDomainStatus::id, LM_DEBUG, lock_, TAO_DDS_DCPSFederationId::overridden(), OpenDDS::DCPS::RepoIdConverter::participantId(), DCPS_IR_Domain::participants(), TheServiceParticipant, um_, and value.

Referenced by Update::Manager::add(), OpenDDS::Federator::ManagerImpl::processCreate(), and receive_image().

 {
   // A value to return.
   OpenDDS::DCPS::AddDomainStatus value;
   value.id        = OpenDDS::DCPS::GUID_UNKNOWN;
   value.federated = this->federation_.overridden();
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, value);
 
   // Grab the domain.
   DCPS_IR_Domain* domainPtr = this->domain(domain);
 
   if (0 == domainPtr) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Obtain a shiny new GUID value.
   OpenDDS::DCPS::GUID_t participantId = domainPtr->get_next_participant_id();
 
   // Determine if this is the 'special' repository internal participant
   // that publishes the built-in topics for a domain.
   bool isBitPart = domainPtr->participants().empty() && TheServiceParticipant->get_BIT();
 
   DCPS_IR_Participant_rch participant =
     OpenDDS::DCPS::make_rch<DCPS_IR_Participant>(
                    this->federation_,
                    participantId,
                    domainPtr,
                    qos, um_, isBitPart);
 
   // We created the participant, now we can return the Id value (eventually).
   value.id = participantId;
 
   if (isBitPart) {
     participant->isBitPublisher() = true;
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(participantId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) (GUID_t)TAO_DDS_DCPSInfo_i::add_domain_participant: ")
                  ACE_TEXT("participant %C in domain %d is BIT publisher for this domain.\n"),
                  std::string(converter).c_str(),
                  domain));
     }
   }
 
   // Assume responsibility for writing back to the participant.
   participant->takeOwnership();
 
   int status = domainPtr->add_participant(participant);
 
   if (0 != status) {
     // Adding the participant failed return the invalid
     // participant Id number.
     participantId = OpenDDS::DCPS::GUID_UNKNOWN;
 
   } else if (this->um_) {
     OpenDDS::DCPS::RepoIdConverter converter(participantId);
     if (participant->isBitPublisher() == false) {
       // Push this participant to interested observers.
       Update::UParticipant updateParticipant(
         domain,
         participant->owner(),
         participantId,
         const_cast<DDS::DomainParticipantQos &>(qos));
       this->um_->create(updateParticipant);
 
       if (OpenDDS::DCPS::DCPS_debug_level > 4) {
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) (GUID_t)TAO_DDS_DCPSInfo_i::add_domain_participant: ")
                    ACE_TEXT("pushing creation of participant %C in domain %d.\n"),
                    std::string(converter).c_str(),
                    domain));
       }
     }
 
     // Update what the last participant id was
     um_->updateLastPartId(converter.participantId());
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::RepoIdConverter converter(participantId);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) (GUID_t)TAO_DDS_DCPSInfo_i::add_domain_participant: ")
                ACE_TEXT("domain %d loaded participant %C at 0x%x.\n"),
                domain,
                std::string(converter).c_str(),
                participant.get()));
   }
   return value;
 }

◆ add_domain_participant() [2/2]

bool TAO_DDS_DCPSInfo_i::add_domain_participant	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const DDS::DomainParticipantQos &	qos
	)

Add a previously existing participant to the repository.

Parameters

domainId	the Domain in which the Participant is contained.
participantId	the GUID Id value to use for the Participant.
qos	the QoS value of the Participant.

Adds a Participant to the repository using a specified Participant GUID Id value. If the ParticipantId indicates that this Participant was created by within this repository (the federation Id is the current repositories federation Id), this method will ensure that any subsequent calls to add a Publication and obtain a newly generated Id value will return an Id value greater than the Id value of the current one.

Definition at line 1105 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domain(), federation_, OpenDDS::DCPS::RepoIdConverter::federationId(), OpenDDS::DCPS::RcHandle< T >::in(), LM_DEBUG, LM_ERROR, LM_WARNING, lock_, DCPS_IR_Domain::participant(), OpenDDS::DCPS::RepoIdConverter::participantId(), DCPS_IR_Domain::participants(), TheServiceParticipant, and um_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, false);
 
   // Grab the domain.
   DCPS_IR_Domain* domainPtr = this->domain(domainId);
 
   if (0 == domainPtr) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: (bool)TAO_DDS_DCPSInfo_i::add_domain_participant: ")
                  ACE_TEXT("invalid domain Id: %d\n"),
                  domainId));
     }
 
     return false;
   }
 
   // Prepare to manipulate the participant's Id value.
   OpenDDS::DCPS::RepoIdConverter converter(participantId);
 
   // Determine if this is the 'special' repository internal participant
   // that publishes the built-in topics for a domain.
   bool isBitPart = domainPtr->participants().empty() && TheServiceParticipant->get_BIT();
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr = domainPtr->participant(participantId);
 
   if (0 != partPtr) {
     if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) (bool)TAO_DDS_DCPSInfo_i::add_domain_participant: ")
                  ACE_TEXT("participant id %C already exists.\n"),
                  std::string(converter).c_str()));
     }
 
     return false;
   }
 
   DCPS_IR_Participant_rch participant =
     OpenDDS::DCPS::make_rch<DCPS_IR_Participant>(this->federation_,
                                      participantId,
                                      domainPtr,
                                      qos, um_, isBitPart);
 
   switch (domainPtr->add_participant(participant)) {
   case -1: {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: (bool)TAO_DDS_DCPSInfo_i::add_domain_participant: ")
                ACE_TEXT("failed to load participant %C in domain %d.\n"),
                std::string(converter).c_str(),
                domainId));
   }
   return false;
 
   case 1:
 
     if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: (bool)TAO_DDS_DCPSInfo_i::add_domain_participant: ")
                  ACE_TEXT("attempt to load duplicate participant %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
 
     return false;
 
   case 0:
   default:
     break;
   }
 
   // See if we are adding a participant that was created within this
   // repository or a different repository.
   if (converter.federationId() == this->federation_.id()) {
     // Ensure the participant GUID values do not conflict.
     domainPtr->last_participant_key(converter.participantId());
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) (bool)TAO_DDS_DCPSInfo_i::add_domain_participant: ")
                  ACE_TEXT("Adjusting highest participant Id value to at least %d.\n"),
                  converter.participantId()));
     }
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) (bool)TAO_DDS_DCPSInfo_i::add_domain_participant: ")
                ACE_TEXT("loaded participant %C at 0x%x in domain %d.\n"),
                std::string(converter).c_str(),
                participant.in(),
                domainId));
   }
 
   return true;
 }

◆ add_publication() [1/2]

OpenDDS::DCPS::GUID_t TAO_DDS_DCPSInfo_i::add_publication	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	topicId,
		OpenDDS::DCPS::DataWriterRemote_ptr	publication,
		const DDS::DataWriterQos &	qos,
		const OpenDDS::DCPS::TransportLocatorSeq &	transInfo,
		const DDS::PublisherQos &	publisherQos,
		const DDS::OctetSeq &	serializedTypeInfo
	)

virtual

Definition at line 374 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), DCPS_IR_Participant::add_publication(), DCPS_IR_Topic::add_publication_reference(), Update::Manager::create(), Update::DataWriter, OpenDDS::DCPS::DCPS_debug_level, dispatchingOrb_, domains_, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), DCPS_IR_Participant::get_next_publication_id(), OpenDDS::DCPS::GUID_UNKNOWN, TAO::String_var< charT >::in(), CORBA::is_nil(), DCPS_IR_Participant::isBitPublisher(), LM_DEBUG, LM_WARNING, lock_, OpenDDS::DCPS::move(), orb_, DCPS_IR_Participant::remove_publication(), OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL::transportContextDefault, and um_.

Referenced by Update::Manager::add(), OpenDDS::Federator::ManagerImpl::processCreate(), OpenDDS::Federator::ManagerImpl::processDeferred(), and receive_image().

 {
   if (CORBA::is_nil(publication)) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_WARNING,
         ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_publication: ")
         ACE_TEXT("invalid publication reference.\n")));
     }
     return OpenDDS::DCPS::GUID_UNKNOWN;
   }
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, OpenDDS::DCPS::GUID_UNKNOWN);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(participantId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   DCPS_IR_Topic* topic = where->second->find_topic(topicId);
 
   if (topic == 0) {
     throw OpenDDS::DCPS::Invalid_Topic();
   }
 
   // Get a Id for the Writer, make it a builtin kind if this is for a BIT
   OpenDDS::DCPS::GUID_t pubId = partPtr->get_next_publication_id(
     OpenDDS::DCPS::RepoIdConverter(topicId).isBuiltinDomainEntity());
 
   OpenDDS::DCPS::DataWriterRemote_var dispatchingPublication =
     OpenDDS::DCPS::DataWriterRemote::_duplicate(publication);
 
   if (dispatchingOrb_) {
     // Remarshall the remote reference onto the dispatching orb.
     CORBA::String_var pubStr = orb_->object_to_string(dispatchingPublication);
     CORBA::Object_var pubObj = dispatchingOrb_->string_to_object(pubStr);
     if (CORBA::is_nil(pubObj))  {
       if (OpenDDS::DCPS::DCPS_debug_level > 4) {
         ACE_DEBUG((LM_WARNING,
                    ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_publication: ")
                    ACE_TEXT("failure marshalling publication on dispatching orb.\n")));
       }
       return OpenDDS::DCPS::GUID_UNKNOWN;
     }
 
     dispatchingPublication = OpenDDS::DCPS::DataWriterRemote::_unchecked_narrow(pubObj);
   }
 
   OpenDDS::DCPS::unique_ptr<DCPS_IR_Publication> pubPtr(
     new DCPS_IR_Publication(
                    pubId,
                    partPtr,
                    topic,
                    dispatchingPublication.in(),
                    qos,
                    transInfo,
                    transportContextDefault,
                    publisherQos,
                    serializedTypeInfo));
 
   DCPS_IR_Publication* pub = pubPtr.get();
   if (partPtr->add_publication(OpenDDS::DCPS::move(pubPtr)) != 0) {
     // failed to add.  we are responsible for the memory.
     pubId = OpenDDS::DCPS::GUID_UNKNOWN;
   } else if (topic->add_publication_reference(pub) != 0) {
     // Failed to add to the topic
     // so remove from participant and fail.
     partPtr->remove_publication(pubId);
     pubId = OpenDDS::DCPS::GUID_UNKNOWN;
   }
 
   if (this->um_ && (partPtr->isBitPublisher() == false)) {
     CORBA::String_var callback = orb_->object_to_string(publication);
     Update::ContentSubscriptionInfo csi;
 
     Update::UWActor actor(domainId, pubId, topicId, participantId, Update::DataWriter
                           , callback.in()
                           , const_cast<DDS::PublisherQos&>(publisherQos)
                           , const_cast<DDS::DataWriterQos&>(qos)
                           , const_cast<OpenDDS::DCPS::TransportLocatorSeq&>(transInfo)
                           , transportContextDefault, csi
                           , const_cast<DDS::OctetSeq&>(serializedTypeInfo));
     this->um_->create(actor);
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(pubId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) (GUID_t)TAO_DDS_DCPSInfo_i::add_publication: ")
                  ACE_TEXT("pushing creation of publication %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 
   where->second->remove_dead_participants();
   return pubId;
 }

◆ add_publication() [2/2]

bool TAO_DDS_DCPSInfo_i::add_publication	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	topicId,
		const OpenDDS::DCPS::GUID_t &	pubId,
		const char *	pub_str,
		const DDS::DataWriterQos &	qos,
		const OpenDDS::DCPS::TransportLocatorSeq &	transInfo,
		ACE_CDR::ULong	transportContext,
		const DDS::PublisherQos &	publisherQos,
		const DDS::OctetSeq &	serializedTypeInfo,
		bool	associate = `false`
	)

Add a previously existing publication to the repository.

Parameters

domainId	the Domain in which the Publication is contained.
participantId	the Participant in which the Publication is contained.
topicId	the Topic of the Publication.
pubId	the GUID Id value to use for the Publication.
pub_str	stringified publication callback to DataWriter.
qos	the QoS value of the DataWriter.
transInfo	the transport information for the Publication.
publisherQos	the QoS value of the Publisher.
associate	indicate whether to create new associations.

Adds a Publication to the repository using a specified Publication GUID Id value. If the PublicationId indicates that this Publication was created by within this repository (the federation Id is the current repositories federation Id), this method will ensure that any subsequent calls to add a Publication and obtain a newly generated Id value will return an Id value greater than the Id value of the current one.

: Check if this is already stored. If so, just clear the callback IOR.

Definition at line 490 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), DCPS_IR_Participant::add_publication(), DCPS_IR_Topic::add_publication_reference(), OpenDDS::DCPS::DCPS_debug_level, dispatchingOrb_, domains_, OpenDDS::DCPS::GuidConverter::entityKey(), federation_, OpenDDS::DCPS::RepoIdConverter::federationId(), TAO_DDS_DCPSFederationId::id(), TAO_Pseudo_Var_T< Object >::in(), CORBA::is_nil(), DCPS_IR_Participant::last_publication_key(), LM_ERROR, LM_WARNING, lock_, OpenDDS::DCPS::move(), orb_, and DCPS_IR_Participant::remove_publication().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, false);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_publication: ")
                  ACE_TEXT("invalid domain %d.\n"),
                  domainId));
     }
 
     return false;
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(participantId);
 
   if (0 == partPtr) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(pubId);
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_publication: ")
                  ACE_TEXT("invalid participant %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
 
     return false;
   }
 
   DCPS_IR_Topic* topic = where->second->find_topic(topicId);
 
   if (topic == 0) {
     OpenDDS::DCPS::RepoIdConverter converter(topicId);
     ACE_DEBUG((LM_WARNING,
                ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_publication: ")
                ACE_TEXT("invalid topic %C in domain %d.\n"),
                std::string(converter).c_str(),
                domainId));
     return false;
   }
 
   /// @TODO: Check if this is already stored.  If so, just clear the callback IOR.
 
   CORBA::Object_var obj = (dispatchingOrb_ ? dispatchingOrb_ : orb_)->string_to_object(pub_str);
   if (CORBA::is_nil(obj.in())) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_publication: ")
                  ACE_TEXT("failure converting string %C to objref\n"),
                  pub_str));
     }
     return false;
   }
 
   OpenDDS::DCPS::DataWriterRemote_var publication = OpenDDS::DCPS::DataWriterRemote::_unchecked_narrow(obj.in());
 
   OpenDDS::DCPS::unique_ptr<DCPS_IR_Publication> pubPtr(
     new DCPS_IR_Publication(
                    pubId,
                    partPtr,
                    topic,
                    publication.in(),
                    qos,
                    transInfo,
                    transportContext,
                    publisherQos,
                    serializedTypeInfo));
 
   DCPS_IR_Publication* pub = pubPtr.get();
   switch (partPtr->add_publication(move(pubPtr))) {
   case -1: {
     OpenDDS::DCPS::RepoIdConverter converter(pubId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::add_publication: ")
                ACE_TEXT("failed to add publication to participant %C.\n"),
                std::string(converter).c_str()));
     return false;
   }
 
   case 1:
     return false;
   case 0:
   default:
     break;
   }
 
   switch (topic->add_publication_reference(pub, associate)) {
   case -1: {
     OpenDDS::DCPS::RepoIdConverter converter(pubId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::add_publication: ")
                ACE_TEXT("failed to add publication to participant %C topic list.\n"),
                std::string(converter).c_str()));
 
     // Remove the publication.
     partPtr->remove_publication(pubId);
 
   }
   return false;
 
   case 1: // This is actually a really really bad place to jump to.
     // This means that we successfully added the new publication
     // to the participant (it had not been inserted before) but
     // that we are adding a duplicate publication to the topic
     // list - which should never ever be able to happen.
     return false;
 
   case 0:
   default:
     break;
   }
 
   OpenDDS::DCPS::RepoIdConverter converter(pubId);
 
   // See if we are adding a publication that was created within this
   // repository or a different repository.
   if (converter.federationId() == federation_.id()) {
     // Ensure the publication GUID_t values do not conflict.
     partPtr->last_publication_key(converter.entityKey());
   }
 
   return true;
 }

◆ add_subscription() [1/2]

OpenDDS::DCPS::GUID_t TAO_DDS_DCPSInfo_i::add_subscription	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	topicId,
		OpenDDS::DCPS::DataReaderRemote_ptr	subscription,
		const DDS::DataReaderQos &	qos,
		const OpenDDS::DCPS::TransportLocatorSeq &	transInfo,
		const DDS::SubscriberQos &	subscriberQos,
		const char *	filterClassName,
		const char *	filterExpression,
		const DDS::StringSeq &	exprParams,
		const DDS::OctetSeq &	serializedTypeInfo
	)

virtual

Definition at line 681 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), DCPS_IR_Participant::add_subscription(), DCPS_IR_Topic::add_subscription_reference(), Update::Manager::create(), Update::DataReader, OpenDDS::DCPS::DCPS_debug_level, dispatchingOrb_, domains_, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), DCPS_IR_Participant::get_next_subscription_id(), OpenDDS::DCPS::GUID_UNKNOWN, TAO::String_var< charT >::in(), TAO_Pseudo_Var_T< Object >::in(), CORBA::is_nil(), DCPS_IR_Participant::isBitPublisher(), LM_DEBUG, LM_ERROR, LM_WARNING, lock_, OpenDDS::DCPS::move(), orb_, DCPS_IR_Domain::participant(), DCPS_IR_Domain::remove_dead_participants(), DCPS_IR_Participant::remove_subscription(), OpenDDS::DCPS::unique_ptr< T, Deleter >::reset(), OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL::transportContextDefault, and um_.

Referenced by Update::Manager::add(), OpenDDS::Federator::ManagerImpl::processCreate(), OpenDDS::Federator::ManagerImpl::processDeferred(), and receive_image().

 {
   if (CORBA::is_nil(subscription)) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_WARNING,
         ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_subscription: ")
         ACE_TEXT("invalid subscription reference.\n")));
     }
     return OpenDDS::DCPS::GUID_UNKNOWN;
   }
 
   DCPS_IR_Domain* domainPtr;
   DCPS_IR_Participant* partPtr;
   DCPS_IR_Topic* topic;
   OpenDDS::DCPS::GUID_t subId;
   OpenDDS::DCPS::unique_ptr<DCPS_IR_Subscription> subPtr;
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, OpenDDS::DCPS::GUID_UNKNOWN);
 
     // Grab the domain.
     DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
     if (where == this->domains_.end()) {
       throw OpenDDS::DCPS::Invalid_Domain();
     }
 
     // Grab the domain and participant.
     domainPtr = where->second.get();
     partPtr = domainPtr->participant(participantId);
 
     if (0 == partPtr) {
       throw OpenDDS::DCPS::Invalid_Participant();
     }
 
     topic = where->second->find_topic(topicId);
 
     if (topic == 0) {
       throw OpenDDS::DCPS::Invalid_Topic();
     }
 
     // Get a Id for the Reader, make it a builtin kind if this is for a BIT
     subId = partPtr->get_next_subscription_id(
       OpenDDS::DCPS::RepoIdConverter(topicId).isBuiltinDomainEntity());
 
     OpenDDS::DCPS::DataReaderRemote_var dispatchingSubscription (
       OpenDDS::DCPS::DataReaderRemote::_duplicate(subscription));
 
     if (dispatchingOrb_) {
       // Remarshall the remote reference onto the dispatching orb.
       CORBA::String_var subStr = orb_->object_to_string(dispatchingSubscription);
       CORBA::Object_var subObj = dispatchingOrb_->string_to_object(subStr);
       if (CORBA::is_nil(subObj.in())) {
         if (OpenDDS::DCPS::DCPS_debug_level > 4) {
           ACE_DEBUG((LM_WARNING,
                      ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_subscription: ")
                      ACE_TEXT("failure marshalling subscription on dispatching orb.\n")));
         }
         return OpenDDS::DCPS::GUID_UNKNOWN;
       }
       dispatchingSubscription = OpenDDS::DCPS::DataReaderRemote::_unchecked_narrow(subObj);
     }
 
     subPtr.reset(
       new DCPS_IR_Subscription(
                      subId,
                      partPtr,
                      topic,
                      dispatchingSubscription.in(),
                      qos,
                      transInfo,
                      transportContextDefault,
                      subscriberQos,
                      filterClassName,
                      filterExpression,
                      exprParams,
                      serializedTypeInfo));
 
     // Release lock
   }
 
   DCPS_IR_Subscription* sub = subPtr.get();
   if (partPtr->add_subscription(move(subPtr)) != 0) {
     // failed to add.  we are responsible for the memory.
     subId = OpenDDS::DCPS::GUID_UNKNOWN;
   } else if (topic->add_subscription_reference(sub) != 0) {
     ACE_ERROR((LM_ERROR, ACE_TEXT("Failed to add subscription to topic list.\n")));
     // No associations were made so remove and fail.
     partPtr->remove_subscription(subId);
     subId = OpenDDS::DCPS::GUID_UNKNOWN;
   }
 
   if (this->um_ && (partPtr->isBitPublisher() == false)) {
     CORBA::String_var callback = orb_->object_to_string(subscription);
     Update::ContentSubscriptionInfo csi(filterClassName, filterExpression, exprParams);
 
     Update::URActor actor(domainId, subId, topicId, participantId, Update::DataReader
                           , callback.in()
                           , const_cast<DDS::SubscriberQos&>(subscriberQos)
                           , const_cast<DDS::DataReaderQos&>(qos)
                           , const_cast<OpenDDS::DCPS::TransportLocatorSeq&>(transInfo)
                           , transportContextDefault, csi
                           , const_cast<DDS::OctetSeq&>(serializedTypeInfo));
 
     this->um_->create(actor);
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(subId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) (GUID_t)TAO_DDS_DCPSInfo_i::add_subscription: ")
                  ACE_TEXT("pushing creation of subscription %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 
   domainPtr->remove_dead_participants();
 
   return subId;
 }

◆ add_subscription() [2/2]

bool TAO_DDS_DCPSInfo_i::add_subscription	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	topicId,
		const OpenDDS::DCPS::GUID_t &	subId,
		const char *	sub_str,
		const DDS::DataReaderQos &	qos,
		const OpenDDS::DCPS::TransportLocatorSeq &	transInfo,
		ACE_CDR::ULong	transportContext,
		const DDS::SubscriberQos &	subscriberQos,
		const char *	filterClassName,
		const char *	filterExpression,
		const DDS::StringSeq &	exprParams,
		const DDS::OctetSeq &	serializedTypeInfo,
		bool	associate = `false`
	)

Add a previously existing subscription to the repository.

Parameters

domainId	the Domain in which the Subscription is contained.
participantId	the Participant in which the Subscription is contained.
topicId	the Topic of the Subscription.
subId	the GUID Id value to use for the Subscription.
sub_str	stringified publication callback to DataReader.
qos	the QoS value of the DataReader.
transInfo	the transport information for the Subscription.
subscriberQos	the QoS value of the Subscriber.
associate	indicate whether to create new associations.

Adds a Subscription to the repository using a specified Subscription GUID Id value. If the SubscriptionId indicates that this Subscription was created by within this repository (the federation Id is the current repositories federation Id), this method will ensure that any subsequent calls to add a Publication and obtain a newly generated Id value will return an Id value greater than the Id value of the current one.

Definition at line 813 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), DCPS_IR_Participant::add_subscription(), DCPS_IR_Topic::add_subscription_reference(), OpenDDS::DCPS::DCPS_debug_level, dispatchingOrb_, domains_, OpenDDS::DCPS::GuidConverter::entityKey(), federation_, OpenDDS::DCPS::RepoIdConverter::federationId(), TAO_DDS_DCPSFederationId::id(), TAO_Pseudo_Var_T< Object >::in(), CORBA::is_nil(), DCPS_IR_Participant::last_subscription_key(), LM_ERROR, LM_WARNING, lock_, OpenDDS::DCPS::move(), orb_, and DCPS_IR_Participant::remove_subscription().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, false);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_subscription: ")
                  ACE_TEXT("invalid domain %d.\n"),
                  domainId));
     }
 
     return false;
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(participantId);
 
   if (0 == partPtr) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(participantId);
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_subscription: ")
                  ACE_TEXT("invalid participant %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
 
     return false;
   }
 
   DCPS_IR_Topic* topic = where->second->find_topic(topicId);
 
   if (topic == 0) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(topicId);
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_subscription: ")
                  ACE_TEXT("invalid topic %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
 
     return false;
   }
 
   CORBA::Object_var obj = (dispatchingOrb_ ? dispatchingOrb_ : orb_) ->string_to_object(sub_str);
   if (CORBA::is_nil(obj.in())) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_subscription: ")
                  ACE_TEXT("failure converting string %C to objref\n"),
                  sub_str));
     }
     return false;
   }
 
   OpenDDS::DCPS::DataReaderRemote_var subscription = OpenDDS::DCPS::DataReaderRemote::_unchecked_narrow(obj.in());
 
   OpenDDS::DCPS::unique_ptr<DCPS_IR_Subscription> subPtr(
     new DCPS_IR_Subscription(
                    subId,
                    partPtr,
                    topic,
                    subscription.in(),
                    qos,
                    transInfo,
                    transportContext,
                    subscriberQos,
                    filterClassName,
                    filterExpression,
                    exprParams,
                    serializedTypeInfo));
 
   DCPS_IR_Subscription* sub = subPtr.get();
   switch (partPtr->add_subscription(OpenDDS::DCPS::move(subPtr))) {
   case -1: {
     OpenDDS::DCPS::RepoIdConverter converter(subId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::add_subscription: ")
                ACE_TEXT("failed to add subscription to participant %C.\n"),
                std::string(converter).c_str()));
     return false;
   }
 
   case 1:
     return false;
 
   case 0:
   default:
     break;
   }
 
   switch (topic->add_subscription_reference(sub, associate)) {
   case -1: {
     OpenDDS::DCPS::RepoIdConverter converter(subId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::add_subscription: ")
                ACE_TEXT("failed to add subscription to participant %C topic list.\n"),
                std::string(converter).c_str()));
 
     // Remove the subscription.
     partPtr->remove_subscription(subId);
 
   }
   return false;
 
   case 1: // This is actually a really really bad place to jump to.
     // This means that we successfully added the new subscription
     // to the participant (it had not been inserted before) but
     // that we are adding a duplicate subscription to the topic
     // list - which should never ever be able to happen.
     return false;
 
   case 0:
   default:
     break;
   }
 
   OpenDDS::DCPS::RepoIdConverter converter(subId);
 
   // See if we are adding a subscription that was created within this
   // repository or a different repository.
   if (converter.federationId() == federation_.id()) {
     // Ensure the subscription GUID_t values do not conflict.
     partPtr->last_subscription_key(converter.entityKey());
   }
 
   return true;
 }

◆ add_topic()

bool TAO_DDS_DCPSInfo_i::add_topic	(	const OpenDDS::DCPS::GUID_t &	topicId,
		DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const char *	topicName,
		const char *	dataTypeName,
		const DDS::TopicQos &	qos
	)

Add a previously existing topic to the repository.

Parameters

topicId	the Topic Entity GUID Id to use.
domainId	the Domain in which the Topic is contained.
participantId	the Participant in which the Topic is contained.
topicName	the name of the Topic.
dataTypeName	the name of the data type.
qos	the QoS value to use for the Topic.

Adds a Topic Entity to the repository using a specified TopicId value. If the TopicId indicates that this Topic was created by within this repository (the federation Id is the current repositories federation Id), this method will ensure that any subsequent calls to add a Topic and obtain a newly generated Id value will return an Id value greater than the Id value of the current one.

Definition at line 232 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::CREATED, OpenDDS::DCPS::DCPS_debug_level, domains_, OpenDDS::DCPS::GuidConverter::entityKey(), federation_, OpenDDS::DCPS::RepoIdConverter::federationId(), TAO_DDS_DCPSFederationId::id(), DCPS_IR_Participant::last_topic_key(), LM_WARNING, and lock_.

Referenced by Update::Manager::add(), OpenDDS::Federator::ManagerImpl::processCreate(), OpenDDS::Federator::ManagerImpl::processDeferred(), and receive_image().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, false);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_topic: ")
                  ACE_TEXT("invalid domain %d.\n"),
                  domainId));
     }
 
     return false;
   }
 
   // Grab the participant.
   DCPS_IR_Participant* participantPtr
   = where->second->participant(participantId);
 
   if (0 == participantPtr) {
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(participantId);
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i:add_topic: ")
                  ACE_TEXT("invalid participant %C.\n"),
                  std::string(converter).c_str()));
     }
 
     return false;
   }
 
   OpenDDS::DCPS::TopicStatus topicStatus
   = where->second->force_add_topic(topicId, topicName, dataTypeName,
                                    qos, participantPtr);
 
   if (topicStatus != OpenDDS::DCPS::CREATED) {
     return false;
   }
 
   OpenDDS::DCPS::RepoIdConverter converter(topicId);
 
   // See if we are adding a topic that was created within this
   // repository or a different repository.
   if (converter.federationId() == federation_.id()) {
     // Ensure the topic GUID_t values do not conflict.
     participantPtr->last_topic_key(converter.entityKey());
   }
 
   return true;
 }

◆ assert_topic()

OpenDDS::DCPS::TopicStatus TAO_DDS_DCPSInfo_i::assert_topic	(	OpenDDS::DCPS::GUID_t_out	topicId,
		DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const char *	topicName,
		const char *	dataTypeName,
		const DDS::TopicQos &	qos,
		bool	hasDcpsKey
	)

virtual

Definition at line 180 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), Update::Manager::create(), OpenDDS::DCPS::DCPS_debug_level, domains_, OpenDDS::DCPS::INTERNAL_ERROR, DCPS_IR_Participant::isBitPublisher(), LM_DEBUG, lock_, and um_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, OpenDDS::DCPS::INTERNAL_ERROR);
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* participantPtr
   = where->second->participant(participantId);
 
   if (0 == participantPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   OpenDDS::DCPS::TopicStatus topicStatus
   = where->second->add_topic(
       topicId,
       topicName,
       dataTypeName,
       qos,
       participantPtr);
 
   if (this->um_ && (participantPtr->isBitPublisher() == false)) {
     Update::UTopic topic(domainId, topicId, participantId
                          , topicName, dataTypeName
                          , const_cast<DDS::TopicQos &>(qos));
     this->um_->create(topic);
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(topicId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::assert_topic: ")
                  ACE_TEXT("pushing creation of topic %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
   return topicStatus;
 }

◆ attach_participant()

CORBA::Boolean TAO_DDS_DCPSInfo_i::attach_participant	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId
	)

virtual

Definition at line 124 of file DCPSInfo_i.cpp.

References ACE_GUARD_RETURN, domains_, lock_, and DCPS_IR_Participant::takeOwnership().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, 0);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* participant
   = where->second->participant(participantId);
 
   if (0 == participant) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   // Establish ownership within the local repository.
   participant->takeOwnership();
 
   return false;
 }

◆ changeOwnership()

bool TAO_DDS_DCPSInfo_i::changeOwnership	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		long	sender,
		long	owner
	)

assert new ownership for a participant and its contained entities.

Parameters

domainId	the domain in which the participant resides.
participantId	the participant to be owned.
sender	the repository sending the update data.
owner	the repository which is to make callbacks for entities within the participant.

Returns: boolean indicating that ownership has been assigned.

This establishes owner as the new owner of the participant. Ownership consists of calling back to the reader and writer remote interfaces when associations are established and removed from a publication or subscription. Owner may be the special value of OWNER_NONE to indicate that the previous owner is no longer available to make callbacks and the application has not indicated which repository is to replace it in this capacity.

The sender of the update is included so that the participant can check that transitions to OWNER_NONE are only honored when initiated by the current owner of the participant.

A return value of false indicates that the ownership was specified for a domain or participant which could not be found.

Definition at line 152 of file DCPSInfo_i.cpp.

References ACE_GUARD_RETURN, DCPS_IR_Participant::changeOwner(), domains_, and lock_.

Referenced by OpenDDS::Federator::ManagerImpl::processCreate(), OpenDDS::Federator::ManagerImpl::processDeferred(), OpenDDS::Federator::ManagerImpl::processDelete(), and OpenDDS::Federator::ManagerImpl::processUpdateQos1().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, false);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     return false;
   }
 
   // Grab the participant.
   DCPS_IR_Participant* participant
   = where->second->participant(participantId);
 
   if (0 == participant) {
     return false;
   }
 
   // Establish the ownership.
   participant->changeOwner(sender, owner);
   return true;
 }

◆ cleanup_all_built_in_topics()

void TAO_DDS_DCPSInfo_i::cleanup_all_built_in_topics ( )

Definition at line 2497 of file DCPSInfo_i.cpp.

References ACE_GUARD, copy(), DCPS_IR_Domain_Map, domains_, in_cleanup_all_built_in_topics_, lock_, and OPENDDS_END_VERSIONED_NAMESPACE_DECL.

Referenced by InfoRepo::finalize(), and InfoRepo::handle_exception().

 {
 #ifndef DDS_HAS_MINIMUM_BIT
   DCPS_IR_Domain_Map copy;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, lock_);
     if (domains_.empty() || in_cleanup_all_built_in_topics_) {
       return;
     }
     copy = domains_;
     in_cleanup_all_built_in_topics_ = true;
   }
 
   for (DCPS_IR_Domain_Map::iterator it = copy.begin(); it != copy.end(); ++it) {
     it->second->cleanup_built_in_topics();
   }
 
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, lock_);
     in_cleanup_all_built_in_topics_ = false;
     copy.clear();
     domains_.clear();
   }
 #endif
 }

◆ disassociate_participant()

void TAO_DDS_DCPSInfo_i::disassociate_participant	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	local_id,
		const OpenDDS::DCPS::GUID_t &	remote_id
	)

virtual

Definition at line 1330 of file DCPSInfo_i.cpp.

References ACE_GUARD, domains_, lock_, DCPS_IR_Participant::publications(), and DCPS_IR_Participant::subscriptions().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   DCPS_IR_Domain_Map::iterator it(this->domains_.find(domainId));
   if (it == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   DCPS_IR_Participant* participant = it->second->participant(local_id);
   if (participant == 0) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   // Disassociate from participant temporarily:
   const DCPS_IR_Subscription_Map& subscriptions = participant->subscriptions();
   for (DCPS_IR_Subscription_Map::const_iterator sub(subscriptions.begin());
        sub != subscriptions.end(); ++sub) {
     sub->second->disassociate_participant(remote_id, true);
   }
 
   const DCPS_IR_Publication_Map& publications = participant->publications();
   for (DCPS_IR_Publication_Map::const_iterator pub(publications.begin());
        pub != publications.end(); ++pub) {
     pub->second->disassociate_participant(remote_id, true);
   }
 
   it->second->remove_dead_participants();
 }

◆ disassociate_publication()

void TAO_DDS_DCPSInfo_i::disassociate_publication	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	local_id,
		const OpenDDS::DCPS::GUID_t &	remote_id
	)

virtual

Definition at line 1406 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, DCPS_IR_Publication::disassociate_subscription(), domains_, DCPS_IR_Participant::find_publication_reference(), LM_ERROR, LM_INFO, and lock_.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   DCPS_IR_Domain_Map::iterator it(this->domains_.find(domainId));
   if (it == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   DCPS_IR_Participant* participant = it->second->participant(participantId);
   if (participant == 0) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) disassociating publication\n"));
   }
 
   DCPS_IR_Publication* publication;
   if (participant->find_publication_reference(local_id, publication)
       != 0 || publication == 0) {
     OpenDDS::DCPS::RepoIdConverter part_converter(participantId);
     OpenDDS::DCPS::RepoIdConverter pub_converter(local_id);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::disassociate_publication: ")
                ACE_TEXT("participant %C could not find publication %C.\n"),
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
     throw OpenDDS::DCPS::Invalid_Publication();
   }
 
   // Disassociate from subscription temporarily:
   publication->disassociate_subscription(remote_id, true);
 
   it->second->remove_dead_participants();
 }

◆ disassociate_subscription()

void TAO_DDS_DCPSInfo_i::disassociate_subscription	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	local_id,
		const OpenDDS::DCPS::GUID_t &	remote_id
	)

virtual

Definition at line 1364 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, DCPS_IR_Subscription::disassociate_publication(), domains_, DCPS_IR_Participant::find_subscription_reference(), LM_ERROR, LM_INFO, and lock_.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   DCPS_IR_Domain_Map::iterator it(this->domains_.find(domainId));
   if (it == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   DCPS_IR_Participant* participant = it->second->participant(participantId);
   if (participant == 0) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) disassociating subscription\n"));
   }
 
   DCPS_IR_Subscription* subscription;
   if (participant->find_subscription_reference(local_id, subscription)
       != 0 || subscription == 0) {
     OpenDDS::DCPS::RepoIdConverter part_converter(participantId);
     OpenDDS::DCPS::RepoIdConverter sub_converter(local_id);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::disassociate_subscription: ")
                ACE_TEXT("participant %C could not find subscription %C.\n"),
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
     throw OpenDDS::DCPS::Invalid_Subscription();
   }
 
   // Disassociate from publication temporarily:
   subscription->disassociate_publication(remote_id, true);
 
   it->second->remove_dead_participants();
 }

◆ domain()

DCPS_IR_Domain * TAO_DDS_DCPSInfo_i::domain ( DDS::DomainId_t domain )

Convert a domain Id into a reference to a DCPS_IR_Domain object.

Definition at line 2129 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::Service_Participant::ANY_DOMAIN, OpenDDS::DCPS::DCPS_debug_level, domains_, federation_, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), DCPS_IR_Domain::init_built_in_topics(), LM_DEBUG, LM_ERROR, OpenDDS::DCPS::move(), TAO_DDS_DCPSFederationId::overridden(), participantIdGenerator_, reincarnate_, TheServiceParticipant, and DCPS_IR_Domain::useBIT().

Referenced by add_domain_participant(), receive_image(), remove_by_owner(), and update_subscription_params().

 {
   if (domain == OpenDDS::DCPS::Service_Participant::ANY_DOMAIN) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::domain: ")
                ACE_TEXT("ANY_DOMAIN not supported for operations.\n")));
     return 0;
   }
 
   // Check if the domain is already in the map.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domain);
 
   if (where == this->domains_.end()) {
     // We will attempt to insert a new domain, go ahead and allocate it.
     OpenDDS::DCPS::unique_ptr<DCPS_IR_Domain> domain_uptr( new
                    DCPS_IR_Domain(domain, this->participantIdGenerator_));
 
     DCPS_IR_Domain* domainPtr = domain_uptr.get();
 
     // We need to insert the domain into the map at this time since it
     // might be looked up during the init_built_in_topics() call.
     this->domains_.insert(
       where,
       DCPS_IR_Domain_Map::value_type(domain, OpenDDS::DCPS::move(domain_uptr)));
 
 #ifndef DDS_HAS_MINIMUM_BIT
     if (TheServiceParticipant->get_BIT() && !domainPtr->useBIT() &&
       domainPtr->init_built_in_topics(federation_.overridden(), reincarnate_)
     ) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::domain: ")
                  ACE_TEXT("failed to initialize the Built-In Topics ")
                  ACE_TEXT("when loading domain %d.\n"),
                  domain));
       this->domains_.erase(domain);
       return 0;
     }
 #endif
 
     if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::domain: ")
                  ACE_TEXT("successfully loaded domain %d at %x.\n"),
                  domain,
                  domainPtr));
     }
     return domainPtr;
 
   } else {
     return where->second.get();
   }
 }

◆ domains()

const DCPS_IR_Domain_Map & TAO_DDS_DCPSInfo_i::domains ( ) const

Expose a readable reference of the domain map.

Definition at line 2473 of file DCPSInfo_i.cpp.

References domains_.

Referenced by OpenDDS::Federator::ManagerImpl::pushState().

 {
   return this->domains_;
 }

◆ dump_to_string()

char * TAO_DDS_DCPSInfo_i::dump_to_string ( )

virtual

Dump the Repos state to string.

Definition at line 2480 of file DCPSInfo_i.cpp.

References domains_, and CORBA::string_dup().

 {
   std::string dump;
 #if !defined (OPENDDS_INFOREPO_REDUCED_FOOTPRINT)
   std::string indent ("    ");
 
   for (DCPS_IR_Domain_Map::const_iterator dm = domains_.begin();
        dm != domains_.end();
        dm++)
   {
     dump += dm->second->dump_to_string(indent, 0);
   }
 #endif // !defined (OPENDDS_INFOREPO_REDUCED_FOOTPRINT)
   return CORBA::string_dup(dump.c_str());
 
 }

◆ finalize()

void TAO_DDS_DCPSInfo_i::finalize ( void )

Cleanup state for shutdown.

Definition at line 2455 of file DCPSInfo_i.cpp.

References ACE_Reactor::cancel_timer(), DCPS_IR_Domain_Map, dispatch_check_timer_id_, orb_, ACE_Event_Handler::reactor(), and reassociate_timer_id_.

Referenced by InfoRepo::finalize(), and InfoRepo::handle_exception().

 {
   if (reassociate_timer_id_ != -1) {
     ACE_Reactor* reactor = this->orb_->orb_core()->reactor();
 
     reactor->cancel_timer(this->reassociate_timer_id_);
     this->reassociate_timer_id_ = -1;
   }
 
   if (dispatch_check_timer_id_ != -1) {
     ACE_Reactor* reactor = this->orb_->orb_core()->reactor();
 
     reactor->cancel_timer(this->dispatch_check_timer_id_);
     this->dispatch_check_timer_id_ = -1;
   }
 }

◆ find_topic()

OpenDDS::DCPS::TopicStatus TAO_DDS_DCPSInfo_i::find_topic	(	DDS::DomainId_t	domainId,
		const char *	topicName,
		CORBA::String_out	dataTypeName,
		DDS::TopicQos_out	qos,
		OpenDDS::DCPS::GUID_t_out	topicId
	)

virtual

Definition at line 291 of file DCPSInfo_i.cpp.

References ACE_GUARD_RETURN, domains_, OpenDDS::DCPS::FOUND, DCPS_IR_Topic_Description::get_dataTypeName(), DCPS_IR_Topic::get_id(), DCPS_IR_Topic::get_topic_description(), DCPS_IR_Topic::get_topic_qos(), OpenDDS::DCPS::INTERNAL_ERROR, lock_, OpenDDS::DCPS::NOT_FOUND, and Update::TopicQos.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, OpenDDS::DCPS::INTERNAL_ERROR);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   OpenDDS::DCPS::TopicStatus status = OpenDDS::DCPS::NOT_FOUND;
 
   DCPS_IR_Topic* topic = 0;
   qos = new DDS::TopicQos;
 
   status = where->second->find_topic(topicName, topic);
 
   if (0 != topic) {
     status = OpenDDS::DCPS::FOUND;
     const DCPS_IR_Topic_Description* desc = topic->get_topic_description();
     dataTypeName = desc->get_dataTypeName();
     *qos = *(topic->get_topic_qos());
     topicId = topic->get_id();
   }
 
   return status;
 }

◆ handle_timeout()

int TAO_DDS_DCPSInfo_i::handle_timeout	(	const ACE_Time_Value &	now,
		const void *	arg
	)

virtual

Reimplemented from ACE_Event_Handler.

Definition at line 70 of file DCPSInfo_i.cpp.

References ACE_GUARD_RETURN, dispatchingOrb_, domains_, and lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, 0);
 
   if (arg == this) {
     if (dispatchingOrb_) {
       if (dispatchingOrb_->work_pending()) {
         // Ten microseconds
         ACE_Time_Value smallval(0, 10);
         dispatchingOrb_->perform_work(smallval);
       }
     }
   } else {
     // NOTE: This is a purposefully naive approach to addressing defunct
     // associations.  In the future, it may be worthwhile to introduce a
     // callback model to fix the heinous runtime cost below:
     for (DCPS_IR_Domain_Map::const_iterator dom(this->domains_.begin());
          dom != this->domains_.end(); ++dom) {
 
       const DCPS_IR_Participant_Map& participants(dom->second->participants());
       for (DCPS_IR_Participant_Map::const_iterator part(participants.begin());
            part != participants.end(); ++part) {
 
         const DCPS_IR_Subscription_Map& subscriptions(part->second->subscriptions());
         for (DCPS_IR_Subscription_Map::const_iterator sub(subscriptions.begin());
              sub != subscriptions.end(); ++sub) {
           sub->second->reevaluate_defunct_associations();
         }
 
         const DCPS_IR_Publication_Map& publications(part->second->publications());
         for (DCPS_IR_Publication_Map::const_iterator pub(publications.begin());
              pub != publications.end(); ++pub) {
           pub->second->reevaluate_defunct_associations();
         }
       }
     }
   }
 
   return 0;
 }

◆ ignore_domain_participant()

void TAO_DDS_DCPSInfo_i::ignore_domain_participant	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	myParticipantId,
		const OpenDDS::DCPS::GUID_t &	ignoreId
	)

virtual

Definition at line 1552 of file DCPSInfo_i.cpp.

References ACE_GUARD, domains_, DCPS_IR_Participant::ignore_participant(), and lock_.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(myParticipantId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   partPtr->ignore_participant(ignoreId);
 
   where->second->remove_dead_participants();
 }

◆ ignore_publication()

void TAO_DDS_DCPSInfo_i::ignore_publication	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	myParticipantId,
		const OpenDDS::DCPS::GUID_t &	ignoreId
	)

virtual

Definition at line 1633 of file DCPSInfo_i.cpp.

References ACE_GUARD, domains_, DCPS_IR_Participant::ignore_publication(), and lock_.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(myParticipantId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   partPtr->ignore_publication(ignoreId);
 
   where->second->remove_dead_participants();
 }

◆ ignore_subscription()

void TAO_DDS_DCPSInfo_i::ignore_subscription	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	myParticipantId,
		const OpenDDS::DCPS::GUID_t &	ignoreId
	)

virtual

Definition at line 1606 of file DCPSInfo_i.cpp.

References ACE_GUARD, domains_, DCPS_IR_Participant::ignore_subscription(), and lock_.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(myParticipantId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   partPtr->ignore_subscription(ignoreId);
 
   where->second->remove_dead_participants();
 }

◆ ignore_topic()

void TAO_DDS_DCPSInfo_i::ignore_topic	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	myParticipantId,
		const OpenDDS::DCPS::GUID_t &	ignoreId
	)

virtual

Definition at line 1579 of file DCPSInfo_i.cpp.

References ACE_GUARD, domains_, DCPS_IR_Participant::ignore_topic(), and lock_.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(myParticipantId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   partPtr->ignore_topic(ignoreId);
 
   where->second->remove_dead_participants();
 }

◆ init_dispatchChecking()

bool TAO_DDS_DCPSInfo_i::init_dispatchChecking ( const ACE_Time_Value & delay )

Definition at line 2444 of file DCPSInfo_i.cpp.

References dispatch_check_timer_id_, orb_, ACE_Event_Handler::reactor(), and ACE_Reactor::schedule_timer().

 {
   if (this->dispatch_check_timer_id_ != -1) return false;  // already scheduled
 
   ACE_Reactor* reactor = this->orb_->orb_core()->reactor();
 
   this->dispatch_check_timer_id_ = reactor->schedule_timer(this, this, delay, delay);
   return this->dispatch_check_timer_id_ != -1;
 }

◆ init_persistence()

bool TAO_DDS_DCPSInfo_i::init_persistence ( )

Definition at line 2411 of file DCPSInfo_i.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), Update::Manager::add(), ACE_Dynamic_Service< class >::instance(), LM_ERROR, reincarnate_, Update::Manager::requestImage(), and um_.

 {
   um_ = ACE_Dynamic_Service<UpdateManagerSvc>::instance
         ("UpdateManagerSvc");
 
   if (um_ != 0) {
     um_->add(this);
 
     // Request persistent image.
     if (reincarnate_) {
       um_->requestImage();
     }
 
   } else {
     ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("TAO_DDS_DCPSInfo_i> Failed to discover ")
                       ACE_TEXT("UpdateManagerSvc.\n")), false);
   }
 
   return true;
 }

◆ init_reassociation()

bool TAO_DDS_DCPSInfo_i::init_reassociation ( const ACE_Time_Value & delay )

Definition at line 2433 of file DCPSInfo_i.cpp.

References orb_, ACE_Event_Handler::reactor(), reassociate_timer_id_, and ACE_Reactor::schedule_timer().

 {
   if (this->reassociate_timer_id_ != -1) return false;  // already scheduled
 
   ACE_Reactor* reactor = this->orb_->orb_core()->reactor();
 
   this->reassociate_timer_id_ = reactor->schedule_timer(this, 0, delay, delay);
   return this->reassociate_timer_id_ != -1;
 }

◆ init_transport()

int TAO_DDS_DCPSInfo_i::init_transport	(	int	listen_address_given,
		const char *	listen_str
	)

Initialize the transport for the Built-In Topics Returns 0 (zero) if succeeds

Definition at line 2182 of file DCPSInfo_i.cpp.

References ACE_TEXT(), OpenDDS::DCPS::TransportRegistry::create_config(), OpenDDS::DCPS::TransportRegistry::create_inst(), ACE_Service_Config::current(), OpenDDS::DCPS::TransportInst::datalink_release_delay_, OpenDDS::DCPS::TransportRegistry::DEFAULT_INST_PREFIX, OpenDDS::DCPS::dynamic_rchandle_cast(), find(), OpenDDS::DCPS::TransportRegistry::instance(), OpenDDS::DCPS::TransportConfig::instances_, and ACE_Service_Config::process_directive().

 {
   int status = 0;
 
 #ifndef DDS_HAS_MINIMUM_BIT
   try {
 
 #ifndef ACE_AS_STATIC_LIBS
     if (ACE_Service_Config::current()->find(ACE_TEXT("OpenDDS_Tcp"))
         < 0 /* not found (-1) or suspended (-2) */) {
       static const ACE_TCHAR directive[] =
         ACE_TEXT("dynamic OpenDDS_Tcp Service_Object * ")
         ACE_TEXT("OpenDDS_Tcp:_make_TcpLoader()");
       ACE_Service_Config::process_directive(directive);
     }
 #endif
 
     const std::string config_name =
       OpenDDS::DCPS::TransportRegistry::DEFAULT_INST_PREFIX
       + std::string("InfoRepoBITTransportConfig");
     OpenDDS::DCPS::TransportConfig_rch config =
       OpenDDS::DCPS::TransportRegistry::instance()->create_config(config_name);
 
     const std::string inst_name =
       OpenDDS::DCPS::TransportRegistry::DEFAULT_INST_PREFIX
       + std::string("InfoRepoBITTCPTransportInst");
     OpenDDS::DCPS::TransportInst_rch inst =
       OpenDDS::DCPS::TransportRegistry::instance()->create_inst(inst_name,
                                                                "tcp");
     config->instances_.push_back(inst);
 
     OpenDDS::DCPS::TcpInst_rch tcp_inst =
       OpenDDS::DCPS::dynamic_rchandle_cast<OpenDDS::DCPS::TcpInst>(inst);
     inst->datalink_release_delay_ = 0;
 
     tcp_inst->conn_retry_attempts_ = 0;
 
     if (listen_address_given) {
       tcp_inst->local_address(listen_str);
     }
 
   } catch (...) {
     // TransportRegistry is extremely varied in the exceptions that
     // it throws on failure; do not allow exceptions to bubble up
     // beyond this point.
     status = 1;
   }
 #else
   ACE_UNUSED_ARG(listen_address_given);
   ACE_UNUSED_ARG(listen_str);
 #endif
 
   return status;
 }

◆ orb()

CORBA::ORB_ptr TAO_DDS_DCPSInfo_i::orb ( void )

Expose the ORB.

Definition at line 119 of file DCPSInfo_i.cpp.

References CORBA::ORB::_duplicate(), TAO_Pseudo_Var_T< ORB >::in(), and orb_.

Referenced by OpenDDS::Federator::ManagerImpl::pushState().

 {
   return CORBA::ORB::_duplicate(this->orb_.in());
 }

◆ receive_image()

bool TAO_DDS_DCPSInfo_i::receive_image ( const Update::UImage & image )

Definition at line 2239 of file DCPSInfo_i.cpp.

Referenced by Update::Manager::pushImage().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::receive_image: ")
                ACE_TEXT("processing persistent data.\n")));
   }
 
   // Initialize builtin topics first so that they always have the same IDs
 #ifndef DDS_HAS_MINIMUM_BIT
   if (TheServiceParticipant->get_BIT()) {
     for (Update::UImage::ParticipantSeq::const_iterator
          iter = image.participants.begin();
          iter != image.participants.end(); iter++) {
       const Update::UParticipant* part = *iter;
       if (!domain(part->domainId)) {
         if (OpenDDS::DCPS::DCPS_debug_level > 4) {
           ACE_DEBUG((LM_WARNING,
                      ACE_TEXT("(%P|%t) WARNING: TAO_DDS_DCPSInfo_i::receive_image: ")
                      ACE_TEXT("invalid domain Id: %d\n"),
                      part->domainId));
         }
         return false;
       }
     }
   }
 #endif
 
   // Ensure that new non-BIT participants do not reuse an id
   participantIdGenerator_.last(image.lastPartId);
 
   for (Update::UImage::ParticipantSeq::const_iterator
        iter = image.participants.begin();
        iter != image.participants.end(); iter++) {
     const Update::UParticipant* part = *iter;
 
     if (!this->add_domain_participant(part->domainId, part->participantId
                                       , part->participantQos)) {
       OpenDDS::DCPS::RepoIdConverter converter(part->participantId);
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::receive_image: ")
                  ACE_TEXT("failed to add participant %C to domain %d.\n"),
                  std::string(converter).c_str(),
                  part->domainId));
       return false;
 
     } else if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       OpenDDS::DCPS::RepoIdConverter converter(part->participantId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::receive_image: ")
                  ACE_TEXT("added participant %C to domain %d.\n"),
                  std::string(converter).c_str(),
                  part->domainId));
     }
   }
 
   for (Update::UImage::TopicSeq::const_iterator iter = image.topics.begin();
        iter != image.topics.end(); iter++) {
     const Update::UTopic* topic = *iter;
 
     if (!this->add_topic(topic->topicId, topic->domainId
                          , topic->participantId, topic->name.c_str()
                          , topic->dataType.c_str(), topic->topicQos)) {
       OpenDDS::DCPS::RepoIdConverter topic_converter(topic->topicId);
       OpenDDS::DCPS::RepoIdConverter part_converter(topic->participantId);
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::receive_image: ")
                  ACE_TEXT("failed to add topic %C to participant %C.\n"),
                  std::string(topic_converter).c_str(),
                  std::string(part_converter).c_str()));
       return false;
 
     } else if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       OpenDDS::DCPS::RepoIdConverter topic_converter(topic->topicId);
       OpenDDS::DCPS::RepoIdConverter part_converter(topic->participantId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::receive_image: ")
                  ACE_TEXT("added topic %C to participant %C.\n"),
                  std::string(topic_converter).c_str(),
                  std::string(part_converter).c_str()));
     }
   }
 
   for (Update::UImage::ReaderSeq::const_iterator iter = image.actors.begin();
        iter != image.actors.end(); iter++) {
     const Update::URActor* sub = *iter;
     // no reason to associate, there are no publishers yet to associate with
     if (!this->add_subscription(sub->domainId, sub->participantId
                                 , sub->topicId, sub->actorId
                                 , sub->callback.c_str(), sub->drdwQos
                                 , sub->transportInterfaceInfo
                                 , sub->transportContext
                                 , sub->pubsubQos
                                 , sub->contentSubscriptionProfile.filterClassName
                                 , sub->contentSubscriptionProfile.filterExpr
                                 , sub->contentSubscriptionProfile.exprParams
                                 , sub->serializedTypeInfo)) {
       OpenDDS::DCPS::RepoIdConverter sub_converter(sub->actorId);
       OpenDDS::DCPS::RepoIdConverter part_converter(sub->participantId);
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::receive_image: ")
                  ACE_TEXT("failed to add subscription %C to participant %C.\n"),
                  std::string(sub_converter).c_str(),
                  std::string(part_converter).c_str()));
       return false;
 
     } else if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       OpenDDS::DCPS::RepoIdConverter sub_converter(sub->actorId);
       OpenDDS::DCPS::RepoIdConverter part_converter(sub->participantId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::receive_image: ")
                  ACE_TEXT("added subscription %C to participant %C.\n"),
                  std::string(sub_converter).c_str(),
                  std::string(part_converter).c_str()));
     }
   }
 
   for (Update::UImage::WriterSeq::const_iterator iter = image.wActors.begin();
        iter != image.wActors.end(); iter++) {
     const Update::UWActor* pub = *iter;
 
     // try to associate with any persisted subscriptions to track any expected
     // existing associations
     if (!this->add_publication(pub->domainId, pub->participantId
                                , pub->topicId, pub->actorId
                                , pub->callback.c_str() , pub->drdwQos
                                , pub->transportInterfaceInfo, pub->transportContext
                                , pub->pubsubQos
                                , pub->serializedTypeInfo
                                , true)) {
       OpenDDS::DCPS::RepoIdConverter pub_converter(pub->actorId);
       OpenDDS::DCPS::RepoIdConverter part_converter(pub->participantId);
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::receive_image: ")
                  ACE_TEXT("failed to add publication %C to participant %C.\n"),
                  std::string(pub_converter).c_str(),
                  std::string(part_converter).c_str()));
       return false;
 
     } else if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       OpenDDS::DCPS::RepoIdConverter pub_converter(pub->actorId);
       OpenDDS::DCPS::RepoIdConverter part_converter(pub->participantId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::receive_image: ")
                  ACE_TEXT("added publication %C to participant %C.\n"),
                  std::string(pub_converter).c_str(),
                  std::string(part_converter).c_str()));
     }
   }
 
 #ifndef DDS_HAS_MINIMUM_BIT
   if (TheServiceParticipant->get_BIT()) {
     for (DCPS_IR_Domain_Map::const_iterator currentDomain = domains_.begin();
          currentDomain != domains_.end();
          ++currentDomain) {
       currentDomain->second->reassociate_built_in_topic_pubs();
     }
   }
 #endif
 
   return true;
 }

◆ remove_by_owner()

bool TAO_DDS_DCPSInfo_i::remove_by_owner	(	DDS::DomainId_t	domain,
		long	owner
	)

Definition at line 1206 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domain(), domains_, LM_DEBUG, lock_, DCPS_IR_Participant::publications(), remove_domain_participant(), DCPS_IR_Participant::remove_publication(), DCPS_IR_Participant::remove_subscription(), DCPS_IR_Participant::remove_topic_reference(), DCPS_IR_Participant::subscriptions(), and DCPS_IR_Participant::topics().

Referenced by OpenDDS::Federator::ManagerImpl::leave_federation().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, false);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domain);
 
   if (where == this->domains_.end()) {
     return false;
   }
 
   std::vector<OpenDDS::DCPS::GUID_t> candidates;
 
   for (DCPS_IR_Participant_Map::const_iterator
        current = where->second->participants().begin();
        current != where->second->participants().end();
        ++current) {
     if (current->second->owner() == owner) {
       candidates.push_back(current->second->get_id());
     }
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) (bool)TAO_DDS_DCPSInfo_i::remove_by_owner: ")
                ACE_TEXT("%d participants to remove from domain %d.\n"),
                candidates.size(),
                domain));
   }
 
   bool status = true;
 
   for (unsigned int index = 0; index < candidates.size(); ++index) {
     DCPS_IR_Participant* participant
     = where->second->participant(candidates[index]);
     if (participant) {
       std::vector<OpenDDS::DCPS::GUID_t> keylist;
 
       // Remove Subscriptions
       for (DCPS_IR_Subscription_Map::const_iterator
         current = participant->subscriptions().begin();
         current != participant->subscriptions().end();
         ++current) {
         keylist.push_back(current->second->get_id());
       }
 
       if (OpenDDS::DCPS::DCPS_debug_level > 0) {
         OpenDDS::DCPS::RepoIdConverter converter(candidates[index]);
         ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) (bool)TAO_DDS_DCPSInfo_i::remove_by_owner: ")
           ACE_TEXT("%d subscriptions to remove from participant %C.\n"),
           keylist.size(),
           std::string(converter).c_str()));
       }
 
       for (unsigned int key = 0; key < keylist.size(); ++key) {
         if (participant->remove_subscription(keylist[key]) != 0) {
           status = false;
         }
       }
 
       // Remove Publications
       keylist.clear();
 
       for (DCPS_IR_Publication_Map::const_iterator
         current = participant->publications().begin();
         current != participant->publications().end();
         ++current) {
         keylist.push_back(current->second->get_id());
       }
 
       if (OpenDDS::DCPS::DCPS_debug_level > 0) {
         OpenDDS::DCPS::RepoIdConverter converter(candidates[index]);
         ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) (bool)TAO_DDS_DCPSInfo_i::remove_by_owner: ")
           ACE_TEXT("%d publications to remove from participant %C.\n"),
           keylist.size(),
           std::string(converter).c_str()));
       }
 
       for (unsigned int key = 0; key < keylist.size(); ++key) {
         if (participant->remove_publication(keylist[key]) != 0) {
           status = false;
         }
       }
 
       // Remove Topics
       keylist.clear();
 
       for (DCPS_IR_Topic_Map::const_iterator
         current = participant->topics().begin();
         current != participant->topics().end();
         ++current) {
         keylist.push_back(current->second->get_id());
       }
 
       if (OpenDDS::DCPS::DCPS_debug_level > 0) {
         OpenDDS::DCPS::RepoIdConverter converter(candidates[index]);
         ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) (bool)TAO_DDS_DCPSInfo_i::remove_by_owner: ")
           ACE_TEXT("%d topics to remove from participant %C.\n"),
           keylist.size(),
           std::string(converter).c_str()));
       }
 
       for (unsigned int key = 0; key < keylist.size(); ++key) {
         DCPS_IR_Topic* discard;
 
         if (participant->remove_topic_reference(keylist[key], discard) != 0) {
           status = false;
         }
       }
     }
 
     // Remove Participant
     this->remove_domain_participant(domain, candidates[ index]);
   }
 
   return status;
 }

◆ remove_domain_participant()

void TAO_DDS_DCPSInfo_i::remove_domain_participant	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId
	)

virtual

Definition at line 1447 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), TAO_DDS_DCPSInfo_i::BIT_Cleanup_Handler::cv_, OpenDDS::DCPS::CvStatus_NoTimeout, OpenDDS::DCPS::DCPS_debug_level, Update::Manager::destroy(), domains_, TAO_DDS_DCPSInfo_i::BIT_Cleanup_Handler::done_, ACE_Event_Handler_var::handler(), in_cleanup_all_built_in_topics_, LM_DEBUG, LM_ERROR, lock_, Update::Participant, TheServiceParticipant, um_, and OpenDDS::DCPS::ConditionVariable< Mutex >::wait().

Referenced by OpenDDS::Federator::ManagerImpl::processDelete(), and remove_by_owner().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   DCPS_IR_Participant_rch participant = where->second->participant_rch(participantId);
   if (!participant) {
     OpenDDS::DCPS::RepoIdConverter converter(participantId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: (bool)TAO_DDS_DCPSInfo_i::remove_domain_participant: ")
                ACE_TEXT("failed to locate participant %C in domain %d.\n"),
                std::string(converter).c_str(),
                domainId));
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   // Determine if we should propagate this event;  we need to cache this
   // result as the participant will be gone by the time we use the result.
   bool sendUpdate = participant->isOwner() && !participant->isBitPublisher();
 
   CORBA::Boolean dont_notify_lost = 0;
   int status = where->second->remove_participant(participantId, dont_notify_lost);
 
   if (0 != status) {
     // Removing the participant failed
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   // Update any concerned observers that the participant was destroyed.
   if (this->um_ && sendUpdate) {
     Update::IdPath path(
       where->second->get_id(),
       participantId,
       participantId);
     this->um_->destroy(path, Update::Participant);
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(participantId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::remove_domain_participant: ")
                  ACE_TEXT("pushing deletion of participant %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 
   if (where->second->participants().empty()
 #ifndef DDS_HAS_MINIMUM_BIT
     && !(participant->isOwner() && participant->isBitPublisher() && in_cleanup_all_built_in_topics_)
     // If this is false, we're running as part of cleanup_all_built_in_topics
     // and we can't remove the domain because we would invalid the iterator
     // we're using in cleanup_all_built_in_topics. cleanup_all_built_in_topics
     // will clear the domains once it's done.
 #endif
     ) {
     domains_.erase(where);
   }
 #ifndef DDS_HAS_MINIMUM_BIT
   else if (where->second->useBIT() &&
            where->second->participants().size() == 1) {
     // The only participant left is the one we created to publish BITs.
     // It can be removed now since no user participants exist in this domain,
     // but it has to be removed on the Service Participant's reactor thread
     // in order to make the locking work properly in delete_participant().
     BIT_Cleanup_Handler* eh_impl = new BIT_Cleanup_Handler(this, domainId);
     const ACE_Event_Handler_var eh = eh_impl;
     TheServiceParticipant->reactor()->notify(eh.handler());
 
     // Wait for that to be finished
     using OpenDDS::DCPS::CvStatus_NoTimeout;
     OpenDDS::DCPS::CvStatus status = CvStatus_NoTimeout;
     OpenDDS::DCPS::ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
     while (status == CvStatus_NoTimeout && !eh_impl->done_) {
       status = eh_impl->cv_.wait(thread_status_manager);
     }
   }
 #endif
 }

◆ remove_publication()

void TAO_DDS_DCPSInfo_i::remove_publication	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	publicationId
	)

virtual

Definition at line 630 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), Update::Actor, Update::DataWriter, OpenDDS::DCPS::DCPS_debug_level, Update::Manager::destroy(), domains_, in_cleanup_all_built_in_topics_, DCPS_IR_Participant::isBitPublisher(), DCPS_IR_Participant::isOwner(), LM_DEBUG, lock_, DCPS_IR_Participant::remove_publication(), and um_.

Referenced by OpenDDS::Federator::ManagerImpl::processDelete().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* const partPtr = where->second->participant(participantId);
   if (!partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   const bool in_cleanup =
 #ifdef DDS_HAS_MINIMUM_BIT
     false;
 #else
     in_cleanup_all_built_in_topics_;
 #endif
 
   if (partPtr->remove_publication(publicationId) != 0) {
     where->second->remove_dead_participants(in_cleanup);
 
     // throw exception because the publication was not removed!
     throw OpenDDS::DCPS::Invalid_Publication();
   }
 
   where->second->remove_dead_participants(in_cleanup);
 
   if (um_ && partPtr->isOwner() && !partPtr->isBitPublisher()) {
     Update::IdPath path(domainId, participantId, publicationId);
     this->um_->destroy(path, Update::Actor, Update::DataWriter);
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(publicationId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::remove_publication: ")
                  ACE_TEXT("pushing deletion of publication %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 }

◆ remove_subscription()

void TAO_DDS_DCPSInfo_i::remove_subscription	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	subscriptionId
	)

virtual

Definition at line 962 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), Update::Actor, Update::DataReader, OpenDDS::DCPS::DCPS_debug_level, Update::Manager::destroy(), domains_, in_cleanup_all_built_in_topics_, DCPS_IR_Participant::isBitPublisher(), DCPS_IR_Participant::isOwner(), LM_DEBUG, lock_, DCPS_IR_Participant::remove_subscription(), and um_.

Referenced by OpenDDS::Federator::ManagerImpl::processDelete().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* const partPtr = where->second->participant(participantId);
   if (!partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   if (partPtr->remove_subscription(subscriptionId) != 0) {
     // throw exception because the subscription was not removed!
     throw OpenDDS::DCPS::Invalid_Subscription();
   }
 
   where->second->remove_dead_participants(
 #ifdef DDS_HAS_MINIMUM_BIT
     false
 #else
     in_cleanup_all_built_in_topics_
 #endif
     );
 
   if (um_ && partPtr->isOwner() && !partPtr->isBitPublisher()) {
     Update::IdPath path(domainId, participantId, subscriptionId);
     this->um_->destroy(path, Update::Actor, Update::DataReader);
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(subscriptionId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::remove_subscription: ")
                  ACE_TEXT("pushing deletion of subscription %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 }

◆ remove_topic()

OpenDDS::DCPS::TopicStatus TAO_DDS_DCPSInfo_i::remove_topic	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	topicId
	)

virtual

Definition at line 325 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, Update::Manager::destroy(), domains_, DCPS_IR_Participant::find_topic_reference(), OpenDDS::DCPS::INTERNAL_ERROR, DCPS_IR_Participant::isBitPublisher(), DCPS_IR_Participant::isOwner(), LM_DEBUG, lock_, Update::Topic, and um_.

Referenced by OpenDDS::Federator::ManagerImpl::processDelete().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, OpenDDS::DCPS::INTERNAL_ERROR);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(participantId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   DCPS_IR_Topic* topic;
 
   if (partPtr->find_topic_reference(topicId, topic) != 0) {
     throw OpenDDS::DCPS::Invalid_Topic();
   }
 
   OpenDDS::DCPS::TopicStatus removedStatus = where->second->remove_topic(partPtr, topic);
 
   if (this->um_
       && (partPtr->isOwner() == true)
       && (partPtr->isBitPublisher() == false)) {
     Update::IdPath path(domainId, participantId, topicId);
     this->um_->destroy(path, Update::Topic);
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(topicId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::remove_topic: ")
                  ACE_TEXT("pushing deletion of topic %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 
   return removedStatus;
 }

◆ shutdown()

void TAO_DDS_DCPSInfo_i::shutdown ( void )

virtual

Cause the entire repository to exit.

Definition at line 113 of file DCPSInfo_i.cpp.

References ShutdownInterface::shutdown(), and shutdown_.

Referenced by OpenDDS::Federator::ManagerImpl::leave_and_shutdown(), and OpenDDS::Federator::ManagerImpl::shutdown().

 {
   this->shutdown_->shutdown();
 }

◆ update_domain_participant_qos()

CORBA::Boolean TAO_DDS_DCPSInfo_i::update_domain_participant_qos	(	DDS::DomainId_t	domain,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const DDS::DomainParticipantQos &	qos
	)

virtual

Definition at line 2084 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domains_, DCPS_IR_Participant::isBitPublisher(), DCPS_IR_Participant::isOwner(), LM_DEBUG, lock_, DCPS_IR_Participant::set_qos(), um_, and Update::Manager::update().

Referenced by OpenDDS::Federator::ManagerImpl::processUpdateQos1().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, 0);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(participantId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   if (partPtr->set_qos(qos) == false)
     return 0;
 
   if (this->um_
       && (partPtr->isOwner() == true)
       && (partPtr->isBitPublisher() == false)) {
     Update::IdPath path(domainId, participantId, participantId);
     this->um_->update(path, qos);
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(participantId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::update_domain_participant_qos: ")
                  ACE_TEXT("pushing update of participant %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 
   return 1;
 }

◆ update_publication_qos() [1/3]

CORBA::Boolean TAO_DDS_DCPSInfo_i::update_publication_qos	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	partId,
		const OpenDDS::DCPS::GUID_t &	dwId,
		const DDS::DataWriterQos &	qos,
		const DDS::PublisherQos &	publisherQos
	)

virtual

Definition at line 1660 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), Update::DataWriterQos, OpenDDS::DCPS::DCPS_debug_level, domains_, DCPS_IR_Participant::find_publication_reference(), DCPS_IR_Participant::isBitPublisher(), LM_DEBUG, LM_ERROR, LM_INFO, lock_, Update::NoQos, Update::PublisherQos, DCPS_IR_Publication::set_qos(), um_, and Update::Manager::update().

Referenced by OpenDDS::Federator::ManagerImpl::processUpdateQos1(), and OpenDDS::Federator::ManagerImpl::processUpdateQos2().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, 0);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(partId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) updating  publication qos 1\n"));
   }
 
   DCPS_IR_Publication* pub;
 
   if (partPtr->find_publication_reference(dwId, pub) != 0 || pub == 0) {
     OpenDDS::DCPS::RepoIdConverter part_converter(partId);
     OpenDDS::DCPS::RepoIdConverter pub_converter(dwId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::update_publication_qos: ")
                ACE_TEXT("participant %C could not find publication %C.\n"),
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
     throw OpenDDS::DCPS::Invalid_Publication();
   }
 
   Update::SpecificQos qosType;
 
   if (pub->set_qos(qos, publisherQos, qosType) == false)  // failed
     return 0;
 
   if (this->um_ && (partPtr->isBitPublisher() == false)) {
     Update::IdPath path(domainId, partId, dwId);
 
     switch (qosType) {
     case Update::DataWriterQos:
       this->um_->update(path, qos);
       break;
 
     case Update::PublisherQos:
       this->um_->update(path, publisherQos);
       break;
 
     case Update::NoQos:
     default:
       break;
     }
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(dwId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::update_publication_qos: ")
                  ACE_TEXT("pushing update of publication %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 
   return 1;
 }

◆ update_publication_qos() [2/3]

void TAO_DDS_DCPSInfo_i::update_publication_qos	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	partId,
		const OpenDDS::DCPS::GUID_t &	dwId,
		const DDS::DataWriterQos &	qos
	)

Entry for federation updates of DataWriterQos values.

Definition at line 1737 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domains_, DCPS_IR_Participant::find_publication_reference(), LM_ERROR, LM_INFO, lock_, and DCPS_IR_Publication::set_qos().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(partId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) updating  publication qos 2\n"));
   }
 
   DCPS_IR_Publication* pub;
 
   if (partPtr->find_publication_reference(dwId, pub) != 0 || pub == 0) {
     OpenDDS::DCPS::RepoIdConverter part_converter(partId);
     OpenDDS::DCPS::RepoIdConverter pub_converter(dwId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::update_publication_qos: ")
                ACE_TEXT("participant %C could not find publication %C.\n"),
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
     throw OpenDDS::DCPS::Invalid_Publication();
   }
 
   pub->set_qos(qos);
 }

◆ update_publication_qos() [3/3]

void TAO_DDS_DCPSInfo_i::update_publication_qos	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	partId,
		const OpenDDS::DCPS::GUID_t &	dwId,
		const DDS::PublisherQos &	qos
	)

Entry for federation updates of PublisherQos values.

Definition at line 1781 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domains_, DCPS_IR_Participant::find_publication_reference(), LM_ERROR, LM_INFO, lock_, and DCPS_IR_Publication::set_qos().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(partId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) updating  publication qos 3\n"));
   }
 
   DCPS_IR_Publication* pub;
 
   if (partPtr->find_publication_reference(dwId, pub) != 0 || pub == 0) {
     OpenDDS::DCPS::RepoIdConverter part_converter(partId);
     OpenDDS::DCPS::RepoIdConverter pub_converter(dwId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::update_publication_qos: ")
                ACE_TEXT("participant %C could not find publication %C.\n"),
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
     throw OpenDDS::DCPS::Invalid_Publication();
   }
 
   pub->set_qos(qos);
 }

◆ update_subscription_params()

CORBA::Boolean TAO_DDS_DCPSInfo_i::update_subscription_params	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const OpenDDS::DCPS::GUID_t &	subscriptionId,
		const DDS::StringSeq &	params
	)

Definition at line 1989 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domain(), domains_, DCPS_IR_Participant::find_subscription_reference(), DCPS_IR_Participant::isBitPublisher(), LM_ERROR, LM_INFO, lock_, um_, Update::Manager::update(), and DCPS_IR_Subscription::update_expr_params().

Referenced by OpenDDS::Federator::ManagerImpl::processUpdateFilterExpressionParams().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, 0);
 
   DCPS_IR_Domain_Map::iterator domain = this->domains_.find(domainId);
   if (domain == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   DCPS_IR_Participant* partPtr = domain->second->participant(participantId);
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) updating subscription params\n"));
   }
 
   DCPS_IR_Subscription* sub;
   if (partPtr->find_subscription_reference(subscriptionId, sub) != 0) {
     OpenDDS::DCPS::RepoIdConverter part_converter(participantId);
     OpenDDS::DCPS::RepoIdConverter sub_converter(subscriptionId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::update_subscription_params: ")
                ACE_TEXT("participant %C could not find subscription %C.\n"),
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
     throw OpenDDS::DCPS::Invalid_Subscription();
   }
 
   sub->update_expr_params(params);  // calls writers via DataWriterRemote
 
   if (this->um_ && !partPtr->isBitPublisher()) {
     Update::IdPath path(domainId, participantId, subscriptionId);
     this->um_->update(path, params);
   }
 
   return true;
 }

◆ update_subscription_qos() [1/3]

CORBA::Boolean TAO_DDS_DCPSInfo_i::update_subscription_qos	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	partId,
		const OpenDDS::DCPS::GUID_t &	drId,
		const DDS::DataReaderQos &	qos,
		const DDS::SubscriberQos &	subscriberQos
	)

virtual

Definition at line 1824 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), Update::DataReaderQos, OpenDDS::DCPS::DCPS_debug_level, domains_, DCPS_IR_Participant::find_subscription_reference(), DCPS_IR_Participant::isBitPublisher(), LM_DEBUG, LM_ERROR, LM_INFO, lock_, Update::NoQos, DCPS_IR_Subscription::set_qos(), Update::SubscriberQos, um_, and Update::Manager::update().

Referenced by OpenDDS::Federator::ManagerImpl::processUpdateQos1(), and OpenDDS::Federator::ManagerImpl::processUpdateQos2().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, 0);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(partId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   DCPS_IR_Subscription* sub;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) updating QOS for subscription 1\n"));
   }
 
   if (partPtr->find_subscription_reference(drId, sub) != 0 || sub == 0) {
     OpenDDS::DCPS::RepoIdConverter part_converter(partId);
     OpenDDS::DCPS::RepoIdConverter sub_converter(drId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::update_subscription_qos: ")
                ACE_TEXT("participant %C could not find subscription %C.\n"),
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
     throw OpenDDS::DCPS::Invalid_Subscription();
   }
 
   Update::SpecificQos qosType;
 
   if (sub->set_qos(qos, subscriberQos, qosType) == false) // failed
     return 0;
 
   if (this->um_ && (partPtr->isBitPublisher() == false)) {
     Update::IdPath path(domainId, partId, drId);
 
     switch (qosType) {
     case Update::DataReaderQos:
       this->um_->update(path, qos);
       break;
 
     case Update::SubscriberQos:
       this->um_->update(path, subscriberQos);
       break;
 
     case Update::NoQos:
     default:
       break;
     }
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(drId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::update_subscription_qos: ")
                  ACE_TEXT("pushing update of subscription %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 
   return 1;
 }

◆ update_subscription_qos() [2/3]

void TAO_DDS_DCPSInfo_i::update_subscription_qos	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	partId,
		const OpenDDS::DCPS::GUID_t &	drId,
		const DDS::DataReaderQos &	qos
	)

Entry for federation updates of DataReaderQos values.

Definition at line 1901 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domains_, DCPS_IR_Participant::find_subscription_reference(), LM_ERROR, LM_INFO, lock_, and DCPS_IR_Subscription::set_qos().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(partId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   DCPS_IR_Subscription* sub;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) updating QOS for subscription 2\n"));
   }
 
   if (partPtr->find_subscription_reference(drId, sub) != 0 || sub == 0) {
     OpenDDS::DCPS::RepoIdConverter part_converter(partId);
     OpenDDS::DCPS::RepoIdConverter sub_converter(drId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::update_subscription_qos: ")
                ACE_TEXT("participant %C could not find subscription %C.\n"),
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
     throw OpenDDS::DCPS::Invalid_Subscription();
   }
 
   sub->set_qos(qos);
 }

◆ update_subscription_qos() [3/3]

void TAO_DDS_DCPSInfo_i::update_subscription_qos	(	DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	partId,
		const OpenDDS::DCPS::GUID_t &	drId,
		const DDS::SubscriberQos &	qos
	)

Entry for federation updates of SubscriberQos values.

Definition at line 1945 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domains_, DCPS_IR_Participant::find_subscription_reference(), LM_ERROR, LM_INFO, lock_, and DCPS_IR_Subscription::set_qos().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(partId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   DCPS_IR_Subscription* sub;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 3) {
     ACE_DEBUG((LM_INFO, "(%P|%t) updating QOS for subscription 3\n"));
   }
 
   if (partPtr->find_subscription_reference(drId, sub) != 0 || sub == 0) {
     OpenDDS::DCPS::RepoIdConverter part_converter(partId);
     OpenDDS::DCPS::RepoIdConverter sub_converter(drId);
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: TAO_DDS_DCPSInfo_i::update_subscription_qos: ")
                ACE_TEXT("participant %C could not find subscription %C.\n"),
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
     throw OpenDDS::DCPS::Invalid_Subscription();
   }
 
   sub->set_qos(qos);
 }

◆ update_topic_qos()

CORBA::Boolean TAO_DDS_DCPSInfo_i::update_topic_qos	(	const OpenDDS::DCPS::GUID_t &	topicId,
		DDS::DomainId_t	domainId,
		const OpenDDS::DCPS::GUID_t &	participantId,
		const DDS::TopicQos &	qos
	)

virtual

Definition at line 2033 of file DCPSInfo_i.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, domains_, DCPS_IR_Participant::find_topic_reference(), DCPS_IR_Participant::isBitPublisher(), DCPS_IR_Participant::isOwner(), LM_DEBUG, lock_, DCPS_IR_Topic::set_topic_qos(), um_, and Update::Manager::update().

Referenced by OpenDDS::Federator::ManagerImpl::processUpdateQos1().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, 0);
 
   // Grab the domain.
   DCPS_IR_Domain_Map::iterator where = this->domains_.find(domainId);
 
   if (where == this->domains_.end()) {
     throw OpenDDS::DCPS::Invalid_Domain();
   }
 
   // Grab the participant.
   DCPS_IR_Participant* partPtr
   = where->second->participant(participantId);
 
   if (0 == partPtr) {
     throw OpenDDS::DCPS::Invalid_Participant();
   }
 
   DCPS_IR_Topic* topic;
 
   if (partPtr->find_topic_reference(topicId, topic) != 0) {
     throw OpenDDS::DCPS::Invalid_Topic();
   }
 
   if (topic->set_topic_qos(qos) == false)
     return 0;
 
   if (this->um_
       && (partPtr->isOwner() == true)
       && (partPtr->isBitPublisher() == false)) {
     Update::IdPath path(domainId, participantId, topicId);
     this->um_->update(path, qos);
 
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       OpenDDS::DCPS::RepoIdConverter converter(topicId);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) TAO_DDS_DCPSInfo_i::update_topic_qos: ")
                  ACE_TEXT("pushing update of topic %C in domain %d.\n"),
                  std::string(converter).c_str(),
                  domainId));
     }
   }
 
   return 1;
 }

Member Data Documentation

◆ dispatch_check_timer_id_

long TAO_DDS_DCPSInfo_i::dispatch_check_timer_id_

private

Definition at line 430 of file DCPSInfo_i.h.

Referenced by finalize(), and init_dispatchChecking().

◆ dispatchingOrb_

CORBA::ORB_var TAO_DDS_DCPSInfo_i::dispatchingOrb_

private

Definition at line 416 of file DCPSInfo_i.h.

Referenced by add_publication(), add_subscription(), handle_timeout(), and TAO_DDS_DCPSInfo_i().

◆ domains_

◆ federation_

const TAO_DDS_DCPSFederationId& TAO_DDS_DCPSInfo_i::federation_

private

Definition at line 418 of file DCPSInfo_i.h.

Referenced by add_domain_participant(), add_publication(), add_subscription(), add_topic(), and domain().

◆ in_cleanup_all_built_in_topics_

bool TAO_DDS_DCPSInfo_i::in_cleanup_all_built_in_topics_

private

Definition at line 450 of file DCPSInfo_i.h.

Referenced by cleanup_all_built_in_topics(), remove_domain_participant(), remove_publication(), and remove_subscription().

◆ lock_

◆ orb_

CORBA::ORB_var TAO_DDS_DCPSInfo_i::orb_

private

Definition at line 415 of file DCPSInfo_i.h.

Referenced by add_publication(), add_subscription(), finalize(), init_dispatchChecking(), init_reassociation(), and orb().

◆ participantIdGenerator_

OpenDDS::DCPS::RepoIdGenerator TAO_DDS_DCPSInfo_i::participantIdGenerator_

private

Definition at line 419 of file DCPSInfo_i.h.

Referenced by domain(), and receive_image().

◆ reassociate_timer_id_

long TAO_DDS_DCPSInfo_i::reassociate_timer_id_

private

Definition at line 429 of file DCPSInfo_i.h.

Referenced by finalize(), and init_reassociation().

◆ reincarnate_

bool TAO_DDS_DCPSInfo_i::reincarnate_

private

Definition at line 422 of file DCPSInfo_i.h.

Referenced by domain(), and init_persistence().

◆ shutdown_

ShutdownInterface* TAO_DDS_DCPSInfo_i::shutdown_

private

Interface to effect shutdown of the process.

Definition at line 425 of file DCPSInfo_i.h.

Referenced by shutdown().

◆ um_

Update::Manager* TAO_DDS_DCPSInfo_i::um_

private

Definition at line 421 of file DCPSInfo_i.h.

Referenced by add(), add_domain_participant(), add_publication(), add_subscription(), assert_topic(), init_persistence(), remove_domain_participant(), remove_publication(), remove_subscription(), remove_topic(), update_domain_participant_qos(), update_publication_qos(), update_subscription_params(), update_subscription_qos(), and update_topic_qos().

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

Classes

Public Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ TAO_DDS_DCPSInfo_i()

◆ ~TAO_DDS_DCPSInfo_i()

Member Function Documentation

◆ add()

◆ add_domain_participant() [1/2]

◆ add_domain_participant() [2/2]

◆ add_publication() [1/2]

◆ add_publication() [2/2]

◆ add_subscription() [1/2]

◆ add_subscription() [2/2]

◆ add_topic()

◆ assert_topic()

◆ attach_participant()

◆ changeOwnership()

◆ cleanup_all_built_in_topics()

◆ disassociate_participant()

◆ disassociate_publication()

◆ disassociate_subscription()

◆ domain()

◆ domains()

◆ dump_to_string()

◆ finalize()

◆ find_topic()

◆ handle_timeout()

◆ ignore_domain_participant()

◆ ignore_publication()

◆ ignore_subscription()

◆ ignore_topic()

◆ init_dispatchChecking()

◆ init_persistence()

◆ init_reassociation()

◆ init_transport()

◆ orb()

◆ receive_image()

◆ remove_by_owner()

◆ remove_domain_participant()

◆ remove_publication()

◆ remove_subscription()

◆ remove_topic()

◆ shutdown()

◆ update_domain_participant_qos()

◆ update_publication_qos() [1/3]

◆ update_publication_qos() [2/3]

◆ update_publication_qos() [3/3]

◆ update_subscription_params()

◆ update_subscription_qos() [1/3]

◆ update_subscription_qos() [2/3]

◆ update_subscription_qos() [3/3]

◆ update_topic_qos()

Member Data Documentation

◆ dispatch_check_timer_id_

◆ dispatchingOrb_

◆ domains_

◆ federation_

◆ in_cleanup_all_built_in_topics_

◆ lock_

◆ orb_

◆ participantIdGenerator_

◆ reassociate_timer_id_

◆ reincarnate_

◆ shutdown_

◆ um_