#include <FederatorManagerImpl.h>

Inheritance diagram for OpenDDS::Federator::ManagerImpl:

[legend]

Collaboration diagram for OpenDDS::Federator::ManagerImpl:

[legend]

Public Member Functions
	ManagerImpl (Config &config)

virtual	~ManagerImpl ()

virtual CORBA::Boolean	discover_federation (const char *ior)

virtual Manager_ptr	join_federation (Manager_ptr peer, FederationDomain federation)

virtual void	leave_federation (RepoKey id)

virtual RepoKey	federation_id ()

virtual OpenDDS::DCPS::DCPSInfo_ptr	repository ()

virtual void	initializeOwner (const OpenDDS::Federator::OwnerUpdate &data)

virtual void	initializeTopic (const OpenDDS::Federator::TopicUpdate &data)

virtual void	initializeParticipant (const OpenDDS::Federator::ParticipantUpdate &data)

virtual void	initializePublication (const OpenDDS::Federator::PublicationUpdate &data)

virtual void	initializeSubscription (const OpenDDS::Federator::SubscriptionUpdate &data)

virtual void	leave_and_shutdown ()

virtual void	shutdown ()

void	initialize ()
	Establish the update publications and subscriptions. More...

void	finalize ()
	Release resources gracefully. More...

TAO_DDS_DCPSInfo_i *&	info ()
	Accessors for the DCPSInfo reference. More...

TAO_DDS_DCPSInfo_i *	info () const

void	localRepo (::OpenDDS::DCPS::DCPSInfo_ptr repo)
	Capture a remote callable reference to the DCPSInfo. More...

const TAO_DDS_DCPSFederationId &	id () const
	Accessors for the federation Id value. More...

CORBA::ORB_ptr	orb ()
	Accessors for the ORB. More...

void	orb (CORBA::ORB_ptr value)

void	pushState (Manager_ptr peer)
	Push our current state to a remote repository. More...

void	processDeferred ()
	Handle any deferred updates that might have become processable. More...

virtual void	unregisterCallback ()

virtual void	requestImage ()

virtual void	create (const Update::UTopic &topic)

virtual void	create (const Update::UParticipant &participant)

virtual void	create (const Update::URActor &reader)

virtual void	create (const Update::UWActor &writer)

virtual void	create (const Update::OwnershipData &data)

virtual void	update (const Update::IdPath &id, const DDS::DomainParticipantQos &qos)

virtual void	update (const Update::IdPath &id, const DDS::TopicQos &qos)

virtual void	update (const Update::IdPath &id, const DDS::DataWriterQos &qos)

virtual void	update (const Update::IdPath &id, const DDS::PublisherQos &qos)

virtual void	update (const Update::IdPath &id, const DDS::DataReaderQos &qos)

virtual void	update (const Update::IdPath &id, const DDS::SubscriberQos &qos)

virtual void	update (const Update::IdPath &id, const DDS::StringSeq &exprParams)

virtual void	destroy (const Update::IdPath &id, Update::ItemType type, Update::ActorType actor)
	Propagate that an entity has been destroyed. More...

void	processCreate (const OwnerUpdate sample, const DDS::SampleInfo info)
	Null implementation for OwnerUpdate samples. More...

void	processCreate (const PublicationUpdate sample, const DDS::SampleInfo info)
	Create a proxy for a new publication. More...

void	processCreate (const SubscriptionUpdate sample, const DDS::SampleInfo info)
	Create a proxy for a new subscription. More...

void	processCreate (const ParticipantUpdate sample, const DDS::SampleInfo info)
	Create a proxy for a new participant. More...

void	processCreate (const TopicUpdate sample, const DDS::SampleInfo info)
	Create a proxy for a new topic. More...

void	processUpdateQos1 (const OwnerUpdate sample, const DDS::SampleInfo info)
	Process ownership changes. More...

void	processUpdateQos1 (const PublicationUpdate sample, const DDS::SampleInfo info)
	Update the proxy DataWriterQos for a publication. More...

void	processUpdateQos2 (const PublicationUpdate sample, const DDS::SampleInfo info)
	Update the proxy PublisherQos for a publication. More...

void	processUpdateQos1 (const SubscriptionUpdate sample, const DDS::SampleInfo info)
	Update the proxy DataReaderQos for a subscription. More...

void	processUpdateQos2 (const SubscriptionUpdate sample, const DDS::SampleInfo info)
	Update the proxy SubscriberQos for a subscription. More...

void	processUpdateFilterExpressionParams (const SubscriptionUpdate sample, const DDS::SampleInfo info)
	Update the proxy filter expression params for a subscription. More...

void	processUpdateQos1 (const ParticipantUpdate sample, const DDS::SampleInfo info)
	Update the proxy ParticipantQos for a participant. More...

void	processUpdateQos1 (const TopicUpdate sample, const DDS::SampleInfo info)
	Update the proxy TopicQos for a topic. More...

void	processDelete (const OwnerUpdate sample, const DDS::SampleInfo info)
	Null implementation for OwnerUpdate samples. More...

void	processDelete (const PublicationUpdate sample, const DDS::SampleInfo info)
	Delete a proxy for a publication. More...

void	processDelete (const SubscriptionUpdate sample, const DDS::SampleInfo info)
	Delete a proxy for a subscription. More...

void	processDelete (const ParticipantUpdate sample, const DDS::SampleInfo info)
	Delete a proxy for a participant. More...

void	processDelete (const TopicUpdate sample, const DDS::SampleInfo info)
	Delete a proxy for a topic. More...

Public Member Functions inherited from Update::Updater
virtual	~Updater ()

virtual void	updateLastPartId (PartIdType partId)
	Update Last Participant Id for the repo. More...

Public Member Functions inherited from OpenDDS::Federator::UpdateProcessor< OwnerUpdate >
	UpdateProcessor ()

virtual	~UpdateProcessor ()

virtual void	processUpdateQos2 (const OwnerUpdate sample, const DDS::SampleInfo info)

virtual void	processUpdateFilterExpressionParams (const OwnerUpdate sample, const DDS::SampleInfo info)
	A default null implementation is provided. More...

void	processSample (const OwnerUpdate sample, const DDS::SampleInfo info)
	Update publication information with sample data. More...

Public Member Functions inherited from OpenDDS::Federator::UpdateProcessor< TopicUpdate >
	UpdateProcessor ()

virtual	~UpdateProcessor ()

virtual void	processUpdateQos2 (const TopicUpdate sample, const DDS::SampleInfo info)

virtual void	processUpdateFilterExpressionParams (const TopicUpdate sample, const DDS::SampleInfo info)
	A default null implementation is provided. More...

void	processSample (const TopicUpdate sample, const DDS::SampleInfo info)
	Update publication information with sample data. More...

Public Member Functions inherited from OpenDDS::Federator::UpdateProcessor< ParticipantUpdate >
	UpdateProcessor ()

virtual	~UpdateProcessor ()

virtual void	processUpdateQos2 (const ParticipantUpdate sample, const DDS::SampleInfo info)

virtual void	processUpdateFilterExpressionParams (const ParticipantUpdate sample, const DDS::SampleInfo info)
	A default null implementation is provided. More...

void	processSample (const ParticipantUpdate sample, const DDS::SampleInfo info)
	Update publication information with sample data. More...

Public Member Functions inherited from OpenDDS::Federator::UpdateProcessor< SubscriptionUpdate >
	UpdateProcessor ()

virtual	~UpdateProcessor ()

void	processSample (const SubscriptionUpdate sample, const DDS::SampleInfo info)
	Update publication information with sample data. More...

Public Member Functions inherited from OpenDDS::Federator::UpdateProcessor< PublicationUpdate >
	UpdateProcessor ()

virtual	~UpdateProcessor ()

virtual void	processUpdateFilterExpressionParams (const PublicationUpdate sample, const DDS::SampleInfo info)
	A default null implementation is provided. More...

void	processSample (const PublicationUpdate sample, const DDS::SampleInfo info)
	Update publication information with sample data. More...

Private Types
typedef std::map< RepoKey, Manager_var >	IdToManagerMap
	Map type to hold references to federated repository Managers. More...

Private Attributes
ACE_SYNCH_MUTEX	lock_
	Critical section MUTEX. More...

ACE_Condition< ACE_SYNCH_MUTEX >	joining_
	Condition used to gate joining activities. More...

RepoKey	joiner_
	Simple recursion avoidance during the join operations. More...

RepoKey	joinRepo_
	Repository to which we joined. More...

bool	federated_

IdToManagerMap	peers_
	The peer with which we have federated. More...

OpenDDS::DCPS::SequenceNumber	sequence_
	The packet sequence number for data that we publish. More...

Config &	config_
	The configuration information for this manager. More...

TAO_DDS_DCPSInfo_i *	info_
	The Info object reference to update. More...

OpenDDS::DCPS::DCPSInfo_var	localRepo_
	Remotely callable reference to the local repository. More...

CORBA::ORB_var	orb_
	The ORB in which we are activated. More...

InfoRepoMulticastResponder	multicastResponder_
	Multicast responder. More...

DDS::DomainParticipant_var	federationParticipant_
	local DomainParticipant More...

UpdateListener< OwnerUpdate, OwnerUpdateDataReader >	ownerListener_
	TopicUpdate listener. More...

UpdateListener< TopicUpdate, TopicUpdateDataReader >	topicListener_
	TopicUpdate listener. More...

UpdateListener< ParticipantUpdate, ParticipantUpdateDataReader >	participantListener_
	ParticipantUpdate listener. More...

UpdateListener< PublicationUpdate, PublicationUpdateDataReader >	publicationListener_
	PublicationUpdate listener. More...

UpdateListener< SubscriptionUpdate, SubscriptionUpdateDataReader >	subscriptionListener_
	SubscriptionUpdate listener. More...

OwnerUpdateDataWriter_var	ownerWriter_
	TopicUpdate writer. More...

TopicUpdateDataWriter_var	topicWriter_
	TopicUpdate writer. More...

ParticipantUpdateDataWriter_var	participantWriter_
	ParticipantUpdate writer. More...

PublicationUpdateDataWriter_var	publicationWriter_
	PublicationUpdate writer. More...

SubscriptionUpdateDataWriter_var	subscriptionWriter_
	SubscriptionUpdate writer. More...

std::list< OwnerUpdate >	deferredOwnerships_
	Deferred ownership updates. More...

std::list< TopicUpdate >	deferredTopics_
	Deferred topic updates. More...

std::list< PublicationUpdate >	deferredPublications_
	Deferred publication updates. More...

std::list< SubscriptionUpdate >	deferredSubscriptions_
	Deferred subscription updates. More...

bool	multicastEnabled_
	Is multicast enabled? More...

ACE_Thread_Mutex	deferred_lock_
	Protect deferred updates. More...

Detailed Description

Definition at line 36 of file FederatorManagerImpl.h.

Member Typedef Documentation

◆ IdToManagerMap

typedef std::map<RepoKey, Manager_var> OpenDDS::Federator::ManagerImpl::IdToManagerMap

private

Map type to hold references to federated repository Managers.

Definition at line 222 of file FederatorManagerImpl.h.

Constructor & Destructor Documentation

◆ ManagerImpl()

OpenDDS::Federator::ManagerImpl::ManagerImpl ( Config & config )

Definition at line 42 of file FederatorManagerImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, ACE_OS::getenv(), LM_DEBUG, and multicastEnabled_.

   : joining_(this->lock_),
     joiner_(NIL_REPOSITORY),
     joinRepo_(NIL_REPOSITORY),
     federated_(false),
     config_(config),
     info_(0),
     ownerListener_(*this),
     topicListener_(*this),
     participantListener_(*this),
     publicationListener_(*this),
     subscriptionListener_(*this),
     multicastEnabled_(false)
 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::ManagerImpl()\n")));
   }
 
   char* mdec = ACE_OS::getenv("MulticastDiscoveryEnabled");
 
   if (mdec != 0) {
     std::string mde(ACE_OS::getenv("MulticastDiscoveryEnabled"));
 
     if (mde != "0") {
       multicastEnabled_ = true;
     }
   }
 }

◆ ~ManagerImpl()

OpenDDS::Federator::ManagerImpl::~ManagerImpl ( )

virtual

Definition at line 72 of file FederatorManagerImpl.cpp.

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

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::~ManagerImpl()\n")));
   }
 }

Member Function Documentation

◆ create() [1/5]

void OpenDDS::Federator::ManagerImpl::create ( const Update::UTopic & topic )

virtual

Propagate an entity has been created.

Implements Update::Updater.

