OpenDDS: OpenDDS::DCPS::Service_Participant Class Reference

anonymous enum

Domain value for the default repository IOR.

Enumerator:

ANY_DOMAIN

Definition at line 62 of file Service_Participant.h.

00062 { ANY_DOMAIN = -1 };

OpenDDS::DCPS::Service_Participant::Service_Participant ( )

Constructor.

Definition at line 149 of file Service_Participant.cpp.

References initialize().

00150   :
00151 #ifndef OPENDDS_SAFETY_PROFILE
00152     ORB_argv_(false /*substitute_env_args*/),
00153 #endif
00154     reactor_(0),
00155     dp_factory_servant_(0),
00156     defaultDiscovery_(DDS_DEFAULT_DISCOVERY_METHOD),
00157     n_chunks_(DEFAULT_NUM_CHUNKS),
00158     association_chunk_multiplier_(DEFAULT_CHUNK_MULTIPLIER),
00159     liveliness_factor_(80),
00160     bit_transport_port_(0),
00161     bit_enabled_(
00162 #ifdef DDS_HAS_MINIMUM_BIT
00163       false
00164 #else
00165       true
00166 #endif
00167     ),
00168     bit_lookup_duration_msec_(BIT_LOOKUP_DURATION_MSEC),
00169     global_transport_config_(ACE_TEXT("")),
00170     monitor_factory_(0),
00171     monitor_(0),
00172     federation_recovery_duration_(DEFAULT_FEDERATION_RECOVERY_DURATION),
00173     federation_initial_backoff_seconds_(DEFAULT_FEDERATION_INITIAL_BACKOFF_SECONDS),
00174     federation_backoff_multiplier_(DEFAULT_FEDERATION_BACKOFF_MULTIPLIER),
00175     federation_liveliness_(DEFAULT_FEDERATION_LIVELINESS),
00176     schedulerQuantum_(ACE_Time_Value::zero),
00177 #if defined OPENDDS_SAFETY_PROFILE && defined ACE_HAS_ALLOC_HOOKS
00178     pool_size_(1024*1024*16),
00179     pool_granularity_(8),
00180 #endif
00181     scheduler_(-1),
00182     priority_min_(0),
00183     priority_max_(0),
00184     publisher_content_filter_(true),
00185 #ifndef OPENDDS_NO_PERSISTENCE_PROFILE
00186     transient_data_cache_(),
00187     persistent_data_cache_(),
00188     persistent_data_dir_(DEFAULT_PERSISTENT_DATA_DIR),
00189 #endif
00190     pending_timeout_(ACE_Time_Value::zero),
00191     shut_down_(false)
00192 {
00193   initialize();
00194 }

OpenDDS::DCPS::Service_Participant::~Service_Participant ( )

Destructor.

Definition at line 196 of file Service_Participant.cpp.

References OpenDDS::DCPS::DCPS_debug_level, monitor_, OPENDDS_MAP(), OPENDDS_STRING, reactor_, and shutdown().

00197 {
00198   shutdown();
00199   ACE_GUARD(TAO_SYNCH_MUTEX, guard, this->factory_lock_);
00200   typedef OPENDDS_MAP(OPENDDS_STRING, Discovery::Config*)::iterator iter_t;
00201   for (iter_t it = discovery_types_.begin(); it != discovery_types_.end(); ++it) {
00202     delete it->second;
00203   }
00204   delete monitor_;
00205   delete reactor_;
00206 
00207   if (DCPS_debug_level > 0) {
00208     ACE_DEBUG((LM_DEBUG,
00209                "%T (%P|%t) Service_Participant::~Service_Participant()\n"));
00210   }
00211 }

void OpenDDS::DCPS::Service_Participant::add_discovery ( Discovery_rch discovery )

Definition at line 1845 of file Service_Participant.cpp.

References discoveryMap_.

Referenced by load_configuration().

01846 {
01847   if (discovery != 0) {
01848     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->maps_lock_);
01849     this->discoveryMap_[discovery->key()] = discovery;
01850   }
01851 }

void OpenDDS::DCPS::Service_Participant::association_chunk_multiplier ( size_t multiplier )

See accessor description.

Definition at line 1237 of file Service_Participant.cpp.

References association_chunk_multiplier_, and OpenDDS::DCPS::got_chunk_association_multiplier.

01238 {
01239   association_chunk_multiplier_ = multiplier;
01240   got_chunk_association_multiplier = true;
01241 }

size_t OpenDDS::DCPS::Service_Participant::association_chunk_multiplier ( ) const

This accessor is to provide the multiplier for allocators that have resources used on a per association basis. Has a default, can be set by the -DCPSChunkAssociationMutltiplier option, or by n_association_chunk_multiplier() setter.

Definition at line 1231 of file Service_Participant.cpp.

References association_chunk_multiplier_.

01232 {
01233   return association_chunk_multiplier_;
01234 }

void OpenDDS::DCPS::Service_Participant::bit_lookup_duration_msec ( int msec )

Accessor for bit_lookup_duration_msec_. The accessor is used for client application to configure the timeout for lookup data from the builtin topic datareader. Value is in milliseconds.

Definition at line 1211 of file Service_Participant.cpp.

References bit_lookup_duration_msec_, and OpenDDS::DCPS::got_bit_lookup_duration_msec.

01212 {
01213   bit_lookup_duration_msec_ = sec;
01214   got_bit_lookup_duration_msec = true;
01215 }

int OpenDDS::DCPS::Service_Participant::bit_lookup_duration_msec ( ) const

Accessor for bit_lookup_duration_msec_. The accessor is used for client application to configure the timeout for lookup data from the builtin topic datareader. Value is in milliseconds.

Definition at line 1205 of file Service_Participant.cpp.

References bit_lookup_duration_msec_.

01206 {
01207   return bit_lookup_duration_msec_;
01208 }

OPENDDS_STRING OpenDDS::DCPS::Service_Participant::bit_transport_ip ( ) const

Definition at line 1185 of file Service_Participant.cpp.

01186 {
01187   return ACE_TEXT_ALWAYS_CHAR(this->bit_transport_ip_.c_str());
01188 }

void OpenDDS::DCPS::Service_Participant::bit_transport_port ( int port )

Accessors for bit_transport_port_.

The accessor is used for client application to configure the local transport listening port number.

Note:: The default port is INVALID. The user needs call this function to setup the desired port number.

Definition at line 1197 of file Service_Participant.cpp.

References bit_transport_port_, and OpenDDS::DCPS::got_bit_transport_port.

01198 {
01199   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->maps_lock_);
01200   this->bit_transport_port_ = port;
01201   got_bit_transport_port = true;
01202 }

int OpenDDS::DCPS::Service_Participant::bit_transport_port ( ) const

Accessors for bit_transport_port_.

The accessor is used for client application to configure the local transport listening port number.

Note:: The default port is INVALID. The user needs call this function to setup the desired port number.

Definition at line 1191 of file Service_Participant.cpp.

References bit_transport_port_.

01192 {
01193   return this->bit_transport_port_;
01194 }

Recorder_ptr OpenDDS::DCPS::Service_Participant::create_recorder	(	DDS::DomainParticipant_ptr	participant,
		DDS::Topic_ptr	a_topic,
		const DDS::SubscriberQos &	subscriber_qos,
		const DDS::DataReaderQos &	datareader_qos,
		const RecorderListener_rch &	a_listener
	)

Create a Recorder object.

Definition at line 1866 of file Service_Participant.cpp.

References OpenDDS::DCPS::DomainParticipantImpl::create_recorder().

01871 {
01872   DomainParticipantImpl* participant_servant = dynamic_cast<DomainParticipantImpl*>(participant);
01873   if (participant_servant)
01874     return participant_servant->create_recorder(a_topic, subscriber_qos, datareader_qos, a_listener, 0);
01875   return 0;
01876 }

Replayer_ptr OpenDDS::DCPS::Service_Participant::create_replayer	(	DDS::DomainParticipant_ptr	participant,
		DDS::Topic_ptr	a_topic,
		const DDS::PublisherQos &	publisher_qos,
		const DDS::DataWriterQos &	datawriter_qos,
		const ReplayerListener_rch &	a_listener
	)

Create a Replayer object

Definition at line 1891 of file Service_Participant.cpp.

References OpenDDS::DCPS::DomainParticipantImpl::create_replayer().

01896 {
01897   ACE_DEBUG((LM_DEBUG, "Service_Participant::create_replayer\n"));
01898   DomainParticipantImpl* participant_servant = dynamic_cast<DomainParticipantImpl*>(participant);
01899   if (participant_servant)
01900     return participant_servant->create_replayer(a_topic, publisher_qos, datawriter_qos, a_listener, 0);
01901   return 0;
01902 }

DDS::Topic_ptr OpenDDS::DCPS::Service_Participant::create_typeless_topic	(	DDS::DomainParticipant_ptr	participant,
		const char *	topic_name,
		const char *	type_name,
		bool	type_has_keys,
		const DDS::TopicQos &	qos,
		DDS::TopicListener_ptr	a_listener = `0`,
		DDS::StatusMask	mask = `0`
	)

Create a topic that does not have the data type registered.

Definition at line 1917 of file Service_Participant.cpp.

References OpenDDS::DCPS::DomainParticipantImpl::create_typeless_topic().

01924 {
01925   DomainParticipantImpl* participant_servant = dynamic_cast<DomainParticipantImpl*>(participant);
01926   if (! participant_servant) {
01927     return 0;
01928   }
01929   return participant_servant->create_typeless_topic(topic_name, type_name, type_has_keys, qos, a_listener, mask);
01930 }

ACE_INLINE ACE_CString OpenDDS::DCPS::Service_Participant::default_address ( ) const

Definition at line 344 of file Service_Participant.inl.

References default_address_.

00345 {
00346   return this->default_address_;
00347 }

DDS::ReturnCode_t OpenDDS::DCPS::Service_Participant::delete_recorder ( Recorder_ptr recorder )

Delete an existing Recorder from its DomainParticipant.

Definition at line 1879 of file Service_Participant.cpp.

References OpenDDS::DCPS::RecorderImpl::cleanup(), OpenDDS::DCPS::DomainParticipantImpl::delete_recorder(), OpenDDS::DCPS::RecorderImpl::participant(), and DDS::RETCODE_ERROR.

01880 {
01881   DDS::ReturnCode_t ret = DDS::RETCODE_ERROR;
01882   RecorderImpl* impl = static_cast<RecorderImpl*>(recorder);
01883   if (impl){
01884     ret = impl->cleanup();
01885     impl->participant()->delete_recorder(recorder);
01886   }
01887   return ret;
01888 }

DDS::ReturnCode_t OpenDDS::DCPS::Service_Participant::delete_replayer ( Replayer_ptr replayer )

Delete an existing Replayer from its DomainParticipant.

Definition at line 1905 of file Service_Participant.cpp.

References OpenDDS::DCPS::ReplayerImpl::cleanup(), OpenDDS::DCPS::DomainParticipantImpl::delete_replayer(), OpenDDS::DCPS::ReplayerImpl::participant(), and DDS::RETCODE_ERROR.

01906 {
01907   DDS::ReturnCode_t ret = DDS::RETCODE_ERROR;
01908   ReplayerImpl* impl = static_cast<ReplayerImpl*>(replayer);
01909   if (impl) {
01910     ret = impl->cleanup();
01911     impl->participant()->delete_replayer(replayer);
01912   }
01913   return ret;
01914 }

const Service_Participant::RepoKeyDiscoveryMap & OpenDDS::DCPS::Service_Participant::discoveryMap ( ) const

Definition at line 1854 of file Service_Participant.cpp.

References discoveryMap_.

01855 {
01856   return this->discoveryMap_;
01857 }

ACE_INLINE Discovery::RepoKey OpenDDS::DCPS::Service_Participant::domain_to_repo ( const DDS::DomainId_t domain ) const

Convert domainId to repository key.

Definition at line 13 of file Service_Participant.inl.

References OpenDDS::DCPS::Discovery::DEFAULT_REPO, and domainRepoMap_.

00014 {
00015   DomainRepoMap::const_iterator where = this->domainRepoMap_.find(domain);
00016 
00017   if (where == this->domainRepoMap_.end()) {
00018     return Discovery::DEFAULT_REPO;
00019 
00020   } else {
00021     return where->second;
00022   }
00023 }

const Service_Participant::DomainRepoMap & OpenDDS::DCPS::Service_Participant::domainRepoMap ( ) const

Definition at line 1860 of file Service_Participant.cpp.

References domainRepoMap_.

01861 {
01862   return this->domainRepoMap_;
01863 }

ACE_INLINE int OpenDDS::DCPS::Service_Participant::federation_backoff_multiplier ( ) const

Accessors for FederationBackoffMultiplier.

Definition at line 267 of file Service_Participant.inl.

References federation_backoff_multiplier_.

00268 {
00269   return this->federation_backoff_multiplier_;
00270 }

ACE_INLINE int & OpenDDS::DCPS::Service_Participant::federation_backoff_multiplier ( )

Accessors for FederationBackoffMultiplier.

Definition at line 260 of file Service_Participant.inl.

References federation_backoff_multiplier_.

Referenced by repository_lost().

00261 {
00262   return this->federation_backoff_multiplier_;
00263 }

ACE_INLINE int OpenDDS::DCPS::Service_Participant::federation_initial_backoff_seconds ( ) const

Accessors for FederationInitialBackoffSeconds.

Definition at line 253 of file Service_Participant.inl.

References federation_initial_backoff_seconds_.

00254 {
00255   return this->federation_initial_backoff_seconds_;
00256 }

ACE_INLINE int & OpenDDS::DCPS::Service_Participant::federation_initial_backoff_seconds ( )

Accessors for FederationInitialBackoffSeconds.

Definition at line 246 of file Service_Participant.inl.

References federation_initial_backoff_seconds_.

Referenced by repository_lost().

00247 {
00248   return this->federation_initial_backoff_seconds_;
00249 }

ACE_INLINE int OpenDDS::DCPS::Service_Participant::federation_liveliness ( ) const

Accessors for FederationLivelinessDuration.

Definition at line 281 of file Service_Participant.inl.

References federation_liveliness_.

00282 {
00283   return this->federation_liveliness_;
00284 }

ACE_INLINE int & OpenDDS::DCPS::Service_Participant::federation_liveliness ( )

Accessors for FederationLivelinessDuration.

Definition at line 274 of file Service_Participant.inl.

References federation_liveliness_.

00275 {
00276   return this->federation_liveliness_;
00277 }

ACE_INLINE int OpenDDS::DCPS::Service_Participant::federation_recovery_duration ( ) const

Accessors for FederationRecoveryDuration in seconds.

Definition at line 239 of file Service_Participant.inl.

References federation_recovery_duration_.

00240 {
00241   return this->federation_recovery_duration_;
00242 }

ACE_INLINE int & OpenDDS::DCPS::Service_Participant::federation_recovery_duration ( )

Accessors for FederationRecoveryDuration in seconds.

Definition at line 232 of file Service_Participant.inl.

References federation_recovery_duration_.

00233 {
00234   return this->federation_recovery_duration_;
00235 }

bool OpenDDS::DCPS::Service_Participant::get_BIT ( ) [inline]

Definition at line 287 of file Service_Participant.h.

00287                  {
00288     return bit_enabled_;
00289   }

DataDurabilityCache * OpenDDS::DCPS::Service_Participant::get_data_durability_cache ( DDS::DurabilityQosPolicy const & durability )

Get the data durability cache corresponding to the given DurabilityQosPolicy and sample list depth.

Definition at line 1787 of file Service_Participant.cpp.

References OpenDDS::DCPS::DCPS_debug_level, DDS::DurabilityQosPolicy::kind, persistent_data_cache_, DDS::PERSISTENT_DURABILITY_QOS, transient_data_cache_, and DDS::TRANSIENT_DURABILITY_QOS.

01789 {
01790   DDS::DurabilityQosPolicyKind const kind =
01791     durability.kind;
01792 
01793   DataDurabilityCache * cache = 0;
01794 
01795   if (kind == DDS::TRANSIENT_DURABILITY_QOS) {
01796     {
01797       ACE_GUARD_RETURN(TAO_SYNCH_MUTEX,
01798                        guard,
01799                        this->factory_lock_,
01800                        0);
01801 
01802       if (this->transient_data_cache_.get() == 0) {
01803         ACE_auto_ptr_reset(this->transient_data_cache_,
01804                            new DataDurabilityCache(kind));
01805       }
01806     }
01807 
01808     cache = this->transient_data_cache_.get();
01809 
01810   } else if (kind == DDS::PERSISTENT_DURABILITY_QOS) {
01811     {
01812       ACE_GUARD_RETURN(TAO_SYNCH_MUTEX,
01813                        guard,
01814                        this->factory_lock_,
01815                        0);
01816 
01817       try {
01818         if (this->persistent_data_cache_.get() == 0) {
01819           ACE_auto_ptr_reset(this->persistent_data_cache_,
01820                              new DataDurabilityCache(kind,
01821                                                      this->persistent_data_dir_));
01822         }
01823 
01824       } catch (const std::exception& ex) {
01825         if (DCPS_debug_level > 0) {
01826           ACE_ERROR((LM_WARNING,
01827                      ACE_TEXT("(%P|%t) WARNING: Service_Participant::get_data_durability_cache ")
01828                      ACE_TEXT("failed to create PERSISTENT cache, falling back on ")
01829                      ACE_TEXT("TRANSIENT behavior: %C\n"), ex.what()));
01830         }
01831 
01832         ACE_auto_ptr_reset(this->persistent_data_cache_,
01833                            new DataDurabilityCache(DDS::TRANSIENT_DURABILITY_QOS));
01834       }
01835     }
01836 
01837     cache = this->persistent_data_cache_.get();
01838   }
01839 
01840   return cache;
01841 }

