OpenDDS::Federator::ManagerImpl Class Reference

#include <FederatorManagerImpl.h>

Inheritance diagram for OpenDDS::Federator::ManagerImpl:

[legend]Collaboration diagram for OpenDDS::Federator::ManagerImpl:

[legend]List of all members.

Public Member Functions
	ManagerImpl (Config &config)
	Default constructor.
virtual	~ManagerImpl ()
	Virtual destructor.
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.
void	finalize ()
	Release resources gracefully.
TAO_DDS_DCPSInfo_i *&	info ()
	Accessors for the DCPSInfo reference.
TAO_DDS_DCPSInfo_i *	info () const
void	localRepo (::OpenDDS::DCPS::DCPSInfo_ptr repo)
	Capture a remote callable reference to the DCPSInfo.
const TAO_DDS_DCPSFederationId &	id () const
	Accessors for the federation Id value.
CORBA::ORB_ptr	orb ()
	Accessors for the ORB.
void	orb (CORBA::ORB_ptr value)
void	pushState (Manager_ptr peer)
	Push our current state to a remote repository.
void	processDeferred ()
	Handle any deferred updates that might have become processable.
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)
void	processCreate (const OwnerUpdate *sample, const DDS::SampleInfo *info)
	Null implementation for OwnerUpdate samples.
void	processCreate (const PublicationUpdate *sample, const DDS::SampleInfo *info)
	Create a proxy for a new publication.
void	processCreate (const SubscriptionUpdate *sample, const DDS::SampleInfo *info)
	Create a proxy for a new subscription.
void	processCreate (const ParticipantUpdate *sample, const DDS::SampleInfo *info)
	Create a proxy for a new participant.
void	processCreate (const TopicUpdate *sample, const DDS::SampleInfo *info)
	Create a proxy for a new topic.
void	processUpdateQos1 (const OwnerUpdate *sample, const DDS::SampleInfo *info)
	Process ownership changes.
void	processUpdateQos1 (const PublicationUpdate *sample, const DDS::SampleInfo *info)
	Update the proxy DataWriterQos for a publication.
void	processUpdateQos2 (const PublicationUpdate *sample, const DDS::SampleInfo *info)
	Update the proxy PublisherQos for a publication.
void	processUpdateQos1 (const SubscriptionUpdate *sample, const DDS::SampleInfo *info)
	Update the proxy DataReaderQos for a subscription.
void	processUpdateQos2 (const SubscriptionUpdate *sample, const DDS::SampleInfo *info)
	Update the proxy SubscriberQos for a subscription.
void	processUpdateFilterExpressionParams (const SubscriptionUpdate *sample, const DDS::SampleInfo *info)
	Update the proxy filter expression params for a subscription.
void	processUpdateQos1 (const ParticipantUpdate *sample, const DDS::SampleInfo *info)
	Update the proxy ParticipantQos for a participant.
void	processUpdateQos1 (const TopicUpdate *sample, const DDS::SampleInfo *info)
	Update the proxy TopicQos for a topic.
void	processDelete (const OwnerUpdate *sample, const DDS::SampleInfo *info)
	Null implementation for OwnerUpdate samples.
void	processDelete (const PublicationUpdate *sample, const DDS::SampleInfo *info)
	Delete a proxy for a publication.
void	processDelete (const SubscriptionUpdate *sample, const DDS::SampleInfo *info)
	Delete a proxy for a subscription.
void	processDelete (const ParticipantUpdate *sample, const DDS::SampleInfo *info)
	Delete a proxy for a participant.
void	processDelete (const TopicUpdate *sample, const DDS::SampleInfo *info)
	Delete a proxy for a topic.
Private Types
typedef std::map< RepoKey, Manager_var >	IdToManagerMap
	Map type to hold references to federated repository Managers.
Private Attributes
ACE_SYNCH_MUTEX	lock_
	Critical section MUTEX.
ACE_Condition< ACE_SYNCH_MUTEX >	joining_
	Condition used to gate joining activities.
RepoKey	joiner_
	Simple recursion avoidance during the join operations.
RepoKey	joinRepo_
	Repository to which we joined.
bool	federated_
IdToManagerMap	peers_
	The peer with which we have federated.
OpenDDS::DCPS::SequenceNumber	sequence_
	The packet sequence number for data that we publish.
