#include <SubscriberImpl.h>

Inheritance diagram for OpenDDS::DCPS::SubscriberImpl:

[legend]

Collaboration diagram for OpenDDS::DCPS::SubscriberImpl:

[legend]

Public Member Functions
	SubscriberImpl (DDS::InstanceHandle_t handle, const DDS::SubscriberQos &qos, DDS::SubscriberListener_ptr a_listener, const DDS::StatusMask &mask, DomainParticipantImpl *participant)

virtual	~SubscriberImpl ()

virtual DDS::InstanceHandle_t	get_instance_handle ()

bool	contains_reader (DDS::InstanceHandle_t a_handle)

virtual DDS::DataReader_ptr	create_datareader (DDS::TopicDescription_ptr a_topic_desc, const DDS::DataReaderQos &qos, DDS::DataReaderListener_ptr a_listener, DDS::StatusMask mask)

virtual DDS::ReturnCode_t	delete_datareader (DDS::DataReader_ptr a_datareader)

virtual DDS::ReturnCode_t	delete_contained_entities ()

virtual DDS::DataReader_ptr	lookup_datareader (const char *topic_name)

virtual DDS::ReturnCode_t	get_datareaders (DDS::DataReaderSeq &readers, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)

virtual DDS::ReturnCode_t	notify_datareaders ()

virtual DDS::ReturnCode_t	set_qos (const DDS::SubscriberQos &qos)

virtual DDS::ReturnCode_t	get_qos (DDS::SubscriberQos &qos)

virtual DDS::ReturnCode_t	set_listener (DDS::SubscriberListener_ptr a_listener, DDS::StatusMask mask)

virtual DDS::SubscriberListener_ptr	get_listener ()

virtual DDS::ReturnCode_t	begin_access ()

virtual DDS::ReturnCode_t	end_access ()

virtual DDS::DomainParticipant_ptr	get_participant ()

virtual DDS::ReturnCode_t	set_default_datareader_qos (const DDS::DataReaderQos &qos)

virtual DDS::ReturnCode_t	get_default_datareader_qos (DDS::DataReaderQos &qos)

virtual DDS::ReturnCode_t	copy_from_topic_qos (DDS::DataReaderQos &a_datareader_qos, const DDS::TopicQos &a_topic_qos)

virtual DDS::ReturnCode_t	enable ()

bool	is_clean (String *leftover_entities=0) const

void	data_received (DataReaderImpl *reader)

DDS::ReturnCode_t	reader_enabled (const char topic_name, DataReaderImpl reader)

DDS::ReturnCode_t	multitopic_reader_enabled (DDS::DataReader_ptr reader)

void	remove_from_datareader_set (DataReaderImpl *reader)

DDS::SubscriberListener_ptr	listener_for (DDS::StatusKind kind)

typedef	OPENDDS_VECTOR (GUID_t) SubscriptionIdVec

void	get_subscription_ids (SubscriptionIdVec &subs)

void	update_ownership_strength (const GUID_t &pub_id, const CORBA::Long &ownership_strength)

void	coherent_change_received (const GUID_t &publisher_id, DataReaderImpl *reader, Coherent_State &group_state)

virtual RcHandle< EntityImpl >	parent () const

Raw Latency Statistics Configuration Interfaces
unsigned int &	raw_latency_buffer_size ()
	Configure the size of the raw data collection buffer. More...

DataCollector< double >::OnFull &	raw_latency_buffer_type ()
	Configure the type of the raw data collection buffer. More...

Public Member Functions inherited from DDS::Subscriber
DataReader	create_datareader (in TopicDescription a_topic, in DataReaderQos qos, in DataReaderListener a_listener, in StatusMask mask)

ReturnCode_t	delete_datareader (in DataReader a_datareader)

DataReader	lookup_datareader (in string topic_name)

ReturnCode_t	get_datareaders (inout DataReaderSeq readers, in SampleStateMask sample_states, in ViewStateMask view_states, in InstanceStateMask instance_states)

ReturnCode_t	set_qos (in SubscriberQos qos)

ReturnCode_t	get_qos (inout SubscriberQos qos)

ReturnCode_t	set_listener (in SubscriberListener a_listener, in StatusMask mask)

ReturnCode_t	set_default_datareader_qos (in DataReaderQos qos)

ReturnCode_t	get_default_datareader_qos (inout DataReaderQos qos)

ReturnCode_t	copy_from_topic_qos (inout DataReaderQos a_datareader_qos, in TopicQos a_topic_qos)

Public Member Functions inherited from OpenDDS::DCPS::LocalObjectBase
virtual void	_add_ref ()

virtual void	_remove_ref ()

virtual CORBA::ULong	_refcount_value () const

Public Member Functions inherited from CORBA::LocalObject
virtual	~LocalObject (void)

virtual CORBA::Boolean	_non_existent (void)

virtual char *	_repository_id (void)

virtual CORBA::InterfaceDef_ptr	_get_interface (void)

virtual CORBA::Object_ptr	_get_component (void)

virtual void	_create_request (CORBA::Context_ptr ctx, const char *operation, CORBA::NVList_ptr arg_list, CORBA::NamedValue_ptr result, CORBA::Request_ptr &request, CORBA::Flags req_flags)

virtual void	_create_request (CORBA::Context_ptr ctx, const char *operation, CORBA::NVList_ptr arg_list, CORBA::NamedValue_ptr result, CORBA::ExceptionList_ptr exclist, CORBA::ContextList_ptr ctxtlist, CORBA::Request_ptr &request, CORBA::Flags req_flags)

virtual CORBA::Request_ptr	_request (const char *operation)

CORBA::Policy_ptr	_get_policy (CORBA::PolicyType type)

CORBA::Policy_ptr	_get_cached_policy (TAO_Cached_Policy_Type type)

CORBA::Object_ptr	_set_policy_overrides (const CORBA::PolicyList &policies, CORBA::SetOverrideType set_add)

CORBA::PolicyList *	_get_policy_overrides (const CORBA::PolicyTypeSeq &types)

CORBA::Boolean	_validate_connection (CORBA::PolicyList_out inconsistent_policies)

virtual CORBA::ULong	_hash (CORBA::ULong maximum)

virtual CORBA::Boolean	_is_equivalent (CORBA::Object_ptr other_obj)

virtual CORBA::ORB_ptr	_get_orb (void)

virtual TAO::ObjectKey *	_key (void)

Public Member Functions inherited from CORBA::Object
virtual	~Object (void)

virtual TAO_Abstract_ServantBase *	_servant (void) const

virtual CORBA::Boolean	_is_collocated (void) const

virtual CORBA::Boolean	_is_local (void) const

	Object (TAO_Stub p, CORBA::Boolean collocated=false, TAO_Abstract_ServantBase servant=0, TAO_ORB_Core *orb_core=0)

	Object (IOP::IOR ior, TAO_ORB_Core orb_core)

virtual TAO_Stub *	_stubobj (void) const

virtual TAO_Stub *	_stubobj (void)

virtual void	_proxy_broker (TAO::Object_Proxy_Broker *proxy_broker)

virtual CORBA::Boolean	marshal (TAO_OutputCDR &cdr)

CORBA::Boolean	is_evaluated (void) const

TAO_ORB_Core *	orb_core (void) const

IOP::IOR *	steal_ior (void)

const IOP::IOR &	ior (void) const

virtual bool	can_convert_to_ior (void) const

virtual char *	convert_to_ior (bool use_omg_ior_format, const char *ior_prefix) const

void	_decr_refcount (void)

virtual CORBA::Boolean	_is_a (const char *logical_type_id)

virtual const char *	_interface_repository_id (void) const

CORBA::Policy_ptr	_get_policy (CORBA::PolicyType type)

CORBA::Policy_ptr	_get_cached_policy (TAO_Cached_Policy_Type type)

CORBA::Object_ptr	_set_policy_overrides (const CORBA::PolicyList &policies, CORBA::SetOverrideType set_add)

CORBA::PolicyList *	_get_policy_overrides (const CORBA::PolicyTypeSeq &types)

CORBA::Boolean	_validate_connection (CORBA::PolicyList_out inconsistent_policies)

Public Member Functions inherited from OpenDDS::DCPS::RcObject
virtual	~RcObject ()

long	ref_count () const

WeakObject *	_get_weak_object () const

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

virtual	~EntityImpl ()

bool	is_enabled () const

virtual DDS::StatusCondition_ptr	get_statuscondition ()

virtual DDS::StatusMask	get_status_changes ()

virtual DDS::DomainId_t	get_domain_id ()

virtual GUID_t	get_id () const

void	set_status_changed_flag (DDS::StatusKind status, bool status_changed_flag)