Discovery::RepoKey OpenDDS::DCPS::Service_Participant::get_default_discovery ( )

Definition at line 1079 of file Service_Participant.cpp.

References defaultDiscovery_.

01080 {
01081   return this->defaultDiscovery_;
01082 }

Discovery_rch OpenDDS::DCPS::Service_Participant::get_discovery ( const DDS::DomainId_t domain )

Accessor of the Discovery object for a given domain.

Definition at line 1085 of file Service_Participant.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::Discovery::DEFAULT_REPO, OpenDDS::DCPS::DEFAULT_REPO_IOR, OpenDDS::DCPS::Discovery::DEFAULT_RTPS, defaultDiscovery_, discoveryMap_, domainRepoMap_, load_discovery_configuration(), OpenDDS::DCPS::RTPS_SECTION_NAME, and set_repo_ior().

Referenced by InfoRepo::init().

01086 {
01087   // Default to the Default InfoRepo-based discovery unless the user has
01088   // changed defaultDiscovery_ using the API or config file
01089   Discovery::RepoKey repo = defaultDiscovery_;
01090 
01091   // Find if this domain has a repo key (really a discovery key)
01092   // mapped to it.
01093   DomainRepoMap::const_iterator where = this->domainRepoMap_.find(domain);
01094   if (where != this->domainRepoMap_.end()) {
01095     repo = where->second;
01096   }
01097 
01098   RepoKeyDiscoveryMap::const_iterator location = this->discoveryMap_.find(repo);
01099 
01100   if (location == this->discoveryMap_.end()) {
01101     if ((repo == Discovery::DEFAULT_REPO) ||
01102         (repo == "-1")) {
01103       // Set the default repository IOR if it hasn't already happened
01104       // by this point.  This is why this can't be const.
01105       bool ok = this->set_repo_ior(DEFAULT_REPO_IOR, Discovery::DEFAULT_REPO);
01106 
01107       if (!ok) {
01108         if (DCPS_debug_level > 0) {
01109           ACE_DEBUG((LM_DEBUG,
01110                      ACE_TEXT("(%P|%t) Service_Participant::get_discovery: ")
01111                      ACE_TEXT("failed attempt to set default IOR for domain %d.\n"),
01112                      domain));
01113         }
01114 
01115       } else {
01116         // Found the default!
01117         if (DCPS_debug_level > 4) {
01118           ACE_DEBUG((LM_DEBUG,
01119                      ACE_TEXT("(%P|%t) Service_Participant::get_discovery: ")
01120                      ACE_TEXT("returning default repository for domain %d.\n"),
01121                      domain));
01122         }
01123 
01124       }
01125       return this->discoveryMap_[Discovery::DEFAULT_REPO];
01126 
01127     } else if (repo == Discovery::DEFAULT_RTPS) {
01128 
01129       ACE_Configuration_Heap cf;
01130       cf.open();
01131       ACE_Configuration_Section_Key k;
01132       cf.open_section(cf.root_section(), RTPS_SECTION_NAME, 1 /*create*/, k);
01133       this->load_discovery_configuration(cf, RTPS_SECTION_NAME);
01134 
01135       // Try to find it again
01136       location = this->discoveryMap_.find(Discovery::DEFAULT_RTPS);
01137 
01138       if (location == this->discoveryMap_.end()) {
01139         // Unable to load DEFAULT_RTPS
01140         if (DCPS_debug_level > 0) {
01141           ACE_DEBUG((LM_DEBUG,
01142                      ACE_TEXT("(%P|%t) Service_Participant::get_discovery: ")
01143                      ACE_TEXT("failed attempt to set default RTPS discovery for domain %d.\n"),
01144                      domain));
01145         }
01146 
01147         return 0;
01148 
01149       } else {
01150         // Found the default!
01151         if (DCPS_debug_level > 4) {
01152           ACE_DEBUG((LM_DEBUG,
01153                      ACE_TEXT("(%P|%t) Service_Participant::get_discovery: ")
01154                      ACE_TEXT("returning default RTPS discovery for domain %d.\n"),
01155                      domain));
01156         }
01157 
01158         return location->second;
01159       }
01160 
01161     } else {
01162       // Non-default repositories _must_ be loaded by application.
01163       if (DCPS_debug_level > 4) {
01164         ACE_DEBUG((LM_DEBUG,
01165                    ACE_TEXT("(%P|%t) Service_Participant::get_discovery: ")
01166                    ACE_TEXT("repository for domain %d was not set.\n"),
01167                    domain));
01168       }
01169 
01170       return 0;
01171     }
01172   }
01173 
01174   if (DCPS_debug_level > 4) {
01175     ACE_DEBUG((LM_DEBUG,
01176                ACE_TEXT("(%P|%t) Service_Participant::get_discovery: ")
01177                ACE_TEXT("returning repository for domain %d, repo %C.\n"),
01178                domain, repo.c_str()));
01179   }
01180 
01181   return location->second;
01182 }

DDS::DomainParticipantFactory_ptr OpenDDS::DCPS::Service_Participant::get_domain_participant_factory	(	int &	argc = `zero_argc`,
		ACE_TCHAR *	argv[] = `0`
	)

Initialize the DDS client environment and get the DomainParticipantFactory.

This method consumes -DCPS* and -ORB* options and their arguments.

: Move ORB intitialization to after the configuration file is processed and the initial scheduling policy and priority are established.

Definition at line 296 of file Service_Participant.cpp.

References OpenDDS::DCPS::config_fname(), OpenDDS::DCPS::MonitorFactory::create_sp_monitor(), OpenDDS::DCPS::DCPS_debug_level, dp_factory_, dp_factory_servant_, initializeScheduling(), monitor_, monitor_factory_, ORB_argv_, parse_args(), reactor_, reactor_task_, and OpenDDS::DCPS::Service_Participant::ReactorTask::wait_for_startup().

00298 {
00299   if (CORBA::is_nil(dp_factory_.in())) {
00300     ACE_GUARD_RETURN(TAO_SYNCH_MUTEX,
00301                      guard,
00302                      this->factory_lock_,
00303                      DDS::DomainParticipantFactory::_nil());
00304 
00305     if (CORBA::is_nil(dp_factory_.in())) {
00306       // This used to be a call to ORB_init().  Since the ORB is now managed
00307       // by InfoRepoDiscovery, just save the -ORB* args for later use.
00308       // The exceptions are -ORBLogFile and -ORBVerboseLogging, which
00309       // are processed by the service participant. This allows log control
00310       // even if an ORB is not being used.
00311 #ifndef OPENDDS_SAFETY_PROFILE
00312       ORB_argv_.add(ACE_TEXT("unused_arg_0"));
00313 #endif
00314       ACE_Arg_Shifter shifter(argc, argv);
00315       while (shifter.is_anything_left()) {
00316         if (shifter.cur_arg_strncasecmp(ACE_TEXT("-ORBLogFile")) == 0) {
00317           shifter.ignore_arg();
00318         } else if (shifter.cur_arg_strncasecmp(ACE_TEXT("-ORBVerboseLogging")) == 0) {
00319           shifter.ignore_arg();
00320         } else if (shifter.cur_arg_strncasecmp(ACE_TEXT("-ORB")) < 0) {
00321           shifter.ignore_arg();
00322         } else {
00323 #ifndef OPENDDS_SAFETY_PROFILE
00324           ORB_argv_.add(shifter.get_current());
00325 #endif
00326           shifter.consume_arg();
00327           if (shifter.is_parameter_next()) {
00328 #ifndef OPENDDS_SAFETY_PROFILE
00329             ORB_argv_.add(shifter.get_current(), true /*quote_arg*/);
00330 #endif
00331             shifter.consume_arg();
00332           }
00333         }
00334       }
00335 
00336       if (parse_args(argc, argv) != 0) {
00337         return DDS::DomainParticipantFactory::_nil();
00338       }
00339 
00340       if (config_fname == ACE_TEXT("")) {
00341         if (DCPS_debug_level) {
00342           ACE_DEBUG((LM_NOTICE,
00343                      ACE_TEXT("(%P|%t) NOTICE: not using file configuration - no configuration ")
00344                      ACE_TEXT("file specified.\n")));
00345         }
00346 
00347       } else {
00348         // Load configuration only if the configuration
00349         // file exists.
00350         FILE* in = ACE_OS::fopen(config_fname.c_str(),
00351                                  ACE_TEXT("r"));
00352 
00353         if (!in) {
00354           ACE_DEBUG((LM_WARNING,
00355                      ACE_TEXT("(%P|%t) WARNING: not using file configuration - ")
00356                      ACE_TEXT("can not open \"%s\" for reading. %p\n"),
00357                      config_fname.c_str(), ACE_TEXT("fopen")));
00358 
00359         } else {
00360           ACE_OS::fclose(in);
00361 
00362           if (this->load_configuration() != 0) {
00363             ACE_ERROR((LM_ERROR,
00364                        ACE_TEXT("(%P|%t) ERROR: Service_Participant::get_domain_participant_factory: ")
00365                        ACE_TEXT("load_configuration() failed.\n")));
00366             return DDS::DomainParticipantFactory::_nil();
00367           }
00368         }
00369       }
00370 
00371 #if defined OPENDDS_SAFETY_PROFILE && defined ACE_HAS_ALLOC_HOOKS
00372       // For non-FACE tests, configure pool
00373       configure_pool();
00374 #endif
00375 
00376       // Establish the default scheduling mechanism and
00377       // priority here.  Sadly, the ORB is already
00378       // initialized so we have no influence over its
00379       // scheduling or thread priority(ies).
00380 
00381       /// @TODO: Move ORB intitialization to after the
00382       ///        configuration file is processed and the
00383       ///        initial scheduling policy and priority are
00384       ///        established.
00385       this->initializeScheduling();
00386 
00387       ACE_NEW_RETURN(dp_factory_servant_,
00388                      DomainParticipantFactoryImpl(),
00389                      DDS::DomainParticipantFactory::_nil());
00390 
00391       dp_factory_ = dp_factory_servant_;
00392 
00393       if (CORBA::is_nil(dp_factory_.in())) {
00394         ACE_ERROR((LM_ERROR,
00395                    ACE_TEXT("(%P|%t) ERROR: ")
00396                    ACE_TEXT("Service_Participant::get_domain_participant_factory, ")
00397                    ACE_TEXT("nil DomainParticipantFactory. \n")));
00398         return DDS::DomainParticipantFactory::_nil();
00399       }
00400 
00401       if (!reactor_)
00402         reactor_ = new ACE_Reactor(new ACE_Select_Reactor, true);
00403 
00404       if (reactor_task_.activate(THR_NEW_LWP | THR_JOINABLE) == -1) {
00405         ACE_ERROR((LM_ERROR,
00406                    ACE_TEXT("ERROR: Service_Participant::get_domain_participant_factory, ")
00407                    ACE_TEXT("Failed to activate the reactor task.")));
00408         return DDS::DomainParticipantFactory::_nil();
00409       }
00410 
00411       reactor_task_.wait_for_startup();
00412 
00413       this->monitor_factory_ =
00414         ACE_Dynamic_Service<MonitorFactory>::instance ("OpenDDS_Monitor");
00415       if (this->monitor_factory_ == 0) {
00416         // Use the stubbed factory
00417         this->monitor_factory_ =
00418           ACE_Dynamic_Service<MonitorFactory>::instance ("OpenDDS_Monitor_Default");
00419       }
00420       this->monitor_ = this->monitor_factory_->create_sp_monitor(this);
00421     }
00422   }
00423 
00424   return DDS::DomainParticipantFactory::_duplicate(dp_factory_.in());
00425 }

ACE_INLINE DDS::DataReaderQos OpenDDS::DCPS::Service_Participant::initial_DataReaderQos ( ) const

Definition at line 218 of file Service_Participant.inl.

References initial_DataReaderQos_.

00219 {
00220   return initial_DataReaderQos_;
00221 }

ACE_INLINE DDS::DataWriterQos OpenDDS::DCPS::Service_Participant::initial_DataWriterQos ( ) const

Definition at line 204 of file Service_Participant.inl.

References initial_DataWriterQos_.

00205 {
00206   return initial_DataWriterQos_;
00207 }

ACE_INLINE DDS::DeadlineQosPolicy OpenDDS::DCPS::Service_Participant::initial_DeadlineQosPolicy ( ) const

Definition at line 83 of file Service_Participant.inl.

References initial_DeadlineQosPolicy_.

00084 {
00085   return initial_DeadlineQosPolicy_;
00086 }

ACE_INLINE DDS::DestinationOrderQosPolicy OpenDDS::DCPS::Service_Participant::initial_DestinationOrderQosPolicy ( ) const

Definition at line 141 of file Service_Participant.inl.

References initial_DestinationOrderQosPolicy_.

00142 {
00143   return initial_DestinationOrderQosPolicy_;
00144 }

ACE_INLINE DDS::DomainParticipantFactoryQos OpenDDS::DCPS::Service_Participant::initial_DomainParticipantFactoryQos ( ) const

Definition at line 183 of file Service_Participant.inl.

References initial_DomainParticipantFactoryQos_.

00184 {
00185   return initial_DomainParticipantFactoryQos_;
00186 }

ACE_INLINE DDS::DomainParticipantQos OpenDDS::DCPS::Service_Participant::initial_DomainParticipantQos ( ) const

Definition at line 190 of file Service_Participant.inl.

References initial_DomainParticipantQos_.

00191 {
00192   return initial_DomainParticipantQos_;
00193 }

ACE_INLINE DDS::DurabilityQosPolicy OpenDDS::DCPS::Service_Participant::initial_DurabilityQosPolicy ( ) const

Definition at line 62 of file Service_Participant.inl.

References initial_DurabilityQosPolicy_.

00063 {
00064   return initial_DurabilityQosPolicy_;
00065 }

ACE_INLINE DDS::DurabilityServiceQosPolicy OpenDDS::DCPS::Service_Participant::initial_DurabilityServiceQosPolicy ( ) const

Definition at line 69 of file Service_Participant.inl.

References initial_DurabilityServiceQosPolicy_.

00070 {
00071   return initial_DurabilityServiceQosPolicy_;
00072 }

ACE_INLINE DDS::EntityFactoryQosPolicy OpenDDS::DCPS::Service_Participant::initial_EntityFactoryQosPolicy ( ) const

Definition at line 162 of file Service_Participant.inl.

References initial_EntityFactoryQosPolicy_.

00163 {
00164   return initial_EntityFactoryQosPolicy_;
00165 }

ACE_INLINE DDS::GroupDataQosPolicy OpenDDS::DCPS::Service_Participant::initial_GroupDataQosPolicy ( ) const

Definition at line 41 of file Service_Participant.inl.

References initial_GroupDataQosPolicy_.

00042 {
00043   return initial_GroupDataQosPolicy_;
00044 }

ACE_INLINE DDS::HistoryQosPolicy OpenDDS::DCPS::Service_Participant::initial_HistoryQosPolicy ( ) const

Definition at line 148 of file Service_Participant.inl.

References initial_HistoryQosPolicy_.

00149 {
00150   return initial_HistoryQosPolicy_;
00151 }

ACE_INLINE DDS::LatencyBudgetQosPolicy OpenDDS::DCPS::Service_Participant::initial_LatencyBudgetQosPolicy ( ) const

Definition at line 90 of file Service_Participant.inl.

References initial_LatencyBudgetQosPolicy_.

00091 {
00092   return initial_LatencyBudgetQosPolicy_;
00093 }

ACE_INLINE DDS::LifespanQosPolicy OpenDDS::DCPS::Service_Participant::initial_LifespanQosPolicy ( ) const

Definition at line 55 of file Service_Participant.inl.

References initial_LifespanQosPolicy_.

00056 {
00057   return initial_LifespanQosPolicy_;
00058 }

ACE_INLINE DDS::LivelinessQosPolicy OpenDDS::DCPS::Service_Participant::initial_LivelinessQosPolicy ( ) const

Definition at line 113 of file Service_Participant.inl.

References initial_LivelinessQosPolicy_.

00114 {
00115   return initial_LivelinessQosPolicy_;
00116 }

ACE_INLINE DDS::OwnershipQosPolicy OpenDDS::DCPS::Service_Participant::initial_OwnershipQosPolicy ( ) const

Definition at line 97 of file Service_Participant.inl.

References initial_OwnershipQosPolicy_.

00098 {
00099   return initial_OwnershipQosPolicy_;
00100 }