Definition at line 40 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::TopicUpdate::action, OpenDDS::Federator::CreateEntity, Update::TopicStrt< Q, S >::dataType, OpenDDS::Federator::TopicUpdate::datatype, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::TopicUpdate::domain, Update::TopicStrt< Q, S >::domainId, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), OpenDDS::Federator::TopicUpdate::id, id(), CORBA::is_nil(), LM_DEBUG, Update::TopicStrt< Q, S >::name, OpenDDS::Federator::TopicUpdate::participant, Update::TopicStrt< Q, S >::participantId, OpenDDS::Federator::TopicUpdate::qos, OpenDDS::Federator::TopicUpdate::sender, OpenDDS::Federator::TopicUpdate::topic, Update::TopicStrt< Q, S >::topicId, Update::TopicStrt< Q, S >::topicQos, and topicWriter_.

Referenced by create().

 {
   if (CORBA::is_nil(this->topicWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   TopicUpdate sample = TopicUpdate();
   sample.sender      = this->id().id();
   sample.action      = CreateEntity;
 
   sample.id          = topic.topicId;
   sample.domain      = topic.domainId;
   sample.participant = topic.participantId;
   sample.topic       = topic.name.c_str();
   sample.datatype    = topic.dataType.c_str();
   sample.qos         = topic.topicQos;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
     OpenDDS::DCPS::RepoIdConverter topic_converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::create( TopicUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ topic %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(part_converter).c_str(),
                std::string(topic_converter).c_str()));
   }
 
   this->topicWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ create() [2/5]

void OpenDDS::Federator::ManagerImpl::create ( const Update::UParticipant & participant )

virtual

Propagate an entity has been created.

Implements Update::Updater.

Definition at line 74 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::ParticipantUpdate::action, OpenDDS::Federator::CreateEntity, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::ParticipantUpdate::domain, Update::ParticipantStrt< Q >::domainId, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), OpenDDS::Federator::ParticipantUpdate::id, id(), CORBA::is_nil(), LM_DEBUG, OpenDDS::Federator::ParticipantUpdate::owner, Update::ParticipantStrt< Q >::owner, Update::ParticipantStrt< Q >::participantId, Update::ParticipantStrt< Q >::participantQos, participantWriter_, OpenDDS::Federator::ParticipantUpdate::qos, and OpenDDS::Federator::ParticipantUpdate::sender.

 {
   if (CORBA::is_nil(this->participantWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   ParticipantUpdate sample = ParticipantUpdate();
   sample.sender = this->id().id();
   sample.action = CreateEntity;
 
   sample.owner  = participant.owner;
   sample.domain = participant.domainId;
   sample.id     = participant.participantId;
   sample.qos    = participant.participantQos;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::create( ParticipantUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(converter).c_str()));
   }
 
   this->participantWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ create() [3/5]

void OpenDDS::Federator::ManagerImpl::create ( const Update::URActor & actor )

virtual

Propagate an entity has been created.

Implements Update::Updater.

Definition at line 104 of file FederatorManagerImpl_updates.cpp.

 {
   if (CORBA::is_nil(this->subscriptionWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   SubscriptionUpdate sample = SubscriptionUpdate();
   sample.sender         = this->id().id();
   sample.action         = CreateEntity;
 
   sample.domain         = reader.domainId;
   sample.participant    = reader.participantId;
   sample.topic          = reader.topicId;
   sample.id             = reader.actorId;
   sample.callback       = reader.callback.c_str();
   sample.datareader_qos = reader.drdwQos;
   sample.subscriber_qos = reader.pubsubQos;
   sample.transport_info = reader.transportInterfaceInfo;
   sample.filter_class_name = reader.contentSubscriptionProfile.filterClassName;
   sample.filter_expression = reader.contentSubscriptionProfile.filterExpr;
   sample.expression_params = reader.contentSubscriptionProfile.exprParams;
   sample.serialized_type_info = reader.serializedTypeInfo;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
     OpenDDS::DCPS::RepoIdConverter sub_converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::create( SubscriptionUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
   }
 
   this->subscriptionWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ create() [4/5]

virtual void OpenDDS::Federator::ManagerImpl::create ( const Update::UWActor & actor )

virtual

Propagate an entity has been created.

Implements Update::Updater.

◆ create() [5/5]

void OpenDDS::Federator::ManagerImpl::create ( const Update::OwnershipData & data )

virtual

Propagate an entity has been created.

Implements Update::Updater.

Definition at line 181 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::OwnerUpdate::action, OpenDDS::Federator::CreateEntity, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::OwnerUpdate::domain, Update::OwnershipData::domain, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), id(), CORBA::is_nil(), LM_DEBUG, OpenDDS::Federator::OwnerUpdate::owner, Update::OwnershipData::owner, ownerWriter_, OpenDDS::Federator::OwnerUpdate::participant, Update::OwnershipData::participant, and OpenDDS::Federator::OwnerUpdate::sender.

 {
   if (CORBA::is_nil(this->ownerWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   OwnerUpdate sample = OwnerUpdate();
   sample.sender      = this->id().id();
   sample.action      = CreateEntity;
 
   sample.domain      = data.domain;
   sample.participant = data.participant;
   sample.owner       = data.owner;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter converter(sample.participant);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::create( OwnerUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ sender %d/ owner %d ]\n"),
                this->id().id(),
                sample.domain,
                std::string(converter).c_str(),
                sample.sender,
                sample.owner));
   }
 
   this->ownerWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ destroy()

void OpenDDS::Federator::ManagerImpl::destroy	(	const Update::IdPath &	id,
		Update::ItemType	type,
		Update::ActorType	actor
	)

virtual

Propagate that an entity has been destroyed.

Implements Update::Updater.

Definition at line 212 of file FederatorManagerImpl_updates.cpp.

 {
   //
   // Do not propagate any destroy() messages within the FederationDomain.
   // This domain will be managed separately.
   //
   if (id.domain == this->config_.federationDomain()) {
     return;
   }
 
   switch (type) {
   case Update::Topic: {
     if (CORBA::is_nil(this->topicWriter_.in())) {
       // Decline to publish data until we can.
       return;
     }
 
     TopicUpdate sample = TopicUpdate();
     sample.sender      = this->id().id();
     sample.action      = DestroyEntity;
 
     sample.id          = id.id;
     sample.domain      = id.domain;
     sample.participant = id.participant;
 
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
       OpenDDS::DCPS::RepoIdConverter topic_converter(sample.id);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::destroy( TopicUpdate): ")
                  ACE_TEXT("repo %d - [ domain %d/ participant %C/ topic %C ]\n"),
                  this->id().id(),
                  sample.domain,
                  std::string(part_converter).c_str(),
                  std::string(topic_converter).c_str()));
     }
 
     this->topicWriter_->write(sample, DDS::HANDLE_NIL);
   }
   break;
 
   case Update::Participant: {
     if (CORBA::is_nil(this->participantWriter_.in())) {
       // Decline to publish data until we can.
       return;
     }
 
     ParticipantUpdate sample = ParticipantUpdate();
     sample.sender = this->id().id();
     sample.action = DestroyEntity;
 
     sample.domain = id.domain;
     sample.id     = id.id;
 
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       OpenDDS::DCPS::RepoIdConverter converter(sample.id);
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::destroy( ParticipantUpdate): ")
                  ACE_TEXT("repo %d - [ domain %d/ participant %C ]\n"),
                  this->id().id(),
                  sample.domain,
                  std::string(converter).c_str()));
     }
 
     this->participantWriter_->write(sample, DDS::HANDLE_NIL);
   }
   break;
 
   case Update::Actor:
 
     // This is VERY annoying.
     switch (actor) {
     case Update::DataWriter: {
       if (CORBA::is_nil(this->publicationWriter_.in())) {
         // Decline to publish data until we can.
         return;
       }
 
       PublicationUpdate sample = PublicationUpdate();
       sample.sender         = this->id().id();
       sample.action         = DestroyEntity;
 
       sample.domain         = id.domain;
       sample.participant    = id.participant;
       sample.id             = id.id;
 
       if (OpenDDS::DCPS::DCPS_debug_level > 9) {
         OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
         OpenDDS::DCPS::RepoIdConverter pub_converter(sample.id);
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) Federator::ManagerImpl::destroy( PublicationUpdate): ")
                    ACE_TEXT("repo %d - [ domain %d/ participant %C/ publication %C ]\n"),
                    this->id().id(),
                    sample.domain,
                    std::string(part_converter).c_str(),
                    std::string(pub_converter).c_str()));
       }
 
       this->publicationWriter_->write(sample, DDS::HANDLE_NIL);
     }
     break;
 
     case Update::DataReader: {
       if (CORBA::is_nil(this->subscriptionWriter_.in())) {
         // Decline to publish data until we can.
         return;
       }
 
       SubscriptionUpdate sample = SubscriptionUpdate();
       sample.sender         = this->id().id();
       sample.action         = DestroyEntity;
 
       sample.domain         = id.domain;
       sample.participant    = id.participant;
       sample.id             = id.id;
 
       if (OpenDDS::DCPS::DCPS_debug_level > 9) {
         OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
         OpenDDS::DCPS::RepoIdConverter sub_converter(sample.id);
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) Federator::ManagerImpl::destroy( SubscriptionUpdate): ")
                    ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                    this->id().id(),
                    sample.domain,
                    std::string(part_converter).c_str(),
                    std::string(sub_converter).c_str()));
       }
 
       this->subscriptionWriter_->write(sample, DDS::HANDLE_NIL);
     }
     break;
     }
 
     break;
   }
 }

◆ discover_federation()

CORBA::Boolean OpenDDS::Federator::ManagerImpl::discover_federation ( const char * ior )

virtual

: Implement this.

Definition at line 907 of file FederatorManagerImpl.cpp.

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

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) ManagerImpl::discover_federation( %C)\n"),
                ior));
   }
 
   ///@TODO: Implement this.
   return false;
 }

◆ federation_id()

RepoKey OpenDDS::Federator::ManagerImpl::federation_id ( )

virtual

Definition at line 870 of file FederatorManagerImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, TAO_DDS_DCPSFederationId::id(), id(), and LM_DEBUG.

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) ManagerImpl::federation_id()\n")));
   }
 
   return this->id().id();
 }

◆ finalize()

void OpenDDS::Federator::ManagerImpl::finalize ( void )

Release resources gracefully.

Definition at line 791 of file FederatorManagerImpl.cpp.

References CORBA::Exception::_tao_print_exception(), ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, ACE_Event_Handler::DONT_CALL, federated_, federationParticipant_, TAO_Pseudo_Var_T< ORB >::in(), CORBA::is_nil(), joinRepo_, LM_DEBUG, LM_ERROR, multicastResponder_, orb_, ownerListener_, participantListener_, peers_, publicationListener_, ACE_Event_Handler::READ_MASK, OpenDDS::DCPS::retcode_to_string(), subscriptionListener_, TheParticipantFactory, and topicListener_.

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

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::finalize()\n")));
   }
 
   ownerListener_.stop();
   topicListener_.stop();
   participantListener_.stop();
   publicationListener_.stop();
   subscriptionListener_.stop();
   ownerListener_.join();
   topicListener_.join();
   participantListener_.join();
   publicationListener_.join();
   subscriptionListener_.join();
 
   if (this->federated_) {
     try {
       IdToManagerMap::iterator where = this->peers_.find(this->joinRepo_);
 
       if (where == this->peers_.end()) {
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) Federator::Manager::finalize: ")
                    ACE_TEXT("repository %d - all attachment to federation left.\n"),
                    this->id().id()));
 
       } else {
         if (CORBA::is_nil(where->second.in())) {
           ACE_ERROR((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: Federator::Manager::finalize: ")
                      ACE_TEXT("repository %d not currently attached to a federation.\n"),
                      this->id().id()));
 
         } else {
           where->second->leave_federation(this->id().id());
           this->federated_ = false;
         }
       }
 
     } catch (const CORBA::Exception& ex) {
       ex._tao_print_exception(
         ACE_TEXT("ERROR: Federator::ManagerImpl::finalize() - ")
         ACE_TEXT("unable to leave remote federation "));
       throw Incomplete();
     }
   }
 
   if (!CORBA::is_nil(this->orb_.in()) && (0 != this->orb_->orb_core())) {
     this->orb_->orb_core()->reactor()->remove_handler(
       &this->multicastResponder_,
       ACE_Event_Handler::READ_MASK | ACE_Event_Handler::DONT_CALL);
   }
 
   // Remove our local participant and contained entities.
   if (federationParticipant_) {
     const DDS::ReturnCode_t entities_error =
       federationParticipant_->delete_contained_entities();
     if (entities_error) {
       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: Federator::Manager: "
                  "unable to release resources for repository %d: %C\n",
                  id().id(), DCPS::retcode_to_string(entities_error)));
       return;
     }
 
     DDS::DomainParticipantFactory_var dpf = TheParticipantFactory;
     const DDS::ReturnCode_t part_error = dpf->delete_participant(federationParticipant_);
     if (part_error) {
       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: Federator::Manager: "
                  "unable to release the participant for repository %d: %C\n",
                  id().id(), DCPS::retcode_to_string(part_error)));
     }
   }
 }