Config &	config_
	The configuration information for this manager.
TAO_DDS_DCPSInfo_i *	info_
	The Info object reference to update.
OpenDDS::DCPS::DCPSInfo_var	localRepo_
	Remotely callable reference to the local repository.
CORBA::ORB_var	orb_
	The ORB in which we are activated.
InfoRepoMulticastResponder	multicastResponder_
	Multicast responder.
DDS::DomainParticipant_var	federationParticipant_
	local DomainParticipant
UpdateListener< OwnerUpdate, OwnerUpdateDataReader >	ownerListener_
	TopicUpdate listener.
UpdateListener< TopicUpdate, TopicUpdateDataReader >	topicListener_
	TopicUpdate listener.
UpdateListener< ParticipantUpdate, ParticipantUpdateDataReader >	participantListener_
	ParticipantUpdate listener.
UpdateListener< PublicationUpdate, PublicationUpdateDataReader >	publicationListener_
	PublicationUpdate listener.
UpdateListener< SubscriptionUpdate, SubscriptionUpdateDataReader >	subscriptionListener_
	SubscriptionUpdate listener.
OwnerUpdateDataWriter_var	ownerWriter_
	TopicUpdate writer.
TopicUpdateDataWriter_var	topicWriter_
	TopicUpdate writer.
ParticipantUpdateDataWriter_var	participantWriter_
	ParticipantUpdate writer.
PublicationUpdateDataWriter_var	publicationWriter_
	PublicationUpdate writer.
SubscriptionUpdateDataWriter_var	subscriptionWriter_
	SubscriptionUpdate writer.
std::list< OwnerUpdate >	deferredOwnerships_
	Deferred ownership updates.
std::list< TopicUpdate >	deferredTopics_
	Deferred topic updates.
std::list< PublicationUpdate >	deferredPublications_
	Deferred publication updates.
std::list< SubscriptionUpdate >	deferredSubscriptions_
	Deferred subscription updates.
bool	multicastEnabled_
	Is multicast enabled?
ACE_Thread_Mutex	deferred_lock_
	Protect deferred updates.

---

Detailed Description

Definition at line 34 of file FederatorManagerImpl.h.

---

Member Typedef Documentation

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

OpenDDS::Federator::ManagerImpl::ManagerImpl

(

Config &

config

)

Default constructor.

Definition at line 37 of file FederatorManagerImpl.cpp.

References OpenDDS::DCPS::DCPS_debug_level, 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;
    }
  }
}

OpenDDS::Federator::ManagerImpl::~ManagerImpl

(

)

[virtual]

Virtual destructor.

Definition at line 67 of file FederatorManagerImpl.cpp.

References OpenDDS::DCPS::DCPS_debug_level.

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

---

Member Function Documentation

void OpenDDS::Federator::ManagerImpl::create

(

const Update::OwnershipData &

data

)

[virtual]

Implements Update::Updater.

Definition at line 177 of file FederatorManagerImpl_updates.cpp.

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

{
  if (CORBA::is_nil(this->ownerWriter_.in())) {
    // Decline to publish data until we can.
    return;
  }

  OwnerUpdate sample;
  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);
}

virtual void OpenDDS::Federator::ManagerImpl::create

(

const Update::UWActor &

writer

)

[virtual]

Implements Update::Updater.

void OpenDDS::Federator::ManagerImpl::create

(

const Update::URActor &

reader

)

[virtual]

Implements Update::Updater.