ACE_INLINE DDS::OwnershipStrengthQosPolicy OpenDDS::DCPS::Service_Participant::initial_OwnershipStrengthQosPolicy ( ) const

Definition at line 105 of file Service_Participant.inl.

References initial_OwnershipStrengthQosPolicy_.

00106 {
00107   return initial_OwnershipStrengthQosPolicy_;
00108 }

ACE_INLINE DDS::PartitionQosPolicy OpenDDS::DCPS::Service_Participant::initial_PartitionQosPolicy ( ) const

Definition at line 127 of file Service_Participant.inl.

References initial_PartitionQosPolicy_.

00128 {
00129   return initial_PartitionQosPolicy_;
00130 }

ACE_INLINE DDS::PresentationQosPolicy OpenDDS::DCPS::Service_Participant::initial_PresentationQosPolicy ( ) const

Definition at line 76 of file Service_Participant.inl.

References initial_PresentationQosPolicy_.

00077 {
00078   return initial_PresentationQosPolicy_;
00079 }

ACE_INLINE DDS::PublisherQos OpenDDS::DCPS::Service_Participant::initial_PublisherQos ( ) const

Definition at line 211 of file Service_Participant.inl.

References initial_PublisherQos_.

00212 {
00213   return initial_PublisherQos_;
00214 }

ACE_INLINE DDS::ReaderDataLifecycleQosPolicy OpenDDS::DCPS::Service_Participant::initial_ReaderDataLifecycleQosPolicy ( ) const

Definition at line 176 of file Service_Participant.inl.

References initial_ReaderDataLifecycleQosPolicy_.

00177 {
00178   return initial_ReaderDataLifecycleQosPolicy_;
00179 }

ACE_INLINE DDS::ReliabilityQosPolicy OpenDDS::DCPS::Service_Participant::initial_ReliabilityQosPolicy ( ) const

Definition at line 134 of file Service_Participant.inl.

References initial_ReliabilityQosPolicy_.

00135 {
00136   return initial_ReliabilityQosPolicy_;
00137 }

ACE_INLINE DDS::ResourceLimitsQosPolicy OpenDDS::DCPS::Service_Participant::initial_ResourceLimitsQosPolicy ( ) const

Definition at line 155 of file Service_Participant.inl.

References initial_ResourceLimitsQosPolicy_.

00156 {
00157   return initial_ResourceLimitsQosPolicy_;
00158 }

ACE_INLINE DDS::SubscriberQos OpenDDS::DCPS::Service_Participant::initial_SubscriberQos ( ) const

Definition at line 225 of file Service_Participant.inl.

References initial_SubscriberQos_.

00226 {
00227   return initial_SubscriberQos_;
00228 }

ACE_INLINE DDS::TimeBasedFilterQosPolicy OpenDDS::DCPS::Service_Participant::initial_TimeBasedFilterQosPolicy ( ) const

Definition at line 120 of file Service_Participant.inl.

References initial_TimeBasedFilterQosPolicy_.

00121 {
00122   return initial_TimeBasedFilterQosPolicy_;
00123 }

ACE_INLINE DDS::TopicDataQosPolicy OpenDDS::DCPS::Service_Participant::initial_TopicDataQosPolicy ( ) const

Definition at line 34 of file Service_Participant.inl.

References initial_TopicDataQosPolicy_.

00035 {
00036   return initial_TopicDataQosPolicy_;
00037 }

ACE_INLINE DDS::TopicQos OpenDDS::DCPS::Service_Participant::initial_TopicQos ( ) const

Definition at line 197 of file Service_Participant.inl.

References initial_TopicQos_.

00198 {
00199   return initial_TopicQos_;
00200 }

ACE_INLINE DDS::TransportPriorityQosPolicy OpenDDS::DCPS::Service_Participant::initial_TransportPriorityQosPolicy ( ) const

Definition at line 48 of file Service_Participant.inl.

References initial_TransportPriorityQosPolicy_.

00049 {
00050   return initial_TransportPriorityQosPolicy_;
00051 }

ACE_INLINE DDS::UserDataQosPolicy OpenDDS::DCPS::Service_Participant::initial_UserDataQosPolicy ( ) const

Accessors of the qos policy initial values.

Definition at line 27 of file Service_Participant.inl.

References initial_UserDataQosPolicy_.

00028 {
00029   return initial_UserDataQosPolicy_;
00030 }

ACE_INLINE DDS::WriterDataLifecycleQosPolicy OpenDDS::DCPS::Service_Participant::initial_WriterDataLifecycleQosPolicy ( ) const

Definition at line 169 of file Service_Participant.inl.

References initial_WriterDataLifecycleQosPolicy_.

00170 {
00171   return initial_WriterDataLifecycleQosPolicy_;
00172 }

void OpenDDS::DCPS::Service_Participant::initialize ( ) [private]

Initialize default qos.

Definition at line 566 of file Service_Participant.cpp.

Referenced by Service_Participant().

00567 {
00568   //NOTE: in the future these initial values may be configurable
00569   //      (to override the Specification's default values
00570   //       hmm - I guess that would be OK since the user
00571   //       is overriding them.)
00572   initial_TransportPriorityQosPolicy_.value = 0;
00573   initial_LifespanQosPolicy_.duration.sec = DDS::DURATION_INFINITE_SEC;
00574   initial_LifespanQosPolicy_.duration.nanosec = DDS::DURATION_INFINITE_NSEC;
00575 
00576   initial_DurabilityQosPolicy_.kind = DDS::VOLATILE_DURABILITY_QOS;
00577 
00578   initial_DurabilityServiceQosPolicy_.service_cleanup_delay.sec =
00579     DDS::DURATION_ZERO_SEC;
00580   initial_DurabilityServiceQosPolicy_.service_cleanup_delay.nanosec =
00581     DDS::DURATION_ZERO_NSEC;
00582   initial_DurabilityServiceQosPolicy_.history_kind =
00583     DDS::KEEP_LAST_HISTORY_QOS;
00584   initial_DurabilityServiceQosPolicy_.history_depth = 1;
00585   initial_DurabilityServiceQosPolicy_.max_samples =
00586     DDS::LENGTH_UNLIMITED;
00587   initial_DurabilityServiceQosPolicy_.max_instances =
00588     DDS::LENGTH_UNLIMITED;
00589   initial_DurabilityServiceQosPolicy_.max_samples_per_instance =
00590     DDS::LENGTH_UNLIMITED;
00591 
00592   initial_PresentationQosPolicy_.access_scope = DDS::INSTANCE_PRESENTATION_QOS;
00593   initial_PresentationQosPolicy_.coherent_access = false;
00594   initial_PresentationQosPolicy_.ordered_access = false;
00595 
00596   initial_DeadlineQosPolicy_.period.sec = DDS::DURATION_INFINITE_SEC;
00597   initial_DeadlineQosPolicy_.period.nanosec = DDS::DURATION_INFINITE_NSEC;
00598 
00599   initial_LatencyBudgetQosPolicy_.duration.sec = DDS::DURATION_ZERO_SEC;
00600   initial_LatencyBudgetQosPolicy_.duration.nanosec = DDS::DURATION_ZERO_NSEC;
00601 
00602   initial_OwnershipQosPolicy_.kind = DDS::SHARED_OWNERSHIP_QOS;
00603 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
00604   initial_OwnershipStrengthQosPolicy_.value = 0;
00605 #endif
00606 
00607   initial_LivelinessQosPolicy_.kind = DDS::AUTOMATIC_LIVELINESS_QOS;
00608   initial_LivelinessQosPolicy_.lease_duration.sec = DDS::DURATION_INFINITE_SEC;
00609   initial_LivelinessQosPolicy_.lease_duration.nanosec = DDS::DURATION_INFINITE_NSEC;
00610 
00611   initial_TimeBasedFilterQosPolicy_.minimum_separation.sec = DDS::DURATION_ZERO_SEC;
00612   initial_TimeBasedFilterQosPolicy_.minimum_separation.nanosec = DDS::DURATION_ZERO_NSEC;
00613 
00614   initial_ReliabilityQosPolicy_.kind = DDS::BEST_EFFORT_RELIABILITY_QOS;
00615   initial_ReliabilityQosPolicy_.max_blocking_time.sec = DDS::DURATION_INFINITE_SEC;
00616   initial_ReliabilityQosPolicy_.max_blocking_time.nanosec = DDS::DURATION_INFINITE_NSEC;
00617 
00618   initial_DestinationOrderQosPolicy_.kind = DDS::BY_RECEPTION_TIMESTAMP_DESTINATIONORDER_QOS;
00619 
00620   initial_HistoryQosPolicy_.kind = DDS::KEEP_LAST_HISTORY_QOS;
00621   initial_HistoryQosPolicy_.depth = 1;
00622 
00623   initial_ResourceLimitsQosPolicy_.max_samples = DDS::LENGTH_UNLIMITED;
00624   initial_ResourceLimitsQosPolicy_.max_instances = DDS::LENGTH_UNLIMITED;
00625   initial_ResourceLimitsQosPolicy_.max_samples_per_instance = DDS::LENGTH_UNLIMITED;
00626 
00627   initial_EntityFactoryQosPolicy_.autoenable_created_entities = true;
00628 
00629   initial_WriterDataLifecycleQosPolicy_.autodispose_unregistered_instances = true;
00630 
00631   initial_ReaderDataLifecycleQosPolicy_.autopurge_nowriter_samples_delay.sec = DDS::DURATION_INFINITE_SEC;
00632   initial_ReaderDataLifecycleQosPolicy_.autopurge_nowriter_samples_delay.nanosec = DDS::DURATION_INFINITE_NSEC;
00633   initial_ReaderDataLifecycleQosPolicy_.autopurge_disposed_samples_delay.sec = DDS::DURATION_INFINITE_SEC;
00634   initial_ReaderDataLifecycleQosPolicy_.autopurge_disposed_samples_delay.nanosec = DDS::DURATION_INFINITE_NSEC;
00635 
00636   initial_DomainParticipantQos_.user_data = initial_UserDataQosPolicy_;
00637   initial_DomainParticipantQos_.entity_factory = initial_EntityFactoryQosPolicy_;
00638   initial_DomainParticipantFactoryQos_.entity_factory = initial_EntityFactoryQosPolicy_;
00639 
00640   initial_TopicQos_.topic_data = initial_TopicDataQosPolicy_;
00641   initial_TopicQos_.durability = initial_DurabilityQosPolicy_;
00642   initial_TopicQos_.durability_service = initial_DurabilityServiceQosPolicy_;
00643   initial_TopicQos_.deadline = initial_DeadlineQosPolicy_;
00644   initial_TopicQos_.latency_budget = initial_LatencyBudgetQosPolicy_;
00645   initial_TopicQos_.liveliness = initial_LivelinessQosPolicy_;
00646   initial_TopicQos_.reliability = initial_ReliabilityQosPolicy_;
00647   initial_TopicQos_.destination_order = initial_DestinationOrderQosPolicy_;
00648   initial_TopicQos_.history = initial_HistoryQosPolicy_;
00649   initial_TopicQos_.resource_limits = initial_ResourceLimitsQosPolicy_;
00650   initial_TopicQos_.transport_priority = initial_TransportPriorityQosPolicy_;
00651   initial_TopicQos_.lifespan = initial_LifespanQosPolicy_;
00652   initial_TopicQos_.ownership = initial_OwnershipQosPolicy_;
00653 
00654   initial_DataWriterQos_.durability = initial_DurabilityQosPolicy_;
00655   initial_DataWriterQos_.durability_service = initial_DurabilityServiceQosPolicy_;
00656   initial_DataWriterQos_.deadline = initial_DeadlineQosPolicy_;
00657   initial_DataWriterQos_.latency_budget = initial_LatencyBudgetQosPolicy_;
00658   initial_DataWriterQos_.liveliness = initial_LivelinessQosPolicy_;
00659   initial_DataWriterQos_.reliability = initial_ReliabilityQosPolicy_;
00660   initial_DataWriterQos_.reliability.kind = DDS::RELIABLE_RELIABILITY_QOS;
00661   initial_DataWriterQos_.reliability.max_blocking_time.sec = 0;
00662   initial_DataWriterQos_.reliability.max_blocking_time.nanosec = 100000000;
00663   initial_DataWriterQos_.destination_order = initial_DestinationOrderQosPolicy_;
00664   initial_DataWriterQos_.history = initial_HistoryQosPolicy_;
00665   initial_DataWriterQos_.resource_limits = initial_ResourceLimitsQosPolicy_;
00666   initial_DataWriterQos_.transport_priority = initial_TransportPriorityQosPolicy_;
00667   initial_DataWriterQos_.lifespan = initial_LifespanQosPolicy_;
00668   initial_DataWriterQos_.user_data = initial_UserDataQosPolicy_;
00669   initial_DataWriterQos_.ownership = initial_OwnershipQosPolicy_;
00670 #ifdef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
00671   initial_DataWriterQos_.ownership_strength.value = 0;
00672 #else
00673   initial_DataWriterQos_.ownership_strength = initial_OwnershipStrengthQosPolicy_;
00674 #endif
00675   initial_DataWriterQos_.writer_data_lifecycle = initial_WriterDataLifecycleQosPolicy_;
00676 
00677   initial_PublisherQos_.presentation = initial_PresentationQosPolicy_;
00678   initial_PublisherQos_.partition = initial_PartitionQosPolicy_;
00679   initial_PublisherQos_.group_data = initial_GroupDataQosPolicy_;
00680   initial_PublisherQos_.entity_factory = initial_EntityFactoryQosPolicy_;
00681 
00682   initial_DataReaderQos_.durability = initial_DurabilityQosPolicy_;
00683   initial_DataReaderQos_.deadline = initial_DeadlineQosPolicy_;
00684   initial_DataReaderQos_.latency_budget = initial_LatencyBudgetQosPolicy_;
00685   initial_DataReaderQos_.liveliness = initial_LivelinessQosPolicy_;
00686   initial_DataReaderQos_.reliability = initial_ReliabilityQosPolicy_;
00687   initial_DataReaderQos_.destination_order = initial_DestinationOrderQosPolicy_;
00688   initial_DataReaderQos_.history = initial_HistoryQosPolicy_;
00689   initial_DataReaderQos_.resource_limits = initial_ResourceLimitsQosPolicy_;
00690   initial_DataReaderQos_.user_data = initial_UserDataQosPolicy_;
00691   initial_DataReaderQos_.time_based_filter = initial_TimeBasedFilterQosPolicy_;
00692   initial_DataReaderQos_.ownership = initial_OwnershipQosPolicy_;
00693   initial_DataReaderQos_.reader_data_lifecycle = initial_ReaderDataLifecycleQosPolicy_;
00694 
00695   initial_SubscriberQos_.presentation = initial_PresentationQosPolicy_;
00696   initial_SubscriberQos_.partition = initial_PartitionQosPolicy_;
00697   initial_SubscriberQos_.group_data = initial_GroupDataQosPolicy_;
00698   initial_SubscriberQos_.entity_factory = initial_EntityFactoryQosPolicy_;
00699 }

void OpenDDS::DCPS::Service_Participant::initializeScheduling ( ) [private]

Initialize the thread scheduling and initial priority.

Definition at line 702 of file Service_Participant.cpp.

References OpenDDS::DCPS::DCPS_debug_level, priority_max_, priority_min_, and scheduler_.

Referenced by get_domain_participant_factory().