void	notify_status_condition ()

virtual void	transport_config (const TransportConfig_rch &cfg)

TransportConfig_rch	transport_config () const

void	set_observer (Observer_rch observer, Observer::Event e)

Observer_rch	get_observer (Observer::Event e)

Static Public Member Functions
static bool	validate_datareader_qos (const DDS::DataReaderQos &qos, const DDS::DataReaderQos &default_qos, DDS::Topic_ptr a_topic, DDS::DataReaderQos &result_qos, bool mt)

Static Public Member Functions inherited from OpenDDS::DCPS::LocalObject< DDS::Subscriber >
static _ptr_type	_narrow (CORBA::Object_ptr obj)

Static Public Member Functions inherited from CORBA::LocalObject
static LocalObject_ptr	_duplicate (LocalObject_ptr obj)

static LocalObject_ptr	_nil (void)

static LocalObject_ptr	_narrow (CORBA::Object_ptr obj)

Static Public Member Functions inherited from CORBA::Object
static CORBA::Boolean	marshal (const Object_ptr x, TAO_OutputCDR &cdr)

static void	_tao_any_destructor (void *)

static CORBA::Boolean	is_nil_i (CORBA::Object_ptr obj)

static void	tao_object_initialize (Object *)

static CORBA::Object_ptr	_duplicate (CORBA::Object_ptr obj)

static CORBA::Object_ptr	_nil (void)

static CORBA::Object_ptr	_narrow (CORBA::Object_ptr obj)

Static Public Member Functions inherited from OpenDDS::DCPS::LocalObject< DDS::Entity >
static _ptr_type	_narrow (CORBA::Object_ptr obj)

Private Member Functions
typedef	OPENDDS_MULTIMAP (OPENDDS_STRING, DataReaderImpl_rch) DataReaderMap

typedef	OPENDDS_SET (DataReaderImpl_rch) DataReaderSet

typedef	OPENDDS_MAP_CMP (GUID_t, DDS::DataReaderQos, GUID_tKeyLessThan) DrIdToQosMap
	DataReader id to qos map. More...

typedef	OPENDDS_MAP (String, DDS::DataReader_var) MultitopicReaderMap

Private Attributes
DDS::InstanceHandle_t	handle_

DDS::SubscriberQos	qos_

DDS::DataReaderQos	default_datareader_qos_

ACE_Thread_Mutex	listener_mutex_

DDS::StatusMask	listener_mask_

DDS::SubscriberListener_var	listener_

DataReaderSet	readers_not_enabled_

DataReaderMap	datareader_map_

DataReaderSet	datareader_set_

MultitopicReaderMap	multitopic_reader_map_

WeakRcHandle< DomainParticipantImpl >	participant_

DDS::DomainId_t	domain_id_

GUID_t	dp_id_

unsigned int	raw_latency_buffer_size_
	Bound (or initial reservation) of raw latency buffers. More...

DataCollector< double >::OnFull	raw_latency_buffer_type_
	Type of raw latency data buffers. More...

ACE_Recursive_Thread_Mutex	dr_set_lock_

ACE_Recursive_Thread_Mutex	si_lock_

unique_ptr< Monitor >	monitor_
	Monitor object for this entity. More...

int	access_depth_

Additional Inherited Members
Public Types inherited from OpenDDS::DCPS::LocalObject< DDS::Subscriber >
typedef DDS::Subscriber ::_ptr_type	_ptr_type

typedef DDS::Subscriber ::_var_type	_var_type

Public Types inherited from CORBA::LocalObject
typedef LocalObject_ptr	_ptr_type

typedef LocalObject_var	_var_type

typedef LocalObject_out	_out_type

Public Types inherited from CORBA::Object
typedef Object_ptr	_ptr_type

typedef Object_var	_var_type

typedef Object_out	_out_type

Public Types inherited from OpenDDS::DCPS::LocalObject< DDS::Entity >
typedef DDS::Entity ::_ptr_type	_ptr_type

typedef DDS::Entity ::_var_type	_var_type

Protected Member Functions inherited from CORBA::LocalObject
	LocalObject (void)

Protected Member Functions inherited from CORBA::Object
	Object (int dummy=0)

TAO::Object_Proxy_Broker *	proxy_broker () const

Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Protected Member Functions inherited from OpenDDS::DCPS::EntityImpl
DDS::ReturnCode_t	set_enabled ()

void	set_deleted (bool state)

bool	get_deleted () const

DDS::InstanceHandle_t	get_entity_instance_handle (const GUID_t &id, const RcHandle< DomainParticipantImpl > &participant)

Protected Attributes inherited from CORBA::Object
ACE_Atomic_Op< TAO_SYNCH_MUTEX, unsigned long >	refcount_

Protected Attributes inherited from OpenDDS::DCPS::EntityImpl
AtomicBool	enabled_
	The flag indicates the entity is enabled. More...

AtomicBool	entity_deleted_
	The flag indicates the entity is being deleted. More...

Detailed Description

Definition at line 36 of file SubscriberImpl.h.

Constructor & Destructor Documentation

◆ SubscriberImpl()

OpenDDS::DCPS::SubscriberImpl::SubscriberImpl	(	DDS::InstanceHandle_t	handle,
		const DDS::SubscriberQos &	qos,
		DDS::SubscriberListener_ptr	a_listener,
		const DDS::StatusMask &	mask,
		DomainParticipantImpl *	participant
	)

Definition at line 41 of file SubscriberImpl.cpp.

References listener_, monitor_, and TheServiceParticipant.

   : handle_(handle),
   qos_(qos),
   default_datareader_qos_(TheServiceParticipant->initial_DataReaderQos()),
   listener_mask_(mask),
   participant_(*participant),
   domain_id_(participant->get_domain_id()),
   raw_latency_buffer_size_(0),
   raw_latency_buffer_type_(DataCollector<double>::KeepOldest),
   access_depth_ (0)
 {
   //Note: OK to duplicate a nil.
   listener_ = DDS::SubscriberListener::_duplicate(a_listener);
 
   monitor_.reset(TheServiceParticipant->monitor_factory_->create_subscriber_monitor(this));
 }

◆ ~SubscriberImpl()

OpenDDS::DCPS::SubscriberImpl::~SubscriberImpl ( )

virtual

Definition at line 62 of file SubscriberImpl.cpp.

References ACE_ERROR, handle_, is_clean(), LM_WARNING, OpenDDS::DCPS::log_level, participant_, and OpenDDS::DCPS::LogLevel::Warning.

 {
   const RcHandle<DomainParticipantImpl> participant = participant_.lock();
   if (participant) {
     participant->return_handle(handle_);
   }
 
   // The datareaders should be deleted already before calling delete
   // subscriber.
   String leftover_entities;
   if (!is_clean(&leftover_entities)) {
     if (log_level >= LogLevel::Warning) {
       ACE_ERROR((LM_WARNING, "(%P|%t) WARNING: SubscriberImpl::~SubscriberImpl: "
                  "%C still exist\n", leftover_entities.c_str()));
     }
   }
 }

Member Function Documentation

◆ begin_access()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::begin_access ( )

virtual

Implements DDS::Subscriber.

Definition at line 728 of file SubscriberImpl.cpp.

References access_depth_, DDS::PresentationQosPolicy::access_scope, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), datareader_set_, OpenDDS::DCPS::DCPS_debug_level, dr_set_lock_, OpenDDS::DCPS::EntityImpl::enabled_, DDS::GROUP_PRESENTATION_QOS, LM_ERROR, DDS::SubscriberQos::presentation, qos_, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, and si_lock_.

 {
   DataReaderSet to_call;
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                      si_guard,
                      si_lock_,
                      DDS::RETCODE_ERROR);
     if (!enabled_) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: SubscriberImpl::begin_access:")
                    ACE_TEXT(" Subscriber is not enabled!\n")));
       }
       return DDS::RETCODE_NOT_ENABLED;
     }
 
     if (qos_.presentation.access_scope != DDS::GROUP_PRESENTATION_QOS) {
       return DDS::RETCODE_OK;
     }
 
     ++access_depth_;
     // We should only notify subscription on the first
     // and last change to the current change set:
     if (access_depth_ == 1) {
       ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                        dr_set_guard,
                        dr_set_lock_,
                        DDS::RETCODE_ERROR);
       to_call = datareader_set_;
     }
   }
 
   for (DataReaderSet::iterator it = to_call.begin(); it != to_call.end(); ++it) {
     (*it)->begin_access();
   }
   return DDS::RETCODE_OK;
 }

◆ coherent_change_received()

void OpenDDS::DCPS::SubscriberImpl::coherent_change_received	(	const GUID_t &	publisher_id,
		DataReaderImpl *	reader,
		Coherent_State &	group_state
	)