Definition at line 102 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::SubscriptionUpdate::action, Update::ActorStrt< PSQ, RWQ, C, T, CSP >::actorId, Update::ActorStrt< PSQ, RWQ, C, T, CSP >::callback, OpenDDS::Federator::SubscriptionUpdate::callback, Update::ActorStrt< PSQ, RWQ, C, T, CSP >::contentSubscriptionProfile, OpenDDS::Federator::CreateEntity, OpenDDS::Federator::SubscriptionUpdate::datareader_qos, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, Update::ActorStrt< PSQ, RWQ, C, T, CSP >::domainId, Update::ActorStrt< PSQ, RWQ, C, T, CSP >::drdwQos, OpenDDS::Federator::SubscriptionUpdate::expression_params, OpenDDS::Federator::SubscriptionUpdate::filter_class_name, OpenDDS::Federator::SubscriptionUpdate::filter_expression, DDS::HANDLE_NIL, OpenDDS::Federator::SubscriptionUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), OpenDDS::Federator::SubscriptionUpdate::participant, Update::ActorStrt< PSQ, RWQ, C, T, CSP >::participantId, Update::ActorStrt< PSQ, RWQ, C, T, CSP >::pubsubQos, OpenDDS::Federator::SubscriptionUpdate::sender, OpenDDS::Federator::SubscriptionUpdate::subscriber_qos, subscriptionWriter_, OpenDDS::Federator::SubscriptionUpdate::topic, Update::ActorStrt< PSQ, RWQ, C, T, CSP >::topicId, OpenDDS::Federator::SubscriptionUpdate::transport_info, and Update::ActorStrt< PSQ, RWQ, C, T, CSP >::transportInterfaceInfo.

{
  if (CORBA::is_nil(this->subscriptionWriter_.in())) {
    // Decline to publish data until we can.
    return;
  }

  SubscriptionUpdate sample;
  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;

  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);
}

void OpenDDS::Federator::ManagerImpl::create

(

const Update::UParticipant &

participant

)

[virtual]

Implements Update::Updater.

Definition at line 72 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::ParticipantUpdate::action, OpenDDS::Federator::CreateEntity, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::ParticipantUpdate::domain, Update::ParticipantStrt< Q >::domainId, DDS::HANDLE_NIL, OpenDDS::Federator::ParticipantUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), Update::ParticipantStrt< Q >::owner, OpenDDS::Federator::ParticipantUpdate::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;
  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);
}

void OpenDDS::Federator::ManagerImpl::create

(

const Update::UTopic &

topic

)

[virtual]

Implements Update::Updater.

Definition at line 38 of file FederatorManagerImpl_updates.cpp.

References 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, OpenDDS::Federator::TopicUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), 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_.

{
  if (CORBA::is_nil(this->topicWriter_.in())) {
    // Decline to publish data until we can.
    return;
  }

  TopicUpdate sample;
  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);
}

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

Implements Update::Updater.

Definition at line 208 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::SubscriptionUpdate::action, OpenDDS::Federator::PublicationUpdate::action, OpenDDS::Federator::ParticipantUpdate::action, OpenDDS::Federator::TopicUpdate::action, Update::Actor, Update::DataReader, Update::DataWriter, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::DestroyEntity, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::PublicationUpdate::domain, OpenDDS::Federator::ParticipantUpdate::domain, OpenDDS::Federator::TopicUpdate::domain, DDS::HANDLE_NIL, OpenDDS::Federator::SubscriptionUpdate::id, OpenDDS::Federator::PublicationUpdate::id, OpenDDS::Federator::ParticipantUpdate::id, OpenDDS::Federator::TopicUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), OpenDDS::Federator::SubscriptionUpdate::participant, OpenDDS::Federator::PublicationUpdate::participant, Update::Participant, OpenDDS::Federator::TopicUpdate::participant, participantWriter_, publicationWriter_, OpenDDS::Federator::SubscriptionUpdate::sender, OpenDDS::Federator::PublicationUpdate::sender, OpenDDS::Federator::ParticipantUpdate::sender, OpenDDS::Federator::TopicUpdate::sender, subscriptionWriter_, Update::Topic, and topicWriter_.

{
  //
  // 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;
    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;
    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;
      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;
      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;
  }
}

CORBA::Boolean OpenDDS::Federator::ManagerImpl::discover_federation

(

const char *

ior

)

[virtual]

Definition at line 902 of file FederatorManagerImpl.cpp.

References OpenDDS::DCPS::DCPS_debug_level.

{
  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;
}

RepoKey OpenDDS::Federator::ManagerImpl::federation_id

(

)

[virtual]

Definition at line 865 of file FederatorManagerImpl.cpp.

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

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

  return this->id().id();
}

void OpenDDS::Federator::ManagerImpl::finalize

(

)

Release resources gracefully.

Definition at line 787 of file FederatorManagerImpl.cpp.