00703 {
00704   //
00705   // Establish the scheduler if specified.
00706   //
00707   if (this->schedulerString_.length() == 0) {
00708     if (DCPS_debug_level > 0) {
00709       ACE_DEBUG((LM_NOTICE,
00710                  ACE_TEXT("(%P|%t) NOTICE: Service_Participant::intializeScheduling() - ")
00711                  ACE_TEXT("no scheduling policy specified, not setting policy.\n")));
00712     }
00713 
00714   } else {
00715     //
00716     // Translate the scheduling policy to a usable value.
00717     //
00718     int ace_scheduler = ACE_SCHED_OTHER;
00719     this->scheduler_  = THR_SCHED_DEFAULT;
00720 
00721     if (this->schedulerString_ == ACE_TEXT("SCHED_RR")) {
00722       this->scheduler_ = THR_SCHED_RR;
00723       ace_scheduler    = ACE_SCHED_RR;
00724 
00725     } else if (this->schedulerString_ == ACE_TEXT("SCHED_FIFO")) {
00726       this->scheduler_ = THR_SCHED_FIFO;
00727       ace_scheduler    = ACE_SCHED_FIFO;
00728 
00729     } else if (this->schedulerString_ == ACE_TEXT("SCHED_OTHER")) {
00730       this->scheduler_ = THR_SCHED_DEFAULT;
00731       ace_scheduler    = ACE_SCHED_OTHER;
00732 
00733     } else {
00734       ACE_DEBUG((LM_WARNING,
00735                  ACE_TEXT("(%P|%t) WARNING: Service_Participant::initializeScheduling() - ")
00736                  ACE_TEXT("unrecognized scheduling policy: %s, set to SCHED_OTHER.\n"),
00737                  this->schedulerString_.c_str()));
00738     }
00739 
00740     //
00741     // Attempt to set the scheduling policy.
00742     //
00743 #ifdef ACE_WIN32
00744     ACE_DEBUG((LM_NOTICE,
00745                ACE_TEXT("(%P|%t) NOTICE: Service_Participant::initializeScheduling() - ")
00746                ACE_TEXT("scheduling is not implemented on Win32.\n")));
00747     ACE_UNUSED_ARG(ace_scheduler);
00748 #else
00749     ACE_Sched_Params params(
00750       ace_scheduler,
00751       ACE_Sched_Params::priority_min(ace_scheduler),
00752       ACE_SCOPE_THREAD,
00753       this->schedulerQuantum_);
00754 
00755     if (ACE_OS::sched_params(params) != 0) {
00756       if (ACE_OS::last_error() == EPERM) {
00757         ACE_DEBUG((LM_WARNING,
00758                    ACE_TEXT("(%P|%t) WARNING: Service_Participant::initializeScheduling() - ")
00759                    ACE_TEXT("user is not superuser, requested scheduler not set.\n")));
00760 
00761       } else {
00762         ACE_ERROR((LM_ERROR,
00763                    ACE_TEXT("(%P|%t) ERROR: Service_Participant::initializeScheduling() - ")
00764                    ACE_TEXT("sched_params failed: %m.\n")));
00765       }
00766 
00767       // Reset the scheduler value(s) if we did not succeed.
00768       this->scheduler_ = -1;
00769       ace_scheduler    = ACE_SCHED_OTHER;
00770 
00771     } else if (DCPS_debug_level > 0) {
00772       ACE_DEBUG((LM_DEBUG,
00773                  ACE_TEXT("(%P|%t) Service_Participant::initializeScheduling() - ")
00774                  ACE_TEXT("scheduling policy set to %s(%d).\n"),
00775                  this->schedulerString_.c_str()));
00776     }
00777 
00778     //
00779     // Setup some scheduler specific information for later use.
00780     //
00781     this->priority_min_ = ACE_Sched_Params::priority_min(ace_scheduler, ACE_SCOPE_THREAD);
00782     this->priority_max_ = ACE_Sched_Params::priority_max(ace_scheduler, ACE_SCOPE_THREAD);
00783 #endif // ACE_WIN32
00784   }
00785 }

Service_Participant * OpenDDS::DCPS::Service_Participant::instance ( ) [static]

Return a singleton instance of this class.

Definition at line 214 of file Service_Participant.cpp.

Referenced by OpenDDS::DCPS::Service_Participant::ReactorTask::svc().

00215 {
00216   // Hide the template instantiation to prevent multiple instances
00217   // from being created.
00218 
00219   return ACE_Singleton<Service_Participant, ACE_SYNCH_MUTEX>::instance();
00220 }

ACE_INLINE bool OpenDDS::DCPS::Service_Participant::is_shut_down ( ) const

Accessor for if the participant has been shutdown.

Definition at line 337 of file Service_Participant.inl.

References shut_down_.

00338 {
00339   return this->shut_down_;
00340 }

int OpenDDS::DCPS::Service_Participant::liveliness_factor ( ) const

Accessor of the Liveliness propagation delay factor.

Returns:: % of lease period before sending a liveliness message.

Definition at line 1251 of file Service_Participant.cpp.

References liveliness_factor_.

01252 {
01253   return liveliness_factor_;
01254 }

void OpenDDS::DCPS::Service_Participant::liveliness_factor ( int factor )

Set the Liveliness propagation delay factor.

Parameters:

factor

% of lease period before sending a liveliness message.

Definition at line 1244 of file Service_Participant.cpp.

References OpenDDS::DCPS::got_liveliness_factor, and liveliness_factor_.

01245 {
01246   liveliness_factor_ = factor;
01247   got_liveliness_factor = true;
01248 }

int OpenDDS::DCPS::Service_Participant::load_common_configuration	(	ACE_Configuration_Heap &	cf,
		const ACE_TCHAR *	filename
	)			`[private]`

Load the common configuration to the Service_Participant singleton.

Note:: The values from command line can overwrite the values in configuration file.

Definition at line 1383 of file Service_Participant.cpp.

Referenced by load_configuration().

01385 {
01386   const ACE_Configuration_Section_Key &root = cf.root_section();
01387   ACE_Configuration_Section_Key sect;
01388 
01389   if (cf.open_section(root, COMMON_SECTION_NAME, 0, sect) != 0) {
01390     if (DCPS_debug_level > 0) {
01391       // This is not an error if the configuration file does not have
01392       // a common section. The code default configuration will be used.
01393       ACE_DEBUG((LM_NOTICE,
01394                  ACE_TEXT("(%P|%t) NOTICE: Service_Participant::load_common_configuration ")
01395                  ACE_TEXT("failed to open section %s\n"),
01396                  COMMON_SECTION_NAME));
01397     }
01398 
01399     return 0;
01400 
01401   } else {
01402     if (got_debug_level) {
01403       ACE_DEBUG((LM_NOTICE,
01404                  ACE_TEXT("(%P|%t) NOTICE: using DCPSDebugLevel value from command option (overrides value if it's in config file).\n")));
01405     } else {
01406       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSDebugLevel"), DCPS_debug_level, int)
01407     }
01408 
01409     if (got_info) {
01410       ACE_DEBUG((LM_NOTICE,
01411                  ACE_TEXT("(%P|%t) NOTICE: using DCPSInfoRepo value from command option (overrides value if it's in config file).\n")));
01412     } else {
01413       ACE_TString value;
01414       GET_CONFIG_TSTRING_VALUE(cf, sect, ACE_TEXT("DCPSInfoRepo"), value)
01415       if (!value.empty()) {
01416         this->set_repo_ior(value.c_str(), Discovery::DEFAULT_REPO);
01417       }
01418     }
01419 
01420     if (got_use_rti_serialization) {
01421       ACE_DEBUG((LM_NOTICE,
01422                  ACE_TEXT("(%P|%t) NOTICE: using DCPSRTISerialization value from command option (overrides value if it's in config file).\n")));
01423     } else {
01424       bool should_use = false;
01425       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSRTISerialization"), should_use, bool)
01426       Serializer::set_use_rti_serialization(should_use);
01427     }
01428 
01429     if (got_chunks) {
01430       ACE_DEBUG((LM_NOTICE,
01431                  ACE_TEXT("(%P|%t) NOTICE: using DCPSChunks value from command option (overrides value if it's in config file).\n")));
01432     } else {
01433       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSChunks"), this->n_chunks_, size_t)
01434     }
01435 
01436     if (got_chunk_association_multiplier) {
01437       ACE_DEBUG((LM_NOTICE,
01438                  ACE_TEXT("(%P|%t) NOTICE: using DCPSChunkAssociationMutltiplier value from command option (overrides value if it's in config file).\n")));
01439     } else {
01440       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSChunkAssociationMutltiplier"), this->association_chunk_multiplier_, size_t)
01441     }
01442 
01443     if (got_bit_transport_port) {
01444       ACE_DEBUG((LM_NOTICE,
01445                  ACE_TEXT("(%P|%t) NOTICE: using DCPSBitTransportPort value from command option (overrides value if it's in config file).\n")));
01446     } else {
01447       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSBitTransportPort"), this->bit_transport_port_, int)
01448     }
01449 
01450     if (got_bit_transport_ip) {
01451       ACE_DEBUG((LM_DEBUG,
01452                  ACE_TEXT("(%P|%t) NOTICE: using DCPSBitTransportIPAddress value from command option (overrides value if it's in config file).\n")));
01453     } else {
01454       GET_CONFIG_TSTRING_VALUE(cf, sect, ACE_TEXT("DCPSBitTransportIPAddress"), this->bit_transport_ip_)
01455     }
01456 
01457     if (got_liveliness_factor) {
01458       ACE_DEBUG((LM_DEBUG,
01459                  ACE_TEXT("(%P|%t) NOTICE: using DCPSLivelinessFactor value from command option (overrides value if it's in config file).\n")));
01460     } else {
01461       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSLivelinessFactor"), this->liveliness_factor_, int)
01462     }
01463 
01464     if (got_bit_lookup_duration_msec) {
01465       ACE_DEBUG((LM_DEBUG,
01466                  ACE_TEXT("(%P|%t) NOTICE: using DCPSBitLookupDurationMsec value from command option (overrides value if it's in config file).\n")));
01467     } else {
01468       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSBitLookupDurationMsec"), this->bit_lookup_duration_msec_, int)
01469     }
01470 
01471     if (got_global_transport_config) {
01472       ACE_DEBUG((LM_DEBUG,
01473                  ACE_TEXT("(%P|%t) NOTICE: using DCPSGlobalTransportConfig value from command option (overrides value if it's in config file).\n")));
01474     } else {
01475       GET_CONFIG_TSTRING_VALUE(cf, sect, ACE_TEXT("DCPSGlobalTransportConfig"), this->global_transport_config_);
01476       if (this->global_transport_config_ == ACE_TEXT("$file")) {
01477         // When the special string of "$file" is used, substitute the file name
01478         this->global_transport_config_ = filename;
01479       }
01480     }
01481 
01482     if (got_bit_flag) {
01483       ACE_DEBUG((LM_DEBUG,
01484                  ACE_TEXT("(%P|%t) NOTICE: using DCPSBit value from command option (overrides value if it's in config file).\n")));
01485     } else {
01486       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSBit"), this->bit_enabled_, int)
01487     }
01488 
01489     if (got_transport_debug_level) {
01490       ACE_DEBUG((LM_NOTICE,
01491                  ACE_TEXT("(%P|%t) NOTICE: using DCPSTransportDebugLevel value from command option (overrides value if it's in config file).\n")));
01492     } else {
01493       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSTransportDebugLevel"), OpenDDS::DCPS::Transport_debug_level, int)
01494     }
01495 
01496 #ifndef OPENDDS_NO_PERSISTENCE_PROFILE
01497     if (got_persistent_data_dir) {
01498       ACE_DEBUG((LM_NOTICE,
01499                  ACE_TEXT("(%P|%t) NOTICE: using DCPSPersistentDataDir value from command option (overrides value if it's in config file).\n")));
01500     } else {
01501       ACE_TString value;
01502       GET_CONFIG_TSTRING_VALUE(cf, sect, ACE_TEXT("DCPSPersistentDataDir"), value)
01503       this->persistent_data_dir_ = ACE_TEXT_ALWAYS_CHAR(value.c_str());
01504     }
01505 #endif
01506 
01507     if (got_pending_timeout) {
01508       ACE_DEBUG((LM_NOTICE,
01509                  ACE_TEXT("(%P|%t) NOTICE: using DCPSPendingTimeout value from command option (overrides value if it's in config file).\n")));
01510     } else {
01511       int timeout = 0;
01512       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSPendingTimeout"), timeout, int)
01513       this->pending_timeout_ = timeout;
01514     }
01515 
01516     if (got_publisher_content_filter) {
01517       ACE_DEBUG((LM_DEBUG,
01518                  ACE_TEXT("(%P|%t) NOTICE: using DCPSPublisherContentFilter ")
01519                  ACE_TEXT("value from command option (overrides value if it's ")
01520                  ACE_TEXT("in config file).\n")));
01521     } else {
01522       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("DCPSPublisherContentFilter"),
01523         this->publisher_content_filter_, bool)
01524     }
01525 
01526     if (got_default_discovery) {
01527       ACE_Configuration::VALUETYPE type;
01528       if (cf.find_value(sect, ACE_TEXT("DCPSDefaultDiscovery"), type) != -1) {
01529         ACE_DEBUG((LM_NOTICE,
01530                    ACE_TEXT("(%P|%t) NOTICE: using DCPSDefaultDiscovery value ")
01531                    ACE_TEXT("from command option, overriding config file\n")));
01532       }
01533     } else {
01534       GET_CONFIG_STRING_VALUE(cf, sect, ACE_TEXT("DCPSDefaultDiscovery"),
01535         this->defaultDiscovery_);
01536     }
01537 
01538     ACE_Configuration::VALUETYPE type;
01539     if (got_log_fname) {
01540       if (cf.find_value(sect, ACE_TEXT("ORBLogFile"), type) != -1) {
01541         ACE_DEBUG((LM_NOTICE,
01542                    ACE_TEXT("(%P|%t) NOTICE: using ORBLogFile value ")
01543                    ACE_TEXT("from command option, overriding config file\n")));
01544       }
01545     } else {
01546       OPENDDS_STRING log_fname;
01547       GET_CONFIG_STRING_VALUE(cf, sect, ACE_TEXT("ORBLogFile"), log_fname);
01548       if (!log_fname.empty()) {
01549         set_log_file_name(log_fname.c_str());
01550       }
01551     }
01552 
01553     if (got_log_verbose) {
01554       if (cf.find_value(sect, ACE_TEXT("ORBVerboseLogging"), type) != -1) {
01555         ACE_DEBUG((LM_NOTICE,
01556                    ACE_TEXT("(%P|%t) NOTICE: using ORBVerboseLogging value ")
01557                    ACE_TEXT("from command option, overriding config file\n")));
01558       }
01559     } else {
01560       unsigned long verbose_logging = 0;
01561       GET_CONFIG_VALUE(cf, sect, ACE_TEXT("ORBVerboseLogging"), verbose_logging, unsigned long);
01562       set_log_verbose(verbose_logging);
01563     }
01564 
01565     if (got_default_address) {
01566       ACE_DEBUG((LM_DEBUG,
01567                  ACE_TEXT("(%P|%t) NOTICE: using DCPSDefaultAddress value from command option (overrides value if it's in config file).\n")));
01568     } else {
01569       GET_CONFIG_STRING_VALUE(cf, sect, ACE_TEXT("DCPSDefaultAddress"), this->default_address_)
01570     }
01571 
01572     // These are not handled on the command line.
01573     GET_CONFIG_VALUE(cf, sect, ACE_TEXT("FederationRecoveryDuration"), this->federation_recovery_duration_, int)
01574     GET_CONFIG_VALUE(cf, sect, ACE_TEXT("FederationInitialBackoffSeconds"), this->federation_initial_backoff_seconds_, int)
01575     GET_CONFIG_VALUE(cf, sect, ACE_TEXT("FederationBackoffMultiplier"), this->federation_backoff_multiplier_, int)
01576     GET_CONFIG_VALUE(cf, sect, ACE_TEXT("FederationLivelinessDuration"), this->federation_liveliness_, int)
01577 
01578 #if defined OPENDDS_SAFETY_PROFILE && defined ACE_HAS_ALLOC_HOOKS
01579     GET_CONFIG_VALUE(cf, sect, ACE_TEXT("pool_size"), pool_size_, size_t)
01580     GET_CONFIG_VALUE(cf, sect, ACE_TEXT("pool_granularity"), pool_granularity_, size_t)
01581 #endif
01582 
01583     //
01584     // Establish the scheduler if specified.
01585     //
01586     GET_CONFIG_TSTRING_VALUE(cf, sect, ACE_TEXT("scheduler"), this->schedulerString_)
01587 
01588     suseconds_t usec(0);
01589 
01590     GET_CONFIG_VALUE(cf, sect, ACE_TEXT("scheduler_slice"), usec, suseconds_t)
01591 
01592     if (usec > 0)
01593       this->schedulerQuantum_.usec(usec);
01594   }
01595 
01596   return 0;
01597 }

int OpenDDS::DCPS::Service_Participant::load_configuration ( ) [private]

Import the configuration file to the ACE_Configuration_Heap object and load common section configuration to the Service_Participant singleton and load the factory and transport section configuration to the TransportRegistry singleton.

Definition at line 1265 of file Service_Participant.cpp.

References cf_, and OpenDDS::DCPS::config_fname().

01266 {
01267   int status = 0;
01268 
01269   if ((status = this->cf_.open()) != 0)
01270     ACE_ERROR_RETURN((LM_ERROR,
01271                       ACE_TEXT("(%P|%t) ERROR: Service_Participant::load_configuration ")
01272                       ACE_TEXT("open() returned %d\n"),
01273                       status),
01274                      -1);
01275 
01276   ACE_Ini_ImpExp import(this->cf_);
01277   status = import.import_config(config_fname.c_str());
01278 
01279   if (status != 0) {
01280     ACE_ERROR_RETURN((LM_ERROR,
01281                       ACE_TEXT("(%P|%t) ERROR: Service_Participant::load_configuration ")
01282                       ACE_TEXT("import_config () returned %d\n"),
01283                       status),
01284                      -1);
01285   } else {
01286     status = this->load_configuration(this->cf_, config_fname.c_str());
01287   }
01288   return status;
01289 }

int OpenDDS::DCPS::Service_Participant::load_configuration	(	ACE_Configuration_Heap &	cf,
		const ACE_TCHAR *	filename
	)

Import the configuration file to the ACE_Configuration_Heap object and load common section configuration to the Service_Participant singleton and load the factory and transport section configuration to the TransportRegistry singleton.

Definition at line 1292 of file Service_Participant.cpp.