◆ id()

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL ACE_INLINE const TAO_DDS_DCPSFederationId & OpenDDS::Federator::ManagerImpl::id ( void ) const

Accessors for the federation Id value.

Definition at line 12 of file FederatorManagerImpl.inl.

References ACE_INLINE, config_, and OpenDDS::Federator::Config::federationId().

Referenced by create(), destroy(), federation_id(), InfoRepo::init(), leave_federation(), pushState(), and update().

 {
   return this->config_.federationId();
 }

◆ info() [1/2]

ACE_INLINE TAO_DDS_DCPSInfo_i *& OpenDDS::Federator::ManagerImpl::info ( )

Accessors for the DCPSInfo reference.

Definition at line 19 of file FederatorManagerImpl.inl.

References ACE_INLINE, and info_.

Referenced by InfoRepo::init().

 {
   return this->info_;
 }

◆ info() [2/2]

ACE_INLINE TAO_DDS_DCPSInfo_i * OpenDDS::Federator::ManagerImpl::info ( ) const

Definition at line 26 of file FederatorManagerImpl.inl.

References ACE_INLINE, and info_.

 {
   return this->info_;
 }

◆ initialize()

void OpenDDS::Federator::ManagerImpl::initialize ( void )

Establish the update publications and subscriptions.

Definition at line 81 of file FederatorManagerImpl.cpp.

Referenced by join_federation().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federation::ManagerImpl::initialize()\n")));
   }
 
   // Let the listeners know which repository we are to filter samples at
   // the earliest opportunity.
   this->ownerListener_.federationId(this->id());
   this->topicListener_.federationId(this->id());
   this->participantListener_.federationId(this->id());
   this->publicationListener_.federationId(this->id());
   this->subscriptionListener_.federationId(this->id());
 
   // Add participant for Federation domain
   DDS::DomainParticipantFactory_var dpf = TheParticipantFactory;
   federationParticipant_ = dpf->create_participant(
       this->config_.federationDomain(),
       PARTICIPANT_QOS_DEFAULT,
       DDS::DomainParticipantListener::_nil(),
       OpenDDS::DCPS::DEFAULT_STATUS_MASK);
   if (!federationParticipant_) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: create_participant failed for ")
                ACE_TEXT("repository %d in federation domain %d.\n"),
                this->id().id(),
                this->config_.federationDomain()));
     throw Incomplete();
   }
 
   //
   // Add type support for update topics
   //
 
   OwnerUpdateTypeSupportImpl::_var_type ownerUpdate = new OwnerUpdateTypeSupportImpl();
 
   if (DDS::RETCODE_OK != ownerUpdate->register_type(
         this->federationParticipant_.in(),
         OWNERUPDATETYPENAME)) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Unable to install ")
                ACE_TEXT("OwnerUpdate type support for repository %d.\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   ParticipantUpdateTypeSupportImpl::_var_type participantUpdate = new ParticipantUpdateTypeSupportImpl();
 
   if (DDS::RETCODE_OK != participantUpdate->register_type(
         this->federationParticipant_.in(),
         PARTICIPANTUPDATETYPENAME)) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Unable to install ")
                ACE_TEXT("ParticipantUpdate type support for repository %d.\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   TopicUpdateTypeSupportImpl::_var_type topicUpdate = new TopicUpdateTypeSupportImpl();
 
   if (DDS::RETCODE_OK != topicUpdate->register_type(
         this->federationParticipant_.in(),
         TOPICUPDATETYPENAME)) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Unable to install ")
                ACE_TEXT("TopicUpdate type support for repository %d.\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   PublicationUpdateTypeSupportImpl::_var_type publicationUpdate = new PublicationUpdateTypeSupportImpl();
 
   if (DDS::RETCODE_OK != publicationUpdate->register_type(
         this->federationParticipant_.in(),
         PUBLICATIONUPDATETYPENAME)) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Unable to install ")
                ACE_TEXT("PublicationUpdate type support for repository %d.\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   SubscriptionUpdateTypeSupportImpl::_var_type subscriptionUpdate = new SubscriptionUpdateTypeSupportImpl();
 
   if (DDS::RETCODE_OK != subscriptionUpdate->register_type(
         this->federationParticipant_.in(),
         SUBSCRIPTIONUPDATETYPENAME)) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Unable to install ")
                ACE_TEXT("SubscriptionUpdate type support for repository %d.\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   //
   // Create a transport config for use with federation entities.
   //
   std::string config_name =
     OpenDDS::DCPS::TransportRegistry::DEFAULT_INST_PREFIX
     + std::string("FederationBITTransportConfig");
   OpenDDS::DCPS::TransportConfig_rch config =
     OpenDDS::DCPS::TransportRegistry::instance()->create_config(config_name);
 
   std::string inst_name = OpenDDS::DCPS::TransportRegistry::DEFAULT_INST_PREFIX
     + std::string("FederationBITTCPTransportInst");
   OpenDDS::DCPS::TransportInst_rch inst =
     OpenDDS::DCPS::TransportRegistry::instance()->create_inst(inst_name,
                                                               "tcp");
   config->instances_.push_back(inst);
 
   //
   // Create the subscriber for the update topics.
   //
 
   DDS::Subscriber_var subscriber
   = this->federationParticipant_->create_subscriber(
       SUBSCRIBER_QOS_DEFAULT,
       DDS::SubscriberListener::_nil(),
       OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(subscriber.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create subscriber for repository %d\n"),
                this->id().id()));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("created federation subscriber for repository %d\n"),
                this->id().id()));
 
   }
 
   // Attach the transport to it.
 
   try {
     OpenDDS::DCPS::TransportRegistry::instance()->bind_config(config,
                                                               subscriber.in());
   } catch (const OpenDDS::DCPS::Transport::Exception&) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to bind transport config to federation subscriber.\n")));
     throw Incomplete();
   }
 
   //
   // Create the publisher for the update topics.
   //
 
   DDS::Publisher_var publisher
   = this->federationParticipant_->create_publisher(
       PUBLISHER_QOS_DEFAULT,
       DDS::PublisherListener::_nil(),
       OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(publisher.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create publisher for repository %d\n"),
                this->id().id()));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("created federation publisher for repository %d\n"),
                this->id().id()));
 
   }
 
   // Attach the transport to it.
 
   try {
     OpenDDS::DCPS::TransportRegistry::instance()->bind_config(config,
                                                               publisher.in());
   } catch (const OpenDDS::DCPS::Transport::Exception&) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to bind transport config to federation publisher.\n")));
     throw Incomplete();
   }
 
   //
   // Some useful items for adding the subscriptions.
   //
   DDS::Topic_var            topic;
   DDS::TopicDescription_var description;
   DDS::DataReader_var       dataReader;
   DDS::DataWriter_var       dataWriter;
 
   DDS::DataReaderQos readerQos;
   subscriber->get_default_datareader_qos(readerQos);
   readerQos.durability.kind                          = DDS::TRANSIENT_LOCAL_DURABILITY_QOS;
   readerQos.history.kind                             = DDS::KEEP_LAST_HISTORY_QOS;
   readerQos.history.depth                            = 50;
   readerQos.reliability.kind                         = DDS::RELIABLE_RELIABILITY_QOS;
   readerQos.reliability.max_blocking_time.sec        = 0;
   readerQos.reliability.max_blocking_time.nanosec    = 0;
 
   DDS::DataWriterQos writerQos;
   publisher->get_default_datawriter_qos(writerQos);
   writerQos.durability.kind                          = DDS::TRANSIENT_LOCAL_DURABILITY_QOS;
   writerQos.history.kind                             = DDS::KEEP_LAST_HISTORY_QOS;
   writerQos.history.depth                            = 50;
   writerQos.reliability.kind                         = DDS::RELIABLE_RELIABILITY_QOS;
   writerQos.reliability.max_blocking_time.sec        = 0;
   writerQos.reliability.max_blocking_time.nanosec    = 0;
 
   //
   // Add update subscriptions
   //
   // NOTE: Its ok to lose the references to the objects here since they
   //       are not needed after this point.  The only thing we will do
   //       with them is to destroy them, and that will be done via a
   //       cascade delete from the participant.  The listeners will
   //       survive and can be used within other participants as well,
   //       since the only state they retain is the manager, which is the
   //       same for all.
   //
 
   topic = this->federationParticipant_->create_topic(
             OWNERUPDATETOPICNAME,
             OWNERUPDATETYPENAME,
             TOPIC_QOS_DEFAULT,
             DDS::TopicListener::_nil(),
             OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   dataWriter = publisher->create_datawriter(
                  topic.in(),
                  writerQos,
                  DDS::DataWriterListener::_nil(),
                  OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataWriter.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create OwnerUpdate writer for repository %d\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   this->ownerWriter_ = OwnerUpdateDataWriter::_narrow(dataWriter.in());
 
   if (::CORBA::is_nil(this->ownerWriter_.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to extract typed OwnerUpdate writer.\n")));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataWriterImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataWriterImpl*>(dataWriter.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed OwnerUpdate writer.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation OwnerUpdate writer %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   description = this->federationParticipant_->lookup_topicdescription(OWNERUPDATETOPICNAME);
   dataReader  = subscriber->create_datareader(
                   description.in(),
                   readerQos,
                   &this->ownerListener_,
                   OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataReader.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create OwnerUpdate reader for repository %d\n"),
                this->id().id()));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataReaderImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataReaderImpl*>(dataReader.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed OwnerUpdate reader.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation OwnerUpdate reader %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   topic = this->federationParticipant_->create_topic(
             TOPICUPDATETOPICNAME,
             TOPICUPDATETYPENAME,
             TOPIC_QOS_DEFAULT,
             DDS::TopicListener::_nil(),
             OpenDDS::DCPS::DEFAULT_STATUS_MASK);
   dataWriter = publisher->create_datawriter(
                  topic.in(),
                  writerQos,
                  DDS::DataWriterListener::_nil(),
                  OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataWriter.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create TopicUpdate writer for repository %d\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   this->topicWriter_
   = TopicUpdateDataWriter::_narrow(dataWriter.in());
 
   if (::CORBA::is_nil(this->topicWriter_.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to extract typed TopicUpdate writer.\n")));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataWriterImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataWriterImpl*>(dataWriter.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed TopicUpdate writer.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation TopicUpdate writer %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   description = this->federationParticipant_->lookup_topicdescription(TOPICUPDATETOPICNAME);
   dataReader  = subscriber->create_datareader(
                   description.in(),
                   readerQos,
                   &this->topicListener_,
                   OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataReader.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create TopicUpdate reader for repository %d\n"),
                this->id().id()));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataReaderImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataReaderImpl*>(dataReader.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed TopicUpdate reader.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation TopicUpdate reader %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   topic = this->federationParticipant_->create_topic(
             PARTICIPANTUPDATETOPICNAME,
             PARTICIPANTUPDATETYPENAME,
             TOPIC_QOS_DEFAULT,
             DDS::TopicListener::_nil(),
             OpenDDS::DCPS::DEFAULT_STATUS_MASK);
   dataWriter = publisher->create_datawriter(
                  topic.in(),
                  writerQos,
                  DDS::DataWriterListener::_nil(),
                  OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataWriter.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create ParticipantUpdate writer for repository %d\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   this->participantWriter_
   = ParticipantUpdateDataWriter::_narrow(dataWriter.in());
 
   if (::CORBA::is_nil(this->participantWriter_.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to extract typed ParticipantUpdate writer.\n")));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataWriterImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataWriterImpl*>(dataWriter.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed ParticipantUpdate writer.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation ParticipantUpdate writer %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   description = this->federationParticipant_->lookup_topicdescription(PARTICIPANTUPDATETOPICNAME);
   dataReader  = subscriber->create_datareader(
                   description.in(),
                   readerQos,
                   &this->participantListener_,
                   OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataReader.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create ParticipantUpdate reader for repository %d\n"),
                this->id().id()));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataReaderImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataReaderImpl*>(dataReader.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed ParticipantUpdate reader.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation ParticipantUpdate reader %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   topic = this->federationParticipant_->create_topic(
             PUBLICATIONUPDATETOPICNAME,
             PUBLICATIONUPDATETYPENAME,
             TOPIC_QOS_DEFAULT,
             DDS::TopicListener::_nil(),
             OpenDDS::DCPS::DEFAULT_STATUS_MASK);
   dataWriter = publisher->create_datawriter(
                  topic.in(),
                  writerQos,
                  DDS::DataWriterListener::_nil(),
                  OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataWriter.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create PublicationUpdate writer for repository %d\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   this->publicationWriter_
   = PublicationUpdateDataWriter::_narrow(dataWriter.in());
 
   if (::CORBA::is_nil(this->publicationWriter_.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to extract typed PublicationUpdate writer.\n")));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataWriterImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataWriterImpl*>(dataWriter.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed PublicationUpdate writer.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation PublicationUpdate writer %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   description = this->federationParticipant_->lookup_topicdescription(PUBLICATIONUPDATETOPICNAME);
   dataReader  = subscriber->create_datareader(
                   description.in(),
                   readerQos,
                   &this->publicationListener_,
                   OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataReader.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create PublicationUpdate reader for repository %d\n"),
                this->id().id()));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataReaderImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataReaderImpl*>(dataReader.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed PublicationUpdate reader.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation PublicationUpdate reader %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   topic = this->federationParticipant_->create_topic(
             SUBSCRIPTIONUPDATETOPICNAME,
             SUBSCRIPTIONUPDATETYPENAME,
             TOPIC_QOS_DEFAULT,
             DDS::TopicListener::_nil(),
             OpenDDS::DCPS::DEFAULT_STATUS_MASK);
   dataWriter = publisher->create_datawriter(
                  topic.in(),
                  writerQos,
                  DDS::DataWriterListener::_nil(),
                  OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataWriter.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create SubscriptionUpdate writer for repository %d\n"),
                this->id().id()));
     throw Incomplete();
   }
 
   this->subscriptionWriter_
   = SubscriptionUpdateDataWriter::_narrow(dataWriter.in());
 
   if (::CORBA::is_nil(this->subscriptionWriter_.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to extract typed SubscriptionUpdate writer.\n")));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataWriterImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataWriterImpl*>(dataWriter.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed SubscriptionUpdate writer.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation SubscriptionUpdate writer %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   description = this->federationParticipant_->lookup_topicdescription(SUBSCRIPTIONUPDATETOPICNAME);
   dataReader  = subscriber->create_datareader(
                   description.in(),
                   readerQos,
                   &this->subscriptionListener_,
                   OpenDDS::DCPS::DEFAULT_STATUS_MASK);
 
   if (CORBA::is_nil(dataReader.in())) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: Federator::ManagerImpl::initialize() - ")
                ACE_TEXT("failed to create SubscriptionUpdate reader for repository %d\n"),
                this->id().id()));
     throw Incomplete();
 
   } else if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     OpenDDS::DCPS::DataReaderImpl* servant
     = dynamic_cast<OpenDDS::DCPS::DataReaderImpl*>(dataReader.in());
 
     if (0 == servant) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("unable to extract typed SubscriptionUpdate reader.\n")));
 
     } else {
       OpenDDS::DCPS::RepoIdConverter converter(servant->get_guid());
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("created federation SubscriptionUpdate reader %C for repository %d\n"),
                  std::string(converter).c_str(),
                  this->id().id()));
     }
   }
 
   // JSP
 #if defined (ACE_HAS_IP_MULTICAST)
 
   if (this->multicastEnabled_) {
     //
     // Install ior multicast handler.
     //
     // Get reactor instance from TAO.
     ACE_Reactor* reactor = this->orb_->orb_core()->reactor();
 
     // See if the -ORBMulticastDiscoveryEndpoint option was specified.
     ACE_CString mde(this->orb_->orb_core()->orb_params()->mcast_discovery_endpoint());
 
     // First, see if the user has given us a multicast port number
     // on the command-line;
     u_short port = 0;
 
     // Check environment var. for multicast port.
     const char *port_number = ACE_OS::getenv("OpenDDSFederationPort");
 
     if (port_number != 0) {
       port = static_cast<u_short>(ACE_OS::atoi(port_number));
     }
 
     // Port wasn't specified on the command-line -
     // use the default.
     if (port == 0)
       port = OpenDDS::Federator::Defaults::DiscoveryRequestPort;
 
     // Initialize the handler
     if (mde.length() != 0) {
       if (this->multicastResponder_.init(
             this->orb_.in(),
             mde.c_str()) == -1) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: Unable to initialize ")
                    ACE_TEXT("the multicast responder for repository %d.\n"),
                    this->id().id()));
         throw Incomplete();
       }
 
     } else {
       if (this->multicastResponder_.init(
             this->orb_.in(),
             port,
 #if defined (ACE_HAS_IPV6)
             ACE_DEFAULT_MULTICASTV6_ADDR
 #else
             ACE_DEFAULT_MULTICAST_ADDR
 #endif /* ACE_HAS_IPV6 */
           )) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: Unable to initialize ")
                    ACE_TEXT("the multicast responder for repository %d.\n"),
                    this->id().id()));
         throw Incomplete();
       }
     }
 
     // Register event handler for the ior multicast.
     if (reactor->register_handler(&this->multicastResponder_,
                                   ACE_Event_Handler::READ_MASK) == -1) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: Unable to register event handler ")
                  ACE_TEXT("for repository %d.\n"),
                  this->id().id()));
       throw Incomplete();
     }
 
     if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::initialize() - ")
                  ACE_TEXT("multicast server setup is complete.\n")));
     }
   }
 
 #else
   ACE_UNUSED_ARG(this->multicastEnabled_);
 #endif /* ACE_HAS_IP_MULTICAST */
 }