References OpenDDS::DCPS::DCPS_debug_level, federated_, orb_, ownerListener_, participantListener_, peers_, publicationListener_, DDS::RETCODE_PRECONDITION_NOT_MET, 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 (0 == CORBA::is_nil(this->federationParticipant_.in())) {
    DDS::DomainParticipantFactory_var dpf = TheParticipantFactory;
    if (DDS::RETCODE_PRECONDITION_NOT_MET
        == this->federationParticipant_->delete_contained_entities()) {
      ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: Federator::Manager ")
                 ACE_TEXT("unable to release resources for repository %d.\n"),
                 this->id().id()));

    } else if (DDS::RETCODE_PRECONDITION_NOT_MET
               == dpf->delete_participant(this->federationParticipant_.in())) {
      ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: Federator::Manager ")
                 ACE_TEXT("unable to release the participant for repository %d.\n"),
                 this->id().id()));
    }
  }
}

ACE_INLINE const TAO_DDS_DCPSFederationId & OpenDDS::Federator::ManagerImpl::id

(

)

const

Accessors for the federation Id value.

Definition at line 19 of file FederatorManagerImpl.inl.

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

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

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

ACE_INLINE TAO_DDS_DCPSInfo_i * OpenDDS::Federator::ManagerImpl::info

(

)

const

Definition at line 33 of file FederatorManagerImpl.inl.

References info_.

{
  return this->info_;
}

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

(

)

Accessors for the DCPSInfo reference.

Definition at line 26 of file FederatorManagerImpl.inl.

References info_.

Referenced by InfoRepo::init().

{
  return this->info_;
}

void OpenDDS::Federator::ManagerImpl::initialize

(

)

Establish the update publications and subscriptions.

Definition at line 76 of file FederatorManagerImpl.cpp.

References config_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TransportRegistry::DEFAULT_INST_PREFIX, OpenDDS::DCPS::DEFAULT_STATUS_MASK, OpenDDS::Federator::Defaults::DiscoveryRequestPort, DDS::DataWriterQos::durability, DDS::DataReaderQos::durability, OpenDDS::Federator::Config::federationDomain(), federationParticipant_, OpenDDS::DCPS::DataWriterImpl::get_publication_id(), OpenDDS::DCPS::DataReaderImpl::get_subscription_id(), DDS::DataWriterQos::history, DDS::DataReaderQos::history, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::TransportRegistry::instance(), DDS::KEEP_LAST_HISTORY_QOS, orb_, ownerListener_, OpenDDS::Federator::OWNERUPDATETOPICNAME, OpenDDS::Federator::OWNERUPDATETYPENAME, ownerWriter_, PARTICIPANT_QOS_DEFAULT, participantListener_, OpenDDS::Federator::PARTICIPANTUPDATETOPICNAME, OpenDDS::Federator::PARTICIPANTUPDATETYPENAME, participantWriter_, publicationListener_, OpenDDS::Federator::PUBLICATIONUPDATETOPICNAME, OpenDDS::Federator::PUBLICATIONUPDATETYPENAME, publicationWriter_, PUBLISHER_QOS_DEFAULT, DDS::DataWriterQos::reliability, DDS::DataReaderQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, DDS::RETCODE_OK, SUBSCRIBER_QOS_DEFAULT, subscriptionListener_, OpenDDS::Federator::SUBSCRIPTIONUPDATETOPICNAME, OpenDDS::Federator::SUBSCRIPTIONUPDATETYPENAME, subscriptionWriter_, TheParticipantFactory, TOPIC_QOS_DEFAULT, topicListener_, OpenDDS::Federator::TOPICUPDATETOPICNAME, OpenDDS::Federator::TOPICUPDATETYPENAME, topicWriter_, and DDS::TRANSIENT_LOCAL_DURABILITY_QOS.

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;
  this->federationParticipant_
  = dpf->create_participant(
      this->config_.federationDomain(),
      PARTICIPANT_QOS_DEFAULT,
      DDS::DomainParticipantListener::_nil(),
      OpenDDS::DCPS::DEFAULT_STATUS_MASK);

  if (CORBA::is_nil(this->federationParticipant_.in())) {
    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* 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* 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* 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* 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* 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_publication_id());
      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_subscription_id());
      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_publication_id());
      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_subscription_id());
      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_publication_id());
      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_subscription_id());
      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_publication_id());
      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_subscription_id());
      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_publication_id());
      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_subscription_id());
      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 */
}