References add_discovery(), OpenDDS::DCPS::TransportRegistry::get_config(), OpenDDS::DCPS::TransportRegistry::global_config(), OpenDDS::DCPS::TransportRegistry::instance(), OpenDDS::DCPS::StaticDiscovery::instance(), load_common_configuration(), load_discovery_configuration(), load_domain_configuration(), OpenDDS::DCPS::TransportRegistry::load_transport_configuration(), OpenDDS::DCPS::REPO_SECTION_NAME, and OpenDDS::DCPS::RTPS_SECTION_NAME.

01295 {
01296   int status = 0;
01297 
01298   status = this->load_common_configuration(config, filename);
01299 
01300   if (status != 0) {
01301     ACE_ERROR_RETURN((LM_ERROR,
01302                       ACE_TEXT("(%P|%t) ERROR: Service_Participant::load_configuration ")
01303                       ACE_TEXT("load_common_configuration () returned %d\n"),
01304                       status),
01305                      -1);
01306   }
01307 
01308   // Register static discovery.
01309   this->add_discovery(static_rchandle_cast<Discovery>(StaticDiscovery::instance()));
01310 
01311   status = this->load_discovery_configuration(config, RTPS_SECTION_NAME);
01312 
01313   if (status != 0) {
01314     ACE_ERROR_RETURN((LM_ERROR,
01315                       ACE_TEXT("(%P|%t) ERROR: Service_Participant::load_configuration ")
01316                       ACE_TEXT("load_discovery_configuration() returned %d\n"),
01317                       status),
01318                      -1);
01319   }
01320 
01321   status = this->load_discovery_configuration(config, REPO_SECTION_NAME);
01322 
01323   if (status != 0) {
01324     ACE_ERROR_RETURN((LM_ERROR,
01325                       ACE_TEXT("(%P|%t) ERROR: Service_Participant::load_configuration ")
01326                       ACE_TEXT("load_discovery_configuration() returned %d\n"),
01327                       status),
01328                      -1);
01329   }
01330 
01331   status = TransportRegistry::instance()->load_transport_configuration(
01332              ACE_TEXT_ALWAYS_CHAR(filename), config);
01333   if (this->global_transport_config_ != ACE_TEXT("")) {
01334     TransportConfig_rch config = TransportRegistry::instance()->get_config(
01335       ACE_TEXT_ALWAYS_CHAR(this->global_transport_config_.c_str()));
01336     if (config == 0) {
01337       ACE_ERROR_RETURN((LM_ERROR,
01338                         ACE_TEXT("(%P|%t) ERROR: Service_Participant::load_configuration ")
01339                         ACE_TEXT("Unable to locate specified global transport config: %s\n"),
01340                         this->global_transport_config_.c_str()),
01341                        -1);
01342     }
01343     TransportRegistry::instance()->global_config(config);
01344   }
01345 
01346   if (status != 0) {
01347     ACE_ERROR_RETURN((LM_ERROR,
01348                       ACE_TEXT("(%P|%t) ERROR: Service_Participant::load_configuration ")
01349                       ACE_TEXT("load_transport_configuration () returned %d\n"),
01350                       status),
01351                      -1);
01352   }
01353 
01354   // Needs to be loaded after the [rtps_discovery/*] and [repository/*]
01355   // sections to allow error reporting on bad discovery config names.
01356   // Also loaded after the transport configuration so that
01357   // DefaultTransportConfig within [domain/*] can use TransportConfig objects.
01358   status = this->load_domain_configuration(config, filename);
01359 
01360   if (status != 0) {
01361     ACE_ERROR_RETURN((LM_ERROR,
01362                       ACE_TEXT("(%P|%t) ERROR: Service_Participant::load_configuration ")
01363                       ACE_TEXT("load_domain_configuration () returned %d\n"),
01364                       status),
01365                      -1);
01366   }
01367 
01368   // Needs to be loaded after transport configs and instances and domains.
01369   status = StaticDiscovery::instance()->load_configuration(config);
01370 
01371   if (status != 0) {
01372     ACE_ERROR_RETURN((LM_ERROR,
01373                       ACE_TEXT("(%P|%t) ERROR: Service_Participant::load_configuration ")
01374                       ACE_TEXT("load_discovery_configuration() returned %d\n"),
01375                       status),
01376                      -1);
01377   }
01378 
01379   return 0;
01380 }

int OpenDDS::DCPS::Service_Participant::load_discovery_configuration	(	ACE_Configuration_Heap &	cf,
		const ACE_TCHAR *	section_name
	)			`[private]`

Load the discovery configuration to the Service_Participant singleton.

Definition at line 1741 of file Service_Participant.cpp.

References OPENDDS_MAP(), OPENDDS_STRING, and TheTransportRegistry.

Referenced by get_discovery(), and load_configuration().

01743 {
01744   const ACE_Configuration_Section_Key &root = cf.root_section();
01745   ACE_Configuration_Section_Key sect;
01746   if (cf.open_section(root, section_name, 0, sect) == 0) {
01747 
01748     const OPENDDS_STRING sect_name = ACE_TEXT_ALWAYS_CHAR(section_name);
01749     OPENDDS_MAP(OPENDDS_STRING, Discovery::Config*)::iterator iter =
01750       this->discovery_types_.find(sect_name);
01751 
01752     if (iter == this->discovery_types_.end()) {
01753       // See if we can dynamically load the required libraries
01754       TheTransportRegistry->load_transport_lib(sect_name);
01755       iter = this->discovery_types_.find(sect_name);
01756     }
01757 
01758     if (iter != this->discovery_types_.end()) {
01759       // discovery code is loaded, process options
01760       return iter->second->discovery_config(cf);
01761     } else {
01762       // No discovery code can be loaded, report an error
01763       ACE_ERROR_RETURN((LM_ERROR,
01764                         ACE_TEXT("(%P|%t) ERROR: Service_Participant::")
01765                         ACE_TEXT("load_discovery_configuration ")
01766                         ACE_TEXT("Unable to load libraries for %s\n"),
01767                         section_name),
01768                        -1);
01769     }
01770   }
01771   return 0;
01772 }

int OpenDDS::DCPS::Service_Participant::load_domain_configuration	(	ACE_Configuration_Heap &	cf,
		const ACE_TCHAR *	filename
	)			`[private]`

Load the domain configuration to the Service_Participant singleton.

Definition at line 1600 of file Service_Participant.cpp.

References OpenDDS::DCPS::convertToInteger(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::Discovery::DEFAULT_REPO, OpenDDS::DCPS::Discovery::DEFAULT_RTPS, OpenDDS::DCPS::Discovery::DEFAULT_STATIC, discoveryMap_, OpenDDS::DCPS::DOMAIN_SECTION_NAME, OpenDDS::DCPS::TransportRegistry::get_config(), OpenDDS::DCPS::TransportRegistry::instance(), OpenDDS::DCPS::RcHandle< T >::is_nil(), OPENDDS_STRING, OpenDDS::DCPS::processSections(), OpenDDS::DCPS::pullValues(), and set_repo_domain().

Referenced by load_configuration().

01602 {
01603   const ACE_Configuration_Section_Key& root = cf.root_section();
01604   ACE_Configuration_Section_Key domain_sect;
01605 
01606   if (cf.open_section(root, DOMAIN_SECTION_NAME, 0, domain_sect) != 0) {
01607     if (DCPS_debug_level > 0) {
01608       // This is not an error if the configuration file does not have
01609       // any domain (sub)section. The code default configuration will be used.
01610       ACE_DEBUG((LM_NOTICE,
01611                  ACE_TEXT("(%P|%t) NOTICE: Service_Participant::load_domain_configuration ")
01612                  ACE_TEXT("failed to open [%s] section - using code default.\n"),
01613                  DOMAIN_SECTION_NAME));
01614     }
01615 
01616     return 0;
01617 
01618   } else {
01619     // Ensure there are no properties in this section
01620     ValueMap vm;
01621     if (pullValues(cf, domain_sect, vm) > 0) {
01622       // There are values inside [domain]
01623       ACE_ERROR_RETURN((LM_ERROR,
01624                         ACE_TEXT("(%P|%t) Service_Participant::load_domain_configuration(): ")
01625                         ACE_TEXT("domain sections must have a subsection name\n")),
01626                        -1);
01627     }
01628     // Process the subsections of this section (the individual domains)
01629     KeyList keys;
01630     if (processSections(cf, domain_sect, keys) != 0) {
01631       ACE_ERROR_RETURN((LM_ERROR,
01632                         ACE_TEXT("(%P|%t) Service_Participant::load_domain_configuration(): ")
01633                         ACE_TEXT("too many nesting layers in the [domain] section.\n")),
01634                        -1);
01635     }
01636 
01637     // Loop through the [domain/*] sections
01638     for (KeyList::const_iterator it = keys.begin(); it != keys.end(); ++it) {
01639       OPENDDS_STRING domain_name = it->first;
01640 
01641       ValueMap values;
01642       pullValues(cf, it->second, values);
01643       DDS::DomainId_t domainId = -1;
01644       Discovery::RepoKey repoKey;
01645       OPENDDS_STRING perDomainDefaultTportConfig;
01646       for (ValueMap::const_iterator it = values.begin(); it != values.end(); ++it) {
01647         OPENDDS_STRING name = it->first;
01648         if (name == "DomainId") {
01649           OPENDDS_STRING value = it->second;
01650           if (!convertToInteger(value, domainId)) {
01651             ACE_ERROR_RETURN((LM_ERROR,
01652                               ACE_TEXT("(%P|%t) Service_Participant::load_domain_configuration(): ")
01653                               ACE_TEXT("Illegal integer value for DomainId (%C) in [domain/%C] section.\n"),
01654                               value.c_str(), domain_name.c_str()),
01655                              -1);
01656           }
01657           if (DCPS_debug_level > 0) {
01658             ACE_DEBUG((LM_DEBUG,
01659                        ACE_TEXT("(%P|%t) [domain/%C]: DomainId == %d\n"),
01660                        domain_name.c_str(), domainId));
01661           }
01662         } else if (name == "DomainRepoKey") {
01663           // We will still process this for backward compatibility, but
01664           // it can now be replaced by "DiscoveryConfig=REPO:<key>"
01665           repoKey = it->second;
01666           if (repoKey == "-1") {
01667             repoKey = Discovery::DEFAULT_REPO;
01668           }
01669 
01670           if (DCPS_debug_level > 0) {
01671             ACE_DEBUG((LM_DEBUG,
01672                        ACE_TEXT("(%P|%t) [domain/%C]: DomainRepoKey == %C\n"),
01673                        domain_name.c_str(), repoKey.c_str()));
01674           }
01675         } else if (name == "DiscoveryConfig") {
01676           repoKey = it->second;
01677 
01678         } else if (name == "DefaultTransportConfig") {
01679           if (it->second == "$file") {
01680             // When the special string of "$file" is used, substitute the file name
01681             perDomainDefaultTportConfig = ACE_TEXT_ALWAYS_CHAR(filename);
01682 
01683           } else {
01684             perDomainDefaultTportConfig = it->second;
01685           }
01686 
01687         } else {
01688           ACE_ERROR_RETURN((LM_ERROR,
01689                             ACE_TEXT("(%P|%t) Service_Participant::load_domain_configuration(): ")
01690                             ACE_TEXT("Unexpected entry (%C) in [domain/%C] section.\n"),
01691                             name.c_str(), domain_name.c_str()),
01692                            -1);
01693         }
01694       }
01695 
01696       if (domainId == -1) {
01697         // DomainId parameter is not set, try using the domain name as an ID
01698         if (!convertToInteger(domain_name, domainId)) {
01699           ACE_ERROR_RETURN((LM_ERROR,
01700                             ACE_TEXT("(%P|%t) Service_Participant::load_domain_configuration(): ")
01701                             ACE_TEXT("Missing DomainId value in [domain/%C] section.\n"),
01702                             domain_name.c_str()),
01703                            -1);
01704         }
01705       }
01706 
01707       if (!perDomainDefaultTportConfig.empty()) {
01708         TransportRegistry* const reg = TransportRegistry::instance();
01709         TransportConfig_rch tc = reg->get_config(perDomainDefaultTportConfig);
01710         if (tc.is_nil()) {
01711           ACE_ERROR_RETURN((LM_ERROR,
01712             ACE_TEXT("(%P|%t) Service_Participant::load_domain_configuration(): ")
01713             ACE_TEXT("Unknown transport config %C in [domain/%C] section.\n"),
01714             perDomainDefaultTportConfig.c_str(), domain_name.c_str()), -1);
01715         } else {
01716           reg->domain_default_config(domainId, tc);
01717         }
01718       }
01719 
01720       // Check to see if the specified discovery configuration has been defined
01721       if (!repoKey.empty()) {
01722         if ((repoKey != Discovery::DEFAULT_REPO) &&
01723             (repoKey != Discovery::DEFAULT_RTPS) &&
01724             (repoKey != Discovery::DEFAULT_STATIC) &&
01725             (this->discoveryMap_.find(repoKey) == this->discoveryMap_.end())) {
01726           ACE_ERROR_RETURN((LM_ERROR,
01727                             ACE_TEXT("(%P|%t) Service_Participant::load_domain_configuration(): ")
01728                             ACE_TEXT("Specified configuration (%C) not found.  Referenced in [domain/%C] section.\n"),
01729                             repoKey.c_str(), domain_name.c_str()),
01730                            -1);
01731         }
01732         this->set_repo_domain(domainId, repoKey);
01733       }
01734     }
01735   }
01736 
01737   return 0;
01738 }

void OpenDDS::DCPS::Service_Participant::n_chunks ( size_t chunks )

Set the value returned by n_chunks() accessor.

See also:: Accessor description.

Definition at line 1224 of file Service_Participant.cpp.

References OpenDDS::DCPS::got_chunks, and n_chunks_.

01225 {
01226   n_chunks_ = chunks;
01227   got_chunks = true;
01228 }

size_t OpenDDS::DCPS::Service_Participant::n_chunks ( ) const

This accessor is to provide the configurable number of chunks that a DataWriter's cached allocator need to allocate when the resource limits are infinite. Has a default, can be set by the -DCPSChunks option, or by n_chunks() setter.

Definition at line 1218 of file Service_Participant.cpp.

References n_chunks_.

01219 {
01220   return n_chunks_;
01221 }

OpenDDS::DCPS::Service_Participant::OPENDDS_MAP	(	OPENDDS_STRING	,
		Discovery::Config *
	)			`[private]`

typedef OpenDDS::DCPS::Service_Participant::OPENDDS_MAP	(	DDS::DomainId_t	,
		Discovery::RepoKey
	)

typedef OpenDDS::DCPS::Service_Participant::OPENDDS_MAP	(	Discovery::RepoKey	,
		Discovery_rch
	)

For internal OpenDDS Use (needed for monitor code).

Referenced by load_discovery_configuration(), shutdown(), and ~Service_Participant().

ACE_ARGV* OpenDDS::DCPS::Service_Participant::ORB_argv ( ) [inline]

Definition at line 314 of file Service_Participant.h.

00314 { return &ORB_argv_; }

int OpenDDS::DCPS::Service_Participant::parse_args	(	int &	argc,
		ACE_TCHAR *	argv[]
	)			`[private]`

Parse the command line for user options. e.g. "-DCPSInfoRepo <iorfile>". It consumes -DCPS* options and their arguments

No need to guard this insertion as we are still single threaded here.

Definition at line 428 of file Service_Participant.cpp.

Referenced by get_domain_participant_factory().

00429 {
00430   ACE_Arg_Shifter arg_shifter(argc, argv);
00431 
00432   while (arg_shifter.is_anything_left()) {
00433     const ACE_TCHAR *currentArg = 0;
00434 
00435     if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSDebugLevel"))) != 0) {
00436       set_DCPS_debug_level(ACE_OS::atoi(currentArg));
00437       arg_shifter.consume_arg();
00438       got_debug_level = true;
00439 
00440     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSInfoRepo"))) != 0) {
00441       this->set_repo_ior(currentArg, Discovery::DEFAULT_REPO);
00442       arg_shifter.consume_arg();
00443       got_info = true;
00444 
00445     } else if (!arg_shifter.cur_arg_strncasecmp(ACE_TEXT("-DCPSRTISerialization"))) {
00446       Serializer::set_use_rti_serialization(true);
00447       arg_shifter.consume_arg();
00448       got_use_rti_serialization = true;
00449 
00450     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSChunks"))) != 0) {
00451       n_chunks_ = ACE_OS::atoi(currentArg);
00452       arg_shifter.consume_arg();
00453       got_chunks = true;
00454 
00455     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSChunkAssociationMultiplier"))) != 0) {
00456       association_chunk_multiplier_ = ACE_OS::atoi(currentArg);
00457       arg_shifter.consume_arg();
00458       got_chunk_association_multiplier = true;
00459 
00460     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSConfigFile"))) != 0) {
00461       config_fname = currentArg;
00462       arg_shifter.consume_arg();
00463 
00464     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSLivelinessFactor"))) != 0) {
00465       liveliness_factor_ = ACE_OS::atoi(currentArg);
00466       arg_shifter.consume_arg();
00467       got_liveliness_factor = true;
00468 
00469     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSBitTransportPort"))) != 0) {
00470       /// No need to guard this insertion as we are still single
00471       /// threaded here.
00472       this->bit_transport_port_ = ACE_OS::atoi(currentArg);
00473       arg_shifter.consume_arg();
00474       got_bit_transport_port = true;
00475 
00476     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSBitTransportIPAddress"))) != 0) {
00477       /// No need to guard this insertion as we are still single
00478       /// threaded here.
00479       this->bit_transport_ip_ = currentArg;
00480       arg_shifter.consume_arg();
00481       got_bit_transport_ip = true;
00482 
00483     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSBitLookupDurationMsec"))) != 0) {
00484       bit_lookup_duration_msec_ = ACE_OS::atoi(currentArg);
00485       arg_shifter.consume_arg();
00486       got_bit_lookup_duration_msec = true;
00487 
00488     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSGlobalTransportConfig"))) != 0) {
00489       global_transport_config_ = currentArg;
00490       arg_shifter.consume_arg();
00491       got_global_transport_config = true;
00492 
00493     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSBit"))) != 0) {
00494       bit_enabled_ = ACE_OS::atoi(currentArg);
00495       arg_shifter.consume_arg();
00496       got_bit_flag = true;
00497 
00498     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSTransportDebugLevel"))) != 0) {
00499       OpenDDS::DCPS::Transport_debug_level = ACE_OS::atoi(currentArg);
00500       arg_shifter.consume_arg();
00501       got_transport_debug_level = true;
00502 
00503 #ifndef OPENDDS_NO_PERSISTENCE_PROFILE
00504     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSPersistentDataDir"))) != 0) {
00505       this->persistent_data_dir_ = ACE_TEXT_ALWAYS_CHAR(currentArg);
00506       arg_shifter.consume_arg();
00507       got_persistent_data_dir = true;
00508 #endif
00509 
00510     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSPendingTimeout"))) != 0) {
00511       this->pending_timeout_ = ACE_OS::atoi(currentArg);
00512       arg_shifter.consume_arg();
00513       got_pending_timeout = true;
00514 
00515     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSPublisherContentFilter"))) != 0) {
00516       this->publisher_content_filter_ = ACE_OS::atoi(currentArg);
00517       arg_shifter.consume_arg();
00518       got_publisher_content_filter = true;
00519 
00520     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSDefaultDiscovery"))) != 0) {
00521       this->defaultDiscovery_ = ACE_TEXT_ALWAYS_CHAR(currentArg);
00522       arg_shifter.consume_arg();
00523       got_default_discovery = true;
00524 
00525     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-FederationRecoveryDuration"))) != 0) {
00526       this->federation_recovery_duration_ = ACE_OS::atoi(currentArg);
00527       arg_shifter.consume_arg();
00528 
00529     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-FederationInitialBackoffSeconds"))) != 0) {
00530       this->federation_initial_backoff_seconds_ = ACE_OS::atoi(currentArg);
00531       arg_shifter.consume_arg();
00532 
00533     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-FederationBackoffMultiplier"))) != 0) {
00534       this->federation_backoff_multiplier_ = ACE_OS::atoi(currentArg);
00535       arg_shifter.consume_arg();
00536 
00537     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-FederationLivelinessDuration"))) != 0) {
00538       this->federation_liveliness_ = ACE_OS::atoi(currentArg);
00539       arg_shifter.consume_arg();
00540 
00541     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-ORBLogFile"))) != 0) {
00542       set_log_file_name(ACE_TEXT_ALWAYS_CHAR(currentArg));
00543       arg_shifter.consume_arg();
00544       got_log_fname = true;
00545 
00546     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-ORBVerboseLogging"))) != 0) {
00547       set_log_verbose(ACE_OS::atoi(currentArg));
00548       arg_shifter.consume_arg();
00549       got_log_verbose = true;
00550 
00551     } else if ((currentArg = arg_shifter.get_the_parameter(ACE_TEXT("-DCPSDefaultAddress"))) != 0) {
00552       this->default_address_ = ACE_TEXT_ALWAYS_CHAR(currentArg);
00553       arg_shifter.consume_arg();
00554       got_default_address = true;
00555 
00556     } else {
00557       arg_shifter.ignore_arg();
00558     }
00559   }
00560 
00561   // Indicates sucessful parsing of the command line
00562   return 0;
00563 }