◆ initializeOwner()

void OpenDDS::Federator::ManagerImpl::initializeOwner ( const OpenDDS::Federator::OwnerUpdate & data )

virtual

Definition at line 1132 of file FederatorManagerImpl.cpp.

References processCreate().

 {
   this->processCreate(&data, 0);
 }

◆ initializeParticipant()

void OpenDDS::Federator::ManagerImpl::initializeParticipant ( const OpenDDS::Federator::ParticipantUpdate & data )

virtual

Definition at line 1146 of file FederatorManagerImpl.cpp.

References processCreate().

 {
   this->processCreate(&data, 0);
 }

◆ initializePublication()

void OpenDDS::Federator::ManagerImpl::initializePublication ( const OpenDDS::Federator::PublicationUpdate & data )

virtual

Definition at line 1153 of file FederatorManagerImpl.cpp.

References processCreate().

 {
   this->processCreate(&data, 0);
 }

◆ initializeSubscription()

void OpenDDS::Federator::ManagerImpl::initializeSubscription ( const OpenDDS::Federator::SubscriptionUpdate & data )

virtual

Definition at line 1160 of file FederatorManagerImpl.cpp.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL, and processCreate().

 {
   this->processCreate(&data, 0);
 }

◆ initializeTopic()

void OpenDDS::Federator::ManagerImpl::initializeTopic ( const OpenDDS::Federator::TopicUpdate & data )

virtual

Definition at line 1139 of file FederatorManagerImpl.cpp.

References processCreate().

 {
   this->processCreate(&data, 0);
 }

◆ join_federation()

Manager_ptr OpenDDS::Federator::ManagerImpl::join_federation	(	Manager_ptr	peer,
		FederationDomain	federation
	)

virtual

Definition at line 920 of file FederatorManagerImpl.cpp.

References CORBA::Exception::_tao_print_exception(), ACE_DEBUG, ACE_GUARD_RETURN, ACE_SYNCH_MUTEX, ACE_TEXT(), config_, OpenDDS::DCPS::DCPS_debug_level, federated_, OpenDDS::Federator::Config::federationDomain(), TAO::String_var< charT >::in(), initialize(), CORBA::is_nil(), joiner_, joining_, joinRepo_, LM_DEBUG, lock_, OpenDDS::Federator::NIL_REPOSITORY, orb(), participantWriter_, peers_, pushState(), ACE_Condition< MUTEX >::signal(), TheServiceParticipant, and ACE_Condition< MUTEX >::wait().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) ManagerImpl::join_federation( peer, %d)\n"),
                federation));
   }
 
   RepoKey remote = NIL_REPOSITORY;
 
   try {
     // Obtain the remote repository federator Id value.
     remote = peer->federation_id();
 
     if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::join_federation() - ")
                  ACE_TEXT("repo id %d entered from repository with id %d.\n"),
                  this->id().id(),
                  remote));
     }
 
   } catch (const CORBA::Exception& ex) {
     ex._tao_print_exception(
       ACE_TEXT("ERROR: Federator::ManagerImpl::join_federation() - ")
       ACE_TEXT("unable to obtain remote federation Id value: "));
     throw Incomplete();
   }
 
   // If we are recursing, then we are done.
   if (this->joiner_ == remote) {
     if (OpenDDS::DCPS::DCPS_debug_level > 0) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::join_federation() - ")
                  ACE_TEXT("repo id %d leaving after reentry from repository with id %d.\n"),
                  this->id().id(),
                  remote));
     }
 
     return this->_this();
 
   } else {
     // Block while any different repository is joining.
     ACE_GUARD_RETURN(ACE_SYNCH_MUTEX, guard, this->lock_, 0);
 
     while (this->joiner_ != NIL_REPOSITORY) {
       // This releases the lock while we block.
       this->joining_.wait();
 
       // We are now recursing - curses!
       if (this->joiner_ == remote) {
         return this->_this();
       }
     }
 
     // Note that we are joining the remote repository now.
     this->joiner_ = remote;
   }
 
   //
   // We only reach this point if:
   //   1) No other repository is processing past this point;
   //   2) We are not recursing.
   //
 
   // Check if we already have Federation repository.
   // Check if we are already federated.
   if (this->federated_ == false) {
     // Go ahead and add the joining repository as our Federation
     // repository.
     try {
       // Mark this repository as the point to which we are joined to
       // the federation.
       this->joinRepo_ = remote;
 
       // Obtain a reference to the remote repository.
       OpenDDS::DCPS::DCPSInfo_var remoteRepo = peer->repository();
 
       CORBA::ORB_var orb = remoteRepo->_get_orb();
       CORBA::String_var remoteRepoIor = orb->object_to_string(remoteRepo.in());
       if (OpenDDS::DCPS::DCPS_debug_level > 4) {
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) FederatorManagerImpl::join_federation() - ")
                    ACE_TEXT("id %d obtained reference to id %d:\n")
                    ACE_TEXT("\t%C\n"),
                    this->id().id(),
                    remote,
                    remoteRepoIor.in()));
       }
 
       // Add remote repository to Service_Participant in the Federation domain
       std::ostringstream oss;
       oss << remote;
       std::string key_string = oss.str();
       TheServiceParticipant->set_repo_ior(remoteRepoIor.in(), key_string);
       TheServiceParticipant->set_repo_domain(this->config_.federationDomain(), key_string);
 
     } catch (const CORBA::Exception& ex) {
       ex._tao_print_exception(
         "ERROR: Federator::ManagerImpl::join_federation() - Unable to join with remote: ");
       throw Incomplete();
     }
   }
 
   // Symmetrical joining behavior.
   try {
     Manager_var self = this->_this();
     Manager_var remoteManager
     = peer->join_federation(self, this->config_.federationDomain());
 
     if (this->joinRepo_ == remote) {
       this->peers_[ this->joinRepo_]
       = OpenDDS::Federator::Manager::_duplicate(remoteManager.in());
     }
 
     //
     // Push our initial state out to the joining repository *after* we call
     // him back to join.  This reduces the amount of duplicate data pushed
     // when a new (empty) repository is joining an existing federation.
     //
     if (OpenDDS::DCPS::DCPS_debug_level > 4) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::join_federation() - ")
                  ACE_TEXT("repo id %d pushing state to repository with id %d.\n"),
                  this->id().id(),
                  remote));
     }
 
     this->pushState(peer);
 
   } catch (const CORBA::Exception& ex) {
     ex._tao_print_exception(
       "ERROR: Federator::ManagerImpl::join_federation() - unsuccessful call to remote->join: ");
     throw Incomplete();
   }
 
   if (CORBA::is_nil(this->participantWriter_.in())) {
     //
     // Establish our update publications and subscriptions *after* we
     // have exchanged internal state with the first joining repository.
     //
     this->initialize();
   }
 
   // Adjust our joining state and give others the opportunity to proceed.
   if (OpenDDS::DCPS::DCPS_debug_level > 4) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::join_federation() - ")
                ACE_TEXT("repo id %d joined to repository with id %d.\n"),
                this->id().id(),
                remote));
   }
 
   this->federated_ = true;
   this->joiner_    = NIL_REPOSITORY;
   this->joining_.signal();
   return this->_this();
 }

◆ leave_and_shutdown()

void OpenDDS::Federator::ManagerImpl::leave_and_shutdown ( void )

virtual

Definition at line 1112 of file FederatorManagerImpl.cpp.

References info_, and TAO_DDS_DCPSInfo_i::shutdown().

 {
   // Shutdown the process via the repository object.
   this->info_->shutdown();
 }

◆ leave_federation()

void OpenDDS::Federator::ManagerImpl::leave_federation ( RepoKey id )

virtual