void OpenDDS::Federator::ManagerImpl::initializeOwner

(

const OpenDDS::Federator::OwnerUpdate &

data

)

[virtual]

Definition at line 1127 of file FederatorManagerImpl.cpp.

References processCreate().

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

void OpenDDS::Federator::ManagerImpl::initializeParticipant

(

const OpenDDS::Federator::ParticipantUpdate &

data

)

[virtual]

Definition at line 1141 of file FederatorManagerImpl.cpp.

References processCreate().

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

void OpenDDS::Federator::ManagerImpl::initializePublication

(

const OpenDDS::Federator::PublicationUpdate &

data

)

[virtual]

Definition at line 1148 of file FederatorManagerImpl.cpp.

References processCreate().

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

void OpenDDS::Federator::ManagerImpl::initializeSubscription

(

const OpenDDS::Federator::SubscriptionUpdate &

data

)

[virtual]

Definition at line 1155 of file FederatorManagerImpl.cpp.

References processCreate().

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

void OpenDDS::Federator::ManagerImpl::initializeTopic

(

const OpenDDS::Federator::TopicUpdate &

data

)

[virtual]

Definition at line 1134 of file FederatorManagerImpl.cpp.

References processCreate().

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

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

Definition at line 915 of file FederatorManagerImpl.cpp.

References config_, OpenDDS::DCPS::DCPS_debug_level, federated_, OpenDDS::Federator::Config::federationDomain(), initialize(), joiner_, joining_, joinRepo_, OpenDDS::Federator::NIL_REPOSITORY, orb(), peers_, pushState(), and TheServiceParticipant.

{
  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 remoteManager
    = peer->join_federation(this->_this(), 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();
}

void OpenDDS::Federator::ManagerImpl::leave_and_shutdown

(

)

[virtual]

Definition at line 1107 of file FederatorManagerImpl.cpp.

References info_, and TAO_DDS_DCPSInfo_i::shutdown().

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

void OpenDDS::Federator::ManagerImpl::leave_federation

(

RepoKey

id

)

[virtual]

Definition at line 1077 of file FederatorManagerImpl.cpp.

References OpenDDS::DCPS::DCPS_debug_level, and peers_.

{
  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 (false
      == this->info_->remove_by_owner(this->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()));
  }
}

ACE_INLINE void OpenDDS::Federator::ManagerImpl::localRepo

(

::OpenDDS::DCPS::DCPSInfo_ptr

repo

)

Capture a remote callable reference to the DCPSInfo.

Definition at line 40 of file FederatorManagerImpl.inl.

References localRepo_.

Referenced by InfoRepo::init().

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

ACE_INLINE void OpenDDS::Federator::ManagerImpl::orb

(

CORBA::ORB_ptr

value

)

Definition at line 54 of file FederatorManagerImpl.inl.

References orb_.

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

ACE_INLINE CORBA::ORB_ptr OpenDDS::Federator::ManagerImpl::orb

(

)

Accessors for the ORB.

Definition at line 47 of file FederatorManagerImpl.inl.

References orb_.

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

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

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

Create a proxy for a new topic.

Definition at line 729 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::TopicUpdate::datatype, OpenDDS::DCPS::DCPS_debug_level, deferredTopics_, OpenDDS::Federator::TopicUpdate::domain, OpenDDS::Federator::TopicUpdate::id, 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();
}

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

Create a proxy for a new participant.

Definition at line 697 of file FederatorManagerImpl_updates.cpp.

References 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_, 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();
}

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

Create a proxy for a new subscription.

Definition at line 653 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::SubscriptionUpdate::callback, OpenDDS::Federator::SubscriptionUpdate::datareader_qos, OpenDDS::DCPS::DCPS_debug_level, deferredSubscriptions_, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::expression_params, OpenDDS::Federator::SubscriptionUpdate::filter_class_name, OpenDDS::Federator::SubscriptionUpdate::filter_expression, OpenDDS::Federator::SubscriptionUpdate::id, OpenDDS::Federator::SubscriptionUpdate::participant, processDeferred(), OpenDDS::Federator::SubscriptionUpdate::subscriber_qos, OpenDDS::Federator::SubscriptionUpdate::topic, and OpenDDS::Federator::SubscriptionUpdate::transport_info.