ACE_INLINE ACE_Time_Value OpenDDS::DCPS::Service_Participant::pending_timeout ( ) const

Accessor for pending data timeout.

Definition at line 302 of file Service_Participant.inl.

References pending_timeout_.

00303 {
00304   return this->pending_timeout_;
00305 }

ACE_INLINE int OpenDDS::DCPS::Service_Participant::priority_max ( ) const

Accessors for priority extremums for the current scheduler.

Definition at line 316 of file Service_Participant.inl.

References priority_max_.

00317 {
00318   return this->priority_max_;
00319 }

ACE_INLINE int OpenDDS::DCPS::Service_Participant::priority_min ( ) const

Accessors for priority extremums for the current scheduler.

Definition at line 309 of file Service_Participant.inl.

References priority_min_.

00310 {
00311   return this->priority_min_;
00312 }

ACE_INLINE bool OpenDDS::DCPS::Service_Participant::publisher_content_filter ( ) const

Accessors for PublisherContentFilter.

Definition at line 330 of file Service_Participant.inl.

References publisher_content_filter_.

00331 {
00332   return this->publisher_content_filter_;
00333 }

ACE_INLINE bool & OpenDDS::DCPS::Service_Participant::publisher_content_filter ( )

Accessors for PublisherContentFilter.

Definition at line 323 of file Service_Participant.inl.

References publisher_content_filter_.

00324 {
00325   return this->publisher_content_filter_;
00326 }

ACE_Reactor * OpenDDS::DCPS::Service_Participant::reactor ( ) const

Definition at line 240 of file Service_Participant.cpp.

References reactor_.

00241 {
00242   return reactor_;
00243 }

ACE_thread_t OpenDDS::DCPS::Service_Participant::reactor_owner ( ) const

Definition at line 246 of file Service_Participant.cpp.

References reactor_owner_.

00247 {
00248   return reactor_owner_;
00249 }

void OpenDDS::DCPS::Service_Participant::register_discovery_type	(	const char *	section_name,
		Discovery::Config *	cfg
	)

Definition at line 1257 of file Service_Participant.cpp.

01259 {
01260   delete discovery_types_[section_name];
01261   discovery_types_[section_name] = cfg;
01262 }

void OpenDDS::DCPS::Service_Participant::remap_domains	(	Discovery::RepoKey	oldKey,
		Discovery::RepoKey	newKey,
		bool	attach_participant = `true`
	)

Rebind a domain from one repository to another.

Definition at line 848 of file Service_Participant.cpp.

References domainRepoMap_, and OpenDDS::DCPS::OPENDDS_VECTOR().

Referenced by repository_lost(), and set_repo_ior().

00851 {
00852   // Search the mappings for any domains mapped to this repository.
00853   OPENDDS_VECTOR(DDS::DomainId_t) domainList;
00854   {
00855     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->maps_lock_);
00856 
00857     for (DomainRepoMap::const_iterator current = this->domainRepoMap_.begin();
00858          current != this->domainRepoMap_.end();
00859          ++current) {
00860       if (current->second == oldKey) {
00861         domainList.push_back(current->first);
00862       }
00863     }
00864   }
00865 
00866   // Remap the domains that were attached to this repository.
00867   for (unsigned int index = 0; index < domainList.size(); ++index) {
00868     // For mapped domains, attach their participants by setting the
00869     // mapping again.
00870     this->set_repo_domain(domainList[ index], newKey, attach_participant);
00871   }
00872 }

void OpenDDS::DCPS::Service_Participant::repository_lost ( Discovery::RepoKey key )

Failover to a new repository.

Definition at line 977 of file Service_Participant.cpp.

References OpenDDS::DCPS::DCPS_debug_level, discoveryMap_, federation_backoff_multiplier(), federation_initial_backoff_seconds(), and remap_domains().

00978 {
00979   // Find the lost repository.
00980   RepoKeyDiscoveryMap::iterator initialLocation = this->discoveryMap_.find(key);
00981   RepoKeyDiscoveryMap::iterator current         = initialLocation;
00982 
00983   if (current == this->discoveryMap_.end()) {
00984     ACE_DEBUG((LM_WARNING,
00985                ACE_TEXT("(%P|%t) WARNING: Service_Participant::repository_lost: ")
00986                ACE_TEXT("lost repository %C was not present, ")
00987                ACE_TEXT("finding another anyway.\n"),
00988                key.c_str()));
00989 
00990   } else {
00991     // Start with the repository *after* the lost one.
00992     ++current;
00993   }
00994 
00995   // Calculate the bounding end time for attempts.
00996   ACE_Time_Value recoveryFailedTime
00997   = ACE_OS::gettimeofday()
00998     + ACE_Time_Value(this->federation_recovery_duration(), 0);
00999 
01000   // Backoff delay.
01001   int backoff = this->federation_initial_backoff_seconds();
01002 
01003   // Keep trying until the total recovery time specified is exceeded.
01004   while (recoveryFailedTime > ACE_OS::gettimeofday()) {
01005 
01006     // Wrap to the beginning at the end of the list.
01007     if (current == this->discoveryMap_.end()) {
01008       // Continue to traverse the list.
01009       current = this->discoveryMap_.begin();
01010     }
01011 
01012     // Handle reaching the lost repository by waiting before trying
01013     // again.
01014     if (current == initialLocation) {
01015       if (DCPS_debug_level > 0) {
01016         ACE_DEBUG((LM_DEBUG,
01017                    ACE_TEXT("(%P|%t) Service_Participant::repository_lost: ")
01018                    ACE_TEXT("waiting %d seconds to traverse the ")
01019                    ACE_TEXT("repository list another time ")
01020                    ACE_TEXT("for lost key %C.\n"),
01021                    backoff,
01022                    key.c_str()));
01023       }
01024 
01025       // Wait to traverse the list and try again.
01026       ACE_OS::sleep(backoff);
01027 
01028       // Exponentially backoff delay.
01029       backoff *= this->federation_backoff_multiplier();
01030 
01031       // Don't increment current to allow us to reattach to the
01032       // original repository if it is restarted.
01033     }
01034 
01035     // Check the availability of the current repository.
01036     if (current->second->active()) {
01037 
01038       if (DCPS_debug_level > 0) {
01039         ACE_DEBUG((LM_DEBUG,
01040                    ACE_TEXT("(%P|%t) Service_Participant::repository_lost: ")
01041                    ACE_TEXT("replacing repository %C with %C.\n"),
01042                    key.c_str(),
01043                    current->first.c_str()));
01044       }
01045 
01046       // If we reach here, the validate_connection() call succeeded
01047       // and the repository is reachable.
01048       this->remap_domains(key, current->first);
01049 
01050       // Now we are done.  This is the only non-failure exit from
01051       // this method.
01052       return;
01053 
01054     } else {
01055       ACE_DEBUG((LM_WARNING,
01056                  ACE_TEXT("(%P|%t) WARNING: Service_Participant::repository_lost: ")
01057                  ACE_TEXT("repository %C was not available to replace %C, ")
01058                  ACE_TEXT("looking for another.\n"),
01059                  current->first.c_str(),
01060                  key.c_str()));
01061     }
01062 
01063     // Move to the next candidate repository.
01064     ++current;
01065   }
01066 
01067   // If we reach here, we have exceeded the total recovery time
01068   // specified.
01069   ACE_ASSERT(recoveryFailedTime == ACE_Time_Value::zero);
01070 }

ACE_INLINE long OpenDDS::DCPS::Service_Participant::scheduler ( ) const

Accessors for scheduling policy value.

Definition at line 295 of file Service_Participant.inl.

References scheduler_.

00296 {
00297   return this->scheduler_;
00298 }

ACE_INLINE long & OpenDDS::DCPS::Service_Participant::scheduler ( )

Accessors for scheduling policy value.

Definition at line 288 of file Service_Participant.inl.

References scheduler_.

00289 {
00290   return this->scheduler_;
00291 }

void OpenDDS::DCPS::Service_Participant::set_BIT ( bool b ) [inline]

Definition at line 291 of file Service_Participant.h.

00291                        {
00292     bit_enabled_ = b;
00293   }

void OpenDDS::DCPS::Service_Participant::set_default_discovery ( const Discovery::RepoKey & defaultDiscovery )

Definition at line 1073 of file Service_Participant.cpp.

References defaultDiscovery_.

01074 {
01075   this->defaultDiscovery_ = key;
01076 }

void OpenDDS::DCPS::Service_Participant::set_repo_domain	(	const DDS::DomainId_t	domain,
		Discovery::RepoKey	repo,
		bool	attach_participant = `true`
	)

Bind DCPSInfoRepo IORs to domains.

Definition at line 875 of file Service_Participant.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::Discovery::DEFAULT_REPO, domainRepoMap_, dp_factory_servant_, OPENDDS_STRING, OpenDDS::DCPS::OPENDDS_VECTOR(), and OpenDDS::DCPS::DomainParticipantFactoryImpl::participants().

Referenced by load_domain_configuration().

00878 {
00879   typedef std::pair<Discovery_rch, RepoId> DiscRepoPair;
00880   OPENDDS_VECTOR(DiscRepoPair) repoList;
00881   {
00882     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->maps_lock_);
00883     DomainRepoMap::const_iterator where = this->domainRepoMap_.find(domain);
00884 
00885     if (key == "-1") {
00886       key = Discovery::DEFAULT_REPO;
00887     }
00888 
00889     if ((where == this->domainRepoMap_.end()) || (where->second != key)) {
00890       // Only assign entries into the map when they change the
00891       // contents.
00892       this->domainRepoMap_[ domain] = key;
00893 
00894       if (DCPS_debug_level > 0) {
00895         ACE_DEBUG((LM_DEBUG,
00896                    ACE_TEXT("(%P|%t) Service_Participant::set_repo_domain: ")
00897                    ACE_TEXT("Domain[ %d] = Repo[ %C].\n"),
00898                    domain, key.c_str()));
00899       }
00900     }
00901 
00902     //
00903     // Make sure that we mark each DomainParticipant for this domain
00904     // using this repository as attached to this repository.
00905     //
00906     // @TODO: Move this note into user documentation.
00907     // N.B. Calling set_repo() or set_repo_ior() will result in this
00908     //      code executing again with the new repository.  It is best
00909     //      to call those routines first when making changes.
00910     //
00911 
00912     // No servant means no participant.  No worries.
00913     if (0 != this->dp_factory_servant_) {
00914       // Map of domains to sets of participants.
00915       const DomainParticipantFactoryImpl::DPMap& participants
00916       = this->dp_factory_servant_->participants();
00917 
00918       // Extract the set of participants for the current domain.
00919       DomainParticipantFactoryImpl::DPMap::const_iterator
00920       which  = participants.find(domain);
00921 
00922       if (which != participants.end()) {
00923         // Extract the repository to attach this domain to.
00924         RepoKeyDiscoveryMap::const_iterator disc_iter = this->discoveryMap_.find(key);
00925 
00926         if (disc_iter != this->discoveryMap_.end()) {
00927           for (DomainParticipantFactoryImpl::DPSet::const_iterator
00928                current  = which->second.begin();
00929                current != which->second.end();
00930                ++current) {
00931             try {
00932               // Attach each DomainParticipant in this domain to this
00933               // repository.
00934               RepoId id = current->svt_->get_id();
00935               repoList.push_back(std::make_pair(disc_iter->second, id));
00936 
00937               if (DCPS_debug_level > 0) {
00938                 GuidConverter converter(id);
00939                 ACE_DEBUG((LM_DEBUG,
00940                            ACE_TEXT("(%P|%t) Service_Participant::set_repo_domain: ")
00941                            ACE_TEXT("participant %C attached to Repo[ %C].\n"),
00942                            OPENDDS_STRING(converter).c_str(),
00943                            key.c_str()));
00944               }
00945 
00946             } catch (const CORBA::Exception& ex) {
00947               ex._tao_print_exception(
00948                 "ERROR: Service_Participant::set_repo_domain: failed to attach repository - ");
00949               return;
00950             }
00951           }
00952         }
00953       }
00954     }
00955   } // End of GUARD scope.
00956 
00957   // Make all of the remote calls after releasing the lock.
00958   for (unsigned int index = 0; index < repoList.size(); ++index) {
00959     if (DCPS_debug_level > 0) {
00960       GuidConverter converter(repoList[ index].second);
00961       ACE_DEBUG((LM_DEBUG,
00962                  ACE_TEXT("(%P|%t) Service_Participant::set_repo_domain: ")
00963                  ACE_TEXT("(%d of %d) attaching domain %d participant %C to Repo[ %C].\n"),
00964                  (1+index), repoList.size(), domain,
00965                  OPENDDS_STRING(converter).c_str(),
00966                  key.c_str()));
00967     }
00968 
00969     if (attach_participant)
00970     {
00971       repoList[ index].first->attach_participant(domain, repoList[ index].second);
00972     }
00973   }
00974 }