Definition at line 1038 of file SubscriberImpl.cpp.

References ACE_GUARD, OpenDDS::DCPS::COMPLETED, datareader_set_, dr_set_lock_, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::NOT_COMPLETED_YET, OpenDDS::DCPS::REJECTED, and state.

 {
   DataReaderSet localdrs;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex,
               guard,
               this->dr_set_lock_);
      localdrs = datareader_set_;
   }
   // Verify if all readers complete the coherent changes. The result
   // is either COMPLETED or REJECTED.
   group_state = COMPLETED;
   for (DataReaderSet::const_iterator iter = localdrs.begin();
        iter != localdrs.end(); ++iter) {
 
     Coherent_State state = COMPLETED;
     (*iter)->coherent_change_received (publisher_id, state);
     if (state == NOT_COMPLETED_YET) {
       group_state = NOT_COMPLETED_YET;
       return;
     }
     else if (state == REJECTED) {
       group_state = REJECTED;
     }
   }
 
   GUID_t writerId = GUID_UNKNOWN;
   for (DataReaderSet::const_iterator iter = localdrs.begin();
        iter != localdrs.end(); ++iter) {
     if (group_state == COMPLETED) {
       (*iter)->accept_coherent (writerId, publisher_id);
     }
     else {   //REJECTED
       (*iter)->reject_coherent (writerId, publisher_id);
     }
   }
 
   if (group_state == COMPLETED) {
     for (DataReaderSet::const_iterator iter = localdrs.begin();
          iter != localdrs.end(); ++iter) {
       (*iter)->coherent_changes_completed (reader);
       (*iter)->reset_coherent_info (writerId, publisher_id);
     }
   }
 }

◆ contains_reader()

bool OpenDDS::DCPS::SubscriberImpl::contains_reader ( DDS::InstanceHandle_t a_handle )

Definition at line 87 of file SubscriberImpl.cpp.

References ACE_GUARD_RETURN, datareader_map_, and si_lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    this->si_lock_,
                    false);
 
   for (DataReaderMap::iterator it(datareader_map_.begin());
        it != datareader_map_.end(); ++it) {
     if (a_handle == it->second->get_instance_handle()) {
       return true;
     }
   }
 
   return false;
 }

◆ copy_from_topic_qos()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::copy_from_topic_qos	(	DDS::DataReaderQos &	a_datareader_qos,
		const DDS::TopicQos &	a_topic_qos
	)

virtual

Definition at line 846 of file SubscriberImpl.cpp.

References OpenDDS::DCPS::Qos_Helper::copy_from_topic_qos(), DDS::RETCODE_INCONSISTENT_POLICY, and DDS::RETCODE_OK.

 {
   if (Qos_Helper::copy_from_topic_qos(a_datareader_qos, a_topic_qos) ) {
     return DDS::RETCODE_OK;
 
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ create_datareader()

DDS::DataReader_ptr OpenDDS::DCPS::SubscriberImpl::create_datareader	(	DDS::TopicDescription_ptr	a_topic_desc,
		const DDS::DataReaderQos &	qos,
		DDS::DataReaderListener_ptr	a_listener,
		DDS::StatusMask	mask
	)

virtual

Definition at line 105 of file SubscriberImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), DDS::EntityFactoryQosPolicy::autoenable_created_entities, OpenDDS::DCPS::DCPS_debug_level, default_datareader_qos_, OpenDDS::DCPS::DataReaderImpl::enable(), OpenDDS::DCPS::DataReaderImpl::enable_filtering(), OpenDDS::DCPS::EntityImpl::enabled_, DDS::SubscriberQos::entity_factory, OpenDDS::DCPS::ContentFilteredTopicImpl::get_related_topic(), OpenDDS::DCPS::TopicDescriptionImpl::get_type_support(), TAO::String_var< charT >::in(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::MultiTopicDataReaderBase::init(), OpenDDS::DCPS::DataReaderImpl::init(), CORBA::is_nil(), LM_ERROR, LM_WARNING, multitopic_reader_enabled(), name, participant_, qos_, OpenDDS::DCPS::DataReaderImpl::raw_latency_buffer_size(), raw_latency_buffer_size_, OpenDDS::DCPS::DataReaderImpl::raw_latency_buffer_type(), raw_latency_buffer_type_, OpenDDS::DCPS::rchandle_from(), readers_not_enabled_, DDS::RETCODE_OK, si_lock_, and validate_datareader_qos().

Referenced by OpenDDS::DCPS::MultiTopicDataReaderBase::init().

 {
   if (CORBA::is_nil(a_topic_desc)) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: ")
                 ACE_TEXT("SubscriberImpl::create_datareader, ")
                 ACE_TEXT("topic desc is nil.\n")));
     }
     return DDS::DataReader::_nil();
   }
 
   DDS::DataReaderQos dr_qos;
   RcHandle<DomainParticipantImpl> participant = this->participant_.lock();
   if (!participant)
     return DDS::DataReader::_nil();
 
   TopicImpl* topic_servant = dynamic_cast<TopicImpl*>(a_topic_desc);
 
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
   ContentFilteredTopicImpl* cft = 0;
 #endif
 #ifndef OPENDDS_NO_MULTI_TOPIC
   MultiTopicImpl* mt = 0;
 #else
   bool mt = false;
 #endif
 
   if (!topic_servant) {
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
     cft = dynamic_cast<ContentFilteredTopicImpl*>(a_topic_desc);
     if (cft) {
       DDS::Topic_var related;
       related = cft->get_related_topic();
       topic_servant = dynamic_cast<TopicImpl*>(related.in());
     }
     else
 #endif
     {
 #ifndef OPENDDS_NO_MULTI_TOPIC
       mt = dynamic_cast<MultiTopicImpl*>(a_topic_desc);
 #endif
     }
   }
 
   if (!validate_datareader_qos (qos, default_datareader_qos_, topic_servant, dr_qos, mt))
     return DDS::DataReader::_nil();
 
 #ifndef OPENDDS_NO_MULTI_TOPIC
   if (mt) {
     try {
       DDS::DataReader_var dr =
         mt->get_type_support()->create_multitopic_datareader();
       MultiTopicDataReaderBase* mtdr =
         dynamic_cast<MultiTopicDataReaderBase*>(dr.in());
       mtdr->init(dr_qos, a_listener, mask, this, mt);
       if (enabled_ && qos_.entity_factory.autoenable_created_entities) {
         if (dr->enable() != DDS::RETCODE_OK) {
           if (DCPS_debug_level > 0) {
             ACE_ERROR((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("SubscriberImpl::create_datareader, ")
                       ACE_TEXT("enable of MultiTopicDataReader failed.\n")));
           }
           return DDS::DataReader::_nil();
         }
         multitopic_reader_enabled(dr);
       }
       return dr._retn();
     } catch (const std::exception& e) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) ERROR: ")
                   ACE_TEXT("SubscriberImpl::create_datareader, ")
                   ACE_TEXT("creation of MultiTopicDataReader failed: %C.\n"),
                   e.what()));
       }
     }
     return DDS::DataReader::_nil();
   }
 #endif
 
   OpenDDS::DCPS::TypeSupport_ptr typesupport =
     topic_servant->get_type_support();
 
   if (0 == typesupport) {
     CORBA::String_var name = a_topic_desc->get_name();
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: ")
                 ACE_TEXT("SubscriberImpl::create_datareader, ")
                 ACE_TEXT("typesupport(topic_name=%C) is nil.\n"),
                 name.in()));
     }
     return DDS::DataReader::_nil();
   }
 
   DDS::DataReader_var dr_obj = typesupport->create_datareader();
 
   DataReaderImpl* dr_servant =
     dynamic_cast<DataReaderImpl*>(dr_obj.in());
 
   if (dr_servant == 0) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("SubscriberImpl::create_datareader, ")
           ACE_TEXT("servant is nil.\n")));
     }
     return DDS::DataReader::_nil();
   }
 
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
   if (cft) {
     dr_servant->enable_filtering(cft);
   }
 #endif
 
   // Propagate the latency buffer data collection configuration.
   // @TODO: Determine whether we want to exclude the Builtin Topic
   //        readers from data gathering.
   dr_servant->raw_latency_buffer_size() = this->raw_latency_buffer_size_;
   dr_servant->raw_latency_buffer_type() = this->raw_latency_buffer_type_;
 
 
   dr_servant->init(topic_servant,
                    dr_qos,
                    a_listener,
                    mask,
                    participant.in(),
                    this);
 
   if (enabled_ && qos_.entity_factory.autoenable_created_entities) {
     const DDS::ReturnCode_t ret = dr_servant->enable();
 
     if (ret != DDS::RETCODE_OK) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_WARNING,
                   ACE_TEXT("(%P|%t) WARNING: ")
                   ACE_TEXT("SubscriberImpl::create_datareader, ")
                   ACE_TEXT("enable failed.\n")));
       }
       return DDS::DataReader::_nil();
     }
   } else {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, si_lock_, 0);
     readers_not_enabled_.insert(rchandle_from(dr_servant));
   }
 
   // add created data reader to this' data reader container -
   // done in enable_reader
   return DDS::DataReader::_duplicate(dr_obj.in());
 }