{
  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->subscriber_qos,
                                             sample->filter_class_name,
                                             sample->filter_expression,
                                             sample->expression_params,
                                             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();
}

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

Create a proxy for a new publication.

Definition at line 612 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::PublicationUpdate::callback, OpenDDS::Federator::PublicationUpdate::datawriter_qos, OpenDDS::DCPS::DCPS_debug_level, deferredPublications_, OpenDDS::Federator::PublicationUpdate::domain, OpenDDS::Federator::PublicationUpdate::id, OpenDDS::Federator::PublicationUpdate::participant, processDeferred(), OpenDDS::Federator::PublicationUpdate::publisher_qos, OpenDDS::Federator::PublicationUpdate::topic, and OpenDDS::Federator::PublicationUpdate::transport_info.

{
  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->publisher_qos,
                                            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();
}

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

Null implementation for OwnerUpdate samples.

Definition at line 575 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, deferredOwnerships_, OpenDDS::Federator::OwnerUpdate::domain, 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();
}

void OpenDDS::Federator::ManagerImpl::processDeferred

(

)

Handle any deferred updates that might have become processable.

Definition at line 767 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, deferredOwnerships_, deferredPublications_, deferredSubscriptions_, and deferredTopics_.

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->publisher_qos,
                                               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->subscriber_qos,
                                                current->filter_class_name,
                                                current->filter_expression,
                                                current->expression_params,
                                                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;
      }
    }
  }

}

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

Delete a proxy for a topic.

Definition at line 1206 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::TopicUpdate::domain, OpenDDS::Federator::TopicUpdate::id, info_, 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")));
    }
  }
}

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

Delete a proxy for a participant.

Definition at line 1181 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::ParticipantUpdate::domain, OpenDDS::Federator::ParticipantUpdate::id, info_, 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")));
    }
  }
}

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

Delete a proxy for a subscription.

Definition at line 1151 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::id, info_, 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")));
    }
  }
}

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

Delete a proxy for a publication.

Definition at line 1121 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, OpenDDS::Federator::PublicationUpdate::id, info_, 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")));
    }
  }
}

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

Null implementation for OwnerUpdate samples.

Definition at line 1089 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, deferredOwnerships_, OpenDDS::Federator::OwnerUpdate::domain, 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")));
  }
}

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

Update the proxy filter expression params for a subscription.

Definition at line 1025 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::expression_params, OpenDDS::Federator::SubscriptionUpdate::id, info_, 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);
}

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

Update the proxy TopicQos for a topic.

Definition at line 1067 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::TopicUpdate::domain, OpenDDS::Federator::TopicUpdate::id, info_, 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);
}

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

Update the proxy ParticipantQos for a participant.

Definition at line 1048 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::ParticipantUpdate::domain, OpenDDS::Federator::ParticipantUpdate::id, info_, 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);
}

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

Update the proxy DataReaderQos for a subscription.

Definition at line 981 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::SubscriptionUpdate::datareader_qos, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::id, info_, 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);
}

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

Update the proxy DataWriterQos for a publication.

Definition at line 937 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::PublicationUpdate::datawriter_qos, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, OpenDDS::Federator::PublicationUpdate::id, info_, 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);
}

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

Process ownership changes.

Definition at line 905 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, deferredOwnerships_, OpenDDS::Federator::OwnerUpdate::domain, 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")));
  }
}

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

Update the proxy SubscriberQos for a subscription.

Definition at line 1003 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::id, info_, 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);
}

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

Update the proxy PublisherQos for a publication.

Definition at line 959 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, OpenDDS::Federator::PublicationUpdate::id, info_, 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);
}

void OpenDDS::Federator::ManagerImpl::pushState

(

Manager_ptr

peer

)

Push our current state to a remote repository.