bool OpenDDS::DCPS::Service_Participant::set_repo_ior	(	const char *	ior,
		Discovery::RepoKey	key = `Discovery::DEFAULT_REPO`,
		bool	attach_participant = `true`
	)

Definition at line 798 of file Service_Participant.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::Discovery::DEFAULT_REPO, OpenDDS::DCPS::got_info, OPENDDS_STRING, remap_domains(), OpenDDS::DCPS::REPO_SECTION_NAME, and TheTransportRegistry.

Referenced by get_discovery(), InfoRepo::init(), and parse_args().

00801 {
00802   if (DCPS_debug_level > 0) {
00803     ACE_DEBUG((LM_DEBUG,
00804                ACE_TEXT("(%P|%t) Service_Participant::set_repo_ior: Repo[ %C] == %C\n"),
00805                key.c_str(), ior));
00806   }
00807 
00808   // This is a global used for the bizzare commandline/configfile
00809   // processing done for this class.
00810   got_info = true;
00811 
00812   if (key == "-1") {
00813     key = Discovery::DEFAULT_REPO;
00814   }
00815 
00816   const OPENDDS_STRING repo_type = ACE_TEXT_ALWAYS_CHAR(REPO_SECTION_NAME);
00817   if (!discovery_types_.count(repo_type)) {
00818     // Re-use a transport registry function to attempt a dynamic load of the
00819     // library that implements the 'repo_type' (InfoRepoDiscovery)
00820     TheTransportRegistry->load_transport_lib(repo_type);
00821   }
00822 
00823   if (discovery_types_.count(repo_type)) {
00824     ACE_Configuration_Heap cf;
00825     cf.open();
00826     ACE_Configuration_Section_Key sect_key;
00827     ACE_TString section = REPO_SECTION_NAME;
00828     section += ACE_TEXT('\\');
00829     section += ACE_TEXT_CHAR_TO_TCHAR(key.c_str());
00830     cf.open_section(cf.root_section(), section.c_str(), 1 /*create*/, sect_key);
00831     cf.set_string_value(sect_key, ACE_TEXT("RepositoryIor"),
00832                         ACE_TEXT_CHAR_TO_TCHAR(ior));
00833 
00834     discovery_types_[repo_type]->discovery_config(cf);
00835 
00836     this->remap_domains(key, key, attach_participant);
00837     return true;
00838   }
00839 
00840   ACE_ERROR_RETURN((LM_ERROR,
00841                     ACE_TEXT("(%P|%t) Service_Participant::set_repo_ior ")
00842                     ACE_TEXT("ERROR - no discovery type registered for ")
00843                     ACE_TEXT("InfoRepoDiscovery\n")),
00844                    false);
00845 }

void OpenDDS::DCPS::Service_Participant::shutdown ( )

Stop being a participant in the service.

Note:: Required Precondition: all DomainParticipants have been deleted.

Definition at line 252 of file Service_Participant.cpp.

References discoveryMap_, domainRepoMap_, dp_factory_, OpenDDS::DCPS::TransportRegistry::instance(), OPENDDS_MAP(), OPENDDS_STRING, persistent_data_cache_, reactor_, reactor_task_, OpenDDS::DCPS::TransportRegistry::release(), shut_down_, and transient_data_cache_.

Referenced by ~Service_Participant().

00253 {
00254   shut_down_ = true;
00255   try {
00256     TransportRegistry::instance()->release();
00257     ACE_GUARD(TAO_SYNCH_MUTEX, guard, this->factory_lock_);
00258 
00259     domainRepoMap_.clear();
00260     discoveryMap_.clear();
00261 
00262     if (0 != reactor_) {
00263       reactor_->end_reactor_event_loop();
00264       reactor_task_.wait();
00265     }
00266 
00267     dp_factory_ = DDS::DomainParticipantFactory::_nil();
00268 
00269 #ifndef OPENDDS_NO_PERSISTENCE_PROFILE
00270     transient_data_cache_.reset();
00271     persistent_data_cache_.reset();
00272 #endif
00273 
00274     typedef OPENDDS_MAP(OPENDDS_STRING, Discovery::Config*)::iterator iter;
00275     for (iter i = discovery_types_.begin(); i != discovery_types_.end(); ++i) {
00276       delete i->second;
00277     }
00278     discovery_types_.clear();
00279   } catch (const CORBA::Exception& ex) {
00280     ex._tao_print_exception("ERROR: Service_Participant::shutdown");
00281   }
00282 }

ACE_Reactor_Timer_Interface * OpenDDS::DCPS::Service_Participant::timer ( ) const

Get the common timer interface. Intended for use by OpenDDS internals only.

Definition at line 234 of file Service_Participant.cpp.

References reactor_.

00235 {
00236   return reactor_;
00237 }

size_t OpenDDS::DCPS::Service_Participant::association_chunk_multiplier_ [private]

The configurable value of maximum number of expected associations for publishers and subscribers. This is used to pre allocate enough memory and reduce heap allocations.

Definition at line 497 of file Service_Participant.h.

Referenced by association_chunk_multiplier(), and parse_args().

bool OpenDDS::DCPS::Service_Participant::bit_enabled_ [private]

Definition at line 508 of file Service_Participant.h.

Referenced by parse_args().

int OpenDDS::DCPS::Service_Participant::bit_lookup_duration_msec_ [private]

The timeout for lookup data from the builtin topic DataReader.

Definition at line 512 of file Service_Participant.h.

Referenced by bit_lookup_duration_msec(), and parse_args().

ACE_TString OpenDDS::DCPS::Service_Participant::bit_transport_ip_ [private]

The builtin topic transport address.

Definition at line 503 of file Service_Participant.h.

Referenced by parse_args().

int OpenDDS::DCPS::Service_Participant::bit_transport_port_ [private]

The builtin topic transport port number.

Definition at line 506 of file Service_Participant.h.

Referenced by bit_transport_port(), and parse_args().

ACE_Configuration_Heap OpenDDS::DCPS::Service_Participant::cf_ [private]

The configuration object that imports the configuration file.

Definition at line 519 of file Service_Participant.h.

Referenced by load_configuration().

ACE_CString OpenDDS::DCPS::Service_Participant::default_address_ [private]

The default network address to use.

Definition at line 515 of file Service_Participant.h.

Referenced by default_address(), and parse_args().

Discovery::RepoKey OpenDDS::DCPS::Service_Participant::defaultDiscovery_ [private]

Definition at line 450 of file Service_Participant.h.

Referenced by get_default_discovery(), get_discovery(), parse_args(), and set_default_discovery().

RepoKeyDiscoveryMap OpenDDS::DCPS::Service_Participant::discoveryMap_ [private]

The RepoKey to Discovery object mapping.

Definition at line 445 of file Service_Participant.h.

Referenced by add_discovery(), discoveryMap(), get_discovery(), load_domain_configuration(), repository_lost(), and shutdown().

DomainRepoMap OpenDDS::DCPS::Service_Participant::domainRepoMap_ [private]

The DomainId to RepoKey mapping.

Definition at line 448 of file Service_Participant.h.

Referenced by domain_to_repo(), domainRepoMap(), get_discovery(), remap_domains(), set_repo_domain(), and shutdown().

DDS::DomainParticipantFactory_var OpenDDS::DCPS::Service_Participant::dp_factory_ [private]

Definition at line 442 of file Service_Participant.h.

Referenced by get_domain_participant_factory(), and shutdown().

DomainParticipantFactoryImpl* OpenDDS::DCPS::Service_Participant::dp_factory_servant_ [private]

Definition at line 441 of file Service_Participant.h.

Referenced by get_domain_participant_factory(), and set_repo_domain().

TAO_SYNCH_MUTEX OpenDDS::DCPS::Service_Participant::factory_lock_ [private]

The lock to serialize DomainParticipantFactory singleton creation and shutdown.

Definition at line 454 of file Service_Participant.h.

int OpenDDS::DCPS::Service_Participant::federation_backoff_multiplier_ [private]

This FederationBackoffMultiplier.

Definition at line 542 of file Service_Participant.h.

Referenced by federation_backoff_multiplier(), and parse_args().

int OpenDDS::DCPS::Service_Participant::federation_initial_backoff_seconds_ [private]

The FederationInitialBackoffSeconds value.

Definition at line 539 of file Service_Participant.h.

Referenced by federation_initial_backoff_seconds(), and parse_args().

int OpenDDS::DCPS::Service_Participant::federation_liveliness_ [private]

This FederationLivelinessDuration.

Definition at line 545 of file Service_Participant.h.

Referenced by federation_liveliness(), and parse_args().

int OpenDDS::DCPS::Service_Participant::federation_recovery_duration_ [private]

The FederationRecoveryDuration value in seconds.

Definition at line 536 of file Service_Participant.h.

Referenced by federation_recovery_duration(), and parse_args().

ACE_TString OpenDDS::DCPS::Service_Participant::global_transport_config_ [private]

Specifies the name of the transport configuration that is used when the entity tree does not specify one. If not set, the default transport configuration is used.

Definition at line 524 of file Service_Participant.h.

Referenced by load_common_configuration(), and parse_args().

DDS::DataReaderQos OpenDDS::DCPS::Service_Participant::initial_DataReaderQos_ [private]

Definition at line 486 of file Service_Participant.h.

Referenced by initial_DataReaderQos(), and initialize().

DDS::DataWriterQos OpenDDS::DCPS::Service_Participant::initial_DataWriterQos_ [private]

Definition at line 484 of file Service_Participant.h.

Referenced by initial_DataWriterQos(), and initialize().

DDS::DeadlineQosPolicy OpenDDS::DCPS::Service_Participant::initial_DeadlineQosPolicy_ [private]

Definition at line 465 of file Service_Participant.h.

Referenced by initial_DeadlineQosPolicy(), and initialize().

DDS::DestinationOrderQosPolicy OpenDDS::DCPS::Service_Participant::initial_DestinationOrderQosPolicy_ [private]

Definition at line 475 of file Service_Participant.h.

Referenced by initial_DestinationOrderQosPolicy(), and initialize().

DDS::DomainParticipantFactoryQos OpenDDS::DCPS::Service_Participant::initial_DomainParticipantFactoryQos_ [private]

Definition at line 488 of file Service_Participant.h.

Referenced by initial_DomainParticipantFactoryQos(), and initialize().

DDS::DomainParticipantQos OpenDDS::DCPS::Service_Participant::initial_DomainParticipantQos_ [private]

Definition at line 482 of file Service_Participant.h.

Referenced by initial_DomainParticipantQos(), and initialize().

DDS::DurabilityQosPolicy OpenDDS::DCPS::Service_Participant::initial_DurabilityQosPolicy_ [private]

Definition at line 462 of file Service_Participant.h.

Referenced by initial_DurabilityQosPolicy(), and initialize().

DDS::DurabilityServiceQosPolicy OpenDDS::DCPS::Service_Participant::initial_DurabilityServiceQosPolicy_ [private]

Definition at line 463 of file Service_Participant.h.

Referenced by initial_DurabilityServiceQosPolicy(), and initialize().

DDS::EntityFactoryQosPolicy OpenDDS::DCPS::Service_Participant::initial_EntityFactoryQosPolicy_ [private]

Definition at line 478 of file Service_Participant.h.

Referenced by initial_EntityFactoryQosPolicy(), and initialize().

DDS::GroupDataQosPolicy OpenDDS::DCPS::Service_Participant::initial_GroupDataQosPolicy_ [private]

Definition at line 459 of file Service_Participant.h.

Referenced by initial_GroupDataQosPolicy(), and initialize().

DDS::HistoryQosPolicy OpenDDS::DCPS::Service_Participant::initial_HistoryQosPolicy_ [private]

Definition at line 476 of file Service_Participant.h.

Referenced by initial_HistoryQosPolicy(), and initialize().

DDS::LatencyBudgetQosPolicy OpenDDS::DCPS::Service_Participant::initial_LatencyBudgetQosPolicy_ [private]

Definition at line 466 of file Service_Participant.h.

Referenced by initial_LatencyBudgetQosPolicy(), and initialize().

DDS::LifespanQosPolicy OpenDDS::DCPS::Service_Participant::initial_LifespanQosPolicy_ [private]

Definition at line 461 of file Service_Participant.h.

Referenced by initial_LifespanQosPolicy(), and initialize().

DDS::LivelinessQosPolicy OpenDDS::DCPS::Service_Participant::initial_LivelinessQosPolicy_ [private]

Definition at line 471 of file Service_Participant.h.

Referenced by initial_LivelinessQosPolicy(), and initialize().

DDS::OwnershipQosPolicy OpenDDS::DCPS::Service_Participant::initial_OwnershipQosPolicy_ [private]

Definition at line 467 of file Service_Participant.h.

Referenced by initial_OwnershipQosPolicy(), and initialize().

DDS::OwnershipStrengthQosPolicy OpenDDS::DCPS::Service_Participant::initial_OwnershipStrengthQosPolicy_ [private]

Definition at line 469 of file Service_Participant.h.

Referenced by initial_OwnershipStrengthQosPolicy(), and initialize().

DDS::PartitionQosPolicy OpenDDS::DCPS::Service_Participant::initial_PartitionQosPolicy_ [private]

Definition at line 473 of file Service_Participant.h.

Referenced by initial_PartitionQosPolicy(), and initialize().

DDS::PresentationQosPolicy OpenDDS::DCPS::Service_Participant::initial_PresentationQosPolicy_ [private]

Definition at line 464 of file Service_Participant.h.

Referenced by initial_PresentationQosPolicy(), and initialize().

DDS::PublisherQos OpenDDS::DCPS::Service_Participant::initial_PublisherQos_ [private]

Definition at line 485 of file Service_Participant.h.

Referenced by initial_PublisherQos(), and initialize().

DDS::ReaderDataLifecycleQosPolicy OpenDDS::DCPS::Service_Participant::initial_ReaderDataLifecycleQosPolicy_ [private]

Definition at line 480 of file Service_Participant.h.

Referenced by initial_ReaderDataLifecycleQosPolicy(), and initialize().

DDS::ReliabilityQosPolicy OpenDDS::DCPS::Service_Participant::initial_ReliabilityQosPolicy_ [private]

Definition at line 474 of file Service_Participant.h.

Referenced by initial_ReliabilityQosPolicy(), and initialize().

DDS::ResourceLimitsQosPolicy OpenDDS::DCPS::Service_Participant::initial_ResourceLimitsQosPolicy_ [private]

Definition at line 477 of file Service_Participant.h.

Referenced by initial_ResourceLimitsQosPolicy(), and initialize().

DDS::SubscriberQos OpenDDS::DCPS::Service_Participant::initial_SubscriberQos_ [private]

Definition at line 487 of file Service_Participant.h.

Referenced by initial_SubscriberQos(), and initialize().

DDS::TimeBasedFilterQosPolicy OpenDDS::DCPS::Service_Participant::initial_TimeBasedFilterQosPolicy_ [private]

Definition at line 472 of file Service_Participant.h.

Referenced by initial_TimeBasedFilterQosPolicy(), and initialize().

DDS::TopicDataQosPolicy OpenDDS::DCPS::Service_Participant::initial_TopicDataQosPolicy_ [private]

Definition at line 458 of file Service_Participant.h.

Referenced by initial_TopicDataQosPolicy(), and initialize().

DDS::TopicQos OpenDDS::DCPS::Service_Participant::initial_TopicQos_ [private]

Definition at line 483 of file Service_Participant.h.

Referenced by initial_TopicQos(), and initialize().

DDS::TransportPriorityQosPolicy OpenDDS::DCPS::Service_Participant::initial_TransportPriorityQosPolicy_ [private]

Definition at line 460 of file Service_Participant.h.

Referenced by initial_TransportPriorityQosPolicy(), and initialize().

DDS::UserDataQosPolicy OpenDDS::DCPS::Service_Participant::initial_UserDataQosPolicy_ [private]

The initial values of qos policies.

Definition at line 457 of file Service_Participant.h.

Referenced by initial_UserDataQosPolicy(), and initialize().

DDS::WriterDataLifecycleQosPolicy OpenDDS::DCPS::Service_Participant::initial_WriterDataLifecycleQosPolicy_ [private]

Definition at line 479 of file Service_Participant.h.

Referenced by initial_WriterDataLifecycleQosPolicy(), and initialize().

int OpenDDS::DCPS::Service_Participant::liveliness_factor_ [private]

The propagation delay factor.

Definition at line 500 of file Service_Participant.h.

Referenced by liveliness_factor(), and parse_args().

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::Service_Participant::maps_lock_ [private]

Guard access to the internal maps.

Definition at line 594 of file Service_Participant.h.

Monitor* OpenDDS::DCPS::Service_Participant::monitor_

Pointer to the monitor object for this object.

Definition at line 532 of file Service_Participant.h.

Referenced by get_domain_participant_factory(), and ~Service_Participant().

MonitorFactory* OpenDDS::DCPS::Service_Participant::monitor_factory_

Pointer to the monitor factory that is used to create monitor objects.

Definition at line 529 of file Service_Participant.h.

Referenced by get_domain_participant_factory().