◆ data_received()

void OpenDDS::DCPS::SubscriberImpl::data_received ( DataReaderImpl * reader )

Definition at line 917 of file SubscriberImpl.cpp.

References ACE_GUARD, datareader_set_, dr_set_lock_, and OpenDDS::DCPS::rchandle_from().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex,
             guard,
             this->dr_set_lock_);
   datareader_set_.insert(rchandle_from(reader));
 }

◆ delete_contained_entities()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::delete_contained_entities ( )

virtual

Implements DDS::Subscriber.

Definition at line 395 of file SubscriberImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), datareader_map_, OpenDDS::DCPS::DCPS_debug_level, delete_datareader(), LM_ERROR, multitopic_reader_map_, OPENDDS_VECTOR(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_deleted(), and si_lock_.

Referenced by OpenDDS::DCPS::DomainParticipantImpl::delete_subscriber().

 {
   // mark that the entity is being deleted
   set_deleted(true);
 
   OPENDDS_VECTOR(DDS::DataReader*) drs;
 
 #ifndef OPENDDS_NO_MULTI_TOPIC
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                      guard,
                      this->si_lock_,
                      DDS::RETCODE_ERROR);
     for (MultitopicReaderMap::iterator mt_iter = multitopic_reader_map_.begin();
          mt_iter != multitopic_reader_map_.end(); ++mt_iter) {
       drs.push_back(mt_iter->second);
     }
   }
 
   for (size_t i = 0; i < drs.size(); ++i) {
     DDS::ReturnCode_t ret = drs[i]->delete_contained_entities();
     if (ret == DDS::RETCODE_OK) {
       ret = delete_datareader(drs[i]);
     }
     if (ret != DDS::RETCODE_OK) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) ERROR: ")
                   ACE_TEXT("SubscriberImpl::delete_contained_entities, ")
                   ACE_TEXT("failed to delete datareader\n")));
       }
       return ret;
     }
   }
   drs.clear();
 #endif
 
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                      guard,
                      this->si_lock_,
                      DDS::RETCODE_ERROR);
     DataReaderMap::iterator it;
     DataReaderMap::iterator itEnd = datareader_map_.end();
 
     for (it = datareader_map_.begin(); it != itEnd; ++it) {
       drs.push_back(it->second.in());
     }
   }
 
   for (size_t i = 0; i < drs.size(); ++i) {
     DDS::ReturnCode_t ret = drs[i]->delete_contained_entities();
     if (ret == DDS::RETCODE_OK) {
       ret = delete_datareader(drs[i]);
     }
     if (ret != DDS::RETCODE_OK) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) ERROR: ")
                   ACE_TEXT("SubscriberImpl::delete_contained_entities, ")
                   ACE_TEXT("failed to delete datareader\n")));
       }
       return ret;
     }
   }
 
   // the subscriber can now start creating new publications
   set_deleted(false);
 
   return DDS::RETCODE_OK;
 }

◆ delete_datareader()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::delete_datareader ( DDS::DataReader_ptr a_datareader )

virtual

Definition at line 264 of file SubscriberImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::MultiTopicDataReaderBase::cleanup(), datareader_map_, datareader_set_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, domain_id_, dr_set_lock_, OpenDDS::DCPS::RcHandle< T >::get(), TAO::String_var< charT >::in(), LM_ERROR, LM_NOTICE, OpenDDS::DCPS::log_level, monitor_, multitopic_reader_map_, OpenDDS::DCPS::LogLevel::Notice, OpenDDS::DCPS::rchandle_from(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, DDS::RETCODE_PRECONDITION_NOT_MET, OpenDDS::DCPS::retcode_to_string(), si_lock_, and TheServiceParticipant.

Referenced by delete_contained_entities().

 {
   DBG_ENTRY_LVL("SubscriberImpl", "delete_datareader", 6);
 
   DataReaderImpl_rch dr_servant = rchandle_from(dynamic_cast<DataReaderImpl*>(a_datareader));
 
   if (dr_servant) { // for MultiTopic this will be false
     const char* reason = " (ERROR: unknown reason)";
     DDS::ReturnCode_t rc = DDS::RETCODE_OK;
     RcHandle<SubscriberImpl> dr_subscriber = dr_servant->get_subscriber_servant();
     if (dr_subscriber.get() != this) {
       reason = "doesn't belong to this subscriber.";
       rc = DDS::RETCODE_PRECONDITION_NOT_MET;
     } else if (dr_servant->has_zero_copies()) {
       reason = "has outstanding zero-copy samples loaned out.";
       rc = DDS::RETCODE_PRECONDITION_NOT_MET;
     } else if (!dr_servant->read_conditions_.empty()) {
       reason = "has read conditions attached.";
       rc = DDS::RETCODE_PRECONDITION_NOT_MET;
     }
     if (rc != DDS::RETCODE_OK) {
       if (log_level >= LogLevel::Notice) {
         DDS::TopicDescription_var topic = a_datareader->get_topicdescription();
         CORBA::String_var topic_name = topic->get_name();
         ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: SubscriberImpl::delete_datareader: "
           "on reader %C (topic \"%C\") will return \"%C\" because it %C\n",
           LogGuid(dr_servant->get_id()).c_str(), topic_name.in(),
           retcode_to_string(rc), reason));
       }
       return rc;
     }
 
     // marks entity as deleted and stops future associating
     dr_servant->prepare_to_delete();
   }
 
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                      si_guard,
                      this->si_lock_,
                      DDS::RETCODE_ERROR);
 
     DataReaderMap::iterator it;
     for (it = datareader_map_.begin(); it != datareader_map_.end(); ++it) {
       if (it->second == dr_servant) {
         break;
       }
     }
 
     if (it == datareader_map_.end()) {
       DDS::TopicDescription_var td = a_datareader->get_topicdescription();
       CORBA::String_var topic_name = td->get_name();
 #ifndef OPENDDS_NO_MULTI_TOPIC
       MultitopicReaderMap::iterator mt_iter = multitopic_reader_map_.find(topic_name.in());
       if (mt_iter != multitopic_reader_map_.end()) {
         DDS::DataReader_ptr ptr = mt_iter->second;
         MultiTopicDataReaderBase* mtdrb = dynamic_cast<MultiTopicDataReaderBase*>(ptr);
         if (!mtdrb) {
           if (DCPS_debug_level > 0) {
             ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) ERROR: ")
               ACE_TEXT("SubscriberImpl::delete_datareader: ")
               ACE_TEXT("datareader(topic_name=%C)")
               ACE_TEXT("failed to obtain MultiTopicDataReaderBase.\n"),
               topic_name.in()));
           }
           return ::DDS::RETCODE_ERROR;
         }
         mtdrb->cleanup();
         multitopic_reader_map_.erase(mt_iter);
         return DDS::RETCODE_OK;
       }
 #endif
       if (!dr_servant) {
         if (DCPS_debug_level > 0) {
           ACE_ERROR((LM_ERROR,
                     ACE_TEXT("(%P|%t) ERROR: ")
                     ACE_TEXT("SubscriberImpl::delete_datareader: ")
                     ACE_TEXT("datareader(topic_name=%C)")
                     ACE_TEXT("for unknown repo id not found.\n"),
                     topic_name.in()));
         }
         return ::DDS::RETCODE_ERROR;
       }
       if (DCPS_debug_level > 0) {
         GUID_t id = dr_servant->get_guid();
         ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) ERROR: ")
                   ACE_TEXT("SubscriberImpl::delete_datareader: ")
                   ACE_TEXT("datareader(topic_name=%C) %C not found.\n"),
                   topic_name.in(),
                   LogGuid(id).c_str()));
       }
       return ::DDS::RETCODE_ERROR;
     }
 
     datareader_map_.erase(it);
 
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                      dr_set_guard,
                      this->dr_set_lock_,
                      DDS::RETCODE_ERROR);
     datareader_set_.erase(dr_servant);
   }
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 
   const GUID_t subscription_id = dr_servant->get_guid();
   Discovery_rch disco = TheServiceParticipant->get_discovery(this->domain_id_);
   if (!disco->remove_subscription(this->domain_id_,
                                   this->dp_id_,
                                   subscription_id)) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: ")
                 ACE_TEXT("SubscriberImpl::delete_datareader: ")
                 ACE_TEXT(" could not remove subscription from discovery.\n")));
     }
     return ::DDS::RETCODE_ERROR;
   }
 
   // Call remove association before unregistering the datareader from the transport,
   // otherwise some callbacks resulted from remove_association may be lost.
   dr_servant->remove_all_associations();
   dr_servant->cleanup();
   return DDS::RETCODE_OK;
 }