Definition at line 1083 of file FederatorManagerImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), config_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::Config::federationDomain(), id(), info_, LM_DEBUG, peers_, and TAO_DDS_DCPSInfo_i::remove_by_owner().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) ManagerImpl::leave_federation( %d)\n"),
                this->id().id()));
   }
 
   // Remove the leaving repository from our outbound mappings.
   IdToManagerMap::iterator where = this->peers_.find(id);
 
   if (where != this->peers_.end()) {
     this->peers_.erase(where);
   }
 
   // Remove all the internal Entities owned by the leaving repository.
   if (!info_->remove_by_owner(config_.federationDomain(), id)) {
     throw Incomplete();
   }
 
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) ManagerImpl::leave_federation( %d) complete.\n"),
                this->id().id()));
   }
 }

◆ localRepo()

ACE_INLINE void OpenDDS::Federator::ManagerImpl::localRepo ( ::OpenDDS::DCPS::DCPSInfo_ptr repo )

Capture a remote callable reference to the DCPSInfo.

Definition at line 33 of file FederatorManagerImpl.inl.

References ACE_INLINE, and localRepo_.

Referenced by InfoRepo::init().

 {
   this->localRepo_ = OpenDDS::DCPS::DCPSInfo::_duplicate(repo);
 }

◆ orb() [1/2]

ACE_INLINE CORBA::ORB_ptr OpenDDS::Federator::ManagerImpl::orb ( void )

Accessors for the ORB.

Definition at line 40 of file FederatorManagerImpl.inl.

References ACE_INLINE, orb_, and TAO_Pseudo_Var_T< ORB >::ptr().

Referenced by InfoRepo::init(), join_federation(), and pushState().

 {
   return this->orb_.ptr();
 }

◆ orb() [2/2]

ACE_INLINE void OpenDDS::Federator::ManagerImpl::orb ( CORBA::ORB_ptr value )

Definition at line 47 of file FederatorManagerImpl.inl.

References CORBA::ORB::_duplicate(), OPENDDS_END_VERSIONED_NAMESPACE_DECL, and orb_.

 {
   this->orb_ = CORBA::ORB::_duplicate(value);
 }

◆ processCreate() [1/5]

void OpenDDS::Federator::ManagerImpl::processCreate	(	const OwnerUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Null implementation for OwnerUpdate samples.

Implements OpenDDS::Federator::UpdateProcessor< OwnerUpdate >.

Definition at line 579 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), TAO_DDS_DCPSInfo_i::changeOwnership(), OpenDDS::DCPS::DCPS_debug_level, deferred_lock_, deferredOwnerships_, OpenDDS::Federator::OwnerUpdate::domain, info_, LM_DEBUG, OpenDDS::Federator::OwnerUpdate::owner, OpenDDS::Federator::OwnerUpdate::participant, processDeferred(), and OpenDDS::Federator::OwnerUpdate::sender.

Referenced by initializeOwner(), initializeParticipant(), initializePublication(), initializeSubscription(), and initializeTopic().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter converter(sample->participant);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processCreate( OwnerUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ sender %d/ owner %d ]\n"),
                this->id().id(),
                sample->domain,
                std::string(converter).c_str(),
                sample->sender,
                sample->owner));
   }
 
   // We could generate an error message here.  Instead we let action be irrelevant.
   if (false == this->info_->changeOwnership(sample->domain,
                                             sample->participant,
                                             sample->sender,
                                             sample->owner)) {
     {
       ACE_GUARD(ACE_Thread_Mutex,
                 guard,
                 this->deferred_lock_);
       this->deferredOwnerships_.push_back(*sample);
     }
 
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::processCreate( OwnerUpdate): ")
                  ACE_TEXT("deferred update.\n")));
     }
   }
 
   this->processDeferred();
 }

◆ processCreate() [2/5]

void OpenDDS::Federator::ManagerImpl::processCreate	(	const PublicationUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Create a proxy for a new publication.

Implements OpenDDS::Federator::UpdateProcessor< PublicationUpdate >.

Definition at line 616 of file FederatorManagerImpl_updates.cpp.

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter pub_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processCreate( PublicationUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ publication %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
   }
 
   if (false == this->info_->add_publication(sample->domain,
                                             sample->participant,
                                             sample->topic,
                                             sample->id,
                                             sample->callback,
                                             sample->datawriter_qos,
                                             sample->transport_info,
                                             sample->transport_context,
                                             sample->publisher_qos,
                                             sample->serialized_type_info,
                                             true)) {
     {
       ACE_GUARD(ACE_Thread_Mutex,
                 guard,
                 this->deferred_lock_);
       this->deferredPublications_.push_back(*sample);
     }
 
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::processCreate( PublicationUpdate): ")
                  ACE_TEXT("deferred update.\n")));
     }
   }
 
   this->processDeferred();
 }

◆ processCreate() [3/5]

void OpenDDS::Federator::ManagerImpl::processCreate	(	const SubscriptionUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Create a proxy for a new subscription.

Implements OpenDDS::Federator::UpdateProcessor< SubscriptionUpdate >.

Definition at line 659 of file FederatorManagerImpl_updates.cpp.

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter sub_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processCreate( SubscriptionUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
   }
 
   if (false == this->info_->add_subscription(sample->domain,
                                              sample->participant,
                                              sample->topic,
                                              sample->id,
                                              sample->callback,
                                              sample->datareader_qos,
                                              sample->transport_info,
                                              sample->transport_context,
                                              sample->subscriber_qos,
                                              sample->filter_class_name,
                                              sample->filter_expression,
                                              sample->expression_params,
                                              sample->serialized_type_info,
                                              true)) {
     {
       ACE_GUARD(ACE_Thread_Mutex,
                 guard,
                 this->deferred_lock_);
       this->deferredSubscriptions_.push_back(*sample);
     }
 
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::processCreate( SubscriptionUpdate): ")
                  ACE_TEXT("deferred update.\n")));
     }
   }
 
   this->processDeferred();
 }

◆ processCreate() [4/5]

void OpenDDS::Federator::ManagerImpl::processCreate	(	const ParticipantUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Create a proxy for a new participant.

Implements OpenDDS::Federator::UpdateProcessor< ParticipantUpdate >.

Definition at line 705 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), TAO_DDS_DCPSInfo_i::add_domain_participant(), TAO_DDS_DCPSInfo_i::changeOwnership(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::ParticipantUpdate::domain, OpenDDS::Federator::ParticipantUpdate::id, info_, LM_DEBUG, LM_ERROR, OpenDDS::Federator::ParticipantUpdate::owner, processDeferred(), OpenDDS::Federator::ParticipantUpdate::qos, and OpenDDS::Federator::ParticipantUpdate::sender.

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processCreate( ParticipantUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ owner %d ]\n"),
                this->id().id(),
                sample->domain,
                std::string(converter).c_str(),
                sample->owner));
   }
 
   this->info_->add_domain_participant(
     sample->domain,
     sample->id,
     sample->qos);
   bool ownershipChanged = this->info_->changeOwnership(
     sample->domain,
     sample->id,
     sample->sender,
     sample->owner);
   if (!ownershipChanged) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: ")
                ACE_TEXT("OpenDDS::Federator::ManagerImpl::processCreate(), ")
                ACE_TEXT("Could not change ownership\n")));
   }
   this->processDeferred();
 }

◆ processCreate() [5/5]

void OpenDDS::Federator::ManagerImpl::processCreate	(	const TopicUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Create a proxy for a new topic.

Implements OpenDDS::Federator::UpdateProcessor< TopicUpdate >.

Definition at line 737 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), TAO_DDS_DCPSInfo_i::add_topic(), OpenDDS::Federator::TopicUpdate::datatype, OpenDDS::DCPS::DCPS_debug_level, deferred_lock_, deferredTopics_, OpenDDS::Federator::TopicUpdate::domain, OpenDDS::Federator::TopicUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::TopicUpdate::participant, processDeferred(), OpenDDS::Federator::TopicUpdate::qos, and OpenDDS::Federator::TopicUpdate::topic.

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter topic_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processCreate( TopicUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ topic %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(topic_converter).c_str()));
   }
 
   if (false == this->info_->add_topic(sample->id,
                                       sample->domain,
                                       sample->participant,
                                       sample->topic,
                                       sample->datatype,
                                       sample->qos)) {
     {
       ACE_GUARD(ACE_Thread_Mutex,
                 guard,
                 this->deferred_lock_);
       this->deferredTopics_.push_back(*sample);
     }
 
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::processCreate( TopicUpdate): ")
                  ACE_TEXT("deferred update.\n")));
     }
   }
 
   this->processDeferred();
 }

◆ processDeferred()

void OpenDDS::Federator::ManagerImpl::processDeferred ( )

Handle any deferred updates that might have become processable.

Definition at line 775 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), TAO_DDS_DCPSInfo_i::add_publication(), TAO_DDS_DCPSInfo_i::add_subscription(), TAO_DDS_DCPSInfo_i::add_topic(), TAO_DDS_DCPSInfo_i::changeOwnership(), OpenDDS::DCPS::DCPS_debug_level, deferred_lock_, deferredOwnerships_, deferredPublications_, deferredSubscriptions_, deferredTopics_, info_, and LM_DEBUG.

Referenced by processCreate().

 {
   ACE_GUARD(ACE_Thread_Mutex,
             guard,
             this->deferred_lock_);
 
   {
     std::list<OwnerUpdate>::iterator current = this->deferredOwnerships_.begin();
 
     while (current != this->deferredOwnerships_.end()) {
       if (this->info_->changeOwnership(current->domain,
                                        current->participant,
                                        current->sender,
                                        current->owner)) {
         if (OpenDDS::DCPS::DCPS_debug_level > 9) {
           OpenDDS::DCPS::RepoIdConverter converter(current->participant);
           ACE_DEBUG((LM_DEBUG,
                      ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDeferred( OwnerUpdate): ")
                      ACE_TEXT("repo %d - [ domain %d/ participant %C/ sender %d/ owner %d ]\n"),
                      this->id().id(),
                      current->domain,
                      std::string(converter).c_str(),
                      current->sender,
                      current->owner));
         }
 
         current = this->deferredOwnerships_.erase(current);
 
       } else {
         ++ current;
       }
     }
   }
 
   {
     std::list<TopicUpdate>::iterator current = this->deferredTopics_.begin();
 
     while (current != this->deferredTopics_.end()) {
       if (true == this->info_->add_topic(current->id,
                                          current->domain,
                                          current->participant,
                                          current->topic,
                                          current->datatype,
                                          current->qos)) {
         if (OpenDDS::DCPS::DCPS_debug_level > 9) {
           OpenDDS::DCPS::RepoIdConverter part_converter(current->participant);
           OpenDDS::DCPS::RepoIdConverter topic_converter(current->id);
           ACE_DEBUG((LM_DEBUG,
                      ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDeferred( TopicUpdate): ")
                      ACE_TEXT("repo %d - [ domain %d/ participant %C/ topic %C ]\n"),
                      this->id().id(),
                      current->domain,
                      std::string(part_converter).c_str(),
                      std::string(topic_converter).c_str()));
         }
 
         current = this->deferredTopics_.erase(current);
 
       } else {
         ++ current;
       }
     }
   }
 
   {
     std::list<PublicationUpdate>::iterator current = this->deferredPublications_.begin();
 
     while (current != this->deferredPublications_.end()) {
 
       if (true == this->info_->add_publication(current->domain,
                                                current->participant,
                                                current->topic,
                                                current->id,
                                                current->callback,
                                                current->datawriter_qos,
                                                current->transport_info,
                                                current->transport_context,
                                                current->publisher_qos,
                                                current->serialized_type_info,
                                                true)) {
         if (OpenDDS::DCPS::DCPS_debug_level > 9) {
           OpenDDS::DCPS::RepoIdConverter part_converter(current->participant);
           OpenDDS::DCPS::RepoIdConverter pub_converter(current->id);
           ACE_DEBUG((LM_DEBUG,
                      ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDeferred( PublicationUpdate): ")
                      ACE_TEXT("repo %d - [ domain %d/ participant %C/ publication %C ]\n"),
                      this->id().id(),
                      current->domain,
                      std::string(part_converter).c_str(),
                      std::string(pub_converter).c_str()));
         }
 
         current = this->deferredPublications_.erase(current);
 
       } else {
         ++ current;
       }
     }
   }
 
   {
     std::list<SubscriptionUpdate>::iterator current = this->deferredSubscriptions_.begin();
 
     while (current != this->deferredSubscriptions_.end()) {
 
       if (true == this->info_->add_subscription(current->domain,
                                                 current->participant,
                                                 current->topic,
                                                 current->id,
                                                 current->callback,
                                                 current->datareader_qos,
                                                 current->transport_info,
                                                 current->transport_context,
                                                 current->subscriber_qos,
                                                 current->filter_class_name,
                                                 current->filter_expression,
                                                 current->expression_params,
                                                 current->serialized_type_info,
                                                 true)) {
         if (OpenDDS::DCPS::DCPS_debug_level > 9) {
           OpenDDS::DCPS::RepoIdConverter part_converter(current->participant);
           OpenDDS::DCPS::RepoIdConverter sub_converter(current->id);
           ACE_DEBUG((LM_DEBUG,
                      ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDeferred( SubscriptionUpdate): ")
                      ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                      this->id().id(),
                      current->domain,
                      std::string(part_converter).c_str(),
                      std::string(sub_converter).c_str()));
         }
 
         current = this->deferredSubscriptions_.erase(current);
 
       } else {
         ++ current;
       }
     }
   }
 
 }