size_t OpenDDS::DCPS::Service_Participant::n_chunks_ [private]

The configurable value of the number chunks that the DataWriter's cached allocator can allocate.

Definition at line 492 of file Service_Participant.h.

Referenced by n_chunks(), and parse_args().

ACE_ARGV OpenDDS::DCPS::Service_Participant::ORB_argv_ [private]

Definition at line 425 of file Service_Participant.h.

Referenced by get_domain_participant_factory().

ACE_Time_Value OpenDDS::DCPS::Service_Participant::pending_timeout_ [private]

Number of seconds to wait on pending samples to be sent or dropped.

Definition at line 588 of file Service_Participant.h.

Referenced by load_common_configuration(), parse_args(), and pending_timeout().

std::auto_ptr<DataDurabilityCache> OpenDDS::DCPS::Service_Participant::persistent_data_cache_ [private]

The PERSISTENT data durability cache.

Definition at line 579 of file Service_Participant.h.

Referenced by get_data_durability_cache(), and shutdown().

ACE_CString OpenDDS::DCPS::Service_Participant::persistent_data_dir_ [private]

The PERSISTENT data durability directory.

Definition at line 582 of file Service_Participant.h.

Referenced by load_common_configuration(), and parse_args().

int OpenDDS::DCPS::Service_Participant::priority_max_ [private]

Maximum priority value for the current scheduling policy.

Definition at line 568 of file Service_Participant.h.

Referenced by initializeScheduling(), and priority_max().

int OpenDDS::DCPS::Service_Participant::priority_min_ [private]

Minimum priority value for the current scheduling policy.

Definition at line 565 of file Service_Participant.h.

Referenced by initializeScheduling(), and priority_min().

bool OpenDDS::DCPS::Service_Participant::publisher_content_filter_ [private]

Allow the publishing side to do content filtering?

Definition at line 571 of file Service_Participant.h.

Referenced by parse_args(), and publisher_content_filter().

ACE_Reactor* OpenDDS::DCPS::Service_Participant::reactor_ [private]

Definition at line 428 of file Service_Participant.h.

Referenced by get_domain_participant_factory(), reactor(), shutdown(), OpenDDS::DCPS::Service_Participant::ReactorTask::svc(), timer(), and ~Service_Participant().

ACE_thread_t OpenDDS::DCPS::Service_Participant::reactor_owner_ [private]

Definition at line 429 of file Service_Participant.h.

Referenced by reactor_owner(), and OpenDDS::DCPS::Service_Participant::ReactorTask::svc().

OpenDDS::DCPS::Service_Participant::ReactorTask OpenDDS::DCPS::Service_Participant::reactor_task_ [private]

Referenced by get_domain_participant_factory(), and shutdown().

long OpenDDS::DCPS::Service_Participant::scheduler_ [private]

Scheduling policy value used for setting thread priorities.

Definition at line 562 of file Service_Participant.h.

Referenced by initializeScheduling(), and scheduler().

ACE_Time_Value OpenDDS::DCPS::Service_Participant::schedulerQuantum_ [private]

Scheduler time slice from configuration file.

Definition at line 551 of file Service_Participant.h.

Referenced by load_common_configuration().

ACE_TString OpenDDS::DCPS::Service_Participant::schedulerString_ [private]

Scheduling policy value from configuration file.

Definition at line 548 of file Service_Participant.h.

bool OpenDDS::DCPS::Service_Participant::shut_down_ [private]

Definition at line 591 of file Service_Participant.h.

Referenced by is_shut_down(), and shutdown().

std::auto_ptr<DataDurabilityCache> OpenDDS::DCPS::Service_Participant::transient_data_cache_ [private]

The TRANSIENT data durability cache.

Definition at line 576 of file Service_Participant.h.

Referenced by get_data_durability_cache(), and shutdown().

int OpenDDS::DCPS::Service_Participant::zero_argc = 0 [static, private]

Definition at line 596 of file Service_Participant.h.

OpenDDS::DCPS::Service_Participant Class Reference

Public Types

Public Member Functions

Static Public Member Functions

Public Attributes

Private Member Functions

Private Attributes

Static Private Attributes

Classes

Detailed Description

Member Enumeration Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation


Public Types
	ANY_DOMAIN = -1
enum	{ ANY_DOMAIN = -1 }
	Domain value for the default repository IOR. More...
Public Member Functions
	Service_Participant ()
	Constructor.
	~Service_Participant ()
	Destructor.
ACE_Reactor_Timer_Interface *	timer () const
ACE_Reactor *	reactor () const
ACE_thread_t	reactor_owner () const
DDS::DomainParticipantFactory_ptr	get_domain_participant_factory (int &argc=zero_argc, ACE_TCHAR *argv[]=0)
void	shutdown ()
bool	is_shut_down () const
	Accessor for if the participant has been shutdown.
Discovery_rch	get_discovery (const DDS::DomainId_t domain)
	Accessor of the Discovery object for a given domain.
DDS::UserDataQosPolicy	initial_UserDataQosPolicy () const
DDS::TopicDataQosPolicy	initial_TopicDataQosPolicy () const
DDS::GroupDataQosPolicy	initial_GroupDataQosPolicy () const
DDS::TransportPriorityQosPolicy	initial_TransportPriorityQosPolicy () const
DDS::LifespanQosPolicy	initial_LifespanQosPolicy () const
DDS::DurabilityQosPolicy	initial_DurabilityQosPolicy () const
DDS::DurabilityServiceQosPolicy	initial_DurabilityServiceQosPolicy () const
DDS::PresentationQosPolicy	initial_PresentationQosPolicy () const
DDS::DeadlineQosPolicy	initial_DeadlineQosPolicy () const
DDS::LatencyBudgetQosPolicy	initial_LatencyBudgetQosPolicy () const
DDS::OwnershipQosPolicy	initial_OwnershipQosPolicy () const
DDS::OwnershipStrengthQosPolicy	initial_OwnershipStrengthQosPolicy () const
DDS::LivelinessQosPolicy	initial_LivelinessQosPolicy () const
DDS::TimeBasedFilterQosPolicy	initial_TimeBasedFilterQosPolicy () const
DDS::PartitionQosPolicy	initial_PartitionQosPolicy () const
DDS::ReliabilityQosPolicy	initial_ReliabilityQosPolicy () const
DDS::DestinationOrderQosPolicy	initial_DestinationOrderQosPolicy () const
DDS::HistoryQosPolicy	initial_HistoryQosPolicy () const
DDS::ResourceLimitsQosPolicy	initial_ResourceLimitsQosPolicy () const
DDS::EntityFactoryQosPolicy	initial_EntityFactoryQosPolicy () const
DDS::WriterDataLifecycleQosPolicy	initial_WriterDataLifecycleQosPolicy () const
DDS::ReaderDataLifecycleQosPolicy	initial_ReaderDataLifecycleQosPolicy () const
DDS::DomainParticipantFactoryQos	initial_DomainParticipantFactoryQos () const
DDS::DomainParticipantQos	initial_DomainParticipantQos () const
DDS::TopicQos	initial_TopicQos () const
DDS::DataWriterQos	initial_DataWriterQos () const
DDS::PublisherQos	initial_PublisherQos () const
DDS::DataReaderQos	initial_DataReaderQos () const
DDS::SubscriberQos	initial_SubscriberQos () const
size_t	n_chunks () const
void	n_chunks (size_t chunks)
	Set the value returned by `n_chunks()` accessor.
size_t	association_chunk_multiplier () const
void	association_chunk_multiplier (size_t multiplier)
void	liveliness_factor (int factor)
int	liveliness_factor () const
void	add_discovery (Discovery_rch discovery)
bool	set_repo_ior (const char *ior, Discovery::RepoKey key=Discovery::DEFAULT_REPO, bool attach_participant=true)
void	remap_domains (Discovery::RepoKey oldKey, Discovery::RepoKey newKey, bool attach_participant=true)
	Rebind a domain from one repository to another.
void	set_repo_domain (const DDS::DomainId_t domain, Discovery::RepoKey repo, bool attach_participant=true)
	Bind DCPSInfoRepo IORs to domains.
void	set_default_discovery (const Discovery::RepoKey &defaultDiscovery)
Discovery::RepoKey	get_default_discovery ()
Discovery::RepoKey	domain_to_repo (const DDS::DomainId_t domain) const
	Convert domainId to repository key.
void	repository_lost (Discovery::RepoKey key)
	Failover to a new repository.
ACE_Time_Value	pending_timeout () const
	Accessor for pending data timeout.
OPENDDS_STRING	bit_transport_ip () const
bool	get_BIT ()
void	set_BIT (bool b)
ACE_CString	default_address () const
DataDurabilityCache *	get_data_durability_cache (DDS::DurabilityQosPolicy const &durability)
typedef	OPENDDS_MAP (Discovery::RepoKey, Discovery_rch) RepoKeyDiscoveryMap
	For internal OpenDDS Use (needed for monitor code).
const RepoKeyDiscoveryMap &	discoveryMap () const
typedef	OPENDDS_MAP (DDS::DomainId_t, Discovery::RepoKey) DomainRepoMap
const DomainRepoMap &	domainRepoMap () const
void	register_discovery_type (const char section_name, Discovery::Config cfg)
ACE_ARGV *	ORB_argv ()
Recorder_ptr	create_recorder (DDS::DomainParticipant_ptr participant, DDS::Topic_ptr a_topic, const DDS::SubscriberQos &subscriber_qos, const DDS::DataReaderQos &datareader_qos, const RecorderListener_rch &a_listener)
DDS::ReturnCode_t	delete_recorder (Recorder_ptr recorder)
Replayer_ptr	create_replayer (DDS::DomainParticipant_ptr participant, DDS::Topic_ptr a_topic, const DDS::PublisherQos &publisher_qos, const DDS::DataWriterQos &datawriter_qos, const ReplayerListener_rch &a_listener)
DDS::ReturnCode_t	delete_replayer (Replayer_ptr replayer)
DDS::Topic_ptr	create_typeless_topic (DDS::DomainParticipant_ptr participant, const char topic_name, const char type_name, bool type_has_keys, const DDS::TopicQos &qos, DDS::TopicListener_ptr a_listener=0, DDS::StatusMask mask=0)
int	load_configuration (ACE_Configuration_Heap &cf, const ACE_TCHAR *filename)

int &	federation_recovery_duration ()
	Accessors for FederationRecoveryDuration in seconds.
int	federation_recovery_duration () const
	Accessors for FederationRecoveryDuration in seconds.

int &	federation_initial_backoff_seconds ()
	Accessors for FederationInitialBackoffSeconds.
int	federation_initial_backoff_seconds () const
	Accessors for FederationInitialBackoffSeconds.

int &	federation_backoff_multiplier ()
	Accessors for FederationBackoffMultiplier.
int	federation_backoff_multiplier () const
	Accessors for FederationBackoffMultiplier.

int &	federation_liveliness ()
	Accessors for FederationLivelinessDuration.
int	federation_liveliness () const
	Accessors for FederationLivelinessDuration.

long &	scheduler ()
	Accessors for scheduling policy value.
long	scheduler () const
	Accessors for scheduling policy value.

bool &	publisher_content_filter ()
	Accessors for PublisherContentFilter.
bool	publisher_content_filter () const
	Accessors for PublisherContentFilter.

int	priority_min () const
	Accessors for priority extremums for the current scheduler.
int	priority_max () const
	Accessors for priority extremums for the current scheduler.

int	bit_transport_port () const
void	bit_transport_port (int port)

int	bit_lookup_duration_msec () const
void	bit_lookup_duration_msec (int msec)
Static Public Member Functions
static Service_Participant *	instance ()
	Return a singleton instance of this class.
Public Attributes
MonitorFactory *	monitor_factory_
Monitor *	monitor_
	Pointer to the monitor object for this object.
Private Member Functions
void	initialize ()
	Initialize default qos.
void	initializeScheduling ()
	Initialize the thread scheduling and initial priority.
int	parse_args (int &argc, ACE_TCHAR *argv[])
int	load_configuration ()
int	load_common_configuration (ACE_Configuration_Heap &cf, const ACE_TCHAR *filename)
int	load_domain_configuration (ACE_Configuration_Heap &cf, const ACE_TCHAR *filename)
int	load_discovery_configuration (ACE_Configuration_Heap &cf, const ACE_TCHAR *section_name)
	OPENDDS_MAP (OPENDDS_STRING, Discovery::Config *) discovery_types_
Private Attributes
ACE_ARGV	ORB_argv_
ACE_Reactor *	reactor_
ACE_thread_t	reactor_owner_
OpenDDS::DCPS::Service_Participant::ReactorTask	reactor_task_
DomainParticipantFactoryImpl *	dp_factory_servant_
DDS::DomainParticipantFactory_var	dp_factory_
RepoKeyDiscoveryMap	discoveryMap_
	The RepoKey to Discovery object mapping.
DomainRepoMap	domainRepoMap_
	The DomainId to RepoKey mapping.
Discovery::RepoKey	defaultDiscovery_
TAO_SYNCH_MUTEX	factory_lock_
DDS::UserDataQosPolicy	initial_UserDataQosPolicy_
	The initial values of qos policies.
DDS::TopicDataQosPolicy	initial_TopicDataQosPolicy_
DDS::GroupDataQosPolicy	initial_GroupDataQosPolicy_
DDS::TransportPriorityQosPolicy	initial_TransportPriorityQosPolicy_
DDS::LifespanQosPolicy	initial_LifespanQosPolicy_
DDS::DurabilityQosPolicy	initial_DurabilityQosPolicy_
DDS::DurabilityServiceQosPolicy	initial_DurabilityServiceQosPolicy_
DDS::PresentationQosPolicy	initial_PresentationQosPolicy_
DDS::DeadlineQosPolicy	initial_DeadlineQosPolicy_
DDS::LatencyBudgetQosPolicy	initial_LatencyBudgetQosPolicy_
DDS::OwnershipQosPolicy	initial_OwnershipQosPolicy_
DDS::OwnershipStrengthQosPolicy	initial_OwnershipStrengthQosPolicy_
DDS::LivelinessQosPolicy	initial_LivelinessQosPolicy_
DDS::TimeBasedFilterQosPolicy	initial_TimeBasedFilterQosPolicy_
DDS::PartitionQosPolicy	initial_PartitionQosPolicy_
DDS::ReliabilityQosPolicy	initial_ReliabilityQosPolicy_
DDS::DestinationOrderQosPolicy	initial_DestinationOrderQosPolicy_
DDS::HistoryQosPolicy	initial_HistoryQosPolicy_
DDS::ResourceLimitsQosPolicy	initial_ResourceLimitsQosPolicy_
DDS::EntityFactoryQosPolicy	initial_EntityFactoryQosPolicy_
DDS::WriterDataLifecycleQosPolicy	initial_WriterDataLifecycleQosPolicy_
DDS::ReaderDataLifecycleQosPolicy	initial_ReaderDataLifecycleQosPolicy_
DDS::DomainParticipantQos	initial_DomainParticipantQos_
DDS::TopicQos	initial_TopicQos_
DDS::DataWriterQos	initial_DataWriterQos_
DDS::PublisherQos	initial_PublisherQos_
DDS::DataReaderQos	initial_DataReaderQos_
DDS::SubscriberQos	initial_SubscriberQos_
DDS::DomainParticipantFactoryQos	initial_DomainParticipantFactoryQos_
size_t	n_chunks_
size_t	association_chunk_multiplier_
int	liveliness_factor_
	The propagation delay factor.
ACE_TString	bit_transport_ip_
	The builtin topic transport address.
int	bit_transport_port_
	The builtin topic transport port number.
bool	bit_enabled_
int	bit_lookup_duration_msec_
ACE_CString	default_address_
	The default network address to use.
ACE_Configuration_Heap	cf_
ACE_TString	global_transport_config_
int	federation_recovery_duration_
	The FederationRecoveryDuration value in seconds.
int	federation_initial_backoff_seconds_
	The FederationInitialBackoffSeconds value.
int	federation_backoff_multiplier_
	This FederationBackoffMultiplier.
int	federation_liveliness_
	This FederationLivelinessDuration.
ACE_TString	schedulerString_
	Scheduling policy value from configuration file.
ACE_Time_Value	schedulerQuantum_
	Scheduler time slice from configuration file.
long	scheduler_
	Scheduling policy value used for setting thread priorities.
int	priority_min_
	Minimum priority value for the current scheduling policy.
int	priority_max_
	Maximum priority value for the current scheduling policy.
bool	publisher_content_filter_
	Allow the publishing side to do content filtering?
std::auto_ptr< DataDurabilityCache >	transient_data_cache_
	The `TRANSIENT` data durability cache.
std::auto_ptr< DataDurabilityCache >	persistent_data_cache_
	The `PERSISTENT` data durability cache.
ACE_CString	persistent_data_dir_
	The `PERSISTENT` data durability directory.
ACE_Time_Value	pending_timeout_
bool	shut_down_
ACE_Recursive_Thread_Mutex	maps_lock_
	Guard access to the internal maps.
Static Private Attributes
static int	zero_argc = 0
Classes
struct	ReactorTask