◆ enable()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::enable ( )

virtual

Implements DDS::Entity.

Definition at line 859 of file SubscriberImpl.cpp.

References ACE_GUARD_RETURN, DDS::EntityFactoryQosPolicy::autoenable_created_entities, dp_id_, DDS::SubscriberQos::entity_factory, OpenDDS::DCPS::EntityImpl::is_enabled(), monitor_, participant_, qos_, readers_not_enabled_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, DDS::RETCODE_PRECONDITION_NOT_MET, OpenDDS::DCPS::EntityImpl::set_enabled(), and si_lock_.

Referenced by OpenDDS::DCPS::DomainParticipantImpl::create_subscriber().

 {
   //According spec:
   // - Calling enable on an already enabled Entity returns OK and has no
   // effect.
   // - Calling enable on an Entity whose factory is not enabled will fail
   // and return PRECONDITION_NOT_MET.
 
   if (this->is_enabled()) {
     return DDS::RETCODE_OK;
   }
 
   RcHandle<DomainParticipantImpl> participant = this->participant_.lock();
   if (!participant || !participant->is_enabled()) {
     return DDS::RETCODE_PRECONDITION_NOT_MET;
   }
 
   dp_id_ = participant->get_id();
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 
   this->set_enabled();
 
   if (qos_.entity_factory.autoenable_created_entities) {
     DataReaderSet readers;
     {
       ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, si_lock_, DDS::RETCODE_ERROR);
       readers_not_enabled_.swap(readers);
     }
     for (DataReaderSet::iterator it = readers.begin(); it != readers.end(); ++it) {
       (*it)->enable();
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ end_access()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::end_access ( )

virtual

Implements DDS::Subscriber.

Definition at line 768 of file SubscriberImpl.cpp.

References access_depth_, DDS::PresentationQosPolicy::access_scope, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), datareader_set_, OpenDDS::DCPS::DCPS_debug_level, dr_set_lock_, OpenDDS::DCPS::EntityImpl::enabled_, DDS::GROUP_PRESENTATION_QOS, LM_ERROR, DDS::SubscriberQos::presentation, qos_, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, DDS::RETCODE_PRECONDITION_NOT_MET, and si_lock_.

 {
   DataReaderSet to_call;
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                      si_guard,
                      si_lock_,
                      DDS::RETCODE_ERROR);
     if (!enabled_) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: SubscriberImpl::end_access:")
                    ACE_TEXT(" Publisher is not enabled!\n")));
       }
       return DDS::RETCODE_NOT_ENABLED;
     }
 
     if (qos_.presentation.access_scope != DDS::GROUP_PRESENTATION_QOS) {
       return DDS::RETCODE_OK;
     }
 
     if (access_depth_ == 0) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
                    ACE_TEXT("(%P|%t) ERROR: SubscriberImpl::end_access:")
                    ACE_TEXT(" No matching call to begin_coherent_changes!\n")));
       }
       return DDS::RETCODE_PRECONDITION_NOT_MET;
     }
 
     --access_depth_;
     // We should only notify subscription on the first
     // and last change to the current change set:
     if (access_depth_ == 0) {
       ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                        dr_set_guard,
                        dr_set_lock_,
                        DDS::RETCODE_ERROR);
       to_call = datareader_set_;
     }
   }
 
   for (DataReaderSet::iterator it = to_call.begin(); it != to_call.end(); ++it) {
     (*it)->end_access();
   }
   return DDS::RETCODE_OK;
 }

◆ get_datareaders()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::get_datareaders	(	DDS::DataReaderSeq &	readers,
		DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states
	)

virtual

Definition at line 504 of file SubscriberImpl.cpp.

References access_depth_, DDS::PresentationQosPolicy::access_scope, ACE_GUARD_RETURN, DDS::PresentationQosPolicy::coherent_access, datareader_set_, dr_set_lock_, OpenDDS::DCPS::GroupRakeData::get_datareaders(), DDS::GROUP_PRESENTATION_QOS, DDS::PresentationQosPolicy::ordered_access, DDS::SubscriberQos::presentation, OpenDDS::DCPS::push_back(), qos_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, and DDS::RETCODE_PRECONDITION_NOT_MET.

 {
   DataReaderSet localreaders;
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                      guard,
                      this->dr_set_lock_,
                      DDS::RETCODE_ERROR);
     localreaders = datareader_set_;
   }
 
 #ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
   // If access_scope is GROUP and ordered_access is true then return readers as
   // list which may contain same readers multiple times. Otherwise return readers
   // as set.
   if (this->qos_.presentation.access_scope == ::DDS::GROUP_PRESENTATION_QOS) {
     if (this->access_depth_ == 0 && this->qos_.presentation.coherent_access) {
       return ::DDS::RETCODE_PRECONDITION_NOT_MET;
     }
     if (this->qos_.presentation.ordered_access) {
 
       GroupRakeData data;
       for (DataReaderSet::const_iterator pos = localreaders.begin();
            pos != localreaders.end(); ++pos) {
         (*pos)->get_ordered_data(data, sample_states, view_states, instance_states);
       }
 
       // Return list of readers in the order of the source timestamp of the received
       // samples from readers.
       data.get_datareaders(readers);
       return DDS::RETCODE_OK;
     }
   }
 #endif
 
   // Return set of datareaders.
   readers.length(0);
   for (DataReaderSet::const_iterator pos = localreaders.begin();
        pos != localreaders.end(); ++pos) {
     if ((*pos)->have_sample_states(sample_states) &&
         (*pos)->have_view_states(view_states) &&
         (*pos)->have_instance_states(instance_states)) {
       push_back(readers, DDS::DataReader::_duplicate(pos->in()));
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ get_default_datareader_qos()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::get_default_datareader_qos ( DDS::DataReaderQos & qos )

virtual

Definition at line 838 of file SubscriberImpl.cpp.

References default_datareader_qos_, and DDS::RETCODE_OK.

Referenced by OpenDDS::RTPS::RtpsDiscovery::init_bit(), and OpenDDS::DCPS::StaticDiscovery::init_bit().

 {
   qos = default_datareader_qos_;
   return DDS::RETCODE_OK;
 }

◆ get_instance_handle()

DDS::InstanceHandle_t OpenDDS::DCPS::SubscriberImpl::get_instance_handle ( )

virtual

Implements OpenDDS::DCPS::EntityImpl.

Definition at line 81 of file SubscriberImpl.cpp.

References handle_.

Referenced by OpenDDS::DCPS::SubscriberMonitorImpl::report().

 {
   return handle_;
 }

◆ get_listener()

DDS::SubscriberListener_ptr OpenDDS::DCPS::SubscriberImpl::get_listener ( )

virtual

Implements DDS::Subscriber.

Definition at line 719 of file SubscriberImpl.cpp.

References listener_, and listener_mutex_.

 {
   ACE_Guard<ACE_Thread_Mutex> g(listener_mutex_);
   return DDS::SubscriberListener::_duplicate(listener_.in());
 }

◆ get_participant()

DDS::DomainParticipant_ptr OpenDDS::DCPS::SubscriberImpl::get_participant ( )

virtual

Implements DDS::Subscriber.

Definition at line 819 of file SubscriberImpl.cpp.

References participant_.

Referenced by OpenDDS::RTPS::RtpsDiscovery::create_bit_dr(), OpenDDS::DCPS::StaticDiscovery::create_bit_dr(), OpenDDS::DCPS::MultiTopicDataReaderBase::init(), and OpenDDS::DCPS::SubscriberMonitorImpl::report().

 {
   return participant_.lock()._retn();
 }

◆ get_qos()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::get_qos ( DDS::SubscriberQos & qos )

virtual

Definition at line 699 of file SubscriberImpl.cpp.

References qos_, and DDS::RETCODE_OK.

Referenced by OpenDDS::DCPS::DataReaderImpl::init().

 {
   qos = qos_;
   return DDS::RETCODE_OK;
 }

◆ get_subscription_ids()

void OpenDDS::DCPS::SubscriberImpl::get_subscription_ids ( SubscriptionIdVec & subs )

Populates a std::vector with the SubscriptionIds (GUIDs) of this Subscriber's Data Readers

Definition at line 1000 of file SubscriberImpl.cpp.

References ACE_GUARD_RETURN, datareader_map_, and si_lock_.

Referenced by OpenDDS::DCPS::SubscriberMonitorImpl::report().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    this->si_lock_,
                    );
 
   subs.reserve(datareader_map_.size());
   for (DataReaderMap::iterator iter = datareader_map_.begin();
        iter != datareader_map_.end();
        ++iter) {
     subs.push_back(iter->second->get_guid());
   }
 }