◆ processDelete() [1/5]

void OpenDDS::Federator::ManagerImpl::processDelete	(	const OwnerUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Null implementation for OwnerUpdate samples.

Implements OpenDDS::Federator::UpdateProcessor< OwnerUpdate >.

Definition at line 1101 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), TAO_DDS_DCPSInfo_i::changeOwnership(), OpenDDS::DCPS::DCPS_debug_level, deferred_lock_, deferredOwnerships_, OpenDDS::Federator::OwnerUpdate::domain, info_, LM_DEBUG, OpenDDS::Federator::OwnerUpdate::owner, OpenDDS::Federator::OwnerUpdate::participant, and OpenDDS::Federator::OwnerUpdate::sender.

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter converter(sample->participant);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( OwnerUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ sender %d/ owner %d ]\n"),
                this->id().id(),
                sample->domain,
                std::string(converter).c_str(),
                sample->sender,
                sample->owner));
   }
 
   // We could generate an error message here.  Instead we let action be irrelevant.
   if (false == this->info_->changeOwnership(sample->domain,
                                             sample->participant,
                                             sample->sender,
                                             sample->owner)) {
     {
       ACE_GUARD(ACE_Thread_Mutex,
                 guard,
                 this->deferred_lock_);
       this->deferredOwnerships_.push_back(*sample);
     }
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( OwnerUpdate): ")
                ACE_TEXT("deferred update.\n")));
   }
 }

◆ processDelete() [2/5]

void OpenDDS::Federator::ManagerImpl::processDelete	(	const PublicationUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Delete a proxy for a publication.

Implements OpenDDS::Federator::UpdateProcessor< PublicationUpdate >.

Definition at line 1133 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, OpenDDS::Federator::PublicationUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::PublicationUpdate::participant, and TAO_DDS_DCPSInfo_i::remove_publication().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter pub_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( PublicationUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ publication %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
   }
 
   try {
     this->info_->remove_publication(
       sample->domain,
       sample->participant,
       sample->id);
 
   } catch (OpenDDS::DCPS::Invalid_Participant&) {
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( PublicationUpdate): ")
                  ACE_TEXT("the participant was already removed.\n")));
     }
   }
 }

◆ processDelete() [3/5]

void OpenDDS::Federator::ManagerImpl::processDelete	(	const SubscriptionUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Delete a proxy for a subscription.

Implements OpenDDS::Federator::UpdateProcessor< SubscriptionUpdate >.

Definition at line 1163 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::SubscriptionUpdate::participant, and TAO_DDS_DCPSInfo_i::remove_subscription().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter sub_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( SubscriptionUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
   }
 
   try {
     this->info_->remove_subscription(
       sample->domain,
       sample->participant,
       sample->id);
 
   } catch (OpenDDS::DCPS::Invalid_Participant&) {
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( SubscriptionUpdate): ")
                  ACE_TEXT("the participant was already removed.\n")));
     }
   }
 }

◆ processDelete() [4/5]

void OpenDDS::Federator::ManagerImpl::processDelete	(	const ParticipantUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Delete a proxy for a participant.

Implements OpenDDS::Federator::UpdateProcessor< ParticipantUpdate >.

Definition at line 1193 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::ParticipantUpdate::domain, OpenDDS::Federator::ParticipantUpdate::id, info_, LM_DEBUG, and TAO_DDS_DCPSInfo_i::remove_domain_participant().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( ParticipantUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(converter).c_str()));
   }
   try {
     this->info_->remove_domain_participant(
       sample->domain,
       sample->id);
   } catch (OpenDDS::DCPS::Invalid_Participant&) {
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( ParticipantUpdate): ")
                  ACE_TEXT("the participant was already removed.\n")));
     }
   }
 }

◆ processDelete() [5/5]

void OpenDDS::Federator::ManagerImpl::processDelete	(	const TopicUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Delete a proxy for a topic.

Implements OpenDDS::Federator::UpdateProcessor< TopicUpdate >.

Definition at line 1218 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::TopicUpdate::domain, OpenDDS::Federator::TopicUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::TopicUpdate::participant, and TAO_DDS_DCPSInfo_i::remove_topic().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter topic_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( TopicUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ topic %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(topic_converter).c_str()));
   }
 
   try {
     this->info_->remove_topic(
       sample->domain,
       sample->participant,
       sample->id);
 
   } catch (OpenDDS::DCPS::Invalid_Participant&) {
     if (OpenDDS::DCPS::DCPS_debug_level > 9) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( TopicUpdate): ")
                  ACE_TEXT("the participant was already removed.\n")));
     }
   } catch (OpenDDS::DCPS::Invalid_Domain&) {
     if (OpenDDS::DCPS::DCPS_debug_level > 1) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) Federator::ManagerImpl::processDelete( TopicUpdate): ")
                  ACE_TEXT("the domain %d no longer exists.\n"),sample->domain));
     }
   }
 }

◆ processUpdateFilterExpressionParams()

void OpenDDS::Federator::ManagerImpl::processUpdateFilterExpressionParams	(	const SubscriptionUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Update the proxy filter expression params for a subscription.

Reimplemented from OpenDDS::Federator::UpdateProcessor< SubscriptionUpdate >.

Definition at line 1037 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::expression_params, OpenDDS::Federator::SubscriptionUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::SubscriptionUpdate::participant, and TAO_DDS_DCPSInfo_i::update_subscription_params().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter sub_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processUpdateFilterExpressionParams(SubscriptionUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
   }
 
   this->info_->update_subscription_params(
     sample->domain,
     sample->participant,
     sample->id,
     sample->expression_params);
 }

◆ processUpdateQos1() [1/5]

void OpenDDS::Federator::ManagerImpl::processUpdateQos1	(	const OwnerUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Process ownership changes.

Implements OpenDDS::Federator::UpdateProcessor< OwnerUpdate >.

Definition at line 917 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), TAO_DDS_DCPSInfo_i::changeOwnership(), OpenDDS::DCPS::DCPS_debug_level, deferred_lock_, deferredOwnerships_, OpenDDS::Federator::OwnerUpdate::domain, info_, LM_DEBUG, OpenDDS::Federator::OwnerUpdate::owner, OpenDDS::Federator::OwnerUpdate::participant, and OpenDDS::Federator::OwnerUpdate::sender.

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter converter(sample->participant);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processUpdateQos1( OwnerUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ sender %d/ owner %d ]\n"),
                this->id().id(),
                sample->domain,
                std::string(converter).c_str(),
                sample->sender,
                sample->owner));
   }
 
   if (false == this->info_->changeOwnership(sample->domain,
                                             sample->participant,
                                             sample->sender,
                                             sample->owner)) {
     {
       ACE_GUARD(ACE_Thread_Mutex,
                 guard,
                 this->deferred_lock_);
 
       this->deferredOwnerships_.push_back(*sample);
     }
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processUpdateQos1( OwnerUpdate): ")
                ACE_TEXT("deferred update.\n")));
   }
 }

◆ processUpdateQos1() [2/5]

void OpenDDS::Federator::ManagerImpl::processUpdateQos1	(	const PublicationUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Update the proxy DataWriterQos for a publication.

Implements OpenDDS::Federator::UpdateProcessor< PublicationUpdate >.

Definition at line 949 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::PublicationUpdate::datawriter_qos, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, OpenDDS::Federator::PublicationUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::PublicationUpdate::participant, and TAO_DDS_DCPSInfo_i::update_publication_qos().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter pub_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processUpdateQos1( PublicationUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ publication %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
   }
 
   this->info_->update_publication_qos(
     sample->domain,
     sample->participant,
     sample->id,
     sample->datawriter_qos);
 }

◆ processUpdateQos1() [3/5]

void OpenDDS::Federator::ManagerImpl::processUpdateQos1	(	const SubscriptionUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Update the proxy DataReaderQos for a subscription.

Implements OpenDDS::Federator::UpdateProcessor< SubscriptionUpdate >.

Definition at line 993 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::SubscriptionUpdate::datareader_qos, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::SubscriptionUpdate::participant, and TAO_DDS_DCPSInfo_i::update_subscription_qos().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter sub_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processUpdateQos1( SubscriptionUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
   }
 
   this->info_->update_subscription_qos(
     sample->domain,
     sample->participant,
     sample->id,
     sample->datareader_qos);
 }

◆ processUpdateQos1() [4/5]

void OpenDDS::Federator::ManagerImpl::processUpdateQos1	(	const ParticipantUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Update the proxy ParticipantQos for a participant.

Implements OpenDDS::Federator::UpdateProcessor< ParticipantUpdate >.

Definition at line 1060 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::ParticipantUpdate::domain, OpenDDS::Federator::ParticipantUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::ParticipantUpdate::qos, and TAO_DDS_DCPSInfo_i::update_domain_participant_qos().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processUpdateQos1( ParticipantUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(converter).c_str()));
   }
 
   this->info_->update_domain_participant_qos(
     sample->domain,
     sample->id,
     sample->qos);
 }

◆ processUpdateQos1() [5/5]

void OpenDDS::Federator::ManagerImpl::processUpdateQos1	(	const TopicUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Update the proxy TopicQos for a topic.

Implements OpenDDS::Federator::UpdateProcessor< TopicUpdate >.

Definition at line 1079 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::TopicUpdate::domain, OpenDDS::Federator::TopicUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::TopicUpdate::participant, OpenDDS::Federator::TopicUpdate::qos, and TAO_DDS_DCPSInfo_i::update_topic_qos().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter topic_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processUpdateQos1( TopicUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ topic %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(topic_converter).c_str()));
   }
 
   this->info_->update_topic_qos(
     sample->id,
     sample->domain,
     sample->participant,
     sample->qos);
 }

◆ processUpdateQos2() [1/2]

void OpenDDS::Federator::ManagerImpl::processUpdateQos2	(	const PublicationUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Update the proxy PublisherQos for a publication.

Reimplemented from OpenDDS::Federator::UpdateProcessor< PublicationUpdate >.

Definition at line 971 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, OpenDDS::Federator::PublicationUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::PublicationUpdate::participant, OpenDDS::Federator::PublicationUpdate::publisher_qos, and TAO_DDS_DCPSInfo_i::update_publication_qos().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter pub_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processUpdateQos2( PublicationUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ publication %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
   }
 
   this->info_->update_publication_qos(
     sample->domain,
     sample->participant,
     sample->id,
     sample->publisher_qos);
 }

◆ processUpdateQos2() [2/2]

void OpenDDS::Federator::ManagerImpl::processUpdateQos2	(	const SubscriptionUpdate *	sample,
		const DDS::SampleInfo *	info
	)

virtual

Update the proxy SubscriberQos for a subscription.

Reimplemented from OpenDDS::Federator::UpdateProcessor< SubscriptionUpdate >.

Definition at line 1015 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::id, info_, LM_DEBUG, OpenDDS::Federator::SubscriptionUpdate::participant, OpenDDS::Federator::SubscriptionUpdate::subscriber_qos, and TAO_DDS_DCPSInfo_i::update_subscription_qos().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample->participant);
     OpenDDS::DCPS::RepoIdConverter sub_converter(sample->id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::processUpdateQos2( SubscriptionUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                this->id().id(),
                sample->domain,
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
   }
 
   this->info_->update_subscription_qos(
     sample->domain,
     sample->participant,
     sample->id,
     sample->subscriber_qos);
 }

◆ pushState()

void OpenDDS::Federator::ManagerImpl::pushState ( Manager_ptr peer )

Push our current state to a remote repository.

Definition at line 1254 of file FederatorManagerImpl_updates.cpp.