Definition at line 1236 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::OwnerUpdate::action, OpenDDS::Federator::ParticipantUpdate::action, config_, OpenDDS::Federator::CreateEntity, OpenDDS::Federator::OwnerUpdate::domain, OpenDDS::Federator::ParticipantUpdate::domain, TAO_DDS_DCPSInfo_i::domains(), OpenDDS::Federator::Config::federationDomain(), DCPS_IR_Subscription::get_datareader_qos(), DCPS_IR_Publication::get_datawriter_qos(), DCPS_IR_Subscription::get_expr_params(), DCPS_IR_Subscription::get_filter_expression(), DCPS_IR_Subscription::get_id(), DCPS_IR_Publication::get_id(), DCPS_IR_Subscription::get_participant_id(), DCPS_IR_Publication::get_participant_id(), DCPS_IR_Publication::get_publisher_qos(), DCPS_IR_Subscription::get_subscriber_qos(), DCPS_IR_Subscription::get_topic_id(), DCPS_IR_Publication::get_topic_id(), DCPS_IR_Subscription::get_transportLocatorSeq(), DCPS_IR_Publication::get_transportLocatorSeq(), OpenDDS::Federator::ParticipantUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), info_, TAO_DDS_DCPSInfo_i::orb(), orb(), OpenDDS::Federator::OwnerUpdate::owner, OpenDDS::Federator::ParticipantUpdate::owner, OpenDDS::Federator::OwnerUpdate::participant, OpenDDS::Federator::ParticipantUpdate::qos, DCPS_IR_Subscription::reader(), OpenDDS::Federator::OwnerUpdate::sender, OpenDDS::Federator::ParticipantUpdate::sender, and DCPS_IR_Publication::writer().

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;
        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;
        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);
      }
    }
  }
}

OpenDDS::DCPS::DCPSInfo_ptr OpenDDS::Federator::ManagerImpl::repository

(

)

[virtual]

Definition at line 876 of file FederatorManagerImpl.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::RcHandle< T >::is_nil(), 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());
  }
}

void OpenDDS::Federator::ManagerImpl::requestImage

(

)

[virtual]

Implements Update::Updater.

Definition at line 26 of file FederatorManagerImpl_updates.cpp.

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

void OpenDDS::Federator::ManagerImpl::shutdown

(

)

[virtual]

Definition at line 1115 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();
}

void OpenDDS::Federator::ManagerImpl::unregisterCallback

(

)

[virtual]

Definition at line 20 of file FederatorManagerImpl_updates.cpp.

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

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

Implements Update::Updater.

Definition at line 506 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::SubscriptionUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, OpenDDS::Federator::SubscriptionUpdate::expression_params, DDS::HANDLE_NIL, OpenDDS::Federator::SubscriptionUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), 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;
  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);
}

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

Implements Update::Updater.

Definition at line 537 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::SubscriptionUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, DDS::HANDLE_NIL, OpenDDS::Federator::SubscriptionUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), 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;
  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);
}

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

Implements Update::Updater.

Definition at line 474 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::SubscriptionUpdate::action, OpenDDS::Federator::SubscriptionUpdate::datareader_qos, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::SubscriptionUpdate::domain, DDS::HANDLE_NIL, OpenDDS::Federator::SubscriptionUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), 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;
  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);
}

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

Implements Update::Updater.

Definition at line 442 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::PublicationUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, DDS::HANDLE_NIL, OpenDDS::Federator::PublicationUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), 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;
  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);
}

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

Implements Update::Updater.

Definition at line 410 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::PublicationUpdate::action, OpenDDS::Federator::PublicationUpdate::datawriter_qos, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::PublicationUpdate::domain, DDS::HANDLE_NIL, OpenDDS::Federator::PublicationUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), 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;
  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);
}

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

Implements Update::Updater.

Definition at line 378 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::TopicUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::TopicUpdate::domain, DDS::HANDLE_NIL, OpenDDS::Federator::TopicUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), 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;
  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);
}

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

Implements Update::Updater.

Definition at line 349 of file FederatorManagerImpl_updates.cpp.

References OpenDDS::Federator::ParticipantUpdate::action, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::Federator::ParticipantUpdate::domain, DDS::HANDLE_NIL, OpenDDS::Federator::ParticipantUpdate::id, TAO_DDS_DCPSFederationId::id(), id(), 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;
  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);
}

---

Member Data Documentation

Config& OpenDDS::Federator::ManagerImpl::config_ [private]

The configuration information for this manager.

Definition at line 231 of file FederatorManagerImpl.h.

Referenced by id(), initialize(), join_federation(), and pushState().

ACE_Thread_Mutex OpenDDS::Federator::ManagerImpl::deferred_lock_ [private]