◆ is_clean()

bool OpenDDS::DCPS::SubscriberImpl::is_clean ( String * leftover_entities = 0 ) const

This method is not defined in the IDL and is defined for internal use. Check if there is any datareader associated with it.

Definition at line 898 of file SubscriberImpl.cpp.

References datareader_map_, OpenDDS::DCPS::NUMBER_OF_BUILT_IN_TOPICS, TheTransientKludge, and OpenDDS::DCPS::to_dds_string().

Referenced by OpenDDS::DCPS::DomainParticipantImpl::delete_subscriber(), and ~SubscriberImpl().

 {
   if (leftover_entities) {
     leftover_entities->clear();
   }
 
   size_t reader_count = datareader_map_.size();
   if (reader_count && !TheTransientKludge->is_enabled()) {
     // BIT datareaders.
     reader_count = reader_count == NUMBER_OF_BUILT_IN_TOPICS ? 0 : reader_count;
   }
   if (leftover_entities && reader_count) {
     *leftover_entities += to_dds_string(reader_count) + " reader(s)";
   }
 
   return reader_count == 0;
 }

◆ listener_for()

DDS::SubscriberListener_ptr OpenDDS::DCPS::SubscriberImpl::listener_for ( DDS::StatusKind kind )

Definition at line 968 of file SubscriberImpl.cpp.

References CORBA::is_nil(), listener_, listener_mask_, listener_mutex_, participant_, and ACE_Guard< ACE_LOCK >::release().

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::finish_store_instance_data().

 {
   // per 2.1.4.3.1 Listener Access to Plain Communication Status
   // use this entities factory if listener is mask not enabled
   // for this kind.
   RcHandle<DomainParticipantImpl> participant = this->participant_.lock();
   if (! participant)
     return 0;
 
   ACE_Guard<ACE_Thread_Mutex> g(listener_mutex_);
   if (CORBA::is_nil(listener_.in()) || (listener_mask_ & kind) == 0) {
     g.release();
     return participant->listener_for(kind);
 
   } else {
     return DDS::SubscriberListener::_duplicate(listener_.in());
   }
 }

◆ lookup_datareader()

DDS::DataReader_ptr OpenDDS::DCPS::SubscriberImpl::lookup_datareader ( const char * topic_name )

virtual

Definition at line 468 of file SubscriberImpl.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), datareader_map_, OpenDDS::DCPS::DCPS_debug_level, LM_DEBUG, multitopic_reader_map_, and si_lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    this->si_lock_,
                    DDS::DataReader::_nil());
 
   // If multiple entries whose key is "topic_name" then which one is
   // returned ? Spec does not limit which one should give.
   DataReaderMap::iterator it = datareader_map_.find(topic_name);
 
   if (it == datareader_map_.end()) {
 #ifndef OPENDDS_NO_MULTI_TOPIC
     MultitopicReaderMap::iterator mt_iter = multitopic_reader_map_.find(topic_name);
     if (mt_iter != multitopic_reader_map_.end()) {
       return DDS::DataReader::_duplicate(mt_iter->second);
     }
 #endif
 
     if (DCPS_debug_level >= 2) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) ")
                  ACE_TEXT("SubscriberImpl::lookup_datareader, ")
                  ACE_TEXT("The datareader(topic_name=%C) is not found\n"),
                  topic_name));
     }
 
     return DDS::DataReader::_nil();
 
   } else {
     return DDS::DataReader::_duplicate(it->second.in());
   }
 }

◆ multitopic_reader_enabled()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::multitopic_reader_enabled ( DDS::DataReader_ptr reader )

Definition at line 951 of file SubscriberImpl.cpp.

References TAO::String_var< charT >::in(), multitopic_reader_map_, and DDS::RETCODE_OK.

Referenced by create_datareader().

 {
   DDS::TopicDescription_var td = reader->get_topicdescription();
   CORBA::String_var topic = td->get_name();
   multitopic_reader_map_[topic.in()] = DDS::DataReader::_duplicate(reader);
   return DDS::RETCODE_OK;
 }

◆ notify_datareaders()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::notify_datareaders ( )

virtual

Implements DDS::Subscriber.

Definition at line 558 of file SubscriberImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), DDS::DATA_AVAILABLE_STATUS, datareader_map_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::MultiTopicDataReaderBase::get_listener(), OpenDDS::DCPS::MultiTopicDataReaderBase::have_sample_states(), CORBA::is_nil(), LM_ERROR, multitopic_reader_map_, DDS::NOT_READ_SAMPLE_STATE, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::MultiTopicDataReaderBase::set_status_changed_flag(), si_lock_, and TheServiceParticipant.

 {
   DataReaderMap localreadermap;
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                     guard,
                     this->si_lock_,
                     DDS::RETCODE_ERROR);
     localreadermap = datareader_map_;
   }
   for (DataReaderMap::iterator it = localreadermap.begin(); it != localreadermap.end(); ++it) {
     if (it->second->have_sample_states(DDS::NOT_READ_SAMPLE_STATE)) {
       DDS::DataReaderListener_var listener = it->second->get_listener();
       if (!it->second->is_bit()) {
         it->second->set_status_changed_flag(DDS::DATA_AVAILABLE_STATUS, false);
         if (listener) {
           listener->on_data_available(it->second.in());
         }
       } else {
         TheServiceParticipant->job_queue()->enqueue(make_rch<DataReaderImpl::OnDataAvailable>(listener, it->second, listener, true, false));
       }
     }
   }
 
 #ifndef OPENDDS_NO_MULTI_TOPIC
   MultitopicReaderMap localmtr;
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                     guard,
                     this->si_lock_,
                     DDS::RETCODE_ERROR);
     localmtr = multitopic_reader_map_;
   }
 
   for (MultitopicReaderMap::iterator it = localmtr.begin();
       it != localmtr.end(); ++it) {
     MultiTopicDataReaderBase* dri =
       dynamic_cast<MultiTopicDataReaderBase*>(it->second.in());
 
     if (!dri) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: SubscriberImpl::notify_datareaders: ")
           ACE_TEXT("failed to obtain MultiTopicDataReaderBase.\n")));
       }
       return ::DDS::RETCODE_ERROR;
     }
 
     if (dri->have_sample_states(DDS::NOT_READ_SAMPLE_STATE)) {
       DDS::DataReaderListener_var listener = dri->get_listener();
       dri->set_status_changed_flag(DDS::DATA_AVAILABLE_STATUS, false);
       if (!CORBA::is_nil(listener)) {
         listener->on_data_available(dri);
       }
     }
   }
 #endif
 
   return DDS::RETCODE_OK;
 }

◆ OPENDDS_MAP()

typedef OpenDDS::DCPS::SubscriberImpl::OPENDDS_MAP	(	String	,
		DDS::DataReader_var
	)

private

◆ OPENDDS_MAP_CMP()

typedef OpenDDS::DCPS::SubscriberImpl::OPENDDS_MAP_CMP	(	GUID_t	,
		DDS::DataReaderQos	,
		GUID_tKeyLessThan
	)

private

DataReader id to qos map.

◆ OPENDDS_MULTIMAP()

typedef OpenDDS::DCPS::SubscriberImpl::OPENDDS_MULTIMAP	(	OPENDDS_STRING	,
		DataReaderImpl_rch
	)

private

Keep track of all the DataReaders attached to this Subscriber: key is the topic_name

◆ OPENDDS_SET()

typedef OpenDDS::DCPS::SubscriberImpl::OPENDDS_SET ( DataReaderImpl_rch )

private

Keep track of DataReaders with data std::set for now, want to encapsulate this so we can switch between a set or list depending on Presentation QoS.

◆ OPENDDS_VECTOR()

typedef OpenDDS::DCPS::SubscriberImpl::OPENDDS_VECTOR ( GUID_t )

Referenced by delete_contained_entities().

◆ parent()

RcHandle< EntityImpl > OpenDDS::DCPS::SubscriberImpl::parent ( void ) const

virtual

Reimplemented from OpenDDS::DCPS::EntityImpl.

Definition at line 1088 of file SubscriberImpl.cpp.