Referenced by join_federation().

 {
   // foreach DCPS_IR_Domain
   //   foreach DCPS_IR_Participant
   //     peer->initializeParticipant(...)
   //     peer->initializeOwner(...)
   //   foreach DCPS_IR_Participant
   //     foreach DCPS_IR_Topic
   //       peer->initializeTopic(...)
   //     foreach DCPS_IR_Publication
   //       peer->initializePublication(...)
   //     foreach DCPS_IR_Subscription
   //       peer->initializeSubscription(...)
 
   // Process each domain within the repository.
   for (DCPS_IR_Domain_Map::const_iterator currentDomain
        = this->info_->domains().begin();
        currentDomain != this->info_->domains().end();
        ++currentDomain) {
 
     if (currentDomain->second->get_id() == this->config_.federationDomain()) {
       // Do not push the Federation domain publications.
       //continue;
     }
 
     // Process each participant within the current domain.
     for (DCPS_IR_Participant_Map::const_iterator currentParticipant
          = currentDomain->second->participants().begin();
          currentParticipant != currentDomain->second->participants().end();
          ++currentParticipant) {
 
       if (currentParticipant->second->isBitPublisher() == true) {
         // Do not push the built-in topic publications.
         continue;
       }
 
       // Initialize the participant on the peer.
       ParticipantUpdate participantSample;
       participantSample.sender = this->id().id();
       participantSample.action = CreateEntity;
 
       participantSample.owner  =  currentParticipant->second->owner();
       participantSample.domain =  currentDomain->second->get_id();
       participantSample.id     =  currentParticipant->second->get_id();
       participantSample.qos    = *currentParticipant->second->get_qos();
 
       peer->initializeParticipant(participantSample);
 
       // Initialize the ownership of the participant on the peer.
       OwnerUpdate ownerSample;
       ownerSample.sender      = this->id().id();
       ownerSample.action      = CreateEntity;
 
       ownerSample.domain      = currentDomain->second->get_id();
       ownerSample.participant = currentParticipant->second->get_id();
       ownerSample.owner       = currentParticipant->second->owner();
 
       peer->initializeOwner(ownerSample);
     }
 
     // Process each participant within the current domain.
     for (DCPS_IR_Participant_Map::const_iterator currentParticipant
          = currentDomain->second->participants().begin();
          currentParticipant != currentDomain->second->participants().end();
          ++currentParticipant) {
 
       if (currentParticipant->second->isBitPublisher() == true) {
         // Do not push the built-in topic publications.
         continue;
       }
 
       // Process each topic within the current particpant.
       for (DCPS_IR_Topic_Map::const_iterator currentTopic
            = currentParticipant->second->topics().begin();
            currentTopic != currentParticipant->second->topics().end();
            ++currentTopic) {
         TopicUpdate topicSample;
         topicSample.sender      = this->id().id();
         topicSample.action      = CreateEntity;
 
         topicSample.id          = currentTopic->second->get_id();
         topicSample.domain      = currentDomain->second->get_id();
         topicSample.participant = currentTopic->second->get_participant_id();
         topicSample.topic       = currentTopic->second->get_topic_description()->get_name();
         topicSample.datatype    = currentTopic->second->get_topic_description()->get_dataTypeName();
         topicSample.qos         = *currentTopic->second->get_topic_qos();
 
         peer->initializeTopic(topicSample);
       }
 
       // Process each publication within the current particpant.
       for (DCPS_IR_Publication_Map::const_iterator currentPublication
            = currentParticipant->second->publications().begin();
            currentPublication != currentParticipant->second->publications().end();
            ++currentPublication) {
         PublicationUpdate publicationSample;
         publicationSample.sender         = this->id().id();
         publicationSample.action         = CreateEntity;
 
         DCPS_IR_Publication* p = currentPublication->second.get();
         CORBA::ORB_var orb = this->info_->orb();
         CORBA::String_var callback = orb->object_to_string(p->writer());
 
         publicationSample.domain         = currentDomain->second->get_id();
         publicationSample.participant    = p->get_participant_id();
         publicationSample.topic          = p->get_topic_id();
         publicationSample.id             = p->get_id();
         publicationSample.callback       = callback.in();
         publicationSample.datawriter_qos = *p->get_datawriter_qos();
         publicationSample.publisher_qos  = *p->get_publisher_qos();
         publicationSample.transport_info = p->get_transportLocatorSeq();
 
         peer->initializePublication(publicationSample);
       }
 
       // Process each subscription within the current particpant.
       for (DCPS_IR_Subscription_Map::const_iterator currentSubscription
            = currentParticipant->second->subscriptions().begin();
            currentSubscription != currentParticipant->second->subscriptions().end();
            ++currentSubscription) {
         SubscriptionUpdate subscriptionSample;
         subscriptionSample.sender         = this->id().id();
         subscriptionSample.action         = CreateEntity;
 
         DCPS_IR_Subscription* s = currentSubscription->second.get();
         CORBA::ORB_var orb = this->info_->orb();
         CORBA::String_var callback = orb->object_to_string(s->reader());
 
         subscriptionSample.domain         = currentDomain->second->get_id();
         subscriptionSample.participant    = s->get_participant_id();
         subscriptionSample.topic          = s->get_topic_id();
         subscriptionSample.id             = s->get_id();
         subscriptionSample.callback       = callback.in();
         subscriptionSample.datareader_qos = *s->get_datareader_qos();
         subscriptionSample.subscriber_qos = *s->get_subscriber_qos();
         subscriptionSample.transport_info = s->get_transportLocatorSeq();
         subscriptionSample.filter_expression = s->get_filter_expression().c_str();
         subscriptionSample.expression_params = s->get_expr_params();
 
         peer->initializeSubscription(subscriptionSample);
       }
     }
   }
 }

◆ repository()

OpenDDS::DCPS::DCPSInfo_ptr OpenDDS::Federator::ManagerImpl::repository ( void )

virtual

Definition at line 881 of file FederatorManagerImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), config_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::Config::federationDomain(), CORBA::is_nil(), OpenDDS::DCPS::RcHandle< T >::is_nil(), LM_DEBUG, localRepo_, OpenDDS::DCPS::static_rchandle_cast(), and TheServiceParticipant.

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) ManagerImpl::repository()\n")));
   }
 
   OpenDDS::DCPS::Discovery_rch disco
   = TheServiceParticipant->get_discovery(
       this->config_.federationDomain());
   OpenDDS::DCPS::DCPSInfo_var repo;
   if (!disco.is_nil()) {
     OpenDDS::DCPS::InfoRepoDiscovery_rch irDisco =
       DCPS::static_rchandle_cast<DCPS::InfoRepoDiscovery>(disco);
     repo = irDisco->get_dcps_info();
   }
 
   if (CORBA::is_nil(repo.in())) {
     return OpenDDS::DCPS::DCPSInfo::_duplicate(this->localRepo_.in());
 
   } else {
     return OpenDDS::DCPS::DCPSInfo::_duplicate(repo.in());
   }
 }

◆ requestImage()

void OpenDDS::Federator::ManagerImpl::requestImage ( )

virtual

Request an image refresh to be sent to the specified callback (asynchronously).

Implements Update::Updater.

Definition at line 28 of file FederatorManagerImpl_updates.cpp.

 {
   /* This method intentionally left unimplemented. */
 }

◆ shutdown()

void OpenDDS::Federator::ManagerImpl::shutdown ( void )

virtual

Definition at line 1120 of file FederatorManagerImpl.cpp.

References federated_, info_, and TAO_DDS_DCPSInfo_i::shutdown().

 {
   // Prevent the removal of this repository from the federation during
   // shutdown processing.
   this->federated_ = false;
 
   // Shutdown the process via the repository object.
   this->info_->shutdown();
 }

◆ unregisterCallback()

void OpenDDS::Federator::ManagerImpl::unregisterCallback ( )

virtual

Definition at line 22 of file FederatorManagerImpl_updates.cpp.

 {
   /* This method intentionally left unimplemented. */
 }

◆ update() [1/7]

void OpenDDS::Federator::ManagerImpl::update	(	const Update::IdPath &	id,
		const DDS::DomainParticipantQos &	qos
	)

virtual

Propagate updated Qos parameters for an entity.

Implements Update::Updater.