Protect deferred updates.

Definition at line 294 of file FederatorManagerImpl.h.

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().

std::list<PublicationUpdate> OpenDDS::Federator::ManagerImpl::deferredPublications_ [private]

Deferred publication updates.

Definition at line 285 of file FederatorManagerImpl.h.

Referenced by processCreate(), and processDeferred().

std::list<SubscriptionUpdate> OpenDDS::Federator::ManagerImpl::deferredSubscriptions_ [private]

Deferred subscription updates.

Definition at line 288 of file FederatorManagerImpl.h.

Referenced by processCreate(), and processDeferred().

std::list<TopicUpdate> OpenDDS::Federator::ManagerImpl::deferredTopics_ [private]

Deferred topic updates.

Definition at line 282 of file FederatorManagerImpl.h.

Referenced by processCreate(), and processDeferred().

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().

DDS::DomainParticipant_var OpenDDS::Federator::ManagerImpl::federationParticipant_ [private]

local DomainParticipant

Definition at line 246 of file FederatorManagerImpl.h.

Referenced by initialize().

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(), processCreate(), processDelete(), processUpdateFilterExpressionParams(), processUpdateQos1(), processUpdateQos2(), pushState(), and shutdown().

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().

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().

RepoKey OpenDDS::Federator::ManagerImpl::joinRepo_ [private]

Repository to which we joined.

Definition at line 215 of file FederatorManagerImpl.h.

Referenced by join_federation().

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().

ACE_SYNCH_MUTEX OpenDDS::Federator::ManagerImpl::lock_ [private]

Critical section MUTEX.

Definition at line 206 of file FederatorManagerImpl.h.

bool OpenDDS::Federator::ManagerImpl::multicastEnabled_ [private]

Is multicast enabled?

Definition at line 291 of file FederatorManagerImpl.h.

Referenced by ManagerImpl().

InfoRepoMulticastResponder OpenDDS::Federator::ManagerImpl::multicastResponder_ [private]

Multicast responder.

Definition at line 243 of file FederatorManagerImpl.h.

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().

UpdateListener<OwnerUpdate, OwnerUpdateDataReader> OpenDDS::Federator::ManagerImpl::ownerListener_ [private]

TopicUpdate listener.

Definition at line 249 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

OwnerUpdateDataWriter_var OpenDDS::Federator::ManagerImpl::ownerWriter_ [private]

TopicUpdate writer.

Definition at line 264 of file FederatorManagerImpl.h.

Referenced by create(), and initialize().

UpdateListener<ParticipantUpdate, ParticipantUpdateDataReader> OpenDDS::Federator::ManagerImpl::participantListener_ [private]

ParticipantUpdate listener.

Definition at line 255 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

ParticipantUpdateDataWriter_var OpenDDS::Federator::ManagerImpl::participantWriter_ [private]

ParticipantUpdate writer.

Definition at line 270 of file FederatorManagerImpl.h.

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

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().

UpdateListener<PublicationUpdate, PublicationUpdateDataReader> OpenDDS::Federator::ManagerImpl::publicationListener_ [private]

PublicationUpdate listener.

Definition at line 258 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

PublicationUpdateDataWriter_var OpenDDS::Federator::ManagerImpl::publicationWriter_ [private]

PublicationUpdate writer.

Definition at line 273 of file FederatorManagerImpl.h.

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

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.

UpdateListener<SubscriptionUpdate, SubscriptionUpdateDataReader> OpenDDS::Federator::ManagerImpl::subscriptionListener_ [private]

SubscriptionUpdate listener.

Definition at line 261 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

SubscriptionUpdateDataWriter_var OpenDDS::Federator::ManagerImpl::subscriptionWriter_ [private]

SubscriptionUpdate writer.

Definition at line 276 of file FederatorManagerImpl.h.

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

UpdateListener<TopicUpdate, TopicUpdateDataReader> OpenDDS::Federator::ManagerImpl::topicListener_ [private]

TopicUpdate listener.

Definition at line 252 of file FederatorManagerImpl.h.

Referenced by finalize(), and initialize().

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:

• FederatorManagerImpl.h
• FederatorManagerImpl.cpp
• FederatorManagerImpl.inl
• FederatorManagerImpl_updates.cpp

---