References participant_.

 {
   return this->participant_.lock();
 }

◆ raw_latency_buffer_size()

unsigned int & OpenDDS::DCPS::SubscriberImpl::raw_latency_buffer_size ( )

Configure the size of the raw data collection buffer.

Definition at line 988 of file SubscriberImpl.cpp.

References raw_latency_buffer_size_.

Referenced by OpenDDS::DCPS::MultiTopicDataReaderBase::init().

 {
   return this->raw_latency_buffer_size_;
 }

◆ raw_latency_buffer_type()

DataCollector< double >::OnFull & OpenDDS::DCPS::SubscriberImpl::raw_latency_buffer_type ( )

Configure the type of the raw data collection buffer.

Definition at line 994 of file SubscriberImpl.cpp.

References raw_latency_buffer_type_.

Referenced by OpenDDS::DCPS::MultiTopicDataReaderBase::init().

 {
   return this->raw_latency_buffer_type_;
 }

◆ reader_enabled()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::reader_enabled	(	const char *	topic_name,
		DataReaderImpl *	reader
	)

Definition at line 926 of file SubscriberImpl.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), datareader_map_, OpenDDS::DCPS::DCPS_debug_level, LM_DEBUG, monitor_, OpenDDS::DCPS::rchandle_from(), readers_not_enabled_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, and si_lock_.

 {
   if (DCPS_debug_level >= 4) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) SubscriberImpl::reader_enabled, ")
                ACE_TEXT("datareader(topic_name=%C) enabled\n"),
                topic_name));
   }
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, si_lock_, DDS::RETCODE_ERROR);
   DataReaderImpl_rch reader = rchandle_from(reader_ptr);
   readers_not_enabled_.erase(reader);
 
   this->datareader_map_.insert(DataReaderMap::value_type(topic_name, reader));
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 
   return DDS::RETCODE_OK;
 }

◆ remove_from_datareader_set()

void OpenDDS::DCPS::SubscriberImpl::remove_from_datareader_set ( DataReaderImpl * reader )

Definition at line 960 of file SubscriberImpl.cpp.

References ACE_GUARD, datareader_set_, dr_set_lock_, and OpenDDS::DCPS::rchandle_from().

Referenced by OpenDDS::DCPS::MultiTopicDataReaderBase::cleanup().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, dr_set_lock_);
   datareader_set_.erase(rchandle_from(reader));
 }

◆ set_default_datareader_qos()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::set_default_datareader_qos ( const DDS::DataReaderQos & qos )

virtual

Definition at line 825 of file SubscriberImpl.cpp.

References OpenDDS::DCPS::Qos_Helper::consistent(), default_datareader_qos_, DDS::RETCODE_INCONSISTENT_POLICY, DDS::RETCODE_OK, and OpenDDS::DCPS::Qos_Helper::valid().

 {
   if (Qos_Helper::valid(qos) && Qos_Helper::consistent(qos)) {
     default_datareader_qos_ = qos;
     return DDS::RETCODE_OK;
 
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ set_listener()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::set_listener	(	DDS::SubscriberListener_ptr	a_listener,
		DDS::StatusMask	mask
	)

virtual

Definition at line 707 of file SubscriberImpl.cpp.

References listener_, listener_mask_, listener_mutex_, and DDS::RETCODE_OK.

 {
   ACE_Guard<ACE_Thread_Mutex> g(listener_mutex_);
   listener_mask_ = mask;
   //note: OK to duplicate  a nil object ref
   listener_ = DDS::SubscriberListener::_duplicate(a_listener);
   return DDS::RETCODE_OK;
 }

◆ set_qos()

DDS::ReturnCode_t OpenDDS::DCPS::SubscriberImpl::set_qos ( const DDS::SubscriberQos & qos )

virtual

Definition at line 620 of file SubscriberImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::Qos_Helper::changeable(), OpenDDS::DCPS::Qos_Helper::consistent(), datareader_map_, OpenDDS::DCPS::DCPS_debug_level, domain_id_, dp_id_, OpenDDS::DCPS::EntityImpl::enabled_, LM_ERROR, OPENDDS_NO_OBJECT_MODEL_PROFILE_COMPATIBILITY_CHECK, qos_, DDS::RETCODE_ERROR, DDS::RETCODE_IMMUTABLE_POLICY, DDS::RETCODE_INCONSISTENT_POLICY, DDS::RETCODE_OK, DDS::RETCODE_UNSUPPORTED, si_lock_, TheServiceParticipant, and OpenDDS::DCPS::Qos_Helper::valid().

 {
 
   OPENDDS_NO_OBJECT_MODEL_PROFILE_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
 
   if (Qos_Helper::valid(qos) && Qos_Helper::consistent(qos)) {
     if (qos_ == qos)
       return DDS::RETCODE_OK;
 
     // for the not changeable qos, it can be changed before enable
     if (!Qos_Helper::changeable(qos_, qos) && enabled_) {
       return DDS::RETCODE_IMMUTABLE_POLICY;
 
     } else {
       qos_ = qos;
 
       DrIdToQosMap idToQosMap;
       {
         ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                          guard,
                          this->si_lock_,
                          DDS::RETCODE_ERROR);
         // after FaceCTS bug 619 is fixed, make endIter and iter const iteratorsx
         DataReaderMap::iterator endIter = datareader_map_.end();
 
         for (DataReaderMap::iterator iter = datareader_map_.begin();
              iter != endIter; ++iter) {
           DataReaderImpl_rch reader = iter->second;
           reader->set_subscriber_qos (qos);
           DDS::DataReaderQos qos = reader->qos_;
           GUID_t id = reader->get_guid();
           std::pair<DrIdToQosMap::iterator, bool> pair
             = idToQosMap.insert(DrIdToQosMap::value_type(id, qos));
 
           if (!pair.second) {
             if (DCPS_debug_level > 0) {
               ACE_ERROR((LM_ERROR,
                         ACE_TEXT("(%P|%t) ERROR: SubscriberImpl::set_qos: ")
                         ACE_TEXT("insert %C to DrIdToQosMap failed.\n"),
                         LogGuid(id).c_str()));
             }
             return ::DDS::RETCODE_ERROR;
           }
         }
       }
 
       DrIdToQosMap::iterator iter = idToQosMap.begin();
 
       while (iter != idToQosMap.end()) {
         Discovery_rch disco = TheServiceParticipant->get_discovery(this->domain_id_);
         const bool status
           = disco->update_subscription_qos(this->domain_id_,
                                            this->dp_id_,
                                            iter->first,
                                            iter->second,
                                            this->qos_);
 
         if (!status) {
           if (DCPS_debug_level > 0) {
             ACE_ERROR((LM_ERROR,
                       ACE_TEXT("(%P|%t) SubscriberImpl::set_qos, ")
                       ACE_TEXT("failed.\n")));
           }
           return DDS::RETCODE_ERROR;
         }
 
         ++iter;
       }
     }
 
     return DDS::RETCODE_OK;
 
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ update_ownership_strength()

void OpenDDS::DCPS::SubscriberImpl::update_ownership_strength	(	const GUID_t &	pub_id,
		const CORBA::Long &	ownership_strength
	)

Definition at line 1017 of file SubscriberImpl.cpp.

References ACE_GUARD_RETURN, datareader_map_, and si_lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    this->si_lock_,
                    );
 
   for (DataReaderMap::iterator iter = datareader_map_.begin();
        iter != datareader_map_.end();
        ++iter) {
     if (!iter->second->is_bit()) {
       iter->second->update_ownership_strength(pub_id, ownership_strength);
     }
   }
 }

◆ validate_datareader_qos()

bool OpenDDS::DCPS::SubscriberImpl::validate_datareader_qos	(	const DDS::DataReaderQos &	qos,
		const DDS::DataReaderQos &	default_qos,
		DDS::Topic_ptr	a_topic,
		DDS::DataReaderQos &	result_qos,
		bool	mt
	)

static

Definition at line 1094 of file SubscriberImpl.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::Qos_Helper::consistent(), OpenDDS::DCPS::Qos_Helper::copy_from_topic_qos(), DATAREADER_QOS_DEFAULT, DATAREADER_QOS_USE_TOPIC_QOS, OpenDDS::DCPS::DCPS_debug_level, LM_ERROR, OPENDDS_END_VERSIONED_NAMESPACE_DECL, OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK, and OpenDDS::DCPS::Qos_Helper::valid().

Referenced by create_datareader(), and OpenDDS::DCPS::DomainParticipantImpl::create_recorder().

 {
 
 
   if (qos == DATAREADER_QOS_DEFAULT) {
     dr_qos = default_qos;
 
   } else if (qos == DATAREADER_QOS_USE_TOPIC_QOS) {
 
 #ifndef OPENDDS_NO_MULTI_TOPIC
     if (mt) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: ")
                    ACE_TEXT("SubscriberImpl::create_datareader, ")
                    ACE_TEXT("DATAREADER_QOS_USE_TOPIC_QOS can not be used ")
                    ACE_TEXT("to create a MultiTopic DataReader.\n")));
       }
       return false;
     }
 #else
     ACE_UNUSED_ARG(mt);
 #endif
     DDS::TopicQos topic_qos;
     a_topic->get_qos(topic_qos);
 
     dr_qos = default_qos;
 
     Qos_Helper::copy_from_topic_qos(dr_qos,
                                     topic_qos);
 
   } else {
     dr_qos = qos;
   }
 
   OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK(dr_qos, false);
   OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK(dr_qos, false);
   OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK(dr_qos, false);
 
   if (!Qos_Helper::valid(dr_qos)) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: ")
                 ACE_TEXT("SubscriberImpl::create_datareader, ")
                 ACE_TEXT("invalid qos.\n")));
     }
     return false;
   }
 
   if (!Qos_Helper::consistent(dr_qos)) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: ")
                 ACE_TEXT("SubscriberImpl::create_datareader, ")
                 ACE_TEXT("inconsistent qos.\n")));
     }
     return false;
   }
 
   return true;
 }