Definition at line 353 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::ParticipantUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::ParticipantUpdate::domain, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), OpenDDS::Federator::ParticipantUpdate::id, id(), CORBA::is_nil(), LM_DEBUG, participantWriter_, OpenDDS::Federator::ParticipantUpdate::qos, OpenDDS::Federator::ParticipantUpdate::sender, and OpenDDS::Federator::UpdateQosValue1.

 {
   if (CORBA::is_nil(this->participantWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   ParticipantUpdate sample = ParticipantUpdate();
   sample.sender = this->id().id();
   sample.action = UpdateQosValue1;
 
   sample.domain = id.domain;
   sample.id     = id.id;
   sample.qos    = qos;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::update( ParticipantUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(converter).c_str()));
   }
 
   this->participantWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ update() [2/7]

void OpenDDS::Federator::ManagerImpl::update	(	const Update::IdPath &	id,
		const DDS::TopicQos &	qos
	)

virtual

Propagate updated Qos parameters for an entity.

Implements Update::Updater.

Definition at line 382 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::TopicUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::TopicUpdate::domain, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), OpenDDS::Federator::TopicUpdate::id, id(), CORBA::is_nil(), LM_DEBUG, OpenDDS::Federator::TopicUpdate::participant, OpenDDS::Federator::TopicUpdate::qos, OpenDDS::Federator::TopicUpdate::sender, topicWriter_, and OpenDDS::Federator::UpdateQosValue1.

 {
   if (CORBA::is_nil(this->topicWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   TopicUpdate sample = TopicUpdate();
   sample.sender      = this->id().id();
   sample.action      = UpdateQosValue1;
 
   sample.id          = id.id;
   sample.domain      = id.domain;
   sample.participant = id.participant;
   sample.qos         = qos;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
     OpenDDS::DCPS::RepoIdConverter topic_converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::update( TopicUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ topic %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(part_converter).c_str(),
                std::string(topic_converter).c_str()));
   }
 
   this->topicWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ update() [3/7]

void OpenDDS::Federator::ManagerImpl::update	(	const Update::IdPath &	id,
		const DDS::DataWriterQos &	qos
	)

virtual

Propagate updated Qos parameters for an entity.

Implements Update::Updater.

Definition at line 414 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::PublicationUpdate::action, OpenDDS::Federator::PublicationUpdate::datawriter_qos, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), id(), OpenDDS::Federator::PublicationUpdate::id, CORBA::is_nil(), LM_DEBUG, OpenDDS::Federator::PublicationUpdate::participant, publicationWriter_, OpenDDS::Federator::PublicationUpdate::sender, and OpenDDS::Federator::UpdateQosValue1.

 {
   if (CORBA::is_nil(this->publicationWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   PublicationUpdate sample = PublicationUpdate();
   sample.sender         = this->id().id();
   sample.action         = UpdateQosValue1;
 
   sample.domain         = id.domain;
   sample.participant    = id.participant;
   sample.id             = id.id;
   sample.datawriter_qos = qos;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
     OpenDDS::DCPS::RepoIdConverter pub_converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::update( WriterUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ publication %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
   }
 
   this->publicationWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ update() [4/7]

void OpenDDS::Federator::ManagerImpl::update	(	const Update::IdPath &	id,
		const DDS::PublisherQos &	qos
	)

virtual

Propagate updated Qos parameters for an entity.

Implements Update::Updater.

Definition at line 446 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::PublicationUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), id(), OpenDDS::Federator::PublicationUpdate::id, CORBA::is_nil(), LM_DEBUG, OpenDDS::Federator::PublicationUpdate::participant, publicationWriter_, OpenDDS::Federator::PublicationUpdate::publisher_qos, OpenDDS::Federator::PublicationUpdate::sender, and OpenDDS::Federator::UpdateQosValue2.

 {
   if (CORBA::is_nil(this->publicationWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   PublicationUpdate sample = PublicationUpdate();
   sample.sender         = this->id().id();
   sample.action         = UpdateQosValue2;
 
   sample.domain         = id.domain;
   sample.participant    = id.participant;
   sample.id             = id.id;
   sample.publisher_qos  = qos;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
     OpenDDS::DCPS::RepoIdConverter pub_converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::update( PublisherUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ publication %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(part_converter).c_str(),
                std::string(pub_converter).c_str()));
   }
 
   this->publicationWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ update() [5/7]

void OpenDDS::Federator::ManagerImpl::update	(	const Update::IdPath &	id,
		const DDS::DataReaderQos &	qos
	)

virtual

Propagate updated Qos parameters for an entity.

Implements Update::Updater.

Definition at line 478 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::SubscriptionUpdate::action, OpenDDS::Federator::SubscriptionUpdate::datareader_qos, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), id(), OpenDDS::Federator::SubscriptionUpdate::id, CORBA::is_nil(), LM_DEBUG, OpenDDS::Federator::SubscriptionUpdate::participant, OpenDDS::Federator::SubscriptionUpdate::sender, subscriptionWriter_, and OpenDDS::Federator::UpdateQosValue1.

 {
   if (CORBA::is_nil(this->subscriptionWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   SubscriptionUpdate sample = SubscriptionUpdate();
   sample.sender         = this->id().id();
   sample.action         = UpdateQosValue1;
 
   sample.domain         = id.domain;
   sample.participant    = id.participant;
   sample.id             = id.id;
   sample.datareader_qos = qos;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
     OpenDDS::DCPS::RepoIdConverter sub_converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::update( ReaderUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
   }
 
   this->subscriptionWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ update() [6/7]

void OpenDDS::Federator::ManagerImpl::update	(	const Update::IdPath &	id,
		const DDS::SubscriberQos &	qos
	)

virtual

Propagate updated Qos parameters for an entity.

Implements Update::Updater.

Definition at line 541 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::SubscriptionUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), id(), OpenDDS::Federator::SubscriptionUpdate::id, CORBA::is_nil(), LM_DEBUG, OpenDDS::Federator::SubscriptionUpdate::participant, OpenDDS::Federator::SubscriptionUpdate::sender, OpenDDS::Federator::SubscriptionUpdate::subscriber_qos, subscriptionWriter_, and OpenDDS::Federator::UpdateQosValue2.

 {
   if (CORBA::is_nil(this->subscriptionWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   SubscriptionUpdate sample = SubscriptionUpdate();
   sample.sender         = this->id().id();
   sample.action         = UpdateQosValue2;
 
   sample.domain         = id.domain;
   sample.participant    = id.participant;
   sample.id             = id.id;
   sample.subscriber_qos = qos;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
     OpenDDS::DCPS::RepoIdConverter sub_converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::update( SubscriberUpdate): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
   }
 
   this->subscriptionWriter_->write(sample, DDS::HANDLE_NIL);
 }

◆ update() [7/7]

void OpenDDS::Federator::ManagerImpl::update	(	const Update::IdPath &	id,
		const DDS::StringSeq &	exprParams
	)

virtual

Propagate updated Qos parameters for an entity.

Implements Update::Updater.

Definition at line 510 of file FederatorManagerImpl_updates.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::Federator::SubscriptionUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::expression_params, DDS::HANDLE_NIL, TAO_DDS_DCPSFederationId::id(), id(), OpenDDS::Federator::SubscriptionUpdate::id, CORBA::is_nil(), LM_DEBUG, OpenDDS::Federator::SubscriptionUpdate::participant, OpenDDS::Federator::SubscriptionUpdate::sender, subscriptionWriter_, and OpenDDS::Federator::UpdateFilterExpressionParams.

 {
   if (CORBA::is_nil(this->subscriptionWriter_.in())) {
     // Decline to publish data until we can.
     return;
   }
 
   SubscriptionUpdate sample = SubscriptionUpdate();
   sample.sender            = this->id().id();
   sample.action            = UpdateFilterExpressionParams;
   sample.domain            = id.domain;
   sample.participant       = id.participant;
   sample.id                = id.id;
   sample.expression_params = params;
 
   if (OpenDDS::DCPS::DCPS_debug_level > 9) {
     OpenDDS::DCPS::RepoIdConverter part_converter(sample.participant);
     OpenDDS::DCPS::RepoIdConverter sub_converter(sample.id);
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) Federator::ManagerImpl::update(FilterParams): ")
                ACE_TEXT("repo %d - [ domain %d/ participant %C/ subscription %C ]\n"),
                this->id().id(),
                sample.domain,
                std::string(part_converter).c_str(),
                std::string(sub_converter).c_str()));
   }
 
   this->subscriptionWriter_->write(sample, DDS::HANDLE_NIL);
 }

Member Data Documentation

◆ config_

Config& OpenDDS::Federator::ManagerImpl::config_

private

The configuration information for this manager.

Definition at line 231 of file FederatorManagerImpl.h.

Referenced by destroy(), id(), initialize(), join_federation(), leave_federation(), pushState(), and repository().

◆ deferred_lock_

ACE_Thread_Mutex OpenDDS::Federator::ManagerImpl::deferred_lock_

private

Protect deferred updates.

Definition at line 294 of file FederatorManagerImpl.h.

Referenced by processCreate(), processDeferred(), processDelete(), and processUpdateQos1().

◆ deferredOwnerships_

std::list<OwnerUpdate> OpenDDS::Federator::ManagerImpl::deferredOwnerships_

private

Deferred ownership updates.

Definition at line 279 of file FederatorManagerImpl.h.

Referenced by processCreate(), processDeferred(), processDelete(), and processUpdateQos1().

◆ deferredPublications_

std::list<PublicationUpdate> OpenDDS::Federator::ManagerImpl::deferredPublications_

private

Deferred publication updates.

Definition at line 285 of file FederatorManagerImpl.h.

Referenced by processCreate(), and processDeferred().

◆ deferredSubscriptions_

std::list<SubscriptionUpdate> OpenDDS::Federator::ManagerImpl::deferredSubscriptions_

private

Deferred subscription updates.

Definition at line 288 of file FederatorManagerImpl.h.

Referenced by processCreate(), and processDeferred().

◆ deferredTopics_

std::list<TopicUpdate> OpenDDS::Federator::ManagerImpl::deferredTopics_

private

Deferred topic updates.

Definition at line 282 of file FederatorManagerImpl.h.

Referenced by processCreate(), and processDeferred().

◆ federated_

bool OpenDDS::Federator::ManagerImpl::federated_

private

Flag indicating that we are actively participating in a federation of repositories.

Definition at line 219 of file FederatorManagerImpl.h.

Referenced by finalize(), join_federation(), and shutdown().

◆ federationParticipant_

DDS::DomainParticipant_var OpenDDS::Federator::ManagerImpl::federationParticipant_

private

local DomainParticipant

Definition at line 246 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

◆ info_

TAO_DDS_DCPSInfo_i* OpenDDS::Federator::ManagerImpl::info_

private

The Info object reference to update.

Definition at line 234 of file FederatorManagerImpl.h.

Referenced by info(), leave_and_shutdown(), leave_federation(), processCreate(), processDeferred(), processDelete(), processUpdateFilterExpressionParams(), processUpdateQos1(), processUpdateQos2(), pushState(), and shutdown().

◆ joiner_

RepoKey OpenDDS::Federator::ManagerImpl::joiner_

private

Simple recursion avoidance during the join operations.

Definition at line 212 of file FederatorManagerImpl.h.

Referenced by join_federation().

◆ joining_

ACE_Condition<ACE_SYNCH_MUTEX> OpenDDS::Federator::ManagerImpl::joining_

private

Condition used to gate joining activities.

Definition at line 209 of file FederatorManagerImpl.h.

Referenced by join_federation().

◆ joinRepo_

RepoKey OpenDDS::Federator::ManagerImpl::joinRepo_

private

Repository to which we joined.

Definition at line 215 of file FederatorManagerImpl.h.

Referenced by finalize(), and join_federation().

◆ localRepo_

OpenDDS::DCPS::DCPSInfo_var OpenDDS::Federator::ManagerImpl::localRepo_

private

Remotely callable reference to the local repository.

Definition at line 237 of file FederatorManagerImpl.h.

Referenced by localRepo(), and repository().

◆ lock_

ACE_SYNCH_MUTEX OpenDDS::Federator::ManagerImpl::lock_

private

Critical section MUTEX.

Definition at line 206 of file FederatorManagerImpl.h.

Referenced by join_federation().

◆ multicastEnabled_

bool OpenDDS::Federator::ManagerImpl::multicastEnabled_

private

Is multicast enabled?

Definition at line 291 of file FederatorManagerImpl.h.

Referenced by initialize(), and ManagerImpl().

◆ multicastResponder_

InfoRepoMulticastResponder OpenDDS::Federator::ManagerImpl::multicastResponder_

private

Multicast responder.

Definition at line 243 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

◆ orb_

CORBA::ORB_var OpenDDS::Federator::ManagerImpl::orb_

private

The ORB in which we are activated.

Definition at line 240 of file FederatorManagerImpl.h.

Referenced by finalize(), initialize(), and orb().

◆ ownerListener_

UpdateListener<OwnerUpdate, OwnerUpdateDataReader> OpenDDS::Federator::ManagerImpl::ownerListener_

private

TopicUpdate listener.

Definition at line 249 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

◆ ownerWriter_

OwnerUpdateDataWriter_var OpenDDS::Federator::ManagerImpl::ownerWriter_

private

TopicUpdate writer.

Definition at line 264 of file FederatorManagerImpl.h.

Referenced by create(), and initialize().

◆ participantListener_

UpdateListener<ParticipantUpdate, ParticipantUpdateDataReader> OpenDDS::Federator::ManagerImpl::participantListener_

private

ParticipantUpdate listener.

Definition at line 255 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

◆ participantWriter_

ParticipantUpdateDataWriter_var OpenDDS::Federator::ManagerImpl::participantWriter_

private

ParticipantUpdate writer.

Definition at line 270 of file FederatorManagerImpl.h.

Referenced by create(), destroy(), initialize(), join_federation(), and update().

◆ peers_

IdToManagerMap OpenDDS::Federator::ManagerImpl::peers_

private

The peer with which we have federated.

Definition at line 225 of file FederatorManagerImpl.h.

Referenced by finalize(), join_federation(), and leave_federation().

◆ publicationListener_

UpdateListener<PublicationUpdate, PublicationUpdateDataReader> OpenDDS::Federator::ManagerImpl::publicationListener_

private

PublicationUpdate listener.

Definition at line 258 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

◆ publicationWriter_

PublicationUpdateDataWriter_var OpenDDS::Federator::ManagerImpl::publicationWriter_

private

PublicationUpdate writer.

Definition at line 273 of file FederatorManagerImpl.h.

Referenced by create(), destroy(), initialize(), and update().

◆ sequence_

OpenDDS::DCPS::SequenceNumber OpenDDS::Federator::ManagerImpl::sequence_

private

The packet sequence number for data that we publish.

Definition at line 228 of file FederatorManagerImpl.h.

◆ subscriptionListener_

UpdateListener<SubscriptionUpdate, SubscriptionUpdateDataReader> OpenDDS::Federator::ManagerImpl::subscriptionListener_

private

SubscriptionUpdate listener.

Definition at line 261 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

◆ subscriptionWriter_

SubscriptionUpdateDataWriter_var OpenDDS::Federator::ManagerImpl::subscriptionWriter_

private

SubscriptionUpdate writer.

Definition at line 276 of file FederatorManagerImpl.h.

Referenced by create(), destroy(), initialize(), and update().

◆ topicListener_

UpdateListener<TopicUpdate, TopicUpdateDataReader> OpenDDS::Federator::ManagerImpl::topicListener_

private

TopicUpdate listener.

Definition at line 252 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

◆ topicWriter_

TopicUpdateDataWriter_var OpenDDS::Federator::ManagerImpl::topicWriter_

private

TopicUpdate writer.

Definition at line 267 of file FederatorManagerImpl.h.

Referenced by create(), destroy(), initialize(), and update().

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

Public Member Functions

Private Types

Private Attributes

Detailed Description

Member Typedef Documentation

◆ IdToManagerMap

Constructor & Destructor Documentation

◆ ManagerImpl()

◆ ~ManagerImpl()

Member Function Documentation

◆ create() [1/5]

◆ create() [2/5]

◆ create() [3/5]

◆ create() [4/5]

◆ create() [5/5]

◆ destroy()

◆ discover_federation()

◆ federation_id()

◆ finalize()

◆ id()

◆ info() [1/2]

◆ info() [2/2]

◆ initialize()

◆ initializeOwner()

◆ initializeParticipant()

◆ initializePublication()

◆ initializeSubscription()

◆ initializeTopic()

◆ join_federation()

◆ leave_and_shutdown()

◆ leave_federation()

◆ localRepo()

◆ orb() [1/2]

◆ orb() [2/2]

◆ processCreate() [1/5]

◆ processCreate() [2/5]

◆ processCreate() [3/5]

◆ processCreate() [4/5]

◆ processCreate() [5/5]

◆ processDeferred()

◆ processDelete() [1/5]

◆ processDelete() [2/5]

◆ processDelete() [3/5]

◆ processDelete() [4/5]

◆ processDelete() [5/5]

◆ processUpdateFilterExpressionParams()

◆ processUpdateQos1() [1/5]

◆ processUpdateQos1() [2/5]

◆ processUpdateQos1() [3/5]

◆ processUpdateQos1() [4/5]

◆ processUpdateQos1() [5/5]

◆ processUpdateQos2() [1/2]

◆ processUpdateQos2() [2/2]

◆ pushState()

◆ repository()

◆ requestImage()

◆ shutdown()

◆ unregisterCallback()

◆ update() [1/7]

◆ update() [2/7]

◆ update() [3/7]

◆ update() [4/7]

◆ update() [5/7]

◆ update() [6/7]

◆ update() [7/7]

Member Data Documentation

◆ config_

◆ deferred_lock_

◆ deferredOwnerships_

◆ deferredPublications_

◆ deferredSubscriptions_

◆ deferredTopics_

◆ federated_

◆ federationParticipant_

◆ info_

◆ joiner_

◆ joining_

◆ joinRepo_

◆ localRepo_

◆ lock_