Member Data Documentation

◆ access_depth_

int OpenDDS::DCPS::SubscriberImpl::access_depth_

private

Definition at line 219 of file SubscriberImpl.h.

Referenced by begin_access(), end_access(), and get_datareaders().

◆ datareader_map_

DataReaderMap OpenDDS::DCPS::SubscriberImpl::datareader_map_

private

Definition at line 188 of file SubscriberImpl.h.

Referenced by contains_reader(), delete_contained_entities(), delete_datareader(), get_subscription_ids(), is_clean(), lookup_datareader(), notify_datareaders(), reader_enabled(), set_qos(), and update_ownership_strength().

◆ datareader_set_

DataReaderSet OpenDDS::DCPS::SubscriberImpl::datareader_set_

private

Definition at line 189 of file SubscriberImpl.h.

Referenced by begin_access(), coherent_change_received(), data_received(), delete_datareader(), end_access(), get_datareaders(), and remove_from_datareader_set().

◆ default_datareader_qos_

DDS::DataReaderQos OpenDDS::DCPS::SubscriberImpl::default_datareader_qos_

private

Definition at line 181 of file SubscriberImpl.h.

Referenced by create_datareader(), get_default_datareader_qos(), and set_default_datareader_qos().

◆ domain_id_

DDS::DomainId_t OpenDDS::DCPS::SubscriberImpl::domain_id_

private

Definition at line 198 of file SubscriberImpl.h.

Referenced by delete_datareader(), and set_qos().

◆ dp_id_

GUID_t OpenDDS::DCPS::SubscriberImpl::dp_id_

private

Definition at line 199 of file SubscriberImpl.h.

Referenced by enable(), and set_qos().

◆ dr_set_lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::SubscriberImpl::dr_set_lock_

private

This lock protects datareader_set_. Only this lock needs to be acquired if only datareader_set_ is accessed.

Definition at line 209 of file SubscriberImpl.h.

Referenced by begin_access(), coherent_change_received(), data_received(), delete_datareader(), end_access(), get_datareaders(), and remove_from_datareader_set().

◆ handle_

DDS::InstanceHandle_t OpenDDS::DCPS::SubscriberImpl::handle_

private

Definition at line 178 of file SubscriberImpl.h.

Referenced by get_instance_handle(), and ~SubscriberImpl().

◆ listener_

DDS::SubscriberListener_var OpenDDS::DCPS::SubscriberImpl::listener_

private

Definition at line 185 of file SubscriberImpl.h.

Referenced by get_listener(), listener_for(), set_listener(), and SubscriberImpl().

◆ listener_mask_

DDS::StatusMask OpenDDS::DCPS::SubscriberImpl::listener_mask_

private

Definition at line 184 of file SubscriberImpl.h.

Referenced by listener_for(), and set_listener().

◆ listener_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::SubscriberImpl::listener_mutex_

private

Definition at line 183 of file SubscriberImpl.h.

Referenced by get_listener(), listener_for(), and set_listener().

◆ monitor_

unique_ptr<Monitor> OpenDDS::DCPS::SubscriberImpl::monitor_

private

Monitor object for this entity.

Definition at line 217 of file SubscriberImpl.h.

Referenced by delete_datareader(), enable(), reader_enabled(), and SubscriberImpl().

◆ multitopic_reader_map_

MultitopicReaderMap OpenDDS::DCPS::SubscriberImpl::multitopic_reader_map_

private

Definition at line 193 of file SubscriberImpl.h.

Referenced by delete_contained_entities(), delete_datareader(), lookup_datareader(), multitopic_reader_enabled(), and notify_datareaders().

◆ participant_

WeakRcHandle<DomainParticipantImpl> OpenDDS::DCPS::SubscriberImpl::participant_

private

Definition at line 196 of file SubscriberImpl.h.

Referenced by create_datareader(), enable(), get_participant(), listener_for(), parent(), and ~SubscriberImpl().

◆ qos_

DDS::SubscriberQos OpenDDS::DCPS::SubscriberImpl::qos_

private

Definition at line 180 of file SubscriberImpl.h.

Referenced by begin_access(), create_datareader(), enable(), end_access(), get_datareaders(), get_qos(), and set_qos().

◆ raw_latency_buffer_size_

unsigned int OpenDDS::DCPS::SubscriberImpl::raw_latency_buffer_size_

private

Bound (or initial reservation) of raw latency buffers.

Definition at line 202 of file SubscriberImpl.h.

Referenced by create_datareader(), and raw_latency_buffer_size().

◆ raw_latency_buffer_type_

DataCollector<double>::OnFull OpenDDS::DCPS::SubscriberImpl::raw_latency_buffer_type_

private

Type of raw latency data buffers.

Definition at line 205 of file SubscriberImpl.h.

Referenced by create_datareader(), and raw_latency_buffer_type().

◆ readers_not_enabled_

DataReaderSet OpenDDS::DCPS::SubscriberImpl::readers_not_enabled_

private

Definition at line 187 of file SubscriberImpl.h.

Referenced by create_datareader(), enable(), and reader_enabled().

◆ si_lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::SubscriberImpl::si_lock_

private

General lock protects the data structures in this class. If datareader_set_ is accessed together with other data members, acquire dr_set_lock_ in the scope of this lock.

Definition at line 214 of file SubscriberImpl.h.

Referenced by begin_access(), contains_reader(), create_datareader(), delete_contained_entities(), delete_datareader(), enable(), end_access(), get_subscription_ids(), lookup_datareader(), notify_datareaders(), reader_enabled(), set_qos(), and update_ownership_strength().

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

Public Member Functions

Static Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ SubscriberImpl()

◆ ~SubscriberImpl()

Member Function Documentation

◆ begin_access()

◆ coherent_change_received()

◆ contains_reader()

◆ copy_from_topic_qos()

◆ create_datareader()

◆ data_received()

◆ delete_contained_entities()

◆ delete_datareader()

◆ enable()

◆ end_access()

◆ get_datareaders()

◆ get_default_datareader_qos()

◆ get_instance_handle()

◆ get_listener()

◆ get_participant()

◆ get_qos()

◆ get_subscription_ids()

◆ is_clean()

◆ listener_for()

◆ lookup_datareader()

◆ multitopic_reader_enabled()

◆ notify_datareaders()

◆ OPENDDS_MAP()

◆ OPENDDS_MAP_CMP()

◆ OPENDDS_MULTIMAP()

◆ OPENDDS_SET()

◆ OPENDDS_VECTOR()

◆ parent()

◆ raw_latency_buffer_size()

◆ raw_latency_buffer_type()

◆ reader_enabled()

◆ remove_from_datareader_set()

◆ set_default_datareader_qos()

◆ set_listener()

◆ set_qos()

◆ update_ownership_strength()

◆ validate_datareader_qos()

Member Data Documentation

◆ access_depth_

◆ datareader_map_

◆ datareader_set_

◆ default_datareader_qos_

◆ domain_id_

◆ dp_id_

◆ dr_set_lock_

◆ handle_

◆ listener_

◆ listener_mask_

◆ listener_mutex_

◆ monitor_

◆ multitopic_reader_map_

◆ participant_

◆ qos_

◆ raw_latency_buffer_size_

◆ raw_latency_buffer_type_

◆ readers_not_enabled_

◆ si_lock_