Implements the DDS::DataReader interface. More...

#include <DataReaderImpl.h>

Inheritance diagram for OpenDDS::DCPS::DataReaderImpl:

[legend]

Collaboration diagram for OpenDDS::DCPS::DataReaderImpl:

[legend]

Classes
struct	GenericBundle

class	LivelinessTimer

class	OnDataAvailable

class	OnDataOnReaders

class	OwnershipManagerPtr

struct	OwnershipManagerScopedAccess

Public Types
typedef std::pair< GUID_t, WriterInfo::WriterState >	WriterStatePair

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

typedef DataReaderEx ::_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

Public Types inherited from OpenDDS::DCPS::TransportClient
enum	{ ASSOC_OK = 1, ASSOC_ACTIVE = 2 }

Public Member Functions
typedef	OPENDDS_MAP (DDS::InstanceHandle_t, SubscriptionInstance_rch) SubscriptionInstanceMapType

typedef	OPENDDS_SET (DDS::InstanceHandle_t) InstanceSet

typedef	OPENDDS_SET (SubscriptionInstance_rch) SubscriptionInstanceSet

typedef	OPENDDS_MAP_CMP (GUID_t, WriterStats, GUID_tKeyLessThan) StatsMapType
	Type of collection of statistics for writers to this reader. More...

	DataReaderImpl ()

virtual	~DataReaderImpl ()

virtual DDS::InstanceHandle_t	get_instance_handle ()

virtual void	add_association (const GUID_t &yourId, const WriterAssociation &writer, bool active)

virtual void	transport_assoc_done (int flags, const GUID_t &remote_id)

virtual void	remove_associations (const WriterIdSeq &writers, bool callback)

virtual void	update_incompatible_qos (const IncompatibleQosStatus &status)

virtual void	signal_liveliness (const GUID_t &remote_participant)

DDS::DataReaderListener_ptr	listener_for (DDS::StatusKind kind)

void	writer_became_alive (WriterInfo &info, const MonotonicTimePoint &when)

void	writer_became_dead (WriterInfo &info)

void	writer_removed (WriterInfo &info)

virtual void	cleanup ()

void	init (TopicDescriptionImpl a_topic_desc, const DDS::DataReaderQos &qos, DDS::DataReaderListener_ptr a_listener, const DDS::StatusMask &mask, DomainParticipantImpl participant, SubscriberImpl *subscriber)

virtual DDS::ReadCondition_ptr	create_readcondition (DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)

virtual DDS::QueryCondition_ptr	create_querycondition (DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, const char *query_expression, const DDS::StringSeq &query_parameters)

virtual DDS::ReturnCode_t	delete_readcondition (DDS::ReadCondition_ptr a_condition)

virtual DDS::ReturnCode_t	delete_contained_entities ()

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

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

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

virtual DDS::DataReaderListener_ptr	get_listener ()

virtual DDS::TopicDescription_ptr	get_topicdescription ()

virtual DDS::Subscriber_ptr	get_subscriber ()

virtual DDS::ReturnCode_t	get_sample_rejected_status (DDS::SampleRejectedStatus &status)

virtual DDS::ReturnCode_t	get_liveliness_changed_status (DDS::LivelinessChangedStatus &status)

virtual DDS::ReturnCode_t	get_requested_deadline_missed_status (DDS::RequestedDeadlineMissedStatus &status)

virtual DDS::ReturnCode_t	get_requested_incompatible_qos_status (DDS::RequestedIncompatibleQosStatus &status)

virtual DDS::ReturnCode_t	get_subscription_matched_status (DDS::SubscriptionMatchedStatus &status)

virtual DDS::ReturnCode_t	get_sample_lost_status (DDS::SampleLostStatus &status)

virtual DDS::ReturnCode_t	wait_for_historical_data (const DDS::Duration_t &max_wait)

virtual DDS::ReturnCode_t	get_matched_publications (DDS::InstanceHandleSeq &publication_handles)

virtual DDS::ReturnCode_t	get_matched_publication_data (DDS::PublicationBuiltinTopicData &publication_data, DDS::InstanceHandle_t publication_handle)

virtual DDS::ReturnCode_t	enable ()

virtual void	get_latency_stats (LatencyStatisticsSeq &stats)

virtual void	reset_latency_stats ()
	Clear any intermediate statistical values. More...

virtual CORBA::Boolean	statistics_enabled ()

virtual void	statistics_enabled (CORBA::Boolean statistics_enabled)

void	writer_activity (const DataSampleHeader &header)
	update liveliness info for this writer. More...

virtual void	data_received (const ReceivedDataSample &sample)
	process a message that has been received - could be control or a data sample. More...

void	transport_discovery_change ()

virtual bool	check_transport_qos (const TransportInst &inst)

bool	have_sample_states (DDS::SampleStateMask sample_states) const

bool	have_view_states (DDS::ViewStateMask view_states) const

bool	have_instance_states (DDS::InstanceStateMask instance_states) const

bool	contains_sample (DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)

virtual bool	contains_sample_filtered (DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, const FilterEvaluator &evaluator, const DDS::StringSeq &params)=0

virtual RcHandle< MessageHolder >	dds_demarshal (const ReceivedDataSample &sample, DDS::InstanceHandle_t publication_handle, SubscriptionInstance_rch &instance, bool &is_new_instance, bool &filtered, MarshalingType marshaling_type, bool full_copy)=0

virtual void	dispose_unregister (const ReceivedDataSample &sample, DDS::InstanceHandle_t publication_handle, SubscriptionInstance_rch &instance)

void	process_latency (const ReceivedDataSample &sample)

void	notify_latency (GUID_t writer)

size_t	get_depth () const

size_t	get_n_chunks () const

void	liveliness_lost ()

void	remove_all_associations ()

void	notify_subscription_disconnected (const WriterIdSeq &pubids)

void	notify_subscription_reconnected (const WriterIdSeq &pubids)

void	notify_subscription_lost (const WriterIdSeq &pubids)

void	notify_liveliness_change ()

bool	is_bit () const

bool	has_zero_copies ()

void	release_instance (DDS::InstanceHandle_t handle)
	Release the instance with the handle. More...

void	state_updated (DDS::InstanceHandle_t handle)

virtual void	release_all_instances ()=0
	Release all instances held by the reader. More...

ACE_Reactor_Timer_Interface *	get_reactor ()

GUID_t	get_topic_id ()

GUID_t	get_dp_id ()

typedef	OPENDDS_VECTOR (DDS::InstanceHandle_t) InstanceHandleVec

void	get_instance_handles (InstanceHandleVec &instance_handles)

typedef	OPENDDS_VECTOR (WriterStatePair) WriterStatePairVec

void	get_writer_states (WriterStatePairVec &writer_states)

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

OwnershipManagerPtr	ownership_manager ()

virtual void	lookup_instance (const ReceivedDataSample &sample, SubscriptionInstance_rch &instance)=0

void	enable_filtering (ContentFilteredTopicImpl *cft)

DDS::ContentFilteredTopic_ptr	get_cf_topic () const

void	enable_multi_topic (MultiTopicImpl *mt)

void	update_subscription_params (const DDS::StringSeq &params) const

typedef	OPENDDS_VECTOR (void *) GenericSeq

virtual DDS::ReturnCode_t	read_generic (GenericBundle &gen, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, bool adjust_ref_count)=0

virtual DDS::ReturnCode_t	take (AbstractSamples &samples, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)=0

virtual DDS::InstanceHandle_t	lookup_instance_generic (const void *data)=0

virtual DDS::ReturnCode_t	read_instance_generic (void *&data, DDS::SampleInfo &info, DDS::InstanceHandle_t instance, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)=0

virtual DDS::ReturnCode_t	read_next_instance_generic (void *&data, DDS::SampleInfo &info, DDS::InstanceHandle_t previous_instance, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)=0

void	set_instance_state (DDS::InstanceHandle_t instance, DDS::InstanceStateKind state, const SystemTimePoint &timestamp=SystemTimePoint::now(), const GUID_t &guid=GUID_UNKNOWN)

void	begin_access ()

void	end_access ()

void	get_ordered_data (GroupRakeData &data, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)

void	accept_coherent (const GUID_t &writer_id, const GUID_t &publisher_id)

void	reject_coherent (const GUID_t &writer_id, const GUID_t &publisher_id)

void	coherent_change_received (const GUID_t &publisher_id, Coherent_State &result)

void	coherent_changes_completed (DataReaderImpl *reader)

void	reset_coherent_info (const GUID_t &writer_id, const GUID_t &publisher_id)

void	set_subscriber_qos (const DDS::SubscriberQos &qos)

void	reset_ownership (DDS::InstanceHandle_t instance)

virtual RcHandle< EntityImpl >	parent () const

void	disable_transport ()

virtual void	register_for_writer (const GUID_t &, const GUID_t &, const GUID_t &, const TransportLocatorSeq &, DiscoveryListener *)

virtual void	unregister_for_writer (const GUID_t &, const GUID_t &, const GUID_t &)

virtual void	update_locators (const GUID_t &remote, const TransportLocatorSeq &locators)

virtual DCPS::WeakRcHandle< ICE::Endpoint >	get_ice_endpoint ()

GUID_t	get_guid () const

void	return_handle (DDS::InstanceHandle_t handle)

const ValueDispatcher *	get_value_dispatcher () const

Raw Latency Statistics Interfaces
const StatsMapType &	raw_latency_statistics () const
	Expose the statistics container. More...

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 OpenDDS::DCPS::DataReaderEx
void	get_latency_stats (inout LatencyStatisticsSeq stats)
	Obtain a sequence of statistics summaries. More...

Public Member Functions inherited from DDS::DataReader
ReadCondition	create_readcondition (in SampleStateMask sample_states, in ViewStateMask view_states, in InstanceStateMask instance_states)

QueryCondition	create_querycondition (in SampleStateMask sample_states, in ViewStateMask view_states, in InstanceStateMask instance_states, in string query_expression, in StringSeq query_parameters)

ReturnCode_t	delete_readcondition (in ReadCondition a_condition)

ReturnCode_t	set_qos (in DataReaderQos qos)

ReturnCode_t	get_qos (inout DataReaderQos qos)

ReturnCode_t	set_listener (in DataReaderListener a_listener, in StatusMask mask)

ReturnCode_t	get_sample_rejected_status (inout SampleRejectedStatus status)

ReturnCode_t	get_liveliness_changed_status (inout LivelinessChangedStatus status)

ReturnCode_t	get_requested_deadline_missed_status (inout RequestedDeadlineMissedStatus status)

ReturnCode_t	get_requested_incompatible_qos_status (inout RequestedIncompatibleQosStatus status)

ReturnCode_t	get_subscription_matched_status (inout SubscriptionMatchedStatus status)

ReturnCode_t	get_sample_lost_status (inout SampleLostStatus status)

ReturnCode_t	wait_for_historical_data (in Duration_t max_wait)

ReturnCode_t	get_matched_publications (inout InstanceHandleSeq publication_handles)

ReturnCode_t	get_matched_publication_data (inout PublicationBuiltinTopicData publication_data, in InstanceHandle_t publication_handle)

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::DataReaderCallbacks
	DataReaderCallbacks ()

virtual	~DataReaderCallbacks ()

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)

Public Member Functions inherited from OpenDDS::DCPS::TransportClient
void	use_datalink (const GUID_t &remote_id, const DataLink_rch &link)

	TransportClient ()

virtual	~TransportClient ()

void	enable_transport (bool reliable, bool durable)

void	enable_transport_using_config (bool reliable, bool durable, const TransportConfig_rch &tc)

bool	swap_bytes () const

bool	cdr_encapsulation () const

const TransportLocatorSeq &	connection_info () const

void	populate_connection_info ()

bool	is_reliable () const

bool	associate (const AssociationData &peer, bool active)

void	disassociate (const GUID_t &peerId)

void	stop_associating ()

void	stop_associating (const GUID_t *repos, CORBA::ULong length)

void	send_final_acks ()

void	transport_stop ()

void	register_for_reader (const GUID_t &participant, const GUID_t &writerid, const GUID_t &readerid, const TransportLocatorSeq &locators, OpenDDS::DCPS::DiscoveryListener *listener)

void	unregister_for_reader (const GUID_t &participant, const GUID_t &writerid, const GUID_t &readerid)

void	register_for_writer (const GUID_t &participant, const GUID_t &readerid, const GUID_t &writerid, const TransportLocatorSeq &locators, DiscoveryListener *listener)

void	unregister_for_writer (const GUID_t &participant, const GUID_t &readerid, const GUID_t &writerid)

void	update_locators (const GUID_t &remote, const TransportLocatorSeq &locators)

WeakRcHandle< ICE::Endpoint >	get_ice_endpoint ()

bool	send_response (const GUID_t &peer, const DataSampleHeader &header, Message_Block_Ptr payload)

void	send (SendStateDataSampleList send_list, ACE_UINT64 transaction_id=0)

SendControlStatus	send_w_control (SendStateDataSampleList send_list, const DataSampleHeader &header, Message_Block_Ptr msg, const GUID_t &destination)

SendControlStatus	send_control (const DataSampleHeader &header, Message_Block_Ptr msg)

SendControlStatus	send_control_to (const DataSampleHeader &header, Message_Block_Ptr msg, const GUID_t &destination)

bool	remove_sample (const DataSampleElement *sample)

bool	remove_all_msgs ()

void	terminate_send_if_suspended ()

bool	associated_with (const GUID_t &remote) const

bool	pending_association_with (const GUID_t &remote) const

GUID_t	repo_id () const

void	data_acked (const GUID_t &remote)

bool	is_leading (const GUID_t &reader_id) const

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

Protected Types
typedef ACE_Reverse_Lock< ACE_Recursive_Thread_Mutex >	Reverse_Lock_t

Protected Member Functions
typedef	OPENDDS_SET (DDS::InstanceHandle_t) HandleSet

typedef	OPENDDS_MAP (CORBA::ULong, HandleSet) LookupMap

void	initialize_lookup_maps ()

void	update_lookup_maps (const SubscriptionInstanceMapType::iterator &input)

void	remove_from_lookup_maps (DDS::InstanceHandle_t handle)

const HandleSet &	lookup_matching_instances (CORBA::ULong sample_states, CORBA::ULong view_states, CORBA::ULong instance_states) const

DataReaderListener_ptr	get_ext_listener ()

virtual void	remove_associations_i (const WriterIdSeq &writers, bool callback)

void	prepare_to_delete ()

DDS::ReturnCode_t	setup_deserialization ()
	Setup deserialization options. More...

RcHandle< SubscriberImpl >	get_subscriber_servant ()

void	post_read_or_take ()

virtual DDS::ReturnCode_t	enable_specific ()=0

void	sample_info (DDS::SampleInfo &sample_info, const ReceivedDataElement *ptr)

CORBA::Long	total_samples () const

void	set_sample_lost_status (const DDS::SampleLostStatus &status)

void	set_sample_rejected_status (const DDS::SampleRejectedStatus &status)

SubscriptionInstance_rch	get_handle_instance (DDS::InstanceHandle_t handle)

DDS::InstanceHandle_t	get_next_handle (const DDS::BuiltinTopicKey_t &key)

virtual void	purge_data (SubscriptionInstance_rch instance)=0

virtual void	release_instance_i (DDS::InstanceHandle_t handle)=0

virtual void	state_updated_i (DDS::InstanceHandle_t handle)=0

bool	has_readcondition (DDS::ReadCondition_ptr a_condition)

bool	filter_sample (const DataSampleHeader &header)

bool	ownership_filter_instance (const SubscriptionInstance_rch &instance, const GUID_t &pubid)

bool	time_based_filter_instance (const SubscriptionInstance_rch &instance, MonotonicTimePoint &now, MonotonicTimePoint &deadline)

void	accept_sample_processing (const SubscriptionInstance_rch &instance, const DataSampleHeader &header, bool is_new_instance)

virtual void	qos_change (const DDS::DataReaderQos &qos)

void	notify_read_conditions ()
	Data has arrived into the cache, unblock waiting ReadConditions. More...

virtual void	add_link (const DataLink_rch &link, const GUID_t &peer)

typedef	OPENDDS_SET (Encoding::Kind) EncodingKinds

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 Member Functions inherited from OpenDDS::DCPS::TransportClient
void	cdr_encapsulation (bool encap)

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

Static Protected Member Functions
static CORBA::ULong	to_combined_states (CORBA::ULong sample_states, CORBA::ULong view_states, CORBA::ULong instance_states)

static void	split_combined_states (CORBA::ULong combined, CORBA::ULong &sample_states, CORBA::ULong &view_states, CORBA::ULong &instance_states)

Protected Attributes
LookupMap	combined_state_lookup_

SubscriptionInstanceMapType	instances_
	: document why the instances_ container is mutable. More...

ACE_Recursive_Thread_Mutex	instances_lock_

bool	has_subscription_id_

ACE_Thread_Mutex	subscription_id_mutex_

ConditionVariable< ACE_Thread_Mutex >	subscription_id_condition_

unique_ptr< ReceivedDataAllocator >	rd_allocator_

DDS::DataReaderQos	qos_

DDS::DataReaderQos	passed_qos_

DDS::SampleRejectedStatus	sample_rejected_status_

DDS::SampleLostStatus	sample_lost_status_

ACE_Recursive_Thread_Mutex	sample_lock_
	lock protecting sample container as well as statuses. More...

Reverse_Lock_t	reverse_sample_lock_

WeakRcHandle< DomainParticipantImpl >	participant_servant_

TopicDescriptionPtr< TopicImpl >	topic_servant_

TypeSupportImpl *	type_support_

GUID_t	topic_id_

bool	is_exclusive_ownership_

ACE_Thread_Mutex	content_filtered_topic_mutex_

TopicDescriptionPtr< ContentFilteredTopicImpl >	content_filtered_topic_

TopicDescriptionPtr< MultiTopicImpl >	multi_topic_

bool	coherent_
	Is accessing to Group coherent changes ? More...

GroupRakeData	group_coherent_ordered_data_
	Ordered group samples. More...

DDS::SubscriberQos	subqos_

EncodingKinds	decoding_modes_

Security::SecurityConfig_rch	security_config_

DDS::DynamicType_var	dynamic_type_

TransportMessageBlockAllocator	mb_alloc_

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

Static Protected Attributes
static const CORBA::ULong	MAX_SAMPLE_STATE_FLAG = DDS::NOT_READ_SAMPLE_STATE

static const CORBA::ULong	MAX_SAMPLE_STATE_MASK = (MAX_SAMPLE_STATE_FLAG << 1) - 1

static const CORBA::ULong	MAX_SAMPLE_STATE_BITS = 2u

static const CORBA::ULong	MAX_VIEW_STATE_FLAG = DDS::NOT_NEW_VIEW_STATE

static const CORBA::ULong	MAX_VIEW_STATE_MASK = (MAX_VIEW_STATE_FLAG << 1) - 1

static const CORBA::ULong	MAX_VIEW_STATE_BITS = 2u

static const CORBA::ULong	MAX_INSTANCE_STATE_FLAG = DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE

static const CORBA::ULong	MAX_INSTANCE_STATE_MASK = (MAX_INSTANCE_STATE_FLAG << 1) - 1

static const CORBA::ULong	MAX_INSTANCE_STATE_BITS = 3u

static const CORBA::ULong	COMBINED_VIEW_STATE_SHIFT = MAX_INSTANCE_STATE_BITS

static const CORBA::ULong	COMBINED_SAMPLE_STATE_SHIFT = COMBINED_VIEW_STATE_SHIFT + MAX_VIEW_STATE_BITS

Private Types
typedef PmfSporadicTask< DataReaderImpl >	DRISporadicTask

typedef VarLess< DDS::ReadCondition >	RCCompLess

Private Member Functions
virtual void	install_type_support (TypeSupportImpl *)

virtual void	set_instance_state_i (DDS::InstanceHandle_t instance, DDS::InstanceHandle_t publication_handle, DDS::InstanceStateKind state, const SystemTimePoint &timestamp, const GUID_t &guid)=0

void	notify_subscription_lost (const DDS::InstanceHandleSeq &handles)

void	lookup_instance_handles (const WriterIdSeq &ids, DDS::InstanceHandleSeq &hdls)
	Lookup the instance handles by the publication repo ids. More...

void	instances_liveliness_update (const GUID_t &writer, DDS::InstanceHandle_t publication_handle)

bool	verify_coherent_changes_completion (WriterInfo *writer)

bool	coherent_change_received (WriterInfo *writer)

RcHandle< BitSubscriber >	get_builtin_subscriber_proxy () const

DDS::DomainId_t	domain_id () const

Priority	get_priority_value (const AssociationData &data) const

DDS::Security::ParticipantCryptoHandle	get_crypto_handle () const

void	resume_sample_processing (const GUID_t &pub_id)
	when done handling historic samples, resume More...

bool	check_historic (const ReceivedDataSample &sample)

void	deliver_historic (OPENDDS_MAP(SequenceNumber, ReceivedDataSample)&samples)
	deliver samples that were held by check_historic() More...

typedef	OPENDDS_MAP_CMP (GUID_t, DDS::InstanceHandle_t, GUID_tKeyLessThan) RepoIdToHandleMap

typedef	OPENDDS_MULTIMAP (MonotonicTimePoint, SubscriptionInstance_rch) DeadlineQueue

void	schedule_deadline (SubscriptionInstance_rch instance, bool timer_called)

void	reset_deadline_period (const TimeDuration &deadline_period)

void	reschedule_deadline (SubscriptionInstance_rch instance, const MonotonicTimePoint &now)

void	cancel_deadline (SubscriptionInstance_rch instance)

void	cancel_all_deadlines ()

void	deadline_task (const MonotonicTimePoint &now)

void	process_deadline (SubscriptionInstance_rch instance, const MonotonicTimePoint &now, bool timer_called)

typedef	OPENDDS_MAP_CMP (GUID_t, WriterInfo_rch, GUID_tKeyLessThan) WriterMapType
	publications writing to this reader. More...

typedef	OPENDDS_SET_CMP (DDS::ReadCondition_var, RCCompLess) ReadConditionSet

Private Member Functions inherited from OpenDDS::DCPS::WriterInfoListener
	WriterInfoListener ()

virtual	~WriterInfoListener ()

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

long	ref_count () const

WeakObject *	_get_weak_object () const

	RcObject ()

Private Attributes
DDS::TopicDescription_var	topic_desc_

ACE_Thread_Mutex	listener_mutex_

DDS::StatusMask	listener_mask_

DDS::DataReaderListener_var	listener_

DDS::DomainId_t	domain_id_

GUID_t	dp_id_

WeakRcHandle< SubscriberImpl >	subscriber_servant_

RcHandle< EndHistoricSamplesMissedSweeper >	end_historic_sweeper_

CORBA::Long	depth_

size_t	n_chunks_

ACE_Recursive_Thread_Mutex	publication_handle_lock_

RepoIdToHandleMap	publication_id_to_handle_map_

DDS::LivelinessChangedStatus	liveliness_changed_status_

DDS::RequestedDeadlineMissedStatus	requested_deadline_missed_status_

DDS::RequestedIncompatibleQosStatus	requested_incompatible_qos_status_

DDS::SubscriptionMatchedStatus	subscription_match_status_

BudgetExceededStatus	budget_exceeded_status_

SubscriptionLostStatus	subscription_lost_status_

ACE_Reactor_Timer_Interface *	reactor_

RcHandle< LivelinessTimer >	liveliness_timer_

CORBA::Long	last_deadline_missed_total_count_

TimeDuration	deadline_period_

DeadlineQueue	deadline_queue_

bool	deadline_queue_enabled_

RcHandle< DRISporadicTask >	deadline_task_

bool	is_bit_

bool	always_get_history_

AtomicBool	statistics_enabled_
	Flag indicating status of statistics gathering. More...

WriterMapType	writers_

ACE_RW_Thread_Mutex	writers_lock_
	RW lock for reading/writing publications. More...

StatsMapType	statistics_
	Statistics for this reader, collected for each writer. More...

ACE_Recursive_Thread_Mutex	statistics_lock_

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

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

ReadConditionSet	read_conditions_

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

unique_ptr< Monitor >	periodic_monitor_
	Periodic Monitor object for this entity. More...

bool	transport_disabled_

Private Attributes inherited from OpenDDS::DCPS::WriterInfoListener
GUID_t	subscription_id_

TimeDuration	liveliness_lease_duration_

Friends
class	RequestedDeadlineWatchdog

class	QueryConditionImpl

class	SubscriberImpl

class	OwnershipManagerPtr

class	InstanceState

class	EndHistoricSamplesMissedSweeper

class	::DDS_TEST

Additional Inherited Members
Static Public Member Functions inherited from OpenDDS::DCPS::LocalObject< DataReaderEx >
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)

Public Attributes inherited from OpenDDS::DCPS::DataReaderEx
attribute boolean	statistics_enabled
	Statistics gathering enable state. More...

Detailed Description

Implements the DDS::DataReader interface.

See the DDS specification, OMG formal/2015-04-10, for a description of the interface this class is implementing.

This class must be inherited by the type-specific datareader which is specific to the data-type associated with the topic.

Definition at line 207 of file DataReaderImpl.h.

Member Typedef Documentation

◆ DRISporadicTask

typedef PmfSporadicTask<DataReaderImpl> OpenDDS::DCPS::DataReaderImpl::DRISporadicTask

private

Definition at line 994 of file DataReaderImpl.h.

◆ RCCompLess

typedef VarLess<DDS::ReadCondition> OpenDDS::DCPS::DataReaderImpl::RCCompLess

private

Definition at line 1037 of file DataReaderImpl.h.

◆ Reverse_Lock_t

typedef ACE_Reverse_Lock<ACE_Recursive_Thread_Mutex> OpenDDS::DCPS::DataReaderImpl::Reverse_Lock_t

protected

Definition at line 782 of file DataReaderImpl.h.

◆ WriterStatePair

typedef std::pair<GUID_t, WriterInfo::WriterState> OpenDDS::DCPS::DataReaderImpl::WriterStatePair

Definition at line 455 of file DataReaderImpl.h.

Constructor & Destructor Documentation

◆ DataReaderImpl()

OpenDDS::DCPS::DataReaderImpl::DataReaderImpl ( )

Definition at line 60 of file DataReaderImpl.cpp.

References DDS::LivelinessChangedStatus::alive_count, DDS::LivelinessChangedStatus::alive_count_change, budget_exceeded_status_, DDS::SubscriptionMatchedStatus::current_count, DDS::SubscriptionMatchedStatus::current_count_change, DDS::HANDLE_NIL, OpenDDS::DCPS::BudgetExceededStatus::last_instance_handle, DDS::SampleRejectedStatus::last_instance_handle, DDS::RequestedDeadlineMissedStatus::last_instance_handle, DDS::RequestedIncompatibleQosStatus::last_policy_id, DDS::LivelinessChangedStatus::last_publication_handle, DDS::SubscriptionMatchedStatus::last_publication_handle, DDS::SampleRejectedStatus::last_reason, liveliness_changed_status_, monitor_, DDS::LivelinessChangedStatus::not_alive_count, DDS::LivelinessChangedStatus::not_alive_count_change, DDS::NOT_REJECTED, periodic_monitor_, DDS::RequestedIncompatibleQosStatus::policies, reactor_, requested_deadline_missed_status_, requested_incompatible_qos_status_, sample_lost_status_, sample_rejected_status_, subscription_match_status_, TheServiceParticipant, OpenDDS::DCPS::BudgetExceededStatus::total_count, DDS::SampleLostStatus::total_count, DDS::SampleRejectedStatus::total_count, DDS::RequestedDeadlineMissedStatus::total_count, DDS::RequestedIncompatibleQosStatus::total_count, DDS::SubscriptionMatchedStatus::total_count, OpenDDS::DCPS::BudgetExceededStatus::total_count_change, DDS::SampleLostStatus::total_count_change, DDS::SampleRejectedStatus::total_count_change, DDS::RequestedDeadlineMissedStatus::total_count_change, DDS::RequestedIncompatibleQosStatus::total_count_change, and DDS::SubscriptionMatchedStatus::total_count_change.

   : has_subscription_id_(false)
   , subscription_id_mutex_()
   , subscription_id_condition_(subscription_id_mutex_)
   , qos_(TheServiceParticipant->initial_DataReaderQos())
   , reverse_sample_lock_(sample_lock_)
   , topic_servant_(0)
   , type_support_(0)
   , topic_id_(GUID_UNKNOWN)
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   , is_exclusive_ownership_(false)
 #endif
   , coherent_(false)
   , subqos_(TheServiceParticipant->initial_SubscriberQos())
   , topic_desc_(0)
   , listener_mask_(DEFAULT_STATUS_MASK)
   , domain_id_(0)
   , end_historic_sweeper_(make_rch<EndHistoricSamplesMissedSweeper>(TheServiceParticipant->reactor(), TheServiceParticipant->reactor_owner(), this))
   , n_chunks_(TheServiceParticipant->n_chunks())
   , reactor_(0)
   , liveliness_timer_(make_rch<LivelinessTimer>(TheServiceParticipant->reactor(), TheServiceParticipant->reactor_owner(), this))
   , last_deadline_missed_total_count_(0)
   , deadline_queue_enabled_(false)
   , deadline_task_(make_rch<DRISporadicTask>(TheServiceParticipant->time_source(), TheServiceParticipant->interceptor(), rchandle_from(this), &DataReaderImpl::deadline_task))
   , is_bit_(false)
   , always_get_history_(false)
   , statistics_enabled_(false)
   , raw_latency_buffer_size_(0)
   , raw_latency_buffer_type_(DataCollector<double>::KeepOldest)
   , transport_disabled_(false)
   , mb_alloc_(DEFAULT_TRANSPORT_RECEIVE_BUFFERS)
 {
   reactor_ = TheServiceParticipant->timer();
 
   liveliness_changed_status_.alive_count = 0;
   liveliness_changed_status_.not_alive_count = 0;
   liveliness_changed_status_.alive_count_change = 0;
   liveliness_changed_status_.not_alive_count_change = 0;
   liveliness_changed_status_.last_publication_handle =
       DDS::HANDLE_NIL;
 
   requested_deadline_missed_status_.total_count = 0;
   requested_deadline_missed_status_.total_count_change = 0;
   requested_deadline_missed_status_.last_instance_handle =
       DDS::HANDLE_NIL;
 
   requested_incompatible_qos_status_.total_count = 0;
   requested_incompatible_qos_status_.total_count_change = 0;
   requested_incompatible_qos_status_.last_policy_id = 0;
   requested_incompatible_qos_status_.policies.length(0);
 
   subscription_match_status_.total_count = 0;
   subscription_match_status_.total_count_change = 0;
   subscription_match_status_.current_count = 0;
   subscription_match_status_.current_count_change = 0;
   subscription_match_status_.last_publication_handle =
       DDS::HANDLE_NIL;
 
   sample_lost_status_.total_count = 0;
   sample_lost_status_.total_count_change = 0;
 
   sample_rejected_status_.total_count = 0;
   sample_rejected_status_.total_count_change = 0;
   sample_rejected_status_.last_reason = DDS::NOT_REJECTED;
   sample_rejected_status_.last_instance_handle = DDS::HANDLE_NIL;
 
   this->budget_exceeded_status_.total_count = 0;
   this->budget_exceeded_status_.total_count_change = 0;
   this->budget_exceeded_status_.last_instance_handle = DDS::HANDLE_NIL;
 
   monitor_.reset(TheServiceParticipant->monitor_factory_->create_data_reader_monitor(this));
   periodic_monitor_.reset(TheServiceParticipant->monitor_factory_->create_data_reader_periodic_monitor(this));
 }

◆ ~DataReaderImpl()

OpenDDS::DCPS::DataReaderImpl::~DataReaderImpl ( )

virtual

Definition at line 136 of file DataReaderImpl.cpp.

References DBG_ENTRY_LVL, deadline_task_, participant_servant_, and OpenDDS::DCPS::WriterInfoListener::subscription_id_.

 {
   DBG_ENTRY_LVL("DataReaderImpl", "~DataReaderImpl", 6);
 
   deadline_task_->cancel();
 
 #ifndef OPENDDS_SAFETY_PROFILE
   RcHandle<DomainParticipantImpl> participant = participant_servant_.lock();
   if (participant) {
     XTypes::TypeLookupService_rch type_lookup_service = participant->get_type_lookup_service();
     if (type_lookup_service) {
       type_lookup_service->remove_guid_from_dynamic_map(subscription_id_);
     }
   }
 #endif
 }

Member Function Documentation

◆ accept_coherent()

void OpenDDS::DCPS::DataReaderImpl::accept_coherent	(	const GUID_t &	writer_id,
		const GUID_t &	publisher_id
	)

Definition at line 2914 of file DataReaderImpl.cpp.

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

 {
   if (::OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) DataReaderImpl::accept_coherent()")
         ACE_TEXT(" reader %C writer %C publisher %C\n"),
         LogGuid(get_guid()).c_str(),
         LogGuid(writer_id).c_str(),
         LogGuid(publisher_id).c_str()));
   }
   SubscriptionInstanceSet localsubs;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, this->instances_lock_);
     for (SubscriptionInstanceMapType::iterator iter = this->instances_.begin();
          iter != this->instances_.end(); ++iter) {
       localsubs.insert(iter->second);
     }
   }
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
   for (SubscriptionInstanceSet::iterator iter = localsubs.begin();
        iter != localsubs.end(); iter++) {
     (*iter)->rcvd_strategy_->accept_coherent(writer_id, publisher_id);
   }
 }

◆ accept_sample_processing()

void OpenDDS::DCPS::DataReaderImpl::accept_sample_processing	(	const SubscriptionInstance_rch &	instance,
		const DataSampleHeader &	header,
		bool	is_new_instance
	)

protected

Definition at line 3345 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_READ_GUARD, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::DataSampleHeader::coherent_change_, OpenDDS::DCPS::SubscriptionInstance::cur_sample_tv_, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::EntityId_t::entityKind, OpenDDS::DCPS::ENTITYKIND_OPENDDS_NIL_WRITER, OpenDDS::DCPS::DataSampleHeader::group_coherent_, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::SubscriptionInstance::last_sample_tv_, LM_WARNING, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::DataSampleHeader::publisher_id_, and OpenDDS::DCPS::TimePoint_T< AceClock >::set_to_now().

Referenced by data_received().

 {
   bool accepted = true;
 #ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
   bool verify_coherent = false;
 #endif
   WriterInfo_rch writer;
 
   if (header.publication_id_.entityId.entityKind != ENTITYKIND_OPENDDS_NIL_WRITER) {
     ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, writers_lock_);
 
     WriterMapType::iterator where = writers_.find(header.publication_id_);
 
     if (where != writers_.end()) {
       if (header.coherent_change_) {
 
 #ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
         // Received coherent change
         where->second->coherent_change(header.group_coherent_, header.publisher_id_);
         verify_coherent = true;
 #endif
         writer = where->second;
       }
     }
     else {
       ACE_DEBUG((LM_WARNING,
         ACE_TEXT("(%P|%t) WARNING: DataReaderImpl::accept_sample_processing - ")
         ACE_TEXT("subscription %C failed to find ")
         ACE_TEXT("publication data for %C.\n"),
         LogGuid(get_guid()).c_str(),
         LogGuid(header.publication_id_).c_str()));
     }
   }
 
 #ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
   if (verify_coherent) {
     accepted = verify_coherent_changes_completion(writer.in());
   }
 #endif
 
   if (instance && deadline_queue_enabled_) {
     instance->last_sample_tv_ = instance->cur_sample_tv_;
     instance->cur_sample_tv_.set_to_now();
 
     if (is_new_instance) {
       schedule_deadline(instance, false);
     } else {
       process_deadline(instance, MonotonicTimePoint::now(), false);
     }
   }
 
   if (accepted) {
     notify_read_conditions();
   }
 }

◆ add_association()

void OpenDDS::DCPS::DataReaderImpl::add_association	(	const GUID_t &	yourId,
		const WriterAssociation &	writer,
		bool	active
	)

virtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 237 of file DataReaderImpl.cpp.

 {
   if (DCPS_debug_level) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) DataReaderImpl::add_association - ")
         ACE_TEXT("bit %d local %C remote %C\n"), is_bit_,
         LogGuid(yourId).c_str(),
         LogGuid(writer.writerId).c_str()));
   }
 
   if (get_deleted()) {
     if (DCPS_debug_level) {
       ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) DataReaderImpl::add_association")
           ACE_TEXT(" This is a deleted datareader, ignoring add.\n")));
     }
     return;
   }
 
   // We are being called back from the repository before we are done
   // processing after our call to the repository that caused this call
   // (from the repository) to be made.
   {
     ACE_Guard<ACE_Thread_Mutex> guard(subscription_id_mutex_);
     if (GUID_UNKNOWN == subscription_id_) {
       subscription_id_ = yourId;
       has_subscription_id_ = true;
       subscription_id_condition_.notify_all();
     }
   }
 
   // For each writer in the list of writers to associate with, we
   // create a WriterInfo and a WriterStats object and store them in
   // our internal maps.
   //
   {
 
     ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, writers_lock_);
 
     const GUID_t& writer_id = writer.writerId;
     WriterInfo_rch info = make_rch<WriterInfo>(rchandle_from<WriterInfoListener>(this), writer_id, writer.writerQos);
     std::pair<WriterMapType::iterator, bool> bpair = writers_.insert(
         // This insertion is idempotent.
         WriterMapType::value_type(
           writer_id,
           info));
 
     // Schedule timer if necessary
     //   - only need to check reader qos - we know the writer must be >= reader
     if (this->qos_.durability.kind > DDS::VOLATILE_DURABILITY_QOS) {
       info->waiting_for_end_historic_samples(true);
     }
 
     {
       ACE_Guard<ACE_Recursive_Thread_Mutex> guard(statistics_lock_);
       statistics_.insert(
         StatsMapType::value_type(
           writer_id,
           WriterStats(raw_latency_buffer_size_, raw_latency_buffer_type_)));
     }
 
     // If this is a durable reader
     if (this->qos_.durability.kind > DDS::VOLATILE_DURABILITY_QOS) {
       // TODO schedule timer for removing flag from writers
     }
 
     if (DCPS_debug_level > 4) {
       ACE_DEBUG((LM_DEBUG,
           "(%P|%t) DataReaderImpl::add_association: "
           "inserted writer %C.return %d\n",
           LogGuid(writer_id).c_str(), bpair.second));
 
       WriterMapType::iterator iter = writers_.find(writer_id);
       if (iter != writers_.end()) {
         // This may not be an error since it could happen that the sample
         // is delivered to the datareader after the write is dis-associated
         // with this datareader.
         ACE_DEBUG((LM_DEBUG,
             ACE_TEXT("(%P|%t) DataReaderImpl::add_association: ")
             ACE_TEXT("reader %C is associated with writer %C.\n"),
             LogGuid(get_guid()).c_str(),
             LogGuid(writer_id).c_str()));
       }
     }
   }
 
   // Propagate the add_associations processing down into the Transport
   // layer here.  This will establish the transport support and reserve
   // usage of an existing connection or initiate creation of a new
   // connection if no suitable connection is available.
   AssociationData data;
   data.remote_id_ = writer.writerId;
   data.remote_data_ = writer.writerTransInfo;
   data.discovery_locator_ = writer.writerDiscInfo;
   data.participant_discovered_at_ = writer.participantDiscoveredAt;
   data.remote_transport_context_ = writer.transportContext;
   data.publication_transport_priority_ =
       writer.writerQos.transport_priority.value;
   data.remote_reliable_ =
       (writer.writerQos.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS);
   data.remote_durable_ =
       (writer.writerQos.durability.kind > DDS::VOLATILE_DURABILITY_QOS);
 
   if (associate(data, active)) {
     const Observer_rch observer = get_observer(Observer::e_ASSOCIATED);
     if (observer) {
       observer->on_associated(this, data.remote_id_);
     }
   } else {
     if (DCPS_debug_level) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) DataReaderImpl::add_association: ")
           ACE_TEXT("ERROR: transport layer failed to associate.\n")));
     }
   }
 }

◆ add_link()

void OpenDDS::DCPS::DataReaderImpl::add_link	(	const DataLink_rch &	link,
		const GUID_t &	peer
	)

protectedvirtual

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 3226 of file DataReaderImpl.cpp.

References ACE_WRITE_GUARD, OpenDDS::DCPS::TransportClient::add_link(), OpenDDS::DCPS::DataLink::impl(), OPENDDS_STRING, OpenDDS::DCPS::TransportImpl::transport_type(), and DDS::VOLATILE_DURABILITY_QOS.

 {
   if (this->qos_.durability.kind > DDS::VOLATILE_DURABILITY_QOS) {
 
     ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, writers_lock_);
 
     WriterMapType::iterator it = writers_.find(peer);
     if (it != writers_.end()) {
       // Schedule timer if necessary
       //   - only need to check reader qos - we know the writer must be >= reader
       end_historic_sweeper_->schedule_timer(it->second);
     }
   }
   TransportClient::add_link(link, peer);
   OPENDDS_STRING type;
   {
     TransportImpl_rch impl = link->impl();
     if (impl) {
       type = impl->transport_type();
     }
   }
 
   if (type == "rtps_udp" || type == "multicast") {
     resume_sample_processing(peer);
   }
 }

◆ begin_access()

void OpenDDS::DCPS::DataReaderImpl::begin_access ( )

Definition at line 3068 of file DataReaderImpl.cpp.

References ACE_GUARD.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
   this->coherent_ = true;
 }

◆ cancel_all_deadlines()

void OpenDDS::DCPS::DataReaderImpl::cancel_all_deadlines ( )

private

Definition at line 3683 of file DataReaderImpl.cpp.

References ACE_GUARD.

Referenced by qos_change().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
   deadline_queue_.clear();
   deadline_task_->cancel();
 }

◆ cancel_deadline()

void OpenDDS::DCPS::DataReaderImpl::cancel_deadline ( SubscriptionInstance_rch instance )

private

Definition at line 3597 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::SubscriptionInstance::deadline_, and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::zero_value.

Referenced by data_received().

 {
   // Should be called with sample_lock_.
   if (instance->deadline_ != MonotonicTimePoint::zero_value) {
     for (DeadlineQueue::iterator pos = deadline_queue_.lower_bound(instance->deadline_), limit = deadline_queue_.upper_bound(instance->deadline_); pos != limit; ++pos) {
       if (pos->second == instance) {
         deadline_queue_.erase(pos);
         break;
       }
     }
     instance->deadline_ = MonotonicTimePoint::zero_value;
   }
 }

◆ check_historic()

bool OpenDDS::DCPS::DataReaderImpl::check_historic ( const ReceivedDataSample & sample )

private

collect samples received before END_HISTORIC_SAMPLES returns false if normal processing of this sample should be skipped

Definition at line 3195 of file DataReaderImpl.cpp.

References ACE_WRITE_GUARD_RETURN, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::DataSampleHeader::historic_sample_, OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::DataSampleHeader::sequence_, and OpenDDS::DCPS::SequenceNumber::SEQUENCENUMBER_UNKNOWN().

Referenced by data_received().

 {
   ACE_WRITE_GUARD_RETURN(ACE_RW_Thread_Mutex, write_guard, writers_lock_, true);
   WriterMapType::iterator iter = writers_.find(sample.header_.publication_id_);
   if (iter != writers_.end()) {
     const SequenceNumber& seq = sample.header_.sequence_;
     SequenceNumber last_historic_seq;
     if (iter->second->check_historic(seq, sample, last_historic_seq)) {
       return false;
     }
     if (last_historic_seq != SequenceNumber::SEQUENCENUMBER_UNKNOWN()
         && !sample.header_.historic_sample_
         && seq <= last_historic_seq) {
       // this sample must have been seen before the END_HISTORIC_SAMPLES control msg
       return false;
     }
   }
   return true;
 }

◆ check_transport_qos()

bool OpenDDS::DCPS::DataReaderImpl::check_transport_qos ( const TransportInst & inst )

virtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 1684 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::TransportInst::is_reliable(), DDS::ReliabilityQosPolicy::kind, qos_, DDS::DataReaderQos::reliability, and DDS::RELIABLE_RELIABILITY_QOS.

 {
   if (this->qos_.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS) {
     return ti.is_reliable();
   }
   return true;
 }

◆ cleanup()

void OpenDDS::DCPS::DataReaderImpl::cleanup ( void )

virtual

Definition at line 155 of file DataReaderImpl.cpp.

References content_filtered_topic_, content_filtered_topic_mutex_, multi_topic_, OpenDDS::DCPS::NO_STATUS_MASK, ownership_manager(), set_listener(), topic_servant_, and OpenDDS::DCPS::OwnershipManager::unregister_reader().

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

 {
   // As first step set our listener to nill which will prevent us from calling
   // back onto the listener at the moment the related DDS entity has been
   // deleted
   set_listener(0, NO_STATUS_MASK);
 
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   OwnershipManagerPtr owner_manager = this->ownership_manager();
   if (owner_manager) {
     owner_manager->unregister_reader(topic_servant_->type_name(), this);
   }
 #endif
 
   topic_servant_ = 0;
 
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
   {
     ACE_Guard<ACE_Thread_Mutex> guard(content_filtered_topic_mutex_);
     content_filtered_topic_ = 0;
   }
 #endif
 
 #ifndef OPENDDS_NO_MULTI_TOPIC
   multi_topic_ = 0;
 #endif
 
 }

◆ coherent_change_received() [1/2]

void OpenDDS::DCPS::DataReaderImpl::coherent_change_received	(	const GUID_t &	publisher_id,
		Coherent_State &	result
	)

Definition at line 2987 of file DataReaderImpl.cpp.

References ACE_READ_GUARD, OpenDDS::DCPS::COMPLETED, OpenDDS::DCPS::NOT_COMPLETED_YET, OpenDDS::DCPS::REJECTED, and state.

 {
   ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);
 
   result = COMPLETED;
   for (WriterMapType::iterator iter = writers_.begin();
       iter != writers_.end();
       ++iter) {
 
     if (iter->second->publisher_id() == publisher_id) {
       const Coherent_State state = iter->second->coherent_change_received();
       if (state == NOT_COMPLETED_YET) {
         result = NOT_COMPLETED_YET;
         break;
       }
       else if (state == REJECTED) {
         result = REJECTED;
       }
     }
   }
 }

◆ coherent_change_received() [2/2]

bool OpenDDS::DCPS::DataReaderImpl::coherent_change_received ( WriterInfo * writer )

private

◆ coherent_changes_completed()

void OpenDDS::DCPS::DataReaderImpl::coherent_changes_completed ( DataReaderImpl * reader )

Definition at line 3011 of file DataReaderImpl.cpp.

References ACE_GUARD, DDS::DATA_AVAILABLE_STATUS, DDS::DATA_ON_READERS_STATUS, OpenDDS::DCPS::RcHandle< T >::in(), CORBA::is_nil(), OpenDDS::DCPS::rchandle_from(), and TheServiceParticipant.

 {
   RcHandle<SubscriberImpl> subscriber = get_subscriber_servant();
   if (!subscriber) {
     return;
   }
 
   subscriber->set_status_changed_flag(::DDS::DATA_ON_READERS_STATUS, true);
   this->set_status_changed_flag(::DDS::DATA_AVAILABLE_STATUS, true);
 
   ::DDS::SubscriberListener_var sub_listener =
       subscriber->listener_for(::DDS::DATA_ON_READERS_STATUS);
   if (!CORBA::is_nil(sub_listener.in()))
   {
     if (!is_bit()) {
       this->set_status_changed_flag(::DDS::DATA_AVAILABLE_STATUS, false);
       subscriber->set_status_changed_flag(::DDS::DATA_ON_READERS_STATUS, false);
       if (reader == this) {
         // Release the sample_lock before listener callback.
         ACE_GUARD(Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
         sub_listener->on_data_on_readers(subscriber.in());
       }
     } else {
       TheServiceParticipant->job_queue()->enqueue(make_rch<OnDataOnReaders>(subscriber, sub_listener, rchandle_from(this), reader == this, true));
     }
   }
   else
   {
     subscriber->notify_status_condition();
 
     ::DDS::DataReaderListener_var listener =
         this->listener_for (::DDS::DATA_AVAILABLE_STATUS);
 
     if (!CORBA::is_nil(listener.in()))
     {
       if (!is_bit()) {
         set_status_changed_flag(::DDS::DATA_AVAILABLE_STATUS, false);
         subscriber->set_status_changed_flag(::DDS::DATA_ON_READERS_STATUS, false);
         if (reader == this) {
           // Release the sample_lock before listener callback.
           ACE_GUARD(Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
           listener->on_data_available(this);
         } else {
           listener->on_data_available(this);
         }
       } else {
         TheServiceParticipant->job_queue()->enqueue(make_rch<OnDataAvailable>(listener, rchandle_from(this), reader == this, true, true));
       }
     }
     else
     {
       this->notify_status_condition();
     }
   }
 }

◆ contains_sample()

bool OpenDDS::DCPS::DataReaderImpl::contains_sample	(	DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states
	)

Fold-in the three separate loops of have_sample_states(), have_view_states(), and have_instance_states(). Takes the sample_lock_.

Definition at line 1743 of file DataReaderImpl.cpp.

References instances_lock_, lookup_matching_instances(), and sample_lock_.

Referenced by OpenDDS::DCPS::ReadConditionImpl::get_trigger_value().

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> sample_guard(sample_lock_);
   ACE_Guard<ACE_Recursive_Thread_Mutex> instance_guard(instances_lock_);
 
   return lookup_matching_instances(sample_states, view_states, instance_states).size();
 }

◆ contains_sample_filtered()

virtual bool OpenDDS::DCPS::DataReaderImpl::contains_sample_filtered	(	DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states,
		const FilterEvaluator &	evaluator,
		const DDS::StringSeq &	params
	)

pure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by OpenDDS::DCPS::QueryConditionImpl::get_trigger_value().

◆ create_querycondition()

DDS::QueryCondition_ptr OpenDDS::DCPS::DataReaderImpl::create_querycondition	(	DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states,
		const char *	query_expression,
		const DDS::StringSeq &	query_parameters
	)

virtual

Definition at line 771 of file DataReaderImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, LM_ERROR, QueryConditionImpl, read_conditions_, DDS::RETCODE_OK, and sample_lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_, 0);
   try {
     DDS::QueryCondition_var qc = new QueryConditionImpl(this, sample_states,
         view_states, instance_states, query_expression);
     if (qc->set_query_parameters(query_parameters) != DDS::RETCODE_OK) {
       return 0;
     }
     DDS::ReadCondition_var rc = DDS::ReadCondition::_duplicate(qc);
     read_conditions_.insert(rc);
     return qc._retn();
   } catch (const std::exception& e) {
     if (DCPS_debug_level) {
       ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ")
           ACE_TEXT("DataReaderImpl::create_querycondition - %C\n"),
           e.what()));
     }
   }
   return 0;
 }

◆ create_readcondition()

DDS::ReadCondition_ptr OpenDDS::DCPS::DataReaderImpl::create_readcondition	(	DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states
	)

virtual

Definition at line 758 of file DataReaderImpl.cpp.

References ACE_GUARD_RETURN, read_conditions_, and sample_lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_, 0);
   DDS::ReadCondition_var rc = new ReadConditionImpl(this, sample_states,
       view_states, instance_states);
   read_conditions_.insert(rc);
   return rc._retn();
 }

◆ data_received()

void OpenDDS::DCPS::DataReaderImpl::data_received ( const ReceivedDataSample & sample )

virtual

process a message that has been received - could be control or a data sample.

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 1371 of file DataReaderImpl.cpp.

 {
   DBG_ENTRY_LVL("DataReaderImpl","data_received",6);
 
   DDS::InstanceHandle_t publication_handle = DDS::HANDLE_NIL;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, publication_handle_lock_);
     RepoIdToHandleMap::const_iterator pos = publication_id_to_handle_map_.find(sample.header_.publication_id_);
     if (pos != publication_id_to_handle_map_.end()) {
       publication_handle = pos->second;
     }
   }
 
   // ensure some other thread is not changing the sample container
   // or statuses related to samples.
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_);
 
   if (get_deleted()) return;
 
   if (DCPS_debug_level > 9) {
     ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) DataReaderImpl::data_received: ")
         ACE_TEXT("%C received sample: %C.\n"),
         LogGuid(get_guid()).c_str(),
         to_string(sample.header_).c_str()));
   }
 
   const ValueDispatcher* vd = get_value_dispatcher();
   const Observer_rch observer = get_observer(Observer::e_SAMPLE_RECEIVED);
 
   RcHandle<MessageHolder> real_data;
   SubscriptionInstance_rch instance;
   switch (sample.header_.message_id_) {
   case SAMPLE_DATA:
   case INSTANCE_REGISTRATION: {
     if (!check_historic(sample)) break;
 
     DataSampleHeader const & header = sample.header_;
 
     this->writer_activity(header);
 
     // Verify data has not exceeded its lifespan.
     if (this->filter_sample(header)) break;
 
     // This adds the reader to the set/list of readers with data.
     RcHandle<SubscriberImpl> subscriber = get_subscriber_servant();
     if (subscriber) {
       subscriber->data_received(this);
     }
 
     // Only gather statistics about real samples, not registration data, etc.
     if (header.message_id_ == SAMPLE_DATA) {
       this->process_latency(sample);
     }
 
     // This also adds to the sample container and makes any callbacks
     // and condition modifications.
 
     bool is_new_instance = false;
     bool filtered = false;
     if (sample.header_.key_fields_only_) {
       dds_demarshal(sample, publication_handle, instance, is_new_instance, filtered, KEY_ONLY_MARSHALING, false);
     } else {
       real_data = dds_demarshal(sample, publication_handle, instance, is_new_instance, filtered, FULL_MARSHALING, observer && vd);
     }
 
     // Per sample logging
     if (DCPS_debug_level >= 8) {
       ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) DataReaderImpl::data_received: reader %C writer %C ")
           ACE_TEXT("instance %d is_new_instance %d filtered %d\n"),
           LogGuid(get_guid()).c_str(),
           LogGuid(header.publication_id_).c_str(),
           instance ? instance->instance_handle_ : 0,
           is_new_instance, filtered));
     }
 
     if (filtered) break; // sample filtered from instance
 
     if (instance) accept_sample_processing(instance, header, is_new_instance);
   }
   break;
 
 #ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
   case END_COHERENT_CHANGES: {
     CoherentChangeControl control;
 
     this->writer_activity(sample.header_);
 
     Message_Block_Ptr payload(sample.data(&mb_alloc_));
     Serializer serializer(
         payload.get(), Encoding::KIND_UNALIGNED_CDR,
         sample.header_.byte_order_ ? ENDIAN_LITTLE : ENDIAN_BIG);
     if (!(serializer >> control)) {
       ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) DataReaderImpl::data_received ")
           ACE_TEXT("deserialization coherent change control failed.\n")));
       return;
     }
 
     if (DCPS_debug_level > 0) {
       std::stringstream buffer;
       buffer << control << std::endl;
 
       ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) DataReaderImpl::data_received: ")
           ACE_TEXT("END_COHERENT_CHANGES %C\n"),
           buffer.str().c_str()));
     }
 
     WriterInfo_rch writer;
     {
       ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);
 
       WriterMapType::iterator it =
           this->writers_.find(sample.header_.publication_id_);
 
       if (it == this->writers_.end()) {
         ACE_DEBUG((LM_WARNING,
             ACE_TEXT("(%P|%t) WARNING: DataReaderImpl::data_received() - ")
             ACE_TEXT(" subscription %C failed to find ")
             ACE_TEXT(" publication data for %C!\n"),
             LogGuid(get_guid()).c_str(),
             LogGuid(sample.header_.publication_id_).c_str()));
         return;
       }
       else {
         writer = it->second;
       }
       it->second->set_group_info(control);
     }
 
     if (this->verify_coherent_changes_completion(writer.in())) {
       this->notify_read_conditions();
     }
   }
   break;
 #endif // OPENDDS_NO_OBJECT_MODEL_PROFILE
 
   case DATAWRITER_LIVELINESS: {
     if (DCPS_debug_level >= 4) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) DataReaderImpl::data_received: ")
                  ACE_TEXT("reader %C got datawriter liveliness from writer %C\n"),
                  LogGuid(get_guid()).c_str(),
                  LogGuid(sample.header_.publication_id_).c_str()));
     }
     this->writer_activity(sample.header_);
 
     // tell all instances they got a liveliness message
     {
       ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, instances_lock_);
       for (SubscriptionInstanceMapType::iterator iter = instances_.begin();
           iter != instances_.end();
           ++iter) {
         if (iter->second->instance_state_->writes_instance(sample.header_.publication_id_)) {
           iter->second->instance_state_->lively(sample.header_.publication_id_);
         }
       }
     }
 
   }
   break;
 
   case DISPOSE_INSTANCE: {
     if (!check_historic(sample)) break;
     this->writer_activity(sample.header_);
     SubscriptionInstance_rch instance;
 
     if (deadline_queue_enabled_) {
       // Find the instance first for timer cancellation since
       // the instance may be deleted during dispose and can
       // not be accessed.
       ReceivedDataSample dup(sample);
       this->lookup_instance(dup, instance);
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
       OwnershipManagerPtr owner_manager = this->ownership_manager();
 
       if (! this->is_exclusive_ownership_
           || (owner_manager
               && (instance)
               && (owner_manager->is_owner(instance->instance_handle_,
                   sample.header_.publication_id_)))) {
 #endif
         cancel_deadline(instance);
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
       }
 #endif
     }
     instance.reset();
     this->dispose_unregister(sample, publication_handle, instance);
   }
   this->notify_read_conditions();
   break;
 
   case UNREGISTER_INSTANCE: {
     if (!check_historic(sample)) break;
     this->writer_activity(sample.header_);
     SubscriptionInstance_rch instance;
 
     if (deadline_queue_enabled_) {
       // Find the instance first for timer cancellation since
       // the instance may be deleted during dispose and can
       // not be accessed.
       ReceivedDataSample dup(sample);
       this->lookup_instance(dup, instance);
       if (instance) {
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
         if (! this->is_exclusive_ownership_
             || (this->is_exclusive_ownership_
                 && instance->instance_state_->is_last(sample.header_.publication_id_))) {
 #endif
           cancel_deadline(instance);
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
         }
 #endif
       }
     }
     instance.reset();
     this->dispose_unregister(sample, publication_handle, instance);
   }
   this->notify_read_conditions();
   break;
 
   case DISPOSE_UNREGISTER_INSTANCE: {
     if (!check_historic(sample)) break;
     this->writer_activity(sample.header_);
     SubscriptionInstance_rch instance;
 
     if (deadline_queue_enabled_) {
       // Find the instance first for timer cancellation since
       // the instance may be deleted during dispose and can
       // not be accessed.
       ReceivedDataSample dup(sample);
       this->lookup_instance(dup, instance);
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
       OwnershipManagerPtr owner_manager = this->ownership_manager();
       if (! this->is_exclusive_ownership_
           || (owner_manager
               && (instance)
               && (owner_manager->is_owner (instance->instance_handle_,
                   sample.header_.publication_id_)))
           || (is_exclusive_ownership_
               && (instance)
               && instance->instance_state_->is_last(sample.header_.publication_id_))) {
 #endif
         if (instance) {
           cancel_deadline(instance);
         }
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
       }
 #endif
     }
     instance.reset();
     this->dispose_unregister(sample, publication_handle, instance);
   }
   this->notify_read_conditions();
   break;
 
   case END_HISTORIC_SAMPLES: {
     if (sample.header_.message_length_ >= sizeof(GUID_t)) {
       Message_Block_Ptr payload(sample.data(&mb_alloc_));
       Serializer ser(payload.get(), Encoding::KIND_UNALIGNED_CDR);
       GUID_t readerId = GUID_UNKNOWN;
       if (!(ser >> readerId)) {
         ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) DataReaderImpl::data_received ")
             ACE_TEXT("deserialization reader failed.\n")));
         return;
       }
       const GUID_t repo_id(get_guid());
       if (readerId != GUID_UNKNOWN && readerId != repo_id) {
         break; // not our message
       }
     }
     if (DCPS_debug_level > 4) {
       ACE_DEBUG((LM_INFO, "(%P|%t) Received END_HISTORIC_SAMPLES control message\n"));
     }
     // Going to acquire writers lock, release samples lock
     guard.release();
     resume_sample_processing(sample.header_.publication_id_);
     if (DCPS_debug_level > 4) {
       ACE_DEBUG((
           LM_INFO,
           "(%P|%t) Resumed sample processing for durable writer %C\n",
           LogGuid(sample.header_.publication_id_).c_str()));
     }
     break;
   }
 
   default:
     ACE_ERROR((LM_ERROR,
         "(%P|%t) ERROR: DataReaderImpl::data_received"
         "unexpected message_id = %d\n",
         sample.header_.message_id_));
     break;
   }
 
   if (observer && real_data && vd) {
     const DDS::Time_t timestamp = {
       sample.header_.source_timestamp_sec_,
       sample.header_.source_timestamp_nanosec_
     };
     Observer::Sample s(instance ? instance->instance_handle_ : DDS::HANDLE_NIL, sample.header_.instance_state(), timestamp, sample.header_.sequence_, real_data->get(), *vd);
     observer->on_sample_received(this, s);
   }
 }

◆ dds_demarshal()

virtual RcHandle<MessageHolder> OpenDDS::DCPS::DataReaderImpl::dds_demarshal	(	const ReceivedDataSample &	sample,
		DDS::InstanceHandle_t	publication_handle,
		SubscriptionInstance_rch &	instance,
		bool &	is_new_instance,
		bool &	filtered,
		MarshalingType	marshaling_type,
		bool	full_copy
	)

pure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by data_received().

◆ deadline_task()

void OpenDDS::DCPS::DataReaderImpl::deadline_task ( const MonotonicTimePoint & now )

private

Definition at line 3739 of file DataReaderImpl.cpp.

References ACE_GUARD, OPENDDS_END_VERSIONED_NAMESPACE_DECL, and TheServiceParticipant.

 {
   ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());
 
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
   for (DeadlineQueue::iterator pos = deadline_queue_.begin(), limit = deadline_queue_.end(); pos != limit && pos->first <= now;) {
     SubscriptionInstance_rch instance = pos->second;
     deadline_queue_.erase(pos++);
     // pos is no longer valid.
     process_deadline(instance, now, true);
   }
 
   if (!deadline_queue_.empty()) {
     deadline_task_->schedule(deadline_queue_.begin()->first - now);
   }
 }

◆ delete_contained_entities()

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

virtual

Implements DDS::DataReader.

Definition at line 816 of file DataReaderImpl.cpp.

References ACE_GUARD_RETURN, read_conditions_, DDS::RETCODE_OK, DDS::RETCODE_OUT_OF_RESOURCES, and sample_lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_,
       DDS::RETCODE_OUT_OF_RESOURCES);
   read_conditions_.clear();
   return DDS::RETCODE_OK;
 }

◆ delete_readcondition()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::delete_readcondition ( DDS::ReadCondition_ptr a_condition )

virtual

Definition at line 806 of file DataReaderImpl.cpp.

References ACE_GUARD_RETURN, read_conditions_, DDS::RETCODE_OK, DDS::RETCODE_OUT_OF_RESOURCES, DDS::RETCODE_PRECONDITION_NOT_MET, and sample_lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_,
       DDS::RETCODE_OUT_OF_RESOURCES);
   DDS::ReadCondition_var rc = DDS::ReadCondition::_duplicate(a_condition);
   return read_conditions_.erase(rc)
       ? DDS::RETCODE_OK : DDS::RETCODE_PRECONDITION_NOT_MET;
 }

◆ deliver_historic()

void OpenDDS::DCPS::DataReaderImpl::deliver_historic ( OPENDDS_MAP(SequenceNumber, ReceivedDataSample)& samples )

private

deliver samples that were held by check_historic()

Definition at line 3215 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::DataSampleHeader::historic_sample_, and OpenDDS::DCPS::OPENDDS_MAP().

 {
   typedef OPENDDS_MAP(SequenceNumber, ReceivedDataSample)::iterator iter_t;
   const iter_t end = samples.end();
   for (iter_t iter = samples.begin(); iter != end; ++iter) {
     iter->second.header_.historic_sample_ = true;
     data_received(iter->second);
   }
 }

◆ disable_transport()

ACE_INLINE void OpenDDS::DCPS::DataReaderImpl::disable_transport ( )

Definition at line 33 of file DataReaderImpl.inl.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL, and transport_disabled_.

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

 {
   this->transport_disabled_ = true;
 }

◆ dispose_unregister()

void OpenDDS::DCPS::DataReaderImpl::dispose_unregister	(	const ReceivedDataSample &	sample,
		DDS::InstanceHandle_t	publication_handle,
		SubscriptionInstance_rch &	instance
	)

virtual

Reimplemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Definition at line 2275 of file DataReaderImpl.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DCPS_debug_level, and LM_DEBUG.

Referenced by data_received().

 {
   if (DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) DataReaderImpl::dispose_unregister()\n"));
   }
 }

◆ domain_id()

DDS::DomainId_t OpenDDS::DCPS::DataReaderImpl::domain_id ( ) const

inlineprivatevirtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 847 of file DataReaderImpl.h.

847 { return this->domain_id_; }

OpenDDS::DCPS::DataReaderImpl::domain_id_

DDS::DomainId_t domain_id_

Definition: DataReaderImpl.h:876

◆ enable()

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

virtual

Implements DDS::Entity.

Definition at line 1115 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_INT32_MAX, ACE_TEXT(), OpenDDS::DCPS::TypeSupportImpl::add_types(), OpenDDS::DCPS::TransportClient::connection_info(), content_filtered_topic_, content_filtered_topic_mutex_, OpenDDS::DCPS::DCPS_debug_level, DDS::DataReaderQos::deadline, deadline_period_, deadline_queue_enabled_, DDS::HistoryQosPolicy::depth, depth_, domain_id_, dp_id_, DDS::DURATION_INFINITE_NSEC, DDS::DURATION_INFINITE_SEC, dynamic_type_, OpenDDS::DCPS::Observer::e_ENABLED, enable_specific(), OpenDDS::DCPS::TransportClient::enable_transport(), get_guid(), OpenDDS::DCPS::EntityImpl::get_observer(), get_subscriber_servant(), OpenDDS::DCPS::TypeSupportImpl::get_type(), OpenDDS::DCPS::GUID_UNKNOWN, has_subscription_id_, DDS::DataReaderQos::history, TAO::String_var< charT >::in(), install_type_support(), OpenDDS::DCPS::EntityImpl::is_enabled(), DDS::KEEP_ALL_HISTORY_QOS, DDS::ReliabilityQosPolicy::kind, DDS::HistoryQosPolicy::kind, DDS::LivelinessQosPolicy::lease_duration, DDS::LENGTH_UNLIMITED, DDS::DataReaderQos::liveliness, OpenDDS::DCPS::WriterInfoListener::liveliness_lease_duration_, LM_DEBUG, LM_ERROR, LM_WARNING, DDS::ResourceLimitsQosPolicy::max_samples, DDS::ResourceLimitsQosPolicy::max_samples_per_instance, monitor_, n_chunks_, name, DDS::Duration_t::nanosec, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), OpenDDS::DCPS::Observer::on_enabled(), participant_servant_, DDS::DeadlineQosPolicy::period, qos_, OpenDDS::DCPS::rchandle_from(), rd_allocator_, DDS::DataReaderQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, DDS::DataReaderQos::representation, DDS::DataReaderQos::resource_limits, DDS::RETCODE_ERROR, DDS::RETCODE_OK, DDS::RETCODE_PRECONDITION_NOT_MET, sample_lock_, DDS::Duration_t::sec, security_config_, OpenDDS::DCPS::EntityImpl::set_enabled(), OpenDDS::DCPS::set_reader_effective_data_rep_qos(), setup_deserialization(), OpenDDS::DCPS::WriterInfoListener::subscription_id_, subscription_id_condition_, subscription_id_mutex_, TheServiceParticipant, OpenDDS::DCPS::TypeSupportImpl::to_type_info(), topic_servant_, transport_disabled_, DDS::DataRepresentationQosPolicy::value, and DDS::VOLATILE_DURABILITY_QOS.

Referenced by OpenDDS::RTPS::RtpsDiscovery::create_bit_dr(), OpenDDS::DCPS::StaticDiscovery::create_bit_dr(), and OpenDDS::DCPS::SubscriberImpl::create_datareader().

 {
   // According to spec:
   // - Calling enable on an already enabled Entity has no effect and returns OK.
   // - 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<SubscriberImpl> subscriber = get_subscriber_servant();
   if (!subscriber) {
     return DDS::RETCODE_ERROR;
   }
 
   if (!subscriber->is_enabled()) {
     return DDS::RETCODE_PRECONDITION_NOT_MET;
   }
 
   if (topic_servant_ && !topic_servant_->is_enabled()) {
     return DDS::RETCODE_PRECONDITION_NOT_MET;
   }
 
   RcHandle<DomainParticipantImpl> participant = participant_servant_.lock();
   if (participant) {
     dp_id_ = participant->get_id();
   }
 
   if (topic_servant_) {
     set_reader_effective_data_rep_qos(qos_.representation.value);
     if (!topic_servant_->check_data_representation(qos_.representation.value, false)) {
       return DDS::RETCODE_ERROR;
     }
   }
 
   if (qos_.history.kind == DDS::KEEP_ALL_HISTORY_QOS) {
     // The spec says qos_.history.depth is "has no effect"
     // when history.kind = KEEP_ALL so use max_samples_per_instance
     depth_ = qos_.resource_limits.max_samples_per_instance;
 
   } else { // qos_.history.kind == DDS::KEEP_LAST_HISTORY_QOS
     depth_ = qos_.history.depth;
   }
 
   if (depth_ == DDS::LENGTH_UNLIMITED) {
     // DDS::LENGTH_UNLIMITED is negative so make it a positive
     // value that is, for all intents and purposes, unlimited
     // and we can use it for comparisons.
     // WARNING: The client risks running out of memory in this case.
     depth_ = ACE_INT32_MAX;
   }
 
   if (qos_.resource_limits.max_samples != DDS::LENGTH_UNLIMITED) {
     n_chunks_ = qos_.resource_limits.max_samples;
   }
 
   //else using value from Service_Participant
 
   // enable the type specific part of this DataReader
   this->enable_specific();
 
   //Note: the QoS used to set n_chunks_ is Changeable=No so
   // it is OK that we cannot change the size of our allocators.
   rd_allocator_.reset(new ReceivedDataAllocator(n_chunks_));
 
   if (DCPS_debug_level >= 2)
     ACE_DEBUG((LM_DEBUG,"(%P|%t) DataReaderImpl::enable"
         " Cached_Allocator_With_Overflow %x with %d chunks\n",
         rd_allocator_.get(), n_chunks_));
 
   if ((qos_.liveliness.lease_duration.sec !=
       DDS::DURATION_INFINITE_SEC) &&
       (qos_.liveliness.lease_duration.nanosec !=
           DDS::DURATION_INFINITE_NSEC)) {
     liveliness_lease_duration_ = TimeDuration(qos_.liveliness.lease_duration);
   }
 
   // Setup the requested deadline watchdog if the configured deadline
   // period is not the default (infinite).
   DDS::Duration_t const deadline_period = this->qos_.deadline.period;
 
   if (!deadline_queue_enabled_
       && (deadline_period.sec != DDS::DURATION_INFINITE_SEC
           || deadline_period.nanosec != DDS::DURATION_INFINITE_NSEC)) {
     deadline_period_ = TimeDuration(qos_.deadline.period);
     deadline_queue_enabled_ = true;
   }
 
   Discovery_rch disco = TheServiceParticipant->get_discovery(domain_id_);
   disco->pre_reader(this);
 
   this->set_enabled();
 
   if (topic_servant_ && !transport_disabled_) {
     try {
       this->enable_transport(this->qos_.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS,
           this->qos_.durability.kind > DDS::VOLATILE_DURABILITY_QOS);
     } catch (const Transport::Exception&) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::enable, ")
           ACE_TEXT("Transport Exception.\n")));
       return DDS::RETCODE_ERROR;
     }
 
     const DDS::ReturnCode_t setup_deserialization_result = setup_deserialization();
     if (setup_deserialization_result != DDS::RETCODE_OK) {
       return setup_deserialization_result;
     }
 
     const TransportLocatorSeq& trans_conf_info = connection_info();
 
     CORBA::String_var filterClassName = "";
     CORBA::String_var filterExpression = "";
     DDS::StringSeq exprParams;
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
     {
       ACE_Guard<ACE_Thread_Mutex> guard(content_filtered_topic_mutex_);
       if (content_filtered_topic_) {
         filterClassName = content_filtered_topic_->get_filter_class_name();
         filterExpression = content_filtered_topic_->get_filter_expression();
         content_filtered_topic_->get_expression_parameters(exprParams);
       }
     }
 #endif
 
     DDS::SubscriberQos sub_qos;
     subscriber->get_qos(sub_qos);
 
     TypeSupportImpl* const typesupport =
       dynamic_cast<TypeSupportImpl*>(topic_servant_->get_type_support());
     if (!typesupport) {
       return DDS::RETCODE_ERROR;
     }
 
     XTypes::TypeInformation type_info;
     typesupport->to_type_info(type_info);
 
     XTypes::TypeLookupService_rch type_lookup_service = participant->get_type_lookup_service();
     typesupport->add_types(type_lookup_service);
 
     install_type_support(typesupport);
 
     const GUID_t subscription_id =
       disco->add_subscription(domain_id_,
         dp_id_,
         topic_servant_->get_id(),
         rchandle_from(this),
         qos_,
         trans_conf_info,
         sub_qos,
         filterClassName,
         filterExpression,
         exprParams,
         type_info);
 
 #if defined(OPENDDS_SECURITY)
     {
       ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, sample_lock_, DDS::RETCODE_ERROR);
       security_config_ = participant->get_security_config();
       dynamic_type_ = typesupport->get_type();
     }
 #endif
 
     {
       ACE_Guard<ACE_Thread_Mutex> guard(subscription_id_mutex_);
       subscription_id_ = subscription_id;
       has_subscription_id_ = true;
       subscription_id_condition_.notify_all();
     }
 
     if (subscription_id == GUID_UNKNOWN) {
       if (DCPS_debug_level >= 1) {
         ACE_DEBUG((LM_WARNING, "(%P|%t) WARNING: DataReaderImpl::enable: "
           "add_subscription failed\n"));
       }
       return DDS::RETCODE_ERROR;
     }
 
     if (DCPS_debug_level >= 2) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) DataReaderImpl::enable: "
         "got GUID %C, subscribed to topic name \"%C\" type \"%C\"\n",
         LogGuid(get_guid()).c_str(),
         topic_servant_->topic_name(), topic_servant_->type_name()));
     }
   }
 
   DDS::ReturnCode_t return_value = DDS::RETCODE_OK;
   if (topic_servant_) {
     const CORBA::String_var name = topic_servant_->get_name();
     return_value = subscriber->reader_enabled(name.in(), this);
 
     if (this->monitor_) {
       this->monitor_->report();
     }
   }
 
   if (return_value == DDS::RETCODE_OK) {
     const Observer_rch observer = get_observer(Observer::e_ENABLED);
     if (observer) {
       observer->on_enabled(this);
     }
   }
 
   return return_value;
 }

◆ enable_filtering()

void OpenDDS::DCPS::DataReaderImpl::enable_filtering ( ContentFilteredTopicImpl * cft )

Definition at line 3124 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::ContentFilteredTopicImpl::add_reader().

Referenced by OpenDDS::DCPS::SubscriberImpl::create_datareader().

 {
   cft->add_reader(*this);
   {
     ACE_Guard<ACE_Thread_Mutex> guard(content_filtered_topic_mutex_);
     content_filtered_topic_ = cft;
   }
 }

◆ enable_multi_topic()

void OpenDDS::DCPS::DataReaderImpl::enable_multi_topic ( MultiTopicImpl * mt )

Definition at line 3143 of file DataReaderImpl.cpp.

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

 {
   multi_topic_ = mt;
 }

◆ enable_specific()

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::enable_specific ( )

protectedpure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by enable().

◆ end_access()

void OpenDDS::DCPS::DataReaderImpl::end_access ( )

Definition at line 3075 of file DataReaderImpl.cpp.

References ACE_GUARD.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
   this->coherent_ = false;
   this->group_coherent_ordered_data_.reset();
   this->post_read_or_take();
 }

◆ filter_sample()

bool OpenDDS::DCPS::DataReaderImpl::filter_sample ( const DataSampleHeader & header )

protected

Check if the received data sample or instance should be filtered.

Note: Filtering will only occur if the application configured a finite duration in the Topic's LIFESPAN QoS policy or DataReader's TIME_BASED_FILTER QoS policy.

Definition at line 2583 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::DataSampleHeader::historic_sample_, OpenDDS::DCPS::DataSampleHeader::lifespan_duration_, OpenDDS::DCPS::DataSampleHeader::lifespan_duration_nanosec_, OpenDDS::DCPS::DataSampleHeader::lifespan_duration_sec_, LM_DEBUG, OpenDDS::DCPS::TimePoint_T< SystemClock >::now(), ACE_Time_Value::sec(), OpenDDS::DCPS::DataSampleHeader::source_timestamp_nanosec_, OpenDDS::DCPS::DataSampleHeader::source_timestamp_sec_, ACE_Time_Value::usec(), OpenDDS::DCPS::TimeDuration::value(), and DDS::VOLATILE_DURABILITY_QOS.

Referenced by data_received().

 {
   const SystemTimePoint now = SystemTimePoint::now();
 
   // Expire historic data if QoS indicates VOLATILE.
   if (!always_get_history_ && header.historic_sample_
       && qos_.durability.kind == DDS::VOLATILE_DURABILITY_QOS) {
     if (DCPS_debug_level >= 8) {
       ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) DataReaderImpl::filter_sample: ")
           ACE_TEXT("Discarded historic data.\n")));
     }
 
     return true; // Data filtered.
   }
 
   // The LIFESPAN_DURATION_FLAG is set when sample data is sent
   // with a non-default LIFESPAN duration value.
   if (header.lifespan_duration_) {
     // Finite lifespan.  Check if data has expired.
 
     const DDS::Time_t expiration_dds_time = {
       header.source_timestamp_sec_ + header.lifespan_duration_sec_,
       header.source_timestamp_nanosec_ + header.lifespan_duration_nanosec_
     };
     const SystemTimePoint expiration_time(expiration_dds_time);
 
     // We assume that the publisher host's clock and subcriber host's
     // clock are synchronized (allowed by the spec).
     if (now >= expiration_time) {
       if (DCPS_debug_level >= 8) {
         const TimeDuration diff(now - expiration_time);
         ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) Received data ")
           ACE_TEXT("expired by %d seconds, %d microseconds.\n"),
           diff.value().sec(),
           diff.value().usec()));
       }
 
       return true; // Data filtered.
     }
   }
 
   return false;
 }

◆ get_builtin_subscriber_proxy()

RcHandle<BitSubscriber> OpenDDS::DCPS::DataReaderImpl::get_builtin_subscriber_proxy ( ) const

inlineprivatevirtual

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 837 of file DataReaderImpl.h.

References OpenDDS::DCPS::WeakRcHandle< T >::lock().

   {
     RcHandle<DomainParticipantImpl> participant_servant = participant_servant_.lock();
     if (participant_servant) {
       return participant_servant->get_builtin_subscriber_proxy();
     }
 
     return RcHandle<BitSubscriber>();
   }

◆ get_cf_topic()

DDS::ContentFilteredTopic_ptr OpenDDS::DCPS::DataReaderImpl::get_cf_topic ( ) const

Definition at line 3134 of file DataReaderImpl.cpp.

 {
   ACE_Guard<ACE_Thread_Mutex> guard(content_filtered_topic_mutex_);
   return DDS::ContentFilteredTopic::_duplicate(content_filtered_topic_.get());
 }

◆ get_crypto_handle()

DDS::Security::ParticipantCryptoHandle OpenDDS::DCPS::DataReaderImpl::get_crypto_handle ( ) const

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 3404 of file DataReaderImpl.cpp.

References DDS::HANDLE_NIL.

 {
   RcHandle<DomainParticipantImpl> participant = participant_servant_.lock();
   return participant ? participant->crypto_handle() : DDS::HANDLE_NIL;
 }

◆ get_depth()

size_t OpenDDS::DCPS::DataReaderImpl::get_depth ( ) const

inline

Definition at line 410 of file DataReaderImpl.h.

   {
     return static_cast<size_t>(depth_);
   }

◆ get_dp_id()

OpenDDS::DCPS::GUID_t OpenDDS::DCPS::DataReaderImpl::get_dp_id ( )

Definition at line 2816 of file DataReaderImpl.cpp.

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

 {
   return dp_id_;
 }

◆ get_ext_listener()

DataReaderListener_ptr OpenDDS::DCPS::DataReaderImpl::get_ext_listener ( )

protected

Definition at line 926 of file DataReaderImpl.cpp.

References listener_, and listener_mutex_.

 {
   ACE_Guard<ACE_Thread_Mutex> g(listener_mutex_);
   return DataReaderListener::_narrow(listener_.in());
 }

◆ get_guid()

GUID_t OpenDDS::DCPS::DataReaderImpl::get_guid ( ) const

inlinevirtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 628 of file DataReaderImpl.h.

References TheServiceParticipant.

Referenced by add_association(), data_received(), enable(), OpenDDS::Federator::ManagerImpl::initialize(), remove_associations(), remove_associations_i(), OpenDDS::DCPS::DRPeriodicMonitorImpl::report(), OpenDDS::DCPS::DRMonitorImpl::report(), transport_assoc_done(), and writer_activity().

   {
     ACE_Guard<ACE_Thread_Mutex> guard(subscription_id_mutex_);
     ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
     while (!has_subscription_id_ && !get_deleted()) {
       subscription_id_condition_.wait(thread_status_manager);
     }
     return subscription_id_;
   }

◆ get_handle_instance()

SubscriptionInstance_rch OpenDDS::DCPS::DataReaderImpl::get_handle_instance ( DDS::InstanceHandle_t handle )

protected

Definition at line 2424 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), and LM_WARNING.

Referenced by release_instance().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, instance_guard, this->instances_lock_, SubscriptionInstance_rch());
 
   SubscriptionInstanceMapType::iterator iter = instances_.find(handle);
   if (iter == instances_.end()) {
     ACE_DEBUG((LM_WARNING,
         ACE_TEXT("(%P|%t) WARNING: ")
         ACE_TEXT("DataReaderImpl::get_handle_instance: ")
         ACE_TEXT("lookup for 0x%x failed\n"),
         handle));
     return SubscriptionInstance_rch();
   } // if (0 != instances_.find(handle, instance))
 
   return iter->second;
 }

◆ get_ice_endpoint()

WeakRcHandle< ICE::Endpoint > OpenDDS::DCPS::DataReaderImpl::get_ice_endpoint ( )

virtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 3286 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::TransportClient::get_ice_endpoint().

 {
   return TransportClient::get_ice_endpoint();
 }

◆ get_instance_handle()

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

virtual

Implements OpenDDS::DCPS::EntityImpl.

Definition at line 230 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::EntityImpl::get_entity_instance_handle(), participant_servant_, and OpenDDS::DCPS::WriterInfoListener::subscription_id_.

 {
   const RcHandle<DomainParticipantImpl> participant = participant_servant_.lock();
   return get_entity_instance_handle(subscription_id_, participant);
 }

◆ get_instance_handles()

void OpenDDS::DCPS::DataReaderImpl::get_instance_handles ( InstanceHandleVec & instance_handles )

Definition at line 2822 of file DataReaderImpl.cpp.

References ACE_GUARD.

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

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
   ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, this->instances_lock_);
 
   for (SubscriptionInstanceMapType::iterator iter = instances_.begin(),
       end = instances_.end(); iter != end; ++iter) {
     instance_handles.push_back(iter->first);
   }
 }

◆ get_latency_stats()

void OpenDDS::DCPS::DataReaderImpl::get_latency_stats ( LatencyStatisticsSeq & stats )

virtual

Definition at line 2369 of file DataReaderImpl.cpp.

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(statistics_lock_);
   stats.length(static_cast<CORBA::ULong>(this->statistics_.size()));
   int index = 0;
 
   for (StatsMapType::const_iterator current = this->statistics_.begin();
       current != this->statistics_.end();
       ++current, ++index) {
     stats[ index] = current->second.get_stats();
     stats[ index].publication = current->first;
   }
 }

◆ get_listener()

DDS::DataReaderListener_ptr OpenDDS::DCPS::DataReaderImpl::get_listener ( )

virtual

Implements DDS::DataReader.

Definition at line 920 of file DataReaderImpl.cpp.

References listener_, and listener_mutex_.

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

◆ get_liveliness_changed_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::get_liveliness_changed_status ( DDS::LivelinessChangedStatus & status )

virtual

Definition at line 963 of file DataReaderImpl.cpp.

References DDS::LivelinessChangedStatus::alive_count_change, DDS::LIVELINESS_CHANGED_STATUS, liveliness_changed_status_, DDS::LivelinessChangedStatus::not_alive_count_change, DDS::RETCODE_OK, sample_lock_, and OpenDDS::DCPS::EntityImpl::set_status_changed_flag().

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> justMe(this->sample_lock_);
 
   set_status_changed_flag(DDS::LIVELINESS_CHANGED_STATUS,
       false);
   status = liveliness_changed_status_;
 
   liveliness_changed_status_.alive_count_change = 0;
   liveliness_changed_status_.not_alive_count_change = 0;
 
   return DDS::RETCODE_OK;
 }

◆ get_matched_publication_data()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::get_matched_publication_data	(	DDS::PublicationBuiltinTopicData &	publication_data,
		DDS::InstanceHandle_t	publication_handle
	)

virtual

Definition at line 1082 of file DataReaderImpl.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), OpenDDS::DCPS::BUILT_IN_PUBLICATION_TOPIC, OpenDDS::DCPS::EntityImpl::enabled_, OpenDDS::DCPS::RcHandle< T >::in(), LM_ERROR, participant_servant_, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, and DDS::RETCODE_OK.

 {
   if (!enabled_) {
     ACE_ERROR_RETURN((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::")
         ACE_TEXT("get_matched_publication_data: ")
         ACE_TEXT("Entity is not enabled.\n")),
         DDS::RETCODE_NOT_ENABLED);
   }
 
   RcHandle<DomainParticipantImpl> participant = this->participant_servant_.lock();
 
   if (!participant)
     return DDS::RETCODE_ERROR;
 
   DDS::PublicationBuiltinTopicDataSeq data;
   const DDS::ReturnCode_t ret = instance_handle_to_bit_data<DDS::PublicationBuiltinTopicDataDataReader_var>(
                                   participant.in(),
                                   BUILT_IN_PUBLICATION_TOPIC,
                                   publication_handle,
                                   data);
 
   if (ret == DDS::RETCODE_OK) {
     publication_data = data[0];
   }
 
   return ret;
 }

◆ get_matched_publications()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::get_matched_publications ( DDS::InstanceHandleSeq & publication_handles )

virtual

Definition at line 1051 of file DataReaderImpl.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::EntityImpl::enabled_, LM_ERROR, publication_handle_lock_, publication_id_to_handle_map_, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, and DDS::RETCODE_OK.

 {
   if (!enabled_) {
     ACE_ERROR_RETURN((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::get_matched_publications: ")
         ACE_TEXT(" Entity is not enabled.\n")),
         DDS::RETCODE_NOT_ENABLED);
   }
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       guard,
       publication_handle_lock_,
       DDS::RETCODE_ERROR);
 
   // Copy out the handles for the current set of publications.
   int index = 0;
   publication_handles.length(static_cast<CORBA::ULong>(this->publication_id_to_handle_map_.size()));
 
   for (RepoIdToHandleMap::iterator
       current = this->publication_id_to_handle_map_.begin();
       current != this->publication_id_to_handle_map_.end();
       ++current, ++index) {
     publication_handles[index] = current->second;
   }
 
   return DDS::RETCODE_OK;
 }

◆ get_n_chunks()

size_t OpenDDS::DCPS::DataReaderImpl::get_n_chunks ( ) const

inline

Definition at line 414 of file DataReaderImpl.h.

References OpenDDS::DCPS::OPENDDS_VECTOR().

   {
     return n_chunks_;
   }

◆ get_next_handle()

DDS::InstanceHandle_t OpenDDS::DCPS::DataReaderImpl::get_next_handle ( const DDS::BuiltinTopicKey_t & key )

protected

Get an instance handle for a new instance.

Definition at line 2442 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::bit_key_to_guid(), and DDS::HANDLE_NIL.

 {
   RcHandle<DomainParticipantImpl> participant = this->participant_servant_.lock();
   if (!participant)
     return DDS::HANDLE_NIL;
 
   if (is_bit()) {
     const GUID_t id = bit_key_to_guid(key);
     return participant->assign_handle(id);
 
   } else {
     return participant->assign_handle();
   }
 }

◆ get_ordered_data()

void OpenDDS::DCPS::DataReaderImpl::get_ordered_data	(	GroupRakeData &	data,
		DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states
	)

Definition at line 3084 of file DataReaderImpl.cpp.

References ACE_GUARD, OpenDDS::DCPS::ReceivedDataElementList::get_next_match(), OpenDDS::DCPS::GroupRakeData::insert_sample(), OpenDDS::DCPS::SubscriptionInstance::instance_state_, OpenDDS::DCPS::InstanceState::match(), and OpenDDS::DCPS::SubscriptionInstance::rcvd_samples_.

 {
   SubscriptionInstanceSet localsubs;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, instances_lock_);
     for (SubscriptionInstanceMapType::iterator iter = instances_.begin();
          iter != instances_.end(); ++iter) {
       localsubs.insert(iter->second);
     }
   }
 
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
 
   for (SubscriptionInstanceSet::iterator iter = localsubs.begin(); iter != localsubs.end(); ++iter) {
     const SubscriptionInstance_rch inst = *iter;
     if (inst->instance_state_->match(view_states, instance_states)) {
       size_t i(0);
       for (ReceivedDataElement* item = inst->rcvd_samples_.get_next_match(sample_states, 0);
            item; item = inst->rcvd_samples_.get_next_match(sample_states, item)) {
           data.insert_sample(item, &inst->rcvd_samples_, *iter, ++i);
           group_coherent_ordered_data_.insert_sample(item, &inst->rcvd_samples_, *iter, ++i);
       }
     }
   }
 }

◆ get_priority_value()

Priority OpenDDS::DCPS::DataReaderImpl::get_priority_value ( const AssociationData & data ) const

inlineprivatevirtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 849 of file DataReaderImpl.h.

References OpenDDS::DCPS::OPENDDS_MAP(), and OpenDDS::DCPS::AssociationData::publication_transport_priority_.

                                                                  {
     return data.publication_transport_priority_;
   }

◆ get_qos()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::get_qos ( DDS::DataReaderQos & qos )

virtual

Definition at line 902 of file DataReaderImpl.cpp.

References passed_qos_, and DDS::RETCODE_OK.

Referenced by OpenDDS::DCPS::StaticDiscovery::pre_reader().

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

◆ get_reactor()

ACE_Reactor_Timer_Interface * OpenDDS::DCPS::DataReaderImpl::get_reactor ( void )

Definition at line 2804 of file DataReaderImpl.cpp.

 {
   return this->reactor_;
 }

◆ get_requested_deadline_missed_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::get_requested_deadline_missed_status ( DDS::RequestedDeadlineMissedStatus & status )

virtual

Definition at line 979 of file DataReaderImpl.cpp.

References last_deadline_missed_total_count_, DDS::REQUESTED_DEADLINE_MISSED_STATUS, requested_deadline_missed_status_, DDS::RETCODE_OK, sample_lock_, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), DDS::RequestedDeadlineMissedStatus::total_count, and DDS::RequestedDeadlineMissedStatus::total_count_change.

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> justMe(this->sample_lock_);
 
   set_status_changed_flag(DDS::REQUESTED_DEADLINE_MISSED_STATUS,
       false);
 
   this->requested_deadline_missed_status_.total_count_change =
       this->requested_deadline_missed_status_.total_count
       - this->last_deadline_missed_total_count_;
 
   // DDS::RequestedDeadlineMissedStatus::last_instance_handle field
   // is updated by the RequestedDeadlineWatchdog.
 
   // Update for next status check.
   this->last_deadline_missed_total_count_ =
       this->requested_deadline_missed_status_.total_count;
 
   status = requested_deadline_missed_status_;
 
   return DDS::RETCODE_OK;
 }

◆ get_requested_incompatible_qos_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::get_requested_incompatible_qos_status ( DDS::RequestedIncompatibleQosStatus & status )

virtual

Definition at line 1004 of file DataReaderImpl.cpp.

References publication_handle_lock_, DDS::REQUESTED_INCOMPATIBLE_QOS_STATUS, requested_incompatible_qos_status_, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), and DDS::RequestedIncompatibleQosStatus::total_count_change.

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> justMe(publication_handle_lock_);
 
   set_status_changed_flag(DDS::REQUESTED_INCOMPATIBLE_QOS_STATUS, false);
   status = requested_incompatible_qos_status_;
   requested_incompatible_qos_status_.total_count_change = 0;
 
   return DDS::RETCODE_OK;
 }

◆ get_sample_lost_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::get_sample_lost_status ( DDS::SampleLostStatus & status )

virtual

Definition at line 1031 of file DataReaderImpl.cpp.

References DDS::RETCODE_OK, sample_lock_, DDS::SAMPLE_LOST_STATUS, sample_lost_status_, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), and DDS::SampleLostStatus::total_count_change.

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> justMe(this->sample_lock_);
 
   set_status_changed_flag(DDS::SAMPLE_LOST_STATUS, false);
   status = sample_lost_status_;
   sample_lost_status_.total_count_change = 0;
   return DDS::RETCODE_OK;
 }

◆ get_sample_rejected_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::get_sample_rejected_status ( DDS::SampleRejectedStatus & status )

virtual

Definition at line 951 of file DataReaderImpl.cpp.

References DDS::RETCODE_OK, sample_lock_, DDS::SAMPLE_REJECTED_STATUS, sample_rejected_status_, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), and DDS::SampleRejectedStatus::total_count_change.

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> justMe(this->sample_lock_);
 
   set_status_changed_flag(DDS::SAMPLE_REJECTED_STATUS, false);
   status = sample_rejected_status_;
   sample_rejected_status_.total_count_change = 0;
   return DDS::RETCODE_OK;
 }

◆ get_subscriber()

DDS::Subscriber_ptr OpenDDS::DCPS::DataReaderImpl::get_subscriber ( )

virtual

Implements DDS::DataReader.

Definition at line 945 of file DataReaderImpl.cpp.

References get_subscriber_servant().

Referenced by OpenDDS::DCPS::StaticDiscovery::pre_reader(), and OpenDDS::DCPS::DRMonitorImpl::report().

 {
   return get_subscriber_servant()._retn();
 }

◆ get_subscriber_servant()

RcHandle< SubscriberImpl > OpenDDS::DCPS::DataReaderImpl::get_subscriber_servant ( )

protected

Definition at line 1712 of file DataReaderImpl.cpp.

References subscriber_servant_.

Referenced by data_received(), enable(), get_subscriber(), listener_for(), and set_qos().

 {
   return subscriber_servant_.lock();
 }

◆ get_subscription_matched_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::get_subscription_matched_status ( DDS::SubscriptionMatchedStatus & status )

virtual

Definition at line 1017 of file DataReaderImpl.cpp.

References DDS::SubscriptionMatchedStatus::current_count_change, publication_handle_lock_, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), subscription_match_status_, DDS::SUBSCRIPTION_MATCHED_STATUS, and DDS::SubscriptionMatchedStatus::total_count_change.

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> justMe(publication_handle_lock_);
 
   set_status_changed_flag(DDS::SUBSCRIPTION_MATCHED_STATUS, false);
   status = subscription_match_status_;
   subscription_match_status_.total_count_change = 0;
   subscription_match_status_.current_count_change = 0;
 
   return DDS::RETCODE_OK;
 }

◆ get_topic_id()

OpenDDS::DCPS::GUID_t OpenDDS::DCPS::DataReaderImpl::get_topic_id ( )

Definition at line 2810 of file DataReaderImpl.cpp.

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

 {
   return topic_id_;
 }

◆ get_topicdescription()

DDS::TopicDescription_ptr OpenDDS::DCPS::DataReaderImpl::get_topicdescription ( )

virtual

Implements DDS::DataReader.

Definition at line 932 of file DataReaderImpl.cpp.

References content_filtered_topic_, content_filtered_topic_mutex_, and topic_desc_.

Referenced by OpenDDS::DCPS::QueryConditionImpl::get_type_support(), and OpenDDS::DCPS::MultiTopicDataReader_T< Sample, TypedDataReader >::join().

 {
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
   {
     ACE_Guard<ACE_Thread_Mutex> guard(content_filtered_topic_mutex_);
     if (content_filtered_topic_) {
       return DDS::TopicDescription::_duplicate(content_filtered_topic_.get());
     }
   }
 #endif
   return DDS::TopicDescription::_duplicate(topic_desc_.in());
 }

◆ get_value_dispatcher()

const ValueDispatcher* OpenDDS::DCPS::DataReaderImpl::get_value_dispatcher ( ) const

inline

Definition at line 640 of file DataReaderImpl.h.

Referenced by data_received().

   {
     TopicDescriptionPtr<TopicImpl> temp(topic_servant_);
     return temp ? dynamic_cast<const ValueDispatcher*>(temp->get_type_support()) : 0;
   }

◆ get_writer_states()

void OpenDDS::DCPS::DataReaderImpl::get_writer_states ( WriterStatePairVec & writer_states )

Definition at line 2834 of file DataReaderImpl.cpp.

References ACE_READ_GUARD.

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

 {
   ACE_READ_GUARD(ACE_RW_Thread_Mutex,
       read_guard,
       this->writers_lock_);
   for (WriterMapType::iterator iter = writers_.begin();
       iter != writers_.end();
       ++iter) {
     writer_states.push_back(WriterStatePair(iter->first,
         iter->second->state()));
   }
 }

◆ has_readcondition()

bool OpenDDS::DCPS::DataReaderImpl::has_readcondition ( DDS::ReadCondition_ptr a_condition )

protected

Definition at line 799 of file DataReaderImpl.cpp.

References read_conditions_.

 {
   //sample lock already held
   DDS::ReadCondition_var rc = DDS::ReadCondition::_duplicate(a_condition);
   return read_conditions_.find(rc) != read_conditions_.end();
 }

◆ has_zero_copies()

bool OpenDDS::DCPS::DataReaderImpl::has_zero_copies ( )

This method is used for a precondition check of delete_datareader.

Return values

true	We have zero-copy samples loaned out
false	We have no zero-copy samples loaned out

Definition at line 2726 of file DataReaderImpl.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::ReceivedDataElementList::has_zero_copies(), and OpenDDS::DCPS::SubscriptionInstance::rcvd_samples_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       guard,
       this->sample_lock_,
       true /* assume we have loans */);
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, instance_guard, this->instances_lock_, true);
 
   for (SubscriptionInstanceMapType::iterator iter = instances_.begin();
       iter != instances_.end();
       ++iter) {
     SubscriptionInstance_rch ptr = iter->second;
 
     if (ptr->rcvd_samples_.has_zero_copies()) {
       return true;
     }
   }
 
   return false;
 }

◆ have_instance_states()

bool OpenDDS::DCPS::DataReaderImpl::have_instance_states ( DDS::InstanceStateMask instance_states ) const

!!caller should have acquired sample_lock_

Definition at line 1733 of file DataReaderImpl.cpp.

References DDS::ANY_SAMPLE_STATE, DDS::ANY_VIEW_STATE, instances_lock_, and lookup_matching_instances().

 {
   //!!!caller should have acquired sample_lock_
   ACE_Guard<ACE_Recursive_Thread_Mutex> instance_guard(instances_lock_);
   return lookup_matching_instances(DDS::ANY_SAMPLE_STATE, DDS::ANY_VIEW_STATE, instance_states).size();
 }

◆ have_sample_states()

bool OpenDDS::DCPS::DataReaderImpl::have_sample_states ( DDS::SampleStateMask sample_states ) const

!!caller should have acquired sample_lock_

Definition at line 1717 of file DataReaderImpl.cpp.

References DDS::ANY_INSTANCE_STATE, DDS::ANY_VIEW_STATE, instances_lock_, and lookup_matching_instances().

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

 {
   //!!!caller should have acquired sample_lock_
   ACE_Guard<ACE_Recursive_Thread_Mutex> instance_guard(instances_lock_);
   return lookup_matching_instances(sample_states, DDS::ANY_VIEW_STATE, DDS::ANY_INSTANCE_STATE).size();
 }

◆ have_view_states()

bool OpenDDS::DCPS::DataReaderImpl::have_view_states ( DDS::ViewStateMask view_states ) const

!!caller should have acquired sample_lock_

Definition at line 1726 of file DataReaderImpl.cpp.

References DDS::ANY_INSTANCE_STATE, DDS::ANY_SAMPLE_STATE, instances_lock_, and lookup_matching_instances().

 {
   //!!!caller should have acquired sample_lock_
   ACE_Guard<ACE_Recursive_Thread_Mutex> instance_guard(instances_lock_);
   return lookup_matching_instances(DDS::ANY_SAMPLE_STATE, view_states, DDS::ANY_INSTANCE_STATE).size();
 }

◆ init()

void OpenDDS::DCPS::DataReaderImpl::init	(	TopicDescriptionImpl *	a_topic_desc,
		const DDS::DataReaderQos &	qos,
		DDS::DataReaderListener_ptr	a_listener,
		const DDS::StatusMask &	mask,
		DomainParticipantImpl *	participant,
		SubscriberImpl *	subscriber
	)

Definition at line 184 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), domain_id_, DDS::EXCLUSIVE_OWNERSHIP_QOS, OpenDDS::DCPS::DomainParticipantImpl::get_domain_id(), OpenDDS::DCPS::TopicDescriptionImpl::get_name(), OpenDDS::DCPS::SubscriberImpl::get_qos(), OpenDDS::DCPS::TopicDescriptionImpl::get_type_name(), is_bit_, is_exclusive_ownership_, DDS::OwnershipQosPolicy::kind, LM_WARNING, DDS::DataReaderQos::ownership, participant_servant_, passed_qos_, qos_, OpenDDS::DCPS::rchandle_from(), DDS::RETCODE_OK, set_listener(), subscriber_servant_, topic_desc_, topic_id_, topic_servant_, OpenDDS::DCPS::topicIsBIT(), and type_support_.

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

 {
   topic_desc_ = DDS::TopicDescription::_duplicate(topic_desc);
   if (TopicImpl* topic = dynamic_cast<TopicImpl*>(topic_desc)) {
     topic_servant_ = topic;
     type_support_ = dynamic_cast<TypeSupportImpl*>(topic->get_type_support());
     topic_id_ = topic->get_id();
   }
 
 #ifndef DDS_HAS_MINIMUM_BIT
   CORBA::String_var topic_name = topic_desc->get_name();
   CORBA::String_var topic_type_name = topic_desc->get_type_name();
   is_bit_ = topicIsBIT(topic_name, topic_type_name);
 #endif // !defined (DDS_HAS_MINIMUM_BIT)
 
   qos_ = qos;
   passed_qos_ = qos;
 
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   is_exclusive_ownership_ = this->qos_.ownership.kind == ::DDS::EXCLUSIVE_OWNERSHIP_QOS;
 #endif
 
   set_listener(listener, mask);
 
   // Only store the participant pointer, since it is our "grand"
   // parent, we will exist as long as it does
   participant_servant_ = *participant;
 
   domain_id_ = participant->get_domain_id();
 
   subscriber_servant_ = rchandle_from(subscriber);
 
   if (subscriber->get_qos(this->subqos_) != ::DDS::RETCODE_OK) {
     ACE_DEBUG((LM_WARNING,
         ACE_TEXT("(%P|%t) WARNING: DataReaderImpl::init() - ")
         ACE_TEXT("failed to get SubscriberQos\n")));
   }
 }

◆ initialize_lookup_maps()

void OpenDDS::DCPS::DataReaderImpl::initialize_lookup_maps ( )

protected

Definition at line 3533 of file DataReaderImpl.cpp.

 {
   // These all start at 1 (0 mask is bogus) and include the full mask (any)
   for (CORBA::ULong is = 1; is <= MAX_SAMPLE_STATE_MASK; ++is) {
     for (CORBA::ULong iv = 1; iv <= MAX_VIEW_STATE_MASK; ++iv) {
       for (CORBA::ULong ii = 1; ii <= MAX_INSTANCE_STATE_MASK; ++ii) {
         combined_state_lookup_[to_combined_states(is, iv, ii)] = HandleSet();
       }
     }
   }
   // catch-all for "bogus" lookups
   combined_state_lookup_[0] = HandleSet();
 }

◆ install_type_support()

virtual void OpenDDS::DCPS::DataReaderImpl::install_type_support ( TypeSupportImpl * )

inlineprivatevirtual

Reimplemented in OpenDDS::XTypes::DynamicDataReaderImpl.

Definition at line 815 of file DataReaderImpl.h.

References timestamp().

Referenced by enable().

815 {}

◆ instances_liveliness_update()

void OpenDDS::DCPS::DataReaderImpl::instances_liveliness_update	(	const GUID_t &	writer,
		DDS::InstanceHandle_t	publication_handle
	)

private

Definition at line 2235 of file DataReaderImpl.cpp.

References ACE_GUARD, DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE, and OpenDDS::DCPS::TimePoint_T< SystemClock >::now().

 {
   // sample_lock_ must be held.
   InstanceSet localinsts;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, instances_lock_);
     if (instances_.size() == 0) {
       return;
     }
     for (SubscriptionInstanceMapType::iterator iter = instances_.begin();
          iter != instances_.end(); ++iter) {
       if (iter->second->instance_state_->writes_instance(writer)) {
         localinsts.insert(iter->first);
       }
     }
   }
 
   for (InstanceSet::iterator iter = localinsts.begin(); iter != localinsts.end(); ++iter) {
     set_instance_state_i(*iter, publication_handle, DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE, SystemTimePoint::now(), writer);
   }
 }

◆ is_bit()

bool OpenDDS::DCPS::DataReaderImpl::is_bit ( ) const

Definition at line 2720 of file DataReaderImpl.cpp.

 {
   return this->is_bit_;
 }

◆ listener_for()

DDS::DataReaderListener_ptr OpenDDS::DCPS::DataReaderImpl::listener_for ( DDS::StatusKind kind )

This is used to retrieve the listener for a certain status change. If this datareader has a registered listener and the status kind is in the listener mask then the listener is returned. Otherwise, the query for the listener is propagated up to the factory/subscriber.

Definition at line 1753 of file DataReaderImpl.cpp.

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

Referenced by remove_associations_i(), transport_assoc_done(), and update_incompatible_qos().

 {
   // 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<SubscriberImpl> subscriber = get_subscriber_servant();
   ACE_Guard<ACE_Thread_Mutex> g(listener_mutex_);
   if (subscriber && (CORBA::is_nil(listener_.in()) || (listener_mask_ & kind) == 0)) {
     g.release();
     return subscriber->listener_for(kind);
 
   } else {
     return DDS::DataReaderListener::_duplicate(listener_.in());
   }
 }

◆ liveliness_lost()

void OpenDDS::DCPS::DataReaderImpl::liveliness_lost ( )

◆ lookup_instance()

virtual void OpenDDS::DCPS::DataReaderImpl::lookup_instance	(	const ReceivedDataSample &	sample,
		SubscriptionInstance_rch &	instance
	)

pure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by data_received().

◆ lookup_instance_generic()

virtual DDS::InstanceHandle_t OpenDDS::DCPS::DataReaderImpl::lookup_instance_generic ( const void * data )

pure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by OpenDDS::DCPS::MultiTopicDataReader_T< Sample, TypedDataReader >::join().

◆ lookup_instance_handles()

void OpenDDS::DCPS::DataReaderImpl::lookup_instance_handles	(	const WriterIdSeq &	ids,
		DDS::InstanceHandleSeq &	hdls
	)

private

Lookup the instance handles by the publication repo ids.

Definition at line 2551 of file DataReaderImpl.cpp.

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

Referenced by remove_associations_i().

 {
   CORBA::ULong const num_wrts = ids.length();
 
   if (DCPS_debug_level > 9) {
     const char* separator = "";
     OPENDDS_STRING guids;
 
     for (CORBA::ULong i = 0; i < num_wrts; ++i) {
       guids += separator;
       guids += LogGuid(ids[i]).conv_;
       separator = ", ";
     }
 
     ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) DataReaderImpl::lookup_instance_handles: ")
         ACE_TEXT("searching for handles for writer Ids: %C.\n"),
         guids.c_str()));
   }
 
   hdls.length(num_wrts);
 
   RcHandle<DomainParticipantImpl> participant = this->participant_servant_.lock();
   if (participant) {
     for (CORBA::ULong i = 0; i < num_wrts; ++i) {
       hdls[i] = participant->lookup_handle(ids[i]);
     }
   }
 }

◆ lookup_matching_instances()

const DataReaderImpl::HandleSet & OpenDDS::DCPS::DataReaderImpl::lookup_matching_instances	(	CORBA::ULong	sample_states,
		CORBA::ULong	view_states,
		CORBA::ULong	instance_states
	)		const

protected

Definition at line 3569 of file DataReaderImpl.cpp.

References OPENDDS_ASSERT.

Referenced by contains_sample(), have_instance_states(), have_sample_states(), and have_view_states().

 {
   const CORBA::ULong combined_states = to_combined_states(sample_states, view_states, instance_states);
   LookupMap::const_iterator ci = combined_state_lookup_.find(combined_states);
   OPENDDS_ASSERT(ci != combined_state_lookup_.end());
   return ci->second;
 }

◆ notify_latency()

void OpenDDS::DCPS::DataReaderImpl::notify_latency ( GUID_t writer )

Definition at line 2337 of file DataReaderImpl.cpp.

References CORBA::is_nil().

 {
   // Narrow to DDS::DCPS::DataReaderListener. If a DDS::DataReaderListener
   // is given to this DataReader then narrow() fails.
   DataReaderListener_var listener = get_ext_listener();
 
   if (!CORBA::is_nil(listener.in())) {
     WriterIdSeq writerIds;
     writerIds.length(1);
     writerIds[ 0] = writer;
 
     DDS::InstanceHandleSeq handles;
     this->lookup_instance_handles(writerIds, handles);
 
     if (handles.length() >= 1) {
       this->budget_exceeded_status_.last_instance_handle = handles[ 0];
 
     } else {
       this->budget_exceeded_status_.last_instance_handle = -1;
     }
 
     ++this->budget_exceeded_status_.total_count;
     ++this->budget_exceeded_status_.total_count_change;
 
     listener->on_budget_exceeded(this, this->budget_exceeded_status_);
 
     this->budget_exceeded_status_.total_count_change = 0;
   }
 }

◆ notify_liveliness_change()

void OpenDDS::DCPS::DataReaderImpl::notify_liveliness_change ( )

Definition at line 2747 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), DDS::LivelinessChangedStatus::alive_count_change, OpenDDS::DCPS::LogGuid::conv_, OpenDDS::DCPS::DCPS_debug_level, CORBA::is_nil(), DDS::LIVELINESS_CHANGED_STATUS, LM_DEBUG, OPENDDS_STRING, and OpenDDS::DCPS::to_dds_string().

 {
   // sample_lock_ must be held.
   // N.B. writers_lock_ should already be acquired when
   //      this method is called.
 
   DDS::DataReaderListener_var listener
   = listener_for(DDS::LIVELINESS_CHANGED_STATUS);
 
   if (!CORBA::is_nil(listener.in())) {
     const DDS::LivelinessChangedStatus status = liveliness_changed_status_;
     liveliness_changed_status_.alive_count_change = 0;
     liveliness_changed_status_.not_alive_count_change = 0;
     ACE_GUARD(Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
     listener->on_liveliness_changed(this, status);
   }
   notify_status_condition();
 
   if (DCPS_debug_level > 9) {
     ACE_Guard<ACE_Thread_Mutex> g(listener_mutex_);
     OPENDDS_STRING output_str;
     output_str += "subscription ";
     output_str += LogGuid(get_guid()).conv_;
     output_str += ", listener at: 0x";
     output_str += to_dds_string(this->listener_.in());
 
     for (WriterMapType::iterator current = this->writers_.begin();
          current != this->writers_.end();
          ++current) {
       const GUID_t id = current->first;
       output_str += "\n\tNOTIFY: writer[ ";
       output_str += LogGuid(id).conv_;
       output_str += "] == ";
       output_str += current->second->get_state_str();
     }
 
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) DataReaderImpl::notify_liveliness_change: ")
                ACE_TEXT("listener at 0x%x, mask 0x%x.\n")
                ACE_TEXT("\tNOTIFY: %C\n"),
                listener.in(),
                listener_mask_,
                output_str.c_str()));
   }
 }

◆ notify_read_conditions()

void OpenDDS::DCPS::DataReaderImpl::notify_read_conditions ( )

protected

Data has arrived into the cache, unblock waiting ReadConditions.

Definition at line 1692 of file DataReaderImpl.cpp.

References ACE_ERROR, ACE_GUARD, ACE_TEXT(), LM_ERROR, read_conditions_, reverse_sample_lock_, and OpenDDS::DCPS::ConditionImpl::signal_all().

Referenced by data_received().

 {
   //sample lock is already held
   ReadConditionSet local_read_conditions = read_conditions_;
   ACE_GUARD(Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
 
   for (ReadConditionSet::iterator it = local_read_conditions.begin(),
       end = local_read_conditions.end(); it != end; ++it) {
     ConditionImpl* ci = dynamic_cast<ConditionImpl*>(it->in());
     if (ci) {
       ci->signal_all();
     } else {
       ACE_ERROR((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::notify_read_conditions: ")
         ACE_TEXT("Failed to obtain ConditionImpl - can't notify.\n")));
     }
   }
 }

◆ notify_subscription_disconnected()

void OpenDDS::DCPS::DataReaderImpl::notify_subscription_disconnected ( const WriterIdSeq & pubids )

virtual

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 2466 of file DataReaderImpl.cpp.

References DBG_ENTRY_LVL, CORBA::is_nil(), and OpenDDS::DCPS::SubscriptionLostStatus::publication_handles.

 {
   DBG_ENTRY_LVL("DataReaderImpl","notify_subscription_disconnected",6);
 
   // Narrow to DDS::DCPS::DataReaderListener. If a DDS::DataReaderListener
   // is given to this DataReader then narrow() fails.
   DataReaderListener_var the_listener = get_ext_listener();
 
   if (!CORBA::is_nil(the_listener.in())) {
     SubscriptionLostStatus status;
 
     // Since this callback may come after remove_association which removes
     // the writer from id_to_handle map, we can ignore this error.
     this->lookup_instance_handles(pubids, status.publication_handles);
     the_listener->on_subscription_disconnected(this, status);
   }
 }

◆ notify_subscription_lost() [1/2]

void OpenDDS::DCPS::DataReaderImpl::notify_subscription_lost ( const WriterIdSeq & pubids )

virtual

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 2531 of file DataReaderImpl.cpp.

References DBG_ENTRY_LVL, CORBA::is_nil(), and OpenDDS::DCPS::SubscriptionLostStatus::publication_handles.

Referenced by remove_associations_i().

 {
   DBG_ENTRY_LVL("DataReaderImpl","notify_subscription_lost",6);
 
   // Narrow to DDS::DCPS::DataReaderListener. If a DDS::DataReaderListener
   // is given to this DataReader then narrow() fails.
   DataReaderListener_var the_listener = get_ext_listener();
 
   if (!CORBA::is_nil(the_listener.in())) {
     SubscriptionLostStatus status;
 
     // Since this callback may come after remove_association which removes
     // the writer from id_to_handle map, we can ignore this error.
     this->lookup_instance_handles(pubids, status.publication_handles);
     the_listener->on_subscription_lost(this, status);
   }
 }

◆ notify_subscription_lost() [2/2]

void OpenDDS::DCPS::DataReaderImpl::notify_subscription_lost ( const DDS::InstanceHandleSeq & handles )

private

Definition at line 2506 of file DataReaderImpl.cpp.

References DBG_ENTRY_LVL, CORBA::is_nil(), and OpenDDS::DCPS::SubscriptionLostStatus::publication_handles.

 {
   DBG_ENTRY_LVL("DataReaderImpl","notify_subscription_lost",6);
 
   if (!this->is_bit_) {
     // Narrow to DDS::DCPS::DataReaderListener. If a DDS::DataReaderListener
     // is given to this DataReader then narrow() fails.
     DataReaderListener_var the_listener = get_ext_listener();
 
     if (!CORBA::is_nil(the_listener.in())) {
       SubscriptionLostStatus status;
 
       CORBA::ULong len = handles.length();
       status.publication_handles.length(len);
 
       for (CORBA::ULong i = 0; i < len; ++ i) {
         status.publication_handles[i] = handles[i];
       }
 
       the_listener->on_subscription_lost(this, status);
     }
   }
 }

◆ notify_subscription_reconnected()

void OpenDDS::DCPS::DataReaderImpl::notify_subscription_reconnected ( const WriterIdSeq & pubids )

virtual

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 2485 of file DataReaderImpl.cpp.

References DBG_ENTRY_LVL, CORBA::is_nil(), and OpenDDS::DCPS::SubscriptionLostStatus::publication_handles.

 {
   DBG_ENTRY_LVL("DataReaderImpl","notify_subscription_reconnected",6);
 
   if (!this->is_bit_) {
     // Narrow to DDS::DCPS::DataReaderListener. If a DDS::DataReaderListener
     // is given to this DataReader then narrow() fails.
     DataReaderListener_var the_listener = get_ext_listener();
 
     if (!CORBA::is_nil(the_listener.in())) {
       SubscriptionLostStatus status;
 
       // If it's reconnected then the reader should be in id_to_handle
       this->lookup_instance_handles(pubids, status.publication_handles);
 
       the_listener->on_subscription_reconnected(this,  status);
     }
   }
 }

◆ OPENDDS_MAP() [1/2]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_MAP	(	DDS::InstanceHandle_t	,
		SubscriptionInstance_rch
	)

◆ OPENDDS_MAP() [2/2]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_MAP	(	CORBA::ULong	,
		HandleSet
	)

protected

◆ OPENDDS_MAP_CMP() [1/3]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_MAP_CMP	(	GUID_t	,
		WriterStats	,
		GUID_tKeyLessThan
	)

Type of collection of statistics for writers to this reader.

◆ OPENDDS_MAP_CMP() [2/3]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_MAP_CMP	(	GUID_t	,
		DDS::InstanceHandle_t	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MAP_CMP() [3/3]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_MAP_CMP	(	GUID_t	,
		WriterInfo_rch	,
		GUID_tKeyLessThan
	)

private

publications writing to this reader.

◆ OPENDDS_MULTIMAP()

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_MULTIMAP	(	MonotonicTimePoint	,
		SubscriptionInstance_rch
	)

private

◆ OPENDDS_SET() [1/4]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_SET ( DDS::InstanceHandle_t )

◆ OPENDDS_SET() [2/4]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_SET ( SubscriptionInstance_rch )

◆ OPENDDS_SET() [3/4]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_SET ( DDS::InstanceHandle_t )

protected

◆ OPENDDS_SET() [4/4]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_SET ( Encoding::Kind )

protected

◆ OPENDDS_SET_CMP()

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_SET_CMP	(	DDS::ReadCondition_var	,
		RCCompLess
	)

private

◆ OPENDDS_VECTOR() [1/3]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_VECTOR ( DDS::InstanceHandle_t )

Referenced by signal_liveliness().

◆ OPENDDS_VECTOR() [2/3]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_VECTOR ( WriterStatePair )

◆ OPENDDS_VECTOR() [3/3]

typedef OpenDDS::DCPS::DataReaderImpl::OPENDDS_VECTOR ( void * )

◆ ownership_filter_instance()

bool OpenDDS::DCPS::DataReaderImpl::ownership_filter_instance	(	const SubscriptionInstance_rch &	instance,
		const GUID_t &	pubid
	)

protected

Definition at line 2630 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), ACE_WRITE_GUARD_RETURN, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::InstanceState::get_owner(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::SubscriptionInstance::instance_handle_, OpenDDS::DCPS::SubscriptionInstance::instance_state_, LM_DEBUG, and OpenDDS::DCPS::OwnershipManager::select_owner().

 {
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   if (this->is_exclusive_ownership_) {
 
     ACE_WRITE_GUARD_RETURN(ACE_RW_Thread_Mutex, write_guard, writers_lock_, true);
     WriterMapType::iterator iter = writers_.find(pubid);
 
     if (iter == writers_.end()) {
       if (DCPS_debug_level > 4) {
         // This may not be an error since it could happen that the sample
         // is delivered to the datareader after the write is dis-associated
         // with this datareader.
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) DataReaderImpl::ownership_filter_instance: ")
                    ACE_TEXT("reader %C is not associated with writer %C.\n"),
                    LogGuid(get_guid()).c_str(),
                    LogGuid(pubid).c_str()));
       }
       return true;
     }
 
 
     // Evaulate the owner of the instance if not selected and filter
     // current message if it's not from owner writer.
     if ( instance->instance_state_->get_owner() == GUID_UNKNOWN
         || ! iter->second->is_owner_evaluated(instance->instance_handle_)) {
       OwnershipManagerPtr owner_manager = this->ownership_manager();
 
       bool is_owner = owner_manager && owner_manager->select_owner (
         instance->instance_handle_,
         iter->second->writer_id(),
         iter->second->writer_qos_ownership_strength(),
         instance->instance_state_);
       iter->second->set_owner_evaluated(instance->instance_handle_, true);
 
       if (! is_owner) {
         if (DCPS_debug_level >= 1) {
           ACE_DEBUG((LM_DEBUG,
                      ACE_TEXT("(%P|%t) DataReaderImpl::ownership_filter_instance: ")
                      ACE_TEXT("reader %C writer %C is not elected as owner %C\n"),
                      LogGuid(get_guid()).c_str(),
                      LogGuid(pubid).c_str(),
                      LogGuid(instance->instance_state_->get_owner()).c_str()));
         }
         return true;
       }
     }
     else if (! (instance->instance_state_->get_owner() == pubid)) {
       if (DCPS_debug_level >= 1) {
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) DataReaderImpl::ownership_filter_instance: ")
                    ACE_TEXT("reader %C writer %C is not owner %C\n"),
                    LogGuid(get_guid()).c_str(),
                    LogGuid(pubid).c_str(),
                    LogGuid(instance->instance_state_->get_owner()).c_str()));
       }
       return true;
     }
   }
 #else
   ACE_UNUSED_ARG(pubid);
   ACE_UNUSED_ARG(instance);
 #endif
   return false;
 }

◆ ownership_manager()

OwnershipManagerPtr OpenDDS::DCPS::DataReaderImpl::ownership_manager ( )

inline

Definition at line 511 of file DataReaderImpl.h.

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

Referenced by cleanup(), data_received(), and release_instance().

511 { return OwnershipManagerPtr(this); }

OpenDDS::DCPS::DataReaderImpl::OwnershipManagerPtr

friend class OwnershipManagerPtr

Definition: DataReaderImpl.h:481

◆ parent()

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

virtual

Reimplemented from OpenDDS::DCPS::EntityImpl.

Definition at line 1678 of file DataReaderImpl.cpp.

References subscriber_servant_.

 {
   return subscriber_servant_.lock();
 }

◆ post_read_or_take()

void OpenDDS::DCPS::DataReaderImpl::post_read_or_take ( )

protected

Definition at line 2793 of file DataReaderImpl.cpp.

References DDS::DATA_AVAILABLE_STATUS, and DDS::DATA_ON_READERS_STATUS.

 {
   set_status_changed_flag(DDS::DATA_AVAILABLE_STATUS, false);
   RcHandle<SubscriberImpl> subscriber = get_subscriber_servant();
   if (subscriber) {
     subscriber->set_status_changed_flag(
       DDS::DATA_ON_READERS_STATUS, false);
   }
 }

◆ prepare_to_delete()

void OpenDDS::DCPS::DataReaderImpl::prepare_to_delete ( )

protected

Definition at line 2410 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::Observer::e_DELETED, and OpenDDS::DCPS::Observer::on_deleted().

 {
   const Observer_rch observer = get_observer(Observer::e_DELETED);
   if (observer) {
     observer->on_deleted(this);
   }
 
   this->set_deleted(true);
   this->stop_associating();
   this->send_final_acks();
   subscription_id_condition_.notify_all();
 }

◆ process_deadline()

void OpenDDS::DCPS::DataReaderImpl::process_deadline	(	SubscriptionInstance_rch	instance,
		const MonotonicTimePoint &	now,
		bool	timer_called
	)

private

Definition at line 3611 of file DataReaderImpl.cpp.

References ACE_GUARD, OpenDDS::DCPS::SubscriptionInstance::cur_sample_tv_, OpenDDS::DCPS::SubscriptionInstance::deadline_, OpenDDS::DCPS::SubscriptionInstance::instance_handle_, OpenDDS::DCPS::SubscriptionInstance::instance_state_, OpenDDS::DCPS::InstanceState::is_exclusive(), CORBA::is_nil(), OpenDDS::DCPS::TimePoint_T< AceClock >::is_zero(), OpenDDS::DCPS::SubscriptionInstance::last_sample_tv_, OpenDDS::DCPS::OwnershipManager::remove_writers(), DDS::REQUESTED_DEADLINE_MISSED_STATUS, and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::zero_value.

 {
   // Should be called with sample_lock_.
 
   if (instance->deadline_ != MonotonicTimePoint::zero_value) {
     bool missed = false;
 
     if (instance->cur_sample_tv_.is_zero()) { // not received any sample.
       missed = true;
 
     } else if (timer_called) { // handle_timeout is called
       missed = (now - instance->cur_sample_tv_) >= deadline_period_;
 
     } else { // upon receiving sample.
       missed = (instance->cur_sample_tv_ - instance->last_sample_tv_) > deadline_period_;
     }
 
     if (missed) {
       ACE_GUARD(ACE_Recursive_Thread_Mutex, monitor, sample_lock_);
       // Only update the status upon timer is called and not
       // when receiving a sample after the interval.
       // Otherwise the counter is doubled.
       if (timer_called) {
         ++requested_deadline_missed_status_.total_count;
         requested_deadline_missed_status_.total_count_change =
           requested_deadline_missed_status_.total_count - last_deadline_missed_total_count_;
         requested_deadline_missed_status_.last_instance_handle = instance->instance_handle_;
 
         set_status_changed_flag(DDS::REQUESTED_DEADLINE_MISSED_STATUS, true);
 
         DDS::DataReaderListener_var listener = listener_for(DDS::REQUESTED_DEADLINE_MISSED_STATUS);
 
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
         if (instance->instance_state_->is_exclusive()) {
           DataReaderImpl::OwnershipManagerPtr owner_manager = ownership_manager();
           if (owner_manager)
             owner_manager->remove_writers (instance->instance_handle_);
         }
 #endif
 
         if (!CORBA::is_nil(listener.in())) {
           // Copy before releasing the lock.
           DDS::RequestedDeadlineMissedStatus const status = requested_deadline_missed_status_;
 
           // Release the lock during the upcall.
           ACE_GUARD(Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
           // @todo Will this operation ever throw?  If so we may want to
           //       catch all exceptions, and act accordingly.
           listener->on_requested_deadline_missed(this, status);
 
           // We need to update the last total count value to our current total
           // so that the next time we will calculate the correct total_count_change;
           last_deadline_missed_total_count_ = requested_deadline_missed_status_.total_count;
         }
 
         notify_status_condition();
       }
     }
 
     // This next part is without status_lock_ held to avoid reactor deadlock.
     if (timer_called) {
       instance->deadline_ = MonotonicTimePoint::zero_value;
       schedule_deadline(instance, timer_called);
     } else {
       cancel_deadline(instance);
       schedule_deadline(instance, timer_called);
     }
   }
 }

◆ process_latency()

void OpenDDS::DCPS::DataReaderImpl::process_latency ( const ReceivedDataSample & sample )

NB: This message is generated contemporaneously with a similar message from writer_activity(). That message is not marked as an error, so we follow that lead and leave this as an informational message, guarded by debug level. This seems to be due to late samples (samples delivered after an association has been torn down). We may want to promote this to a warning if other conditions causing this symptom are discovered.

Definition at line 2284 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::DCPS_debug_level, DDS::DURATION_ZERO_NSEC, DDS::DURATION_ZERO_SEC, OpenDDS::DCPS::ReceivedDataSample::header_, LM_DEBUG, OpenDDS::DCPS::TimePoint_T< SystemClock >::now(), OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::DataSampleHeader::source_timestamp_nanosec_, OpenDDS::DCPS::DataSampleHeader::source_timestamp_sec_, OpenDDS::DCPS::TimeDuration::str(), and timestamp().

Referenced by data_received().

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(statistics_lock_);
   StatsMapType::iterator location = this->statistics_.find(sample.header_.publication_id_);
 
   if (location != this->statistics_.end()) {
     const DDS::Duration_t zero = { DDS::DURATION_ZERO_SEC, DDS::DURATION_ZERO_NSEC };
 
     // Only when the user has specified a latency budget or statistics
     // are enabled we need to calculate our latency
     if ((this->statistics_enabled()) ||
         (this->qos_.latency_budget.duration > zero)) {
       const DDS::Time_t timestamp = {
         sample.header_.source_timestamp_sec_,
         sample.header_.source_timestamp_nanosec_
       };
       const TimeDuration latency = SystemTimePoint::now() - SystemTimePoint(timestamp);
 
       if (this->statistics_enabled()) {
         location->second.add_stat(latency);
       }
 
       if (DCPS_debug_level > 9) {
         ACE_DEBUG((LM_DEBUG,
             ACE_TEXT("(%P|%t) DataReaderImpl::process_latency() - ")
             ACE_TEXT("measured latency of %C for current sample.\n"),
             latency.str().c_str()));
       }
 
       if (this->qos_.latency_budget.duration > zero) {
         // Check latency against the budget.
         if (latency > TimeDuration(this->qos_.latency_budget.duration)) {
           this->notify_latency(sample.header_.publication_id_);
         }
       }
     }
   } else if (DCPS_debug_level > 0) {
     /// NB: This message is generated contemporaneously with a similar
     ///     message from writer_activity().  That message is not marked
     ///     as an error, so we follow that lead and leave this as an
     ///     informational message, guarded by debug level.  This seems
     ///     to be due to late samples (samples delivered after an
     ///     association has been torn down).  We may want to promote this
     ///     to a warning if other conditions causing this symptom are
     ///     discovered.
     ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) DataReaderImpl::process_latency() - ")
         ACE_TEXT("reader %C is not associated with writer %C (late sample?).\n"),
         LogGuid(get_guid()).c_str(),
         LogGuid(sample.header_.publication_id_).c_str()));
   }
 }

◆ purge_data()

virtual void OpenDDS::DCPS::DataReaderImpl::purge_data ( SubscriptionInstance_rch instance )

protectedpure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by release_instance().

◆ qos_change()

void OpenDDS::DCPS::DataReaderImpl::qos_change ( const DDS::DataReaderQos & qos )

protectedvirtual

Reimplemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Definition at line 882 of file DataReaderImpl.cpp.

References cancel_all_deadlines(), DDS::DataReaderQos::deadline, deadline_period_, deadline_queue_enabled_, DDS::DURATION_INFINITE_NSEC, DDS::DURATION_INFINITE_SEC, DDS::Duration_t::nanosec, DDS::DeadlineQosPolicy::period, qos_, reset_deadline_period(), and DDS::Duration_t::sec.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::qos_change(), and set_qos().

 {
   // Reset the deadline timer if the period has changed.
   if (qos_.deadline.period.sec != qos.deadline.period.sec ||
       qos_.deadline.period.nanosec != qos.deadline.period.nanosec) {
     if (qos_.deadline.period.sec == DDS::DURATION_INFINITE_SEC &&
         qos_.deadline.period.nanosec == DDS::DURATION_INFINITE_NSEC) {
       deadline_period_ = TimeDuration(qos.deadline.period);
       deadline_queue_enabled_ = true;
     } else if (qos.deadline.period.sec == DDS::DURATION_INFINITE_SEC &&
                qos.deadline.period.nanosec == DDS::DURATION_INFINITE_NSEC) {
       cancel_all_deadlines();
       deadline_queue_enabled_ = false;
     } else {
       reset_deadline_period(TimeDuration(qos.deadline.period));
     }
   }
 }

◆ raw_latency_buffer_size()

ACE_INLINE unsigned int & OpenDDS::DCPS::DataReaderImpl::raw_latency_buffer_size ( )

Configure the size of the raw data collection buffer.

Definition at line 19 of file DataReaderImpl.inl.

References ACE_INLINE, and raw_latency_buffer_size_.

Referenced by OpenDDS::DCPS::SubscriberImpl::create_datareader(), and OpenDDS::DCPS::MultiTopicDataReaderBase::init().

 {
   return this->raw_latency_buffer_size_;
 }

◆ raw_latency_buffer_type()

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

Configure the type of the raw data collection buffer.

Definition at line 26 of file DataReaderImpl.inl.

References ACE_INLINE, and raw_latency_buffer_type_.

Referenced by OpenDDS::DCPS::SubscriberImpl::create_datareader(), and OpenDDS::DCPS::MultiTopicDataReaderBase::init().

 {
   return this->raw_latency_buffer_type_;
 }

◆ raw_latency_statistics()

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL ACE_INLINE const OpenDDS::DCPS::DataReaderImpl::StatsMapType & OpenDDS::DCPS::DataReaderImpl::raw_latency_statistics ( ) const

Expose the statistics container.

Definition at line 12 of file DataReaderImpl.inl.

References ACE_INLINE, and statistics_.

 {
   return this->statistics_;
 }

◆ read_generic()

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::read_generic	(	GenericBundle &	gen,
		DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states,
		bool	adjust_ref_count
	)

pure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >, OpenDDS::DCPS::DataReaderImpl_T< MessageType >, OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >, OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

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

◆ read_instance_generic()

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::read_instance_generic	(	void *&	data,
		DDS::SampleInfo &	info,
		DDS::InstanceHandle_t	instance,
		DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states
	)

pure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by OpenDDS::DCPS::MultiTopicDataReader_T< Sample, TypedDataReader >::join().

◆ read_next_instance_generic()

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::read_next_instance_generic	(	void *&	data,
		DDS::SampleInfo &	info,
		DDS::InstanceHandle_t	previous_instance,
		DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states
	)

pure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by OpenDDS::DCPS::MultiTopicDataReader_T< Sample, TypedDataReader >::join().

◆ register_for_writer()

void OpenDDS::DCPS::DataReaderImpl::register_for_writer	(	const GUID_t &	participant,
		const GUID_t &	readerid,
		const GUID_t &	writerid,
		const TransportLocatorSeq &	locators,
		DiscoveryListener *	listener
	)

virtual

Reimplemented from OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 3254 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::TransportClient::register_for_writer().

 {
   TransportClient::register_for_writer(participant, readerid, writerid, locators, listener);
 }

◆ reject_coherent()

void OpenDDS::DCPS::DataReaderImpl::reject_coherent	(	const GUID_t &	writer_id,
		const GUID_t &	publisher_id
	)

Definition at line 2941 of file DataReaderImpl.cpp.

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

 {
   if (::OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) DataReaderImpl::reject_coherent()")
         ACE_TEXT(" reader %C writer %C publisher %C\n"),
         LogGuid(get_guid()).c_str(),
         LogGuid(writer_id).c_str(),
         LogGuid(publisher_id).c_str()));
   }
 
   SubscriptionInstanceSet localsubs;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, this->instances_lock_);
     for (SubscriptionInstanceMapType::iterator iter = this->instances_.begin();
          iter != this->instances_.end(); ++iter) {
       localsubs.insert(iter->second);
     }
   }
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
   for (SubscriptionInstanceSet::iterator iter = localsubs.begin();
        iter != localsubs.end(); iter++) {
     (*iter)->rcvd_strategy_->reject_coherent(writer_id, publisher_id);
   }
   this->reset_coherent_info(writer_id, publisher_id);
 }

◆ release_all_instances()

virtual void OpenDDS::DCPS::DataReaderImpl::release_all_instances ( )

pure virtual

Release all instances held by the reader.

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

◆ release_instance()

void OpenDDS::DCPS::DataReaderImpl::release_instance ( DDS::InstanceHandle_t handle )

Release the instance with the handle.

Definition at line 1944 of file DataReaderImpl.cpp.

References ACE_ERROR, ACE_GUARD, get_handle_instance(), instances_, instances_lock_, LM_ERROR, monitor_, ownership_manager(), purge_data(), release_instance_i(), OpenDDS::DCPS::OwnershipManager::remove_writers(), and sample_lock_.

 {
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   OwnershipManagerPtr owner_manager = this->ownership_manager();
   if (owner_manager) {
     owner_manager->remove_writers(handle);
   }
 #endif
 
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_);
   SubscriptionInstance_rch instance = this->get_handle_instance(handle);
 
   if (!instance) {
     ACE_ERROR((LM_ERROR, "(%P|%t) DataReaderImpl::release_instance "
         "could not find the instance by handle 0x%x\n", handle));
     return;
   }
 
   this->purge_data(instance);
 
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, instances_lock_);
     instances_.erase(handle);
   }
 
   this->release_instance_i(handle);
   if (this->monitor_) {
     this->monitor_->report();
   }
 }

◆ release_instance_i()

virtual void OpenDDS::DCPS::DataReaderImpl::release_instance_i ( DDS::InstanceHandle_t handle )

protectedpure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by release_instance().

◆ remove_all_associations()

void OpenDDS::DCPS::DataReaderImpl::remove_all_associations ( )

Definition at line 640 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_READ_GUARD, ACE_TEXT(), DBG_ENTRY_LVL, LM_WARNING, remove_associations(), OpenDDS::DCPS::TransportClient::stop_associating(), OpenDDS::DCPS::TransportClient::transport_stop(), writers_, and writers_lock_.

 {
   DBG_ENTRY_LVL("DataReaderImpl","remove_all_associations",6);
   stop_associating();
 
   OpenDDS::DCPS::WriterIdSeq writers;
   int size;
 
   {
     ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);
 
     size = static_cast<int>(writers_.size());
     writers.length(size);
 
     WriterMapType::iterator curr_writer = writers_.begin();
     WriterMapType::iterator end_writer = writers_.end();
 
     int i = 0;
 
     while (curr_writer != end_writer) {
       writers[i++] = curr_writer->first;
       ++curr_writer;
     }
   }
 
   try {
     if (0 < size) {
       remove_associations(writers, false);
     }
   } catch (const CORBA::Exception&) {
     ACE_DEBUG((LM_WARNING,
                ACE_TEXT("(%P|%t) WARNING: DataReaderImpl::remove_all_associations() - ")
                ACE_TEXT("caught exception from remove_associations.\n")));
   }
 
   transport_stop();
 }

◆ remove_associations()

void OpenDDS::DCPS::DataReaderImpl::remove_associations	(	const WriterIdSeq &	writers,
		bool	callback
	)

virtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 452 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), ACE_WRITE_GUARD, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::Observer::e_DISASSOCIATED, OpenDDS::DCPS::EntityImpl::get_deleted(), get_guid(), OpenDDS::DCPS::EntityImpl::get_observer(), is_bit_, LM_DEBUG, OpenDDS::DCPS::Observer::on_disassociated(), remove_associations_i(), statistics_, statistics_lock_, OpenDDS::DCPS::TransportClient::stop_associating(), and writers_lock_.

Referenced by remove_all_associations().

 {
   DBG_ENTRY_LVL("DataReaderImpl", "remove_associations", 6);
 
   if (writers.length() == 0) {
     return;
   }
 
   const Observer_rch observer = get_observer(Observer::e_DISASSOCIATED);
   if (observer) {
     for (CORBA::ULong i = 0; i < writers.length(); ++i) {
       observer->on_disassociated(this, writers[i]);
     }
   }
 
   if (DCPS_debug_level >= 1) {
     ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) DataReaderImpl::remove_associations: ")
         ACE_TEXT("bit %d local %C remote %C num remotes %d\n"),
         is_bit_,
         LogGuid(get_guid()).c_str(),
         LogGuid(writers[0]).c_str(),
         writers.length()));
   }
   if (!get_deleted()) {
     // stop pending associations for these writer ids
     this->stop_associating(writers.get_buffer(), writers.length());
 
     {
       CORBA::ULong wr_len = writers.length();
       ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, this->writers_lock_);
 
       for (CORBA::ULong i = 0; i < wr_len; i++) {
         const GUID_t writer_id = writers[i];
         {
           ACE_Guard<ACE_Recursive_Thread_Mutex> guard(statistics_lock_);
           statistics_.erase(writer_id);
         }
       }
     }
   }
 
   remove_associations_i(writers, notify_lost);
 }

◆ remove_associations_i()

void OpenDDS::DCPS::DataReaderImpl::remove_associations_i	(	const WriterIdSeq &	writers,
		bool	callback
	)

protectedvirtual

Section 7.1.4.1: total_count will not decrement.

: Reconcile this with the verbiage in section 7.1.4.1

Definition at line 499 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), ACE_WRITE_GUARD, DDS::SubscriptionMatchedStatus::current_count, DDS::SubscriptionMatchedStatus::current_count_change, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TransportClient::disassociate(), end_historic_sweeper_, get_guid(), is_bit_, CORBA::is_nil(), DDS::SubscriptionMatchedStatus::last_publication_handle, listener_for(), LM_DEBUG, lookup_instance_handles(), monitor_, OpenDDS::DCPS::EntityImpl::notify_status_condition(), notify_subscription_lost(), publication_handle_lock_, publication_id_to_handle_map_, OpenDDS::DCPS::push_back(), resume_sample_processing(), OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), subscription_match_status_, DDS::SUBSCRIPTION_MATCHED_STATUS, DDS::SubscriptionMatchedStatus::total_count_change, writers_, and writers_lock_.

Referenced by remove_associations().

 {
   DBG_ENTRY_LVL("DataReaderImpl", "remove_associations_i", 6);
 
   if (writers.length() == 0) {
     return;
   }
 
   if (DCPS_debug_level >= 1) {
     ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) DataReaderImpl::remove_associations_i: ")
         ACE_TEXT("bit %d local %C remote %C num remotes %d\n"),
         is_bit_,
         LogGuid(get_guid()).c_str(),
         LogGuid(writers[0]).c_str(),
         writers.length()));
   }
   DDS::InstanceHandleSeq handles;
 
   CORBA::ULong wr_len = writers.length();
 
   // Flush historic samples and/or allow in-progress delivery of historic samples to complete
   for (CORBA::ULong i = 0; i < wr_len; i++) {
     resume_sample_processing(writers[i]);
   }
 
   // This is used to hold the list of writers which were actually
   // removed, which is a proper subset of the writers which were
   // requested to be removed.
   WriterIdSeq updated_writers;
   WriterMapType removed_writers;
 
   //Remove the writers from writer list. If the supplied writer
   //is not in the cached writers list then it is already removed.
   //We just need remove the writers in the list that have not been
   //removed.
   {
     ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, this->writers_lock_);
 
     for (CORBA::ULong i = 0; i < wr_len; i++) {
       const GUID_t writer_id = writers[i];
 
       WriterMapType::iterator it = this->writers_.find(writer_id);
 
       if (it != this->writers_.end()) {
         removed_writers.insert(*it);
         end_historic_sweeper_->cancel_timer(it->second);
       }
 
       if (this->writers_.erase(writer_id) == 0) {
         if (DCPS_debug_level >= 1) {
           ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) DataReaderImpl::remove_associations_i: ")
               ACE_TEXT("the writer local %C was already removed.\n"),
               LogGuid(writer_id).c_str()));
         }
 
       } else {
         push_back(updated_writers, writer_id);
       }
     }
   }
 
   for (WriterMapType::iterator it = removed_writers.begin(); it != removed_writers.end(); ++it) {
     it->second->removed();
   }
   removed_writers.clear();
 
   wr_len = updated_writers.length();
 
   // Return now if the supplied writers have been removed already.
   if (wr_len == 0) {
     return;
   }
 
   if (!is_bit_) {
     // The writer should be in the id_to_handle map at this time.
     this->lookup_instance_handles(updated_writers, handles);
 
     ACE_Guard<ACE_Recursive_Thread_Mutex> guard(publication_handle_lock_);
 
     for (CORBA::ULong i = 0; i < wr_len; ++i) {
       publication_id_to_handle_map_.erase(updated_writers[i]);
     }
   }
 
   for (CORBA::ULong i = 0; i < updated_writers.length(); ++i) {
     {
       this->disassociate(updated_writers[i]);
     }
   }
 
   // Mirror the add_associations SUBSCRIPTION_MATCHED_STATUS processing.
   if (!this->is_bit_) {
     ACE_Guard<ACE_Recursive_Thread_Mutex> justMe(publication_handle_lock_);
 
     // Derive the change in the number of publications writing to this reader.
     int matchedPublications = static_cast<int>(this->publication_id_to_handle_map_.size());
     this->subscription_match_status_.current_count_change
     = matchedPublications - this->subscription_match_status_.current_count;
 
     // Only process status if the number of publications has changed.
     if (this->subscription_match_status_.current_count_change != 0) {
       this->subscription_match_status_.current_count = matchedPublications;
 
       /// Section 7.1.4.1: total_count will not decrement.
 
       /// @TODO: Reconcile this with the verbiage in section 7.1.4.1
       this->subscription_match_status_.last_publication_handle
       = handles[ wr_len - 1];
 
       set_status_changed_flag(DDS::SUBSCRIPTION_MATCHED_STATUS, true);
 
       DDS::DataReaderListener_var listener
       = listener_for(DDS::SUBSCRIPTION_MATCHED_STATUS);
 
       if (!CORBA::is_nil(listener.in())) {
         listener->on_subscription_matched(this, this->subscription_match_status_);
 
         // Client will look at it so next time it looks the change should be 0
         this->subscription_match_status_.total_count_change = 0;
         this->subscription_match_status_.current_count_change = 0;
       }
       notify_status_condition();
     }
   }
 
   // If this remove_association is invoked when the InfoRepo
   // detects a lost writer then make a callback to notify
   // subscription lost.
   if (notify_lost) {
     this->notify_subscription_lost(handles);
   }
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 }

◆ remove_from_lookup_maps()

void OpenDDS::DCPS::DataReaderImpl::remove_from_lookup_maps ( DDS::InstanceHandle_t handle )

protected

Definition at line 3561 of file DataReaderImpl.cpp.

 {
   for (LookupMap::iterator it = combined_state_lookup_.begin(), the_end = combined_state_lookup_.end(); it != the_end; ++it) {
     if (it->first == 0) continue;
     it->second.erase(handle);
   }
 }

◆ reschedule_deadline()

void OpenDDS::DCPS::DataReaderImpl::reschedule_deadline	(	SubscriptionInstance_rch	instance,
		const MonotonicTimePoint &	now
	)

private

Definition at line 3709 of file DataReaderImpl.cpp.

References ACE_GUARD, OpenDDS::DCPS::SubscriptionInstance::deadline_, and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::zero_value.

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
 
   // So the datareader can call back into us.
   if (instance->deadline_ != MonotonicTimePoint::zero_value) {
 
     // Remove.
     for (DeadlineQueue::iterator pos = deadline_queue_.lower_bound(instance->deadline_), limit = deadline_queue_.upper_bound(instance->deadline_); pos != limit; ++pos) {
       if (pos->second == instance) {
         deadline_queue_.erase(pos);
         break;
       }
     }
 
     instance->deadline_ = now + (deadline_period_ - (instance->deadline_ - now));
 
     const bool schedule = deadline_queue_.empty();
     deadline_queue_.insert(std::make_pair(instance->deadline_, instance));
     if (schedule) {
       deadline_task_->schedule(deadline_period_);
     } else if (deadline_queue_.begin()->second == instance) {
       // Moved to front.
       deadline_task_->cancel();
       deadline_task_->schedule(deadline_period_);
     }
   }
 }

◆ reset_coherent_info()

void OpenDDS::DCPS::DataReaderImpl::reset_coherent_info	(	const GUID_t &	writer_id,
		const GUID_t &	publisher_id
	)

Definition at line 2970 of file DataReaderImpl.cpp.

References ACE_READ_GUARD.

 {
   ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);
 
   WriterMapType::iterator itEnd = this->writers_.end();
   for (WriterMapType::iterator it = this->writers_.begin();
       it != itEnd; ++it) {
     if (it->second->writer_id() == writer_id
         && it->second->publisher_id() == publisher_id) {
       it->second->reset_coherent_info();
     }
   }
 }

◆ reset_deadline_period()

void OpenDDS::DCPS::DataReaderImpl::reset_deadline_period ( const TimeDuration & deadline_period )

private

Definition at line 3690 of file DataReaderImpl.cpp.

References ACE_GUARD, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::zero_value.

Referenced by qos_change().

 {
   if (deadline_period_ != deadline_period) {
     deadline_period_ = deadline_period;
 
     if (deadline_queue_enabled_) {
       ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, this->instances_lock_);
       const MonotonicTimePoint now = MonotonicTimePoint::now();
       for (SubscriptionInstanceMapType::iterator iter = this->instances_.begin();
            iter != this->instances_.end();
            ++iter) {
         if (iter->second->deadline_ != MonotonicTimePoint::zero_value) {
           reschedule_deadline(iter->second, now);
         }
       }
     }
   }
 }

◆ reset_latency_stats()

void OpenDDS::DCPS::DataReaderImpl::reset_latency_stats ( )

virtual

Clear any intermediate statistical values.

Implements OpenDDS::DCPS::DataReaderEx.

Definition at line 2386 of file DataReaderImpl.cpp.

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(statistics_lock_);
   for (StatsMapType::iterator current = this->statistics_.begin();
       current != this->statistics_.end();
       ++current) {
     current->second.reset_stats();
   }
 }

◆ reset_ownership()

void OpenDDS::DCPS::DataReaderImpl::reset_ownership ( DDS::InstanceHandle_t instance )

Definition at line 3163 of file DataReaderImpl.cpp.

References ACE_WRITE_GUARD.

 {
   ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, this->writers_lock_);
   for (WriterMapType::iterator iter = writers_.begin();
       iter != writers_.end();
       ++iter) {
     iter->second->set_owner_evaluated(instance, false);
   }
 }

◆ resume_sample_processing()

void OpenDDS::DCPS::DataReaderImpl::resume_sample_processing ( const GUID_t & pub_id )

private

when done handling historic samples, resume

Definition at line 3174 of file DataReaderImpl.cpp.

References ACE_WRITE_GUARD, OpenDDS::DCPS::WriterInfo::check_end_historic_samples(), OpenDDS::DCPS::WriterInfo::finished_delivering_historic(), and OpenDDS::DCPS::OPENDDS_MAP().

Referenced by data_received(), and remove_associations_i().

 {
   WriterInfo_rch info;
   {
     ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, this->writers_lock_);
     WriterMapType::iterator where = writers_.find(pub_id);
     if (writers_.end() != where) {
       info = where->second;
     }
   }
 
   if (info) {
     OPENDDS_MAP(SequenceNumber, ReceivedDataSample) to_deliver;
     // Stop filtering these
     if (info->check_end_historic_samples(end_historic_sweeper_.in(), to_deliver)) {
       deliver_historic(to_deliver);
       info->finished_delivering_historic();
     }
   }
 }

◆ return_handle()

void OpenDDS::DCPS::DataReaderImpl::return_handle ( DDS::InstanceHandle_t handle )

Definition at line 2457 of file DataReaderImpl.cpp.

 {
   const RcHandle<DomainParticipantImpl> participant = participant_servant_.lock();
   if (participant) {
     participant->return_handle(handle);
   }
 }

◆ sample_info()

void OpenDDS::DCPS::DataReaderImpl::sample_info	(	DDS::SampleInfo &	sample_info,
		const ReceivedDataElement *	ptr
	)

protected

Definition at line 1769 of file DataReaderImpl.cpp.

References DDS::SampleInfo::absolute_generation_rank, DDS::SampleInfo::disposed_generation_count, OpenDDS::DCPS::ReceivedDataElement::disposed_generation_count_, DDS::SampleInfo::generation_rank, OpenDDS::DCPS::SequenceNumber::getValue(), DDS::SampleInfo::no_writers_generation_count, OpenDDS::DCPS::ReceivedDataElement::no_writers_generation_count_, DDS::SampleInfo::opendds_reserved_publication_seq, DDS::SampleInfo::sample_rank, and OpenDDS::DCPS::ReceivedDataElement::sequence_.

 {
 
   sample_info.sample_rank = 0;
 
   // generation_rank =
   //    (MRSIC.disposed_generation_count +
   //     MRSIC.no_writers_generation_count)
   //  - (S.disposed_generation_count +
   //     S.no_writers_generation_count)
   //
   sample_info.generation_rank =
       (sample_info.disposed_generation_count +
           sample_info.no_writers_generation_count) -
           sample_info.generation_rank;
 
   // absolute_generation_rank =
   //     (MRS.disposed_generation_count +
   //      MRS.no_writers_generation_count)
   //   - (S.disposed_generation_count +
   //      S.no_writers_generation_count)
   //
   sample_info.absolute_generation_rank =
       (static_cast<CORBA::Long>(ptr->disposed_generation_count_) +
           static_cast<CORBA::Long>(ptr->no_writers_generation_count_)) -
           sample_info.absolute_generation_rank;
 
   sample_info.opendds_reserved_publication_seq = ptr->sequence_.getValue();
 }

◆ schedule_deadline()

void OpenDDS::DCPS::DataReaderImpl::schedule_deadline	(	SubscriptionInstance_rch	instance,
		bool	timer_called
	)

private

Definition at line 3577 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::SubscriptionInstance::deadline_, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::zero_value.

 {
   // Should be called with sample_lock_.
   if (instance->deadline_ == MonotonicTimePoint::zero_value) {
     instance->deadline_ = MonotonicTimePoint::now() + deadline_period_;
     const bool schedule = deadline_queue_.empty();
     deadline_queue_.insert(std::make_pair(instance->deadline_, instance));
     if (!timer_called) {
       if (schedule) {
         deadline_task_->schedule(deadline_period_);
       } else if (deadline_queue_.begin()->second == instance) {
         // Moved to front.
         deadline_task_->cancel();
         deadline_task_->schedule(deadline_period_);
       }
     }
   }
 }

◆ set_instance_state()

void OpenDDS::DCPS::DataReaderImpl::set_instance_state	(	DDS::InstanceHandle_t	instance,
		DDS::InstanceStateKind	state,
		const SystemTimePoint &	timestamp = `SystemTimePoint::now()`,
		const GUID_t &	guid = `GUID_UNKNOWN`
	)

inline

Definition at line 565 of file DataReaderImpl.h.

References ACE_GUARD, DDS::HANDLE_NIL, instance_states, sample_states, timestamp(), and view_states.

Referenced by OpenDDS::DCPS::MultiTopicDataReaderBase::data_available(), OpenDDS::DCPS::BitSubscriber::remove_connection_record(), OpenDDS::DCPS::StaticParticipant::remove_discovered_participant(), OpenDDS::DCPS::BitSubscriber::remove_i(), and OpenDDS::DCPS::BitSubscriber::remove_thread_status().

   {
     DDS::InstanceHandle_t publication_handle = DDS::HANDLE_NIL;
     {
       ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, publication_handle_lock_);
       RepoIdToHandleMap::const_iterator pos = publication_id_to_handle_map_.find(guid);
       if (pos != publication_id_to_handle_map_.end()) {
         publication_handle = pos->second;
       }
     }
 
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
     set_instance_state_i(instance, publication_handle, state, timestamp, guid);
   }

◆ set_instance_state_i()

virtual void OpenDDS::DCPS::DataReaderImpl::set_instance_state_i	(	DDS::InstanceHandle_t	instance,
		DDS::InstanceHandle_t	publication_handle,
		DDS::InstanceStateKind	state,
		const SystemTimePoint &	timestamp,
		const GUID_t &	guid
	)

privatepure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

◆ set_listener()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::set_listener	(	DDS::DataReaderListener_ptr	a_listener,
		DDS::StatusMask	mask
	)

virtual

Definition at line 909 of file DataReaderImpl.cpp.

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

Referenced by cleanup(), and init().

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

◆ set_qos()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::set_qos ( const DDS::DataReaderQos & qos )

virtual

Definition at line 824 of file DataReaderImpl.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), OpenDDS::DCPS::Qos_Helper::changeable(), OpenDDS::DCPS::Qos_Helper::consistent(), domain_id_, dp_id_, OpenDDS::DCPS::Observer::e_QOS_CHANGED, OpenDDS::DCPS::EntityImpl::enabled_, OpenDDS::DCPS::EntityImpl::get_observer(), get_subscriber_servant(), LM_ERROR, OpenDDS::DCPS::Observer::on_qos_changed(), OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK, passed_qos_, qos_, qos_change(), DDS::DataReaderQos::representation, DDS::RETCODE_ERROR, DDS::RETCODE_IMMUTABLE_POLICY, DDS::RETCODE_INCONSISTENT_POLICY, DDS::RETCODE_OK, DDS::RETCODE_UNSUPPORTED, OpenDDS::DCPS::WriterInfoListener::subscription_id_, TheServiceParticipant, OpenDDS::DCPS::Qos_Helper::valid(), and DDS::DataRepresentationQosPolicy::value.

 {
   OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
   OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
   OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
 
   DDS::DataReaderQos new_qos = qos;
   new_qos.representation.value = qos_.representation.value;
   if (Qos_Helper::valid(new_qos) && Qos_Helper::consistent(new_qos)) {
 
     if (qos_ == new_qos)
       return DDS::RETCODE_OK;
 
     if (enabled_) {
       if (!Qos_Helper::changeable(qos_, new_qos)) {
         return DDS::RETCODE_IMMUTABLE_POLICY;
 
       } else {
         Discovery_rch disco = TheServiceParticipant->get_discovery(domain_id_);
         DDS::SubscriberQos subscriberQos;
 
         RcHandle<SubscriberImpl> subscriber = get_subscriber_servant();
         bool status = false;
         if (subscriber) {
           subscriber->get_qos(subscriberQos);
           status =
             disco->update_subscription_qos(
               domain_id_,
               dp_id_,
               subscription_id_,
               new_qos,
               subscriberQos);
         }
         if (!status) {
           ACE_ERROR_RETURN((LM_ERROR,
                             ACE_TEXT("(%P|%t) DataReaderImpl::set_qos, ")
                             ACE_TEXT("qos not updated.\n")),
                             DDS::RETCODE_ERROR);
         }
       }
     }
 
     qos_change(new_qos);
     qos_ = new_qos;
     passed_qos_ = qos;
 
     const Observer_rch observer = get_observer(Observer::e_QOS_CHANGED);
     if (observer) {
       observer->on_qos_changed(this);
     }
 
     return DDS::RETCODE_OK;
 
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ set_sample_lost_status()

void OpenDDS::DCPS::DataReaderImpl::set_sample_lost_status ( const DDS::SampleLostStatus & status )

protected

!!caller should have acquired sample_lock_

Definition at line 2260 of file DataReaderImpl.cpp.

 {
   //!!!caller should have acquired sample_lock_
   sample_lost_status_ = status;
 }

◆ set_sample_rejected_status()

void OpenDDS::DCPS::DataReaderImpl::set_sample_rejected_status ( const DDS::SampleRejectedStatus & status )

protected

!!caller should have acquired sample_lock_

Definition at line 2268 of file DataReaderImpl.cpp.

 {
   //!!!caller should have acquired sample_lock_
   sample_rejected_status_ = status;
 }

◆ set_subscriber_qos()

void OpenDDS::DCPS::DataReaderImpl::set_subscriber_qos ( const DDS::SubscriberQos & qos )

Definition at line 3116 of file DataReaderImpl.cpp.

 {
   this->subqos_ = qos;
 }

◆ setup_deserialization()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::setup_deserialization ( )

protected

Setup deserialization options.

Definition at line 3291 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_ERROR, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::Encoding::kind_to_string(), OpenDDS::DCPS::Encoding::KIND_UNALIGNED_CDR, OpenDDS::DCPS::Encoding::KIND_XCDR1, LM_DEBUG, LM_ERROR, LM_WARNING, OpenDDS::DCPS::MUTABLE, OPENDDS_STRING, OpenDDS::DCPS::repr_to_encoding_kind(), OpenDDS::DCPS::repr_to_string(), DDS::RETCODE_ERROR, and DDS::RETCODE_OK.

Referenced by enable().

 {
   bool xcdr1_mutable = false;
   bool illegal_unaligned = false;
   for (CORBA::ULong i = 0; i < qos_.representation.value.length(); ++i) {
     Encoding::Kind encoding_kind;
     if (repr_to_encoding_kind(qos_.representation.value[i], encoding_kind)) {
       if (encoding_kind == Encoding::KIND_XCDR1 && type_support_->max_extensibility() == MUTABLE) {
         xcdr1_mutable = true;
       } else if (encoding_kind == Encoding::KIND_UNALIGNED_CDR && cdr_encapsulation()) {
         illegal_unaligned = true;
       } else {
         decoding_modes_.insert(encoding_kind);
       }
     } else if (DCPS_debug_level) {
       ACE_DEBUG((LM_WARNING, "(%P|%t) WARNING: "
                  "DataReaderImpl::setup_deserialization: "
                  "Encountered unsupported or unknown data representation: %C\n",
                  repr_to_string(qos_.representation.value[i]).c_str()));
     }
   }
   if (decoding_modes_.empty()) {
     if (DCPS_debug_level) {
       DCPS::String error_message;
       if (xcdr1_mutable) {
         error_message = " Unsupported combination of XCDR1 and mutable";
       } else if (illegal_unaligned) {
         error_message = " Unaligned CDR is not allowed in rtps_udp transport";
       }
       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: "
                  "DataReaderImpl::setup_deserialization: "
                  "Could not find a valid data representation.%C\n",
                  error_message.c_str()));
     }
     return DDS::RETCODE_ERROR;
   }
   if (DCPS_debug_level >= 2) {
     OPENDDS_STRING encodings;
     EncodingKinds::iterator it = decoding_modes_.begin();
     for (; it != decoding_modes_.end(); ++it) {
       if (!encodings.empty()) {
         encodings += ", ";
       }
       encodings += Encoding::kind_to_string(*it);
     }
     ACE_DEBUG((LM_DEBUG, "(%P|%t) DataReaderImpl::setup_deserialization: "
                "Setup successfully with the following data representation%C: %C\n",
                encodings.size() != 1 ? "s" : "",
                encodings.c_str()));
   }
 
   return DDS::RETCODE_OK;
 }

◆ signal_liveliness()

void OpenDDS::DCPS::DataReaderImpl::signal_liveliness ( const GUID_t & remote_participant )

virtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 715 of file DataReaderImpl.cpp.

References ACE_GUARD, ACE_READ_GUARD, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::equal_guid_prefixes(), OpenDDS::DCPS::SubscriptionInstance::instance_state_, instances_, instances_lock_, OpenDDS::DCPS::InstanceState::lively(), OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OPENDDS_VECTOR(), sample_lock_, writers_, and writers_lock_.

 {
   GUID_t prefix = remote_participant;
   prefix.entityId = EntityId_t();
 
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_);
 
   typedef std::pair<GUID_t, WriterInfo_rch> RepoWriterPair;
   typedef OPENDDS_VECTOR(RepoWriterPair) WriterSet;
   WriterSet writers;
 
   {
     ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);
     for (WriterMapType::iterator pos = writers_.lower_bound(prefix),
            limit = writers_.end();
          pos != limit && equal_guid_prefixes(pos->first, prefix);
          ++pos) {
       writers.push_back(std::make_pair(pos->first, pos->second));
     }
   }
 
   const MonotonicTimePoint when = MonotonicTimePoint::now();
   for (WriterSet::iterator pos = writers.begin(), limit = writers.end();
        pos != limit;
        ++pos) {
     pos->second->received_activity(when);
   }
 
   if (!writers.empty()) {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, this->instances_lock_);
     for (WriterSet::iterator pos = writers.begin(), limit = writers.end();
          pos != limit;
          ++pos) {
       for (SubscriptionInstanceMapType::iterator iter = instances_.begin();
            iter != instances_.end();
            ++iter) {
         SubscriptionInstance_rch ptr = iter->second;
         ptr->instance_state_->lively(pos->first);
       }
     }
   }
 }

◆ split_combined_states()

static void OpenDDS::DCPS::DataReaderImpl::split_combined_states	(	CORBA::ULong	combined,
		CORBA::ULong &	sample_states,
		CORBA::ULong &	view_states,
		CORBA::ULong &	instance_states
	)

inlinestaticprotected

Definition at line 682 of file DataReaderImpl.h.

   {
     sample_states = (combined >> COMBINED_SAMPLE_STATE_SHIFT) & MAX_SAMPLE_STATE_MASK;
     view_states = (combined >> COMBINED_VIEW_STATE_SHIFT) & MAX_VIEW_STATE_MASK;
     instance_states = combined & MAX_INSTANCE_STATE_MASK;
   }

◆ state_updated()

void OpenDDS::DCPS::DataReaderImpl::state_updated ( DDS::InstanceHandle_t handle )

Definition at line 1976 of file DataReaderImpl.cpp.

References ACE_GUARD, sample_lock_, and state_updated_i().

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
   state_updated_i(handle);
 }

◆ state_updated_i()

virtual void OpenDDS::DCPS::DataReaderImpl::state_updated_i ( DDS::InstanceHandle_t handle )

protectedpure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

Referenced by state_updated().

◆ statistics_enabled() [1/2]

CORBA::Boolean OpenDDS::DCPS::DataReaderImpl::statistics_enabled ( )

virtual

Definition at line 2397 of file DataReaderImpl.cpp.

 {
   return statistics_enabled_;
 }

◆ statistics_enabled() [2/2]

void OpenDDS::DCPS::DataReaderImpl::statistics_enabled ( CORBA::Boolean statistics_enabled )

virtual

Definition at line 2403 of file DataReaderImpl.cpp.

 {
   statistics_enabled_ = statistics_enabled;
 }

◆ take()

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::take	(	AbstractSamples &	samples,
		DDS::SampleStateMask	sample_states,
		DDS::ViewStateMask	view_states,
		DDS::InstanceStateMask	instance_states
	)

pure virtual

Implemented in OpenDDS::DCPS::DataReaderImpl_T< MessageType >, OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >, OpenDDS::DCPS::DataReaderImpl_T< MessageType >, OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >, OpenDDS::DCPS::DataReaderImpl_T< MessageType >, and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >.

◆ time_based_filter_instance()

bool OpenDDS::DCPS::DataReaderImpl::time_based_filter_instance	(	const SubscriptionInstance_rch &	instance,
		MonotonicTimePoint &	now,
		MonotonicTimePoint &	deadline
	)

protected

Definition at line 2699 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::TimeDuration::is_zero(), OpenDDS::DCPS::SubscriptionInstance::last_accepted_, and OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now().

 {
   now = MonotonicTimePoint::now();
   const TimeDuration minimum_separation(qos_.time_based_filter.minimum_separation);
 
   // TIME_BASED_FILTER processing; expire data samples
   // if minimum separation is not met for instance.
   if (!minimum_separation.is_zero()) {
     if (now - instance->last_accepted_ < minimum_separation) {
       deadline = now + minimum_separation;
       return true; // Data filtered.
     }
   }
 
   instance->last_accepted_ = now;
 
   return false;
 }

◆ to_combined_states()

static CORBA::ULong OpenDDS::DCPS::DataReaderImpl::to_combined_states	(	CORBA::ULong	sample_states,
		CORBA::ULong	view_states,
		CORBA::ULong	instance_states
	)

inlinestaticprotected

Definition at line 670 of file DataReaderImpl.h.

   {
     sample_states &= MAX_SAMPLE_STATE_MASK;
     view_states &= MAX_VIEW_STATE_MASK;
     instance_states &= MAX_INSTANCE_STATE_MASK;
     if (!(sample_states && view_states && instance_states)) {
       // catch-all for "bogus" lookups
       return 0;
     }
     return (sample_states << COMBINED_SAMPLE_STATE_SHIFT) | (view_states << COMBINED_VIEW_STATE_SHIFT) | instance_states;
   }

◆ total_samples()

CORBA::Long OpenDDS::DCPS::DataReaderImpl::total_samples ( ) const

protected

!!caller should have acquired sample_lock_

Definition at line 1800 of file DataReaderImpl.cpp.

References ACE_GUARD_RETURN, instances_, instances_lock_, OpenDDS::DCPS::SubscriptionInstance::rcvd_samples_, and OpenDDS::DCPS::ReceivedDataElementList::size().

 {
   //!!!caller should have acquired sample_lock_
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, instance_guard, this->instances_lock_,0);
 
   CORBA::Long count(0);
 
   for (SubscriptionInstanceMapType::iterator iter = instances_.begin();
       iter != instances_.end();
       ++iter) {
     SubscriptionInstance_rch ptr = iter->second;
 
     count += static_cast<CORBA::Long>(ptr->rcvd_samples_.size());
   }
 
   return count;
 }

◆ transport_assoc_done()

void OpenDDS::DCPS::DataReaderImpl::transport_assoc_done	(	int	flags,
		const GUID_t &	remote_id
	)

virtual

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 355 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), ACE_WRITE_GUARD, OpenDDS::DCPS::TransportClient::ASSOC_OK, DDS::SubscriptionMatchedStatus::current_count, DDS::SubscriptionMatchedStatus::current_count_change, OpenDDS::DCPS::DCPS_debug_level, get_guid(), is_bit_, CORBA::is_nil(), OpenDDS::DCPS::TimeDuration::is_zero(), DDS::SubscriptionMatchedStatus::last_publication_handle, listener_for(), OpenDDS::DCPS::WriterInfoListener::liveliness_lease_duration_, liveliness_timer_, LM_DEBUG, LM_ERROR, monitor_, OpenDDS::DCPS::EntityImpl::notify_status_condition(), participant_servant_, publication_handle_lock_, publication_id_to_handle_map_, sample_lock_, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), subscription_match_status_, DDS::SUBSCRIPTION_MATCHED_STATUS, DDS::SubscriptionMatchedStatus::total_count, DDS::SubscriptionMatchedStatus::total_count_change, writers_, and writers_lock_.

 {
   if (!(flags & ASSOC_OK)) {
     if (DCPS_debug_level) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) DataReaderImpl::transport_assoc_done: ")
           ACE_TEXT("ERROR: transport layer failed to associate %C\n"),
           LogGuid(remote_id).c_str()));
     }
     return;
   }
 
   // LIVELINESS policy timers are managed here.
   if (!liveliness_lease_duration_.is_zero()) {
     if (DCPS_debug_level >= 5) {
       ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) DataReaderImpl::transport_assoc_done: ")
           ACE_TEXT("starting/resetting liveliness timer for reader %C\n"),
           LogGuid(get_guid()).c_str()));
     }
     // this call will start the timer if it is not already set
     liveliness_timer_->check_liveliness();
   }
 
   const RcHandle<DomainParticipantImpl> participant = participant_servant_.lock();
 
   if (!participant)
     return;
 
   const DDS::InstanceHandle_t handle = participant->assign_handle(remote_id);
 
   if (!is_bit_) {
     // We acquire the publication_handle_lock_ for the remainder of our
     // processing.
     {
       ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, publication_handle_lock_);
 
       // This insertion is idempotent.
       publication_id_to_handle_map_.insert(RepoIdToHandleMap::value_type(remote_id, handle));
 
       if (DCPS_debug_level > 4) {
         ACE_DEBUG((LM_DEBUG,
             ACE_TEXT("(%P|%t) DataReaderImpl::transport_assoc_done: ")
             ACE_TEXT("id_to_handle_map_[ %C] = 0x%x.\n"),
             LogGuid(remote_id).c_str(),
             handle));
       }
 
       // We need to adjust these after the insertions have all completed
       // since insertions are not guaranteed to increase the number of
       // currently matched publications.
       const int matchedPublications = static_cast<int>(publication_id_to_handle_map_.size());
       subscription_match_status_.current_count_change =
           matchedPublications - subscription_match_status_.current_count;
       subscription_match_status_.current_count = matchedPublications;
 
       ++subscription_match_status_.total_count;
       ++subscription_match_status_.total_count_change;
 
       subscription_match_status_.last_publication_handle = handle;
 
       set_status_changed_flag(DDS::SUBSCRIPTION_MATCHED_STATUS, true);
 
       DDS::DataReaderListener_var listener =
           listener_for(DDS::SUBSCRIPTION_MATCHED_STATUS);
 
       if (!CORBA::is_nil(listener)) {
         listener->on_subscription_matched(this, subscription_match_status_);
 
         // TBD - why does the spec say to change this but not change
         //       the ChangeFlagStatus after a listener call?
 
         // Client will look at it so next time it looks the change should be 0
         subscription_match_status_.total_count_change = 0;
         subscription_match_status_.current_count_change = 0;
       }
 
       notify_status_condition();
     }
 
     {
       ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
       ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, writers_lock_);
 
       if (!writers_.count(remote_id)) {
         return;
       }
       writers_[remote_id]->handle(handle);
     }
   }
 
   if (monitor_) {
     monitor_->report();
   }
 }

◆ transport_discovery_change()

void OpenDDS::DCPS::DataReaderImpl::transport_discovery_change ( )

virtual

Reimplemented from OpenDDS::DCPS::TransportReceiveListener.

Definition at line 3483 of file DataReaderImpl.cpp.

References TheServiceParticipant.

 {
   populate_connection_info();
   const TransportLocatorSeq& trans_conf_info = connection_info();
   const GUID_t dp_id_copy = dp_id_;
   Discovery_rch disco = TheServiceParticipant->get_discovery(domain_id_);
   disco->update_subscription_locators(domain_id_,
                                       dp_id_copy,
                                       get_guid(),
                                       trans_conf_info);
 }

◆ unregister_for_writer()

void OpenDDS::DCPS::DataReaderImpl::unregister_for_writer	(	const GUID_t &	participant,
		const GUID_t &	readerid,
		const GUID_t &	writerid
	)

virtual

Reimplemented from OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 3264 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::TransportClient::unregister_for_writer().

 {
   TransportClient::unregister_for_writer(participant, readerid, writerid);
 }

◆ update_incompatible_qos()

void OpenDDS::DCPS::DataReaderImpl::update_incompatible_qos ( const IncompatibleQosStatus & status )

virtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 679 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::IncompatibleQosStatus::count_since_last_send, CORBA::is_nil(), OpenDDS::DCPS::IncompatibleQosStatus::last_policy_id, DDS::RequestedIncompatibleQosStatus::last_policy_id, listener_for(), OpenDDS::DCPS::EntityImpl::notify_status_condition(), OpenDDS::DCPS::IncompatibleQosStatus::policies, DDS::RequestedIncompatibleQosStatus::policies, DDS::REQUESTED_INCOMPATIBLE_QOS_STATUS, requested_incompatible_qos_status_, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), OpenDDS::DCPS::IncompatibleQosStatus::total_count, DDS::RequestedIncompatibleQosStatus::total_count, and DDS::RequestedIncompatibleQosStatus::total_count_change.

 {
   DDS::DataReaderListener_var listener =
       listener_for(DDS::REQUESTED_INCOMPATIBLE_QOS_STATUS);
 
   if (this->requested_incompatible_qos_status_.total_count == status.total_count) {
     // This test should make the method idempotent.
     return;
   }
 
   set_status_changed_flag(DDS::REQUESTED_INCOMPATIBLE_QOS_STATUS,
       true);
 
   // copy status and increment change
   requested_incompatible_qos_status_.total_count = status.total_count;
   requested_incompatible_qos_status_.total_count_change +=
       status.count_since_last_send;
   requested_incompatible_qos_status_.last_policy_id =
       status.last_policy_id;
   requested_incompatible_qos_status_.policies = status.policies;
 
   if (!CORBA::is_nil(listener.in())) {
     listener->on_requested_incompatible_qos(this, requested_incompatible_qos_status_);
 
     // TBD - why does the spec say to change total_count_change but not
     // change the ChangeFlagStatus after a listener call?
 
     // client just looked at it so next time it looks the
     // change should be 0
     requested_incompatible_qos_status_.total_count_change = 0;
   }
 
   notify_status_condition();
 }

◆ update_locators()

void OpenDDS::DCPS::DataReaderImpl::update_locators	(	const GUID_t &	remote,
		const TransportLocatorSeq &	locators
	)

virtual

Reimplemented from OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 3272 of file DataReaderImpl.cpp.

References ACE_READ_GUARD, and OpenDDS::DCPS::TransportClient::update_locators().

 {
   {
     ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, writers_lock_);
     WriterMapType::const_iterator iter = writers_.find(writerId);
     if (iter == writers_.end()) {
       return;
     }
   }
   TransportClient::update_locators(writerId, locators);
 }

◆ update_lookup_maps()

void OpenDDS::DCPS::DataReaderImpl::update_lookup_maps ( const SubscriptionInstanceMapType::iterator & input )

protected

Definition at line 3547 of file DataReaderImpl.cpp.

References instance_states, sample_states, and view_states.

 {
   for (LookupMap::iterator it = combined_state_lookup_.begin(); it != combined_state_lookup_.end(); ++it) {
     if (it->first == 0) continue;
     CORBA::ULong sample_states, view_states, instance_states;
     split_combined_states(it->first, sample_states, view_states, instance_states);
     if (input->second->matches(sample_states, view_states, instance_states)) {
       it->second.insert(input->first);
     } else {
       it->second.erase(input->first);
     }
   }
 }

◆ update_ownership_strength()

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

Definition at line 2849 of file DataReaderImpl.cpp.

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

 {
   ACE_READ_GUARD(ACE_RW_Thread_Mutex,
       read_guard,
       this->writers_lock_);
   for (WriterMapType::iterator iter = writers_.begin();
       iter != writers_.end();
       ++iter) {
     if (iter->second->writer_id() == pub_id) {
       if (ownership_strength != iter->second->writer_qos_ownership_strength()) {
         if (DCPS_debug_level >= 1) {
           ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) DataReaderImpl::update_ownership_strength - ")
               ACE_TEXT("local %C update remote %C strength from %d to %d\n"),
               LogGuid(get_guid()).c_str(),
               LogGuid(pub_id).c_str(),
               iter->second->writer_qos_ownership_strength(), ownership_strength));
         }
         iter->second->writer_qos_ownership_strength(ownership_strength);
         iter->second->clear_owner_evaluated();
       }
       break;
     }
   }
 }

◆ update_subscription_params()

void OpenDDS::DCPS::DataReaderImpl::update_subscription_params ( const DDS::StringSeq & params ) const

Definition at line 3152 of file DataReaderImpl.cpp.

References TheServiceParticipant.

 {
   Discovery_rch disco = TheServiceParticipant->get_discovery(domain_id_);
   disco->update_subscription_params(domain_id_,
       dp_id_,
       subscription_id_,
       params);
 }

◆ verify_coherent_changes_completion()

bool OpenDDS::DCPS::DataReaderImpl::verify_coherent_changes_completion ( WriterInfo * writer )

private

Definition at line 2878 of file DataReaderImpl.cpp.

References OpenDDS::DCPS::WriterInfo::coherent_change_received(), OpenDDS::DCPS::COMPLETED, OpenDDS::DCPS::WriterInfo::group_coherent(), DDS::INSTANCE_PRESENTATION_QOS, OpenDDS::DCPS::NOT_COMPLETED_YET, OpenDDS::DCPS::WriterInfo::publisher_id(), OpenDDS::DCPS::REJECTED, OpenDDS::DCPS::WriterInfo::reset_coherent_info(), state, and OpenDDS::DCPS::WriterInfo::writer_id().

Referenced by data_received().

 {
   Coherent_State state = COMPLETED;
   bool accept_here = true;
 
   const GUID_t writer_id = writer->writer_id();
   const GUID_t publisher_id = writer->publisher_id();
 
   if (subqos_.presentation.access_scope != ::DDS::INSTANCE_PRESENTATION_QOS &&
       subqos_.presentation.coherent_access) {
     // verify current coherent changes from single writer
     state = writer->coherent_change_received();
     if (writer->group_coherent()) { // GROUP coherent any state
       RcHandle<SubscriberImpl> subscriber = get_subscriber_servant();
       if (subscriber && state != NOT_COMPLETED_YET) {
         // verify if all readers received complete coherent changes in a group.
         subscriber->coherent_change_received(publisher_id, this, state);
         accept_here = false; // coherent_change_received does that itself
       }
     } else if (state != NOT_COMPLETED_YET) { // TOPIC coherent with final state
       if (state == REJECTED) {
         reject_coherent(writer_id, publisher_id);
       }
       writer->reset_coherent_info();
     }
   }
 
   if (state == COMPLETED && accept_here) {
     accept_coherent(writer_id, publisher_id);
     coherent_changes_completed(this);
   }
 
   return state == COMPLETED;
 }

◆ wait_for_historical_data()

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl::wait_for_historical_data ( const DDS::Duration_t & max_wait )

virtual

Definition at line 1043 of file DataReaderImpl.cpp.

References DDS::RETCODE_OK.

 {
   // Add your implementation here
   return DDS::RETCODE_OK;
 }

◆ writer_activity()

void OpenDDS::DCPS::DataReaderImpl::writer_activity ( const DataSampleHeader & header )

update liveliness info for this writer.

Definition at line 1323 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_READ_GUARD, ACE_TEXT(), OpenDDS::DCPS::WriterInfo::add_coherent_samples(), OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::DataSampleHeader::coherent_change_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::DISPOSE_INSTANCE, OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE, get_guid(), OpenDDS::DCPS::INSTANCE_REGISTRATION, OpenDDS::DCPS::RcHandle< T >::is_nil(), LM_DEBUG, OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::WriterInfo::received_activity(), OpenDDS::DCPS::SAMPLE_DATA, OpenDDS::DCPS::DataSampleHeader::sequence_, OpenDDS::DCPS::UNREGISTER_INSTANCE, writers_, and writers_lock_.

Referenced by data_received().

 {
   // caller should have the sample_lock_ !!!
 
   WriterInfo_rch writer;
 
   // The received_activity() has to be called outside the writers_lock_
   // because it probably acquire writers_lock_ read lock recursively
   // (in handle_timeout). This could cause deadlock when there are writers
   // waiting.
   {
     ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);
     WriterMapType::iterator iter = writers_.find(header.publication_id_);
 
     if (iter != writers_.end()) {
       writer = iter->second;
 
     } else if (DCPS_debug_level > 4) {
       // This may not be an error since it could happen that the sample
       // is delivered to the datareader after the write is dis-associated
       // with this datareader.
       ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) DataReaderImpl::writer_activity: ")
           ACE_TEXT("reader %C is not associated with writer %C.\n"),
           LogGuid(get_guid()).c_str(),
           LogGuid(header.publication_id_).c_str()));
     }
   }
 
   if (!writer.is_nil()) {
     writer->received_activity(MonotonicTimePoint::now());
 
     if ((header.message_id_ == SAMPLE_DATA) ||
         (header.message_id_ == INSTANCE_REGISTRATION) ||
         (header.message_id_ == UNREGISTER_INSTANCE) ||
         (header.message_id_ == DISPOSE_INSTANCE) ||
         (header.message_id_ == DISPOSE_UNREGISTER_INSTANCE)) {
 
 #ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
       if (header.coherent_change_) {
         writer->add_coherent_samples(header.sequence_);
       }
 #endif
     }
   }
 }

◆ writer_became_alive()

void OpenDDS::DCPS::DataReaderImpl::writer_became_alive	(	WriterInfo &	info,
		const MonotonicTimePoint &	when
	)

virtual

tell instances when a DataWriter transitions to being alive The writer state is inout parameter, it has to be set ALIVE before handle_timeout is called since some subroutine use the state.

Reimplemented from OpenDDS::DCPS::WriterInfoListener.

Definition at line 2083 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::WriterInfo::ALIVE, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::WriterInfo::DEAD, OpenDDS::DCPS::WriterInfo::get_state_str(), OpenDDS::DCPS::WriterInfo::handle(), DDS::LIVELINESS_CHANGED_STATUS, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::WriterInfo::state(), and OpenDDS::DCPS::WriterInfo::writer_id().

 {
   const GUID_t info_writer_id = info.writer_id();
 
   if (DCPS_debug_level >= 5) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) DataReaderImpl::writer_became_alive: ")
                ACE_TEXT("reader %C from writer %C previous state %C.\n"),
                LogGuid(get_guid()).c_str(),
                LogGuid(info_writer_id).c_str(),
                info.get_state_str()));
   }
 
   // NOTE: each instance will change to ALIVE_STATE when they receive a sample
 
   const WriterInfo::WriterState info_state = info.state();
 
   {
     bool liveliness_changed = false;
 
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
 
     if (info_state != WriterInfo::ALIVE) {
       liveliness_changed_status_.alive_count++;
       liveliness_changed_status_.alive_count_change++;
       liveliness_changed = true;
     }
 
     if (info_state == WriterInfo::DEAD) {
       liveliness_changed_status_.not_alive_count--;
       liveliness_changed_status_.not_alive_count_change--;
     }
 
     if (liveliness_changed_status_.alive_count < 0) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::writer_became_alive: ")
                  ACE_TEXT("invalid liveliness_changed_status alive count - %d.\n"),
                  liveliness_changed_status_.alive_count));
       return;
     }
 
     if (liveliness_changed_status_.not_alive_count < 0) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::writer_became_alive: ")
                  ACE_TEXT("invalid liveliness_changed_status not alive count - %d.\n"),
                  liveliness_changed_status_.not_alive_count));
       return;
     }
 
     liveliness_changed_status_.last_publication_handle = info.handle();
 
     // Change the state to ALIVE since handle_timeout may call writer_became_dead
     // which need the current state info.
     info.state(WriterInfo::ALIVE);
 
     if (this->monitor_) {
       this->monitor_->report();
     }
 
     // Call listener only when there are liveliness status changes.
     if (liveliness_changed) {
       set_status_changed_flag(DDS::LIVELINESS_CHANGED_STATUS, true);
       this->notify_liveliness_change();
     }
   }
 
   // this call will start the liveliness timer if it is not already set
   liveliness_timer_->check_liveliness();
 }

◆ writer_became_dead()

void OpenDDS::DCPS::DataReaderImpl::writer_became_dead ( WriterInfo & info )

virtual

tell instances when a DataWriter transitions to DEAD The writer state is inout parameter, the state is set to DEAD when it returns.

Reimplemented from OpenDDS::DCPS::WriterInfoListener.

Definition at line 2153 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::WriterInfo::ALIVE, OpenDDS::DCPS::WriterInfo::clear_owner_evaluated(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::WriterInfo::DEAD, OpenDDS::DCPS::WriterInfo::get_state_str(), OpenDDS::DCPS::WriterInfo::handle(), DDS::HANDLE_NIL, DDS::LIVELINESS_CHANGED_STATUS, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::OwnershipManager::remove_writer(), OpenDDS::DCPS::WriterInfo::state(), and OpenDDS::DCPS::WriterInfo::writer_id().

 {
   const GUID_t info_writer_id = info.writer_id();
 
   if (DCPS_debug_level >= 5) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) DataReaderImpl::writer_became_dead: ")
                ACE_TEXT("reader %C from writer %C previous state %C.\n"),
                LogGuid(get_guid()).c_str(),
                LogGuid(info_writer_id).c_str(),
                info.get_state_str()));
   }
 
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   OwnershipManagerPtr owner_manager = this->ownership_manager();
   if (owner_manager) {
     owner_manager->remove_writer(info_writer_id);
     info.clear_owner_evaluated();
   }
 #endif
 
   bool liveliness_changed = false;
 
   const WriterInfo::WriterState info_state = info.state();
 
   DDS::InstanceHandle_t publication_handle = DDS::HANDLE_NIL;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, publication_handle_lock_);
     RepoIdToHandleMap::const_iterator pos = publication_id_to_handle_map_.find(info_writer_id);
     if (pos != publication_id_to_handle_map_.end()) {
       publication_handle = pos->second;
     }
   }
 
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
 
     if (info_state != WriterInfo::DEAD) {
       ++liveliness_changed_status_.not_alive_count;
       ++liveliness_changed_status_.not_alive_count_change;
       liveliness_changed = true;
     }
 
     if (info_state == WriterInfo::ALIVE) {
       --liveliness_changed_status_.alive_count;
       --liveliness_changed_status_.alive_count_change;
     }
 
     if (liveliness_changed_status_.alive_count < 0) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::writer_became_dead: ")
                  ACE_TEXT("invalid liveliness_changed_status alive count - %d.\n"),
                  liveliness_changed_status_.alive_count));
       return;
     }
 
     if (liveliness_changed_status_.not_alive_count < 0) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::writer_became_dead: ")
                  ACE_TEXT("invalid liveliness_changed_status not alive count - %d.\n"),
                  liveliness_changed_status_.not_alive_count));
       return;
     }
 
     liveliness_changed_status_.last_publication_handle = info.handle();
 
     info.state(WriterInfo::DEAD);
 
     if (this->monitor_) {
       this->monitor_->report();
     }
 
     instances_liveliness_update(info_writer_id, publication_handle);
 
     // Call listener only when there are liveliness status changes.
     if (liveliness_changed) {
       set_status_changed_flag(DDS::LIVELINESS_CHANGED_STATUS, true);
       this->notify_liveliness_change();
     }
   }
 }

◆ writer_removed()

void OpenDDS::DCPS::DataReaderImpl::writer_removed ( WriterInfo & info )

virtual

tell instance when a DataWriter is removed. The liveliness status need update.

Reimplemented from OpenDDS::DCPS::WriterInfoListener.

Definition at line 2024 of file DataReaderImpl.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::WriterInfo::ALIVE, OpenDDS::DCPS::WriterInfo::clear_owner_evaluated(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::WriterInfo::DEAD, OpenDDS::DCPS::WriterInfo::handle(), DDS::HANDLE_NIL, DDS::LIVELINESS_CHANGED_STATUS, LM_DEBUG, OpenDDS::DCPS::OwnershipManager::remove_writer(), OpenDDS::DCPS::WriterInfo::state(), and OpenDDS::DCPS::WriterInfo::writer_id().

 {
   const GUID_t info_writer_id = info.writer_id();
 
   if (DCPS_debug_level >= 5) {
     ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) DataReaderImpl::writer_removed: ")
         ACE_TEXT("reader %C from writer %C.\n"),
         LogGuid(get_guid()).c_str(),
         LogGuid(info_writer_id).c_str()));
   }
 
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   OwnershipManagerPtr owner_manager = this->ownership_manager();
   if (owner_manager) {
     owner_manager->remove_writer(info_writer_id);
     info.clear_owner_evaluated();
   }
 #endif
 
   {
     DDS::InstanceHandle_t publication_handle = DDS::HANDLE_NIL;
     {
       ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, publication_handle_lock_);
       RepoIdToHandleMap::const_iterator pos = publication_id_to_handle_map_.find(info_writer_id);
       if (pos != publication_id_to_handle_map_.end()) {
         publication_handle = pos->second;
       }
     }
 
     bool liveliness_changed = false;
 
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
 
     const WriterInfo::WriterState info_state = info.state();
 
     if (info_state == WriterInfo::ALIVE) {
       --liveliness_changed_status_.alive_count;
       --liveliness_changed_status_.alive_count_change;
       liveliness_changed = true;
     }
 
     if (info_state == WriterInfo::DEAD) {
       --liveliness_changed_status_.not_alive_count;
       --liveliness_changed_status_.not_alive_count_change;
       liveliness_changed = true;
     }
 
     liveliness_changed_status_.last_publication_handle = info.handle();
     instances_liveliness_update(info_writer_id, publication_handle);
 
     if (liveliness_changed) {
       set_status_changed_flag(DDS::LIVELINESS_CHANGED_STATUS, true);
       this->notify_liveliness_change();
     }
   }
 }

Friends And Related Function Documentation

◆ ::DDS_TEST

friend class ::DDS_TEST

friend

Definition at line 870 of file DataReaderImpl.h.

◆ EndHistoricSamplesMissedSweeper

friend class EndHistoricSamplesMissedSweeper

friend

Definition at line 868 of file DataReaderImpl.h.

◆ InstanceState

friend class InstanceState

friend

Definition at line 867 of file DataReaderImpl.h.

◆ OwnershipManagerPtr

friend class OwnershipManagerPtr

friend

Definition at line 481 of file DataReaderImpl.h.

◆ QueryConditionImpl

friend class QueryConditionImpl

friend

Definition at line 216 of file DataReaderImpl.h.

Referenced by create_querycondition().

◆ RequestedDeadlineWatchdog

friend class RequestedDeadlineWatchdog

friend

Definition at line 215 of file DataReaderImpl.h.

◆ SubscriberImpl

friend class SubscriberImpl

friend

Definition at line 217 of file DataReaderImpl.h.

Member Data Documentation

◆ always_get_history_

bool OpenDDS::DCPS::DataReaderImpl::always_get_history_

private

Definition at line 1013 of file DataReaderImpl.h.

◆ budget_exceeded_status_

BudgetExceededStatus OpenDDS::DCPS::DataReaderImpl::budget_exceeded_status_

private

Definition at line 900 of file DataReaderImpl.h.

Referenced by DataReaderImpl().

◆ coherent_

bool OpenDDS::DCPS::DataReaderImpl::coherent_

protected

Is accessing to Group coherent changes ?

Definition at line 805 of file DataReaderImpl.h.

◆ COMBINED_SAMPLE_STATE_SHIFT

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::COMBINED_SAMPLE_STATE_SHIFT = COMBINED_VIEW_STATE_SHIFT + MAX_VIEW_STATE_BITS

staticprotected

Definition at line 665 of file DataReaderImpl.h.

◆ combined_state_lookup_

LookupMap OpenDDS::DCPS::DataReaderImpl::combined_state_lookup_

protected

Definition at line 694 of file DataReaderImpl.h.

◆ COMBINED_VIEW_STATE_SHIFT

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::COMBINED_VIEW_STATE_SHIFT = MAX_INSTANCE_STATE_BITS

staticprotected

Definition at line 664 of file DataReaderImpl.h.

◆ content_filtered_topic_

TopicDescriptionPtr<ContentFilteredTopicImpl> OpenDDS::DCPS::DataReaderImpl::content_filtered_topic_

protected

Definition at line 797 of file DataReaderImpl.h.

Referenced by cleanup(), enable(), and get_topicdescription().

◆ content_filtered_topic_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::DataReaderImpl::content_filtered_topic_mutex_

mutableprotected

Definition at line 796 of file DataReaderImpl.h.

Referenced by cleanup(), enable(), and get_topicdescription().

◆ deadline_period_

TimeDuration OpenDDS::DCPS::DataReaderImpl::deadline_period_

private

Watchdog responsible for reporting missed offered deadlines.

Definition at line 990 of file DataReaderImpl.h.

Referenced by enable(), and qos_change().

◆ deadline_queue_

DeadlineQueue OpenDDS::DCPS::DataReaderImpl::deadline_queue_

private

Definition at line 992 of file DataReaderImpl.h.

◆ deadline_queue_enabled_

bool OpenDDS::DCPS::DataReaderImpl::deadline_queue_enabled_

private

Definition at line 993 of file DataReaderImpl.h.

Referenced by data_received(), enable(), and qos_change().

◆ deadline_task_

RcHandle<DRISporadicTask> OpenDDS::DCPS::DataReaderImpl::deadline_task_

private

Definition at line 995 of file DataReaderImpl.h.

Referenced by ~DataReaderImpl().

◆ decoding_modes_

EncodingKinds OpenDDS::DCPS::DataReaderImpl::decoding_modes_

protected

Definition at line 1051 of file DataReaderImpl.h.

◆ depth_

CORBA::Long OpenDDS::DCPS::DataReaderImpl::depth_

private

Definition at line 884 of file DataReaderImpl.h.

Referenced by enable().

◆ domain_id_

DDS::DomainId_t OpenDDS::DCPS::DataReaderImpl::domain_id_

private

Definition at line 876 of file DataReaderImpl.h.

Referenced by enable(), init(), and set_qos().

◆ dp_id_

GUID_t OpenDDS::DCPS::DataReaderImpl::dp_id_

private

Definition at line 877 of file DataReaderImpl.h.

Referenced by enable(), and set_qos().

◆ dynamic_type_

DDS::DynamicType_var OpenDDS::DCPS::DataReaderImpl::dynamic_type_

protected

Definition at line 1105 of file DataReaderImpl.h.

Referenced by enable().

◆ end_historic_sweeper_

RcHandle<EndHistoricSamplesMissedSweeper> OpenDDS::DCPS::DataReaderImpl::end_historic_sweeper_

private

Definition at line 882 of file DataReaderImpl.h.

Referenced by remove_associations_i().

◆ group_coherent_ordered_data_

GroupRakeData OpenDDS::DCPS::DataReaderImpl::group_coherent_ordered_data_

protected

Ordered group samples.

Definition at line 808 of file DataReaderImpl.h.

◆ has_subscription_id_

bool OpenDDS::DCPS::DataReaderImpl::has_subscription_id_

protected

Definition at line 767 of file DataReaderImpl.h.

Referenced by add_association(), and enable().

◆ instances_

SubscriptionInstanceMapType OpenDDS::DCPS::DataReaderImpl::instances_

mutableprotected

: document why the instances_ container is mutable.

Definition at line 738 of file DataReaderImpl.h.

Referenced by data_received(), release_instance(), signal_liveliness(), and total_samples().

◆ instances_lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::DataReaderImpl::instances_lock_

mutableprotected

Assume since the container is mutable(?!!?) it may need to use the lock while const. : remove the recursive nature of the instances_lock if not needed.

Definition at line 743 of file DataReaderImpl.h.

Referenced by contains_sample(), data_received(), have_instance_states(), have_sample_states(), have_view_states(), release_instance(), signal_liveliness(), and total_samples().

◆ is_bit_

bool OpenDDS::DCPS::DataReaderImpl::is_bit_

private

Flag indicates that this datareader is a builtin topic datareader.

Definition at line 1011 of file DataReaderImpl.h.

Referenced by add_association(), init(), remove_associations(), remove_associations_i(), and transport_assoc_done().

◆ is_exclusive_ownership_

bool OpenDDS::DCPS::DataReaderImpl::is_exclusive_ownership_

protected

Definition at line 791 of file DataReaderImpl.h.

Referenced by data_received(), and init().

◆ last_deadline_missed_total_count_

CORBA::Long OpenDDS::DCPS::DataReaderImpl::last_deadline_missed_total_count_

private

Definition at line 987 of file DataReaderImpl.h.

Referenced by get_requested_deadline_missed_status().

◆ listener_

DDS::DataReaderListener_var OpenDDS::DCPS::DataReaderImpl::listener_

private

Definition at line 875 of file DataReaderImpl.h.

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

◆ listener_mask_

DDS::StatusMask OpenDDS::DCPS::DataReaderImpl::listener_mask_

private

Definition at line 874 of file DataReaderImpl.h.

Referenced by listener_for(), and set_listener().

◆ listener_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::DataReaderImpl::listener_mutex_

private

Definition at line 873 of file DataReaderImpl.h.

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

◆ liveliness_changed_status_

DDS::LivelinessChangedStatus OpenDDS::DCPS::DataReaderImpl::liveliness_changed_status_

private

Definition at line 894 of file DataReaderImpl.h.

Referenced by DataReaderImpl(), and get_liveliness_changed_status().

◆ liveliness_timer_

RcHandle<LivelinessTimer> OpenDDS::DCPS::DataReaderImpl::liveliness_timer_

private

Definition at line 985 of file DataReaderImpl.h.

Referenced by transport_assoc_done().

◆ MAX_INSTANCE_STATE_BITS

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::MAX_INSTANCE_STATE_BITS = 3u

staticprotected

Definition at line 661 of file DataReaderImpl.h.

◆ MAX_INSTANCE_STATE_FLAG

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::MAX_INSTANCE_STATE_FLAG = DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE

staticprotected

Definition at line 659 of file DataReaderImpl.h.

◆ MAX_INSTANCE_STATE_MASK

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::MAX_INSTANCE_STATE_MASK = (MAX_INSTANCE_STATE_FLAG << 1) - 1

staticprotected

Definition at line 660 of file DataReaderImpl.h.

◆ MAX_SAMPLE_STATE_BITS

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::MAX_SAMPLE_STATE_BITS = 2u

staticprotected

Definition at line 651 of file DataReaderImpl.h.

◆ MAX_SAMPLE_STATE_FLAG

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::MAX_SAMPLE_STATE_FLAG = DDS::NOT_READ_SAMPLE_STATE

staticprotected

Definition at line 649 of file DataReaderImpl.h.

◆ MAX_SAMPLE_STATE_MASK

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::MAX_SAMPLE_STATE_MASK = (MAX_SAMPLE_STATE_FLAG << 1) - 1

staticprotected

Definition at line 650 of file DataReaderImpl.h.

◆ MAX_VIEW_STATE_BITS

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::MAX_VIEW_STATE_BITS = 2u

staticprotected

Definition at line 656 of file DataReaderImpl.h.

◆ MAX_VIEW_STATE_FLAG

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::MAX_VIEW_STATE_FLAG = DDS::NOT_NEW_VIEW_STATE

staticprotected

Definition at line 654 of file DataReaderImpl.h.

◆ MAX_VIEW_STATE_MASK

const CORBA::ULong OpenDDS::DCPS::DataReaderImpl::MAX_VIEW_STATE_MASK = (MAX_VIEW_STATE_FLAG << 1) - 1

staticprotected

Definition at line 655 of file DataReaderImpl.h.

◆ mb_alloc_

TransportMessageBlockAllocator OpenDDS::DCPS::DataReaderImpl::mb_alloc_

protected

Definition at line 1108 of file DataReaderImpl.h.

Referenced by data_received().

◆ monitor_

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

private

Monitor object for this entity.

Definition at line 1042 of file DataReaderImpl.h.

Referenced by DataReaderImpl(), enable(), release_instance(), remove_associations_i(), and transport_assoc_done().

◆ multi_topic_

TopicDescriptionPtr<MultiTopicImpl> OpenDDS::DCPS::DataReaderImpl::multi_topic_

protected

Definition at line 801 of file DataReaderImpl.h.

Referenced by cleanup().

◆ n_chunks_

size_t OpenDDS::DCPS::DataReaderImpl::n_chunks_

private

Definition at line 885 of file DataReaderImpl.h.

Referenced by enable().

◆ participant_servant_

WeakRcHandle<DomainParticipantImpl> OpenDDS::DCPS::DataReaderImpl::participant_servant_

protected

Definition at line 785 of file DataReaderImpl.h.

Referenced by enable(), get_instance_handle(), get_matched_publication_data(), init(), transport_assoc_done(), and ~DataReaderImpl().

◆ passed_qos_

DDS::DataReaderQos OpenDDS::DCPS::DataReaderImpl::passed_qos_

protected

Definition at line 773 of file DataReaderImpl.h.

Referenced by get_qos(), init(), and set_qos().

◆ periodic_monitor_

unique_ptr<Monitor> OpenDDS::DCPS::DataReaderImpl::periodic_monitor_

private

Periodic Monitor object for this entity.

Definition at line 1045 of file DataReaderImpl.h.

Referenced by DataReaderImpl().

◆ publication_handle_lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::DataReaderImpl::publication_handle_lock_

private

Definition at line 888 of file DataReaderImpl.h.

Referenced by data_received(), get_matched_publications(), get_requested_incompatible_qos_status(), get_subscription_matched_status(), remove_associations_i(), and transport_assoc_done().

◆ publication_id_to_handle_map_

RepoIdToHandleMap OpenDDS::DCPS::DataReaderImpl::publication_id_to_handle_map_

private

Definition at line 891 of file DataReaderImpl.h.

Referenced by data_received(), get_matched_publications(), remove_associations_i(), and transport_assoc_done().

◆ qos_

DDS::DataReaderQos OpenDDS::DCPS::DataReaderImpl::qos_

protected

Definition at line 772 of file DataReaderImpl.h.

Referenced by add_association(), check_transport_qos(), enable(), init(), qos_change(), and set_qos().

◆ raw_latency_buffer_size_

unsigned int OpenDDS::DCPS::DataReaderImpl::raw_latency_buffer_size_

private

Bound (or initial reservation) of raw latency buffer.

Definition at line 1032 of file DataReaderImpl.h.

Referenced by add_association(), and raw_latency_buffer_size().

◆ raw_latency_buffer_type_

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

private

Type of raw latency data buffer.

Definition at line 1035 of file DataReaderImpl.h.

Referenced by add_association(), and raw_latency_buffer_type().

◆ rd_allocator_

unique_ptr<ReceivedDataAllocator> OpenDDS::DCPS::DataReaderImpl::rd_allocator_

protected

Definition at line 771 of file DataReaderImpl.h.

Referenced by enable().

◆ reactor_

ACE_Reactor_Timer_Interface* OpenDDS::DCPS::DataReaderImpl::reactor_

private

The orb's reactor to be used to register the liveliness timer.

Definition at line 916 of file DataReaderImpl.h.

Referenced by DataReaderImpl().

◆ read_conditions_

ReadConditionSet OpenDDS::DCPS::DataReaderImpl::read_conditions_

private

Definition at line 1039 of file DataReaderImpl.h.

Referenced by create_querycondition(), create_readcondition(), delete_contained_entities(), delete_readcondition(), has_readcondition(), and notify_read_conditions().

◆ requested_deadline_missed_status_

DDS::RequestedDeadlineMissedStatus OpenDDS::DCPS::DataReaderImpl::requested_deadline_missed_status_

private

Definition at line 895 of file DataReaderImpl.h.

Referenced by DataReaderImpl(), and get_requested_deadline_missed_status().

◆ requested_incompatible_qos_status_

DDS::RequestedIncompatibleQosStatus OpenDDS::DCPS::DataReaderImpl::requested_incompatible_qos_status_

private

Definition at line 896 of file DataReaderImpl.h.

Referenced by DataReaderImpl(), get_requested_incompatible_qos_status(), and update_incompatible_qos().

◆ reverse_sample_lock_

Reverse_Lock_t OpenDDS::DCPS::DataReaderImpl::reverse_sample_lock_

protected

Definition at line 783 of file DataReaderImpl.h.

Referenced by notify_read_conditions().

◆ sample_lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::DataReaderImpl::sample_lock_

protected

lock protecting sample container as well as statuses.

Definition at line 780 of file DataReaderImpl.h.

Referenced by contains_sample(), create_querycondition(), create_readcondition(), data_received(), delete_contained_entities(), delete_readcondition(), enable(), get_liveliness_changed_status(), get_requested_deadline_missed_status(), get_sample_lost_status(), get_sample_rejected_status(), OpenDDS::DCPS::QueryConditionImpl::get_trigger_value(), release_instance(), signal_liveliness(), state_updated(), and transport_assoc_done().

◆ sample_lost_status_

DDS::SampleLostStatus OpenDDS::DCPS::DataReaderImpl::sample_lost_status_

protected

Definition at line 777 of file DataReaderImpl.h.

Referenced by DataReaderImpl(), and get_sample_lost_status().

◆ sample_rejected_status_

DDS::SampleRejectedStatus OpenDDS::DCPS::DataReaderImpl::sample_rejected_status_

protected

Definition at line 776 of file DataReaderImpl.h.

Referenced by DataReaderImpl(), and get_sample_rejected_status().

◆ security_config_

Security::SecurityConfig_rch OpenDDS::DCPS::DataReaderImpl::security_config_

protected

Definition at line 1104 of file DataReaderImpl.h.

Referenced by enable().

◆ statistics_

StatsMapType OpenDDS::DCPS::DataReaderImpl::statistics_

private

Statistics for this reader, collected for each writer.

Definition at line 1028 of file DataReaderImpl.h.

Referenced by add_association(), raw_latency_statistics(), and remove_associations().

◆ statistics_enabled_

AtomicBool OpenDDS::DCPS::DataReaderImpl::statistics_enabled_

private

Flag indicating status of statistics gathering.

Definition at line 1016 of file DataReaderImpl.h.

◆ statistics_lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::DataReaderImpl::statistics_lock_

private

Definition at line 1029 of file DataReaderImpl.h.

Referenced by add_association(), and remove_associations().

◆ subqos_

DDS::SubscriberQos OpenDDS::DCPS::DataReaderImpl::subqos_

protected

Definition at line 810 of file DataReaderImpl.h.

◆ subscriber_servant_

WeakRcHandle<SubscriberImpl> OpenDDS::DCPS::DataReaderImpl::subscriber_servant_

private

Definition at line 881 of file DataReaderImpl.h.

Referenced by get_subscriber_servant(), init(), and parent().

◆ subscription_id_condition_

ConditionVariable<ACE_Thread_Mutex> OpenDDS::DCPS::DataReaderImpl::subscription_id_condition_

mutableprotected

Definition at line 769 of file DataReaderImpl.h.

Referenced by add_association(), and enable().

◆ subscription_id_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::DataReaderImpl::subscription_id_mutex_

mutableprotected

Definition at line 768 of file DataReaderImpl.h.

Referenced by add_association(), and enable().

◆ subscription_lost_status_

SubscriptionLostStatus OpenDDS::DCPS::DataReaderImpl::subscription_lost_status_

private

Todo:: The subscription_lost_status_ and subscription_reconnecting_status_ are left here for future use when we add get_subscription_lost_status() and get_subscription_reconnecting_status() methods.

Definition at line 909 of file DataReaderImpl.h.

◆ subscription_match_status_

DDS::SubscriptionMatchedStatus OpenDDS::DCPS::DataReaderImpl::subscription_match_status_

private

Definition at line 897 of file DataReaderImpl.h.

Referenced by DataReaderImpl(), get_subscription_matched_status(), remove_associations_i(), and transport_assoc_done().

◆ topic_desc_

DDS::TopicDescription_var OpenDDS::DCPS::DataReaderImpl::topic_desc_

private

Definition at line 872 of file DataReaderImpl.h.

Referenced by get_topicdescription(), and init().

◆ topic_id_

GUID_t OpenDDS::DCPS::DataReaderImpl::topic_id_

protected

Definition at line 788 of file DataReaderImpl.h.

Referenced by init().

◆ topic_servant_

TopicDescriptionPtr<TopicImpl> OpenDDS::DCPS::DataReaderImpl::topic_servant_

protected

Definition at line 786 of file DataReaderImpl.h.

Referenced by cleanup(), enable(), and init().

◆ transport_disabled_

bool OpenDDS::DCPS::DataReaderImpl::transport_disabled_

private

Definition at line 1047 of file DataReaderImpl.h.

Referenced by disable_transport(), and enable().

◆ type_support_

TypeSupportImpl* OpenDDS::DCPS::DataReaderImpl::type_support_

protected

Definition at line 787 of file DataReaderImpl.h.

Referenced by init().

◆ writers_

WriterMapType OpenDDS::DCPS::DataReaderImpl::writers_

private

Definition at line 1022 of file DataReaderImpl.h.

Referenced by add_association(), data_received(), remove_all_associations(), remove_associations_i(), signal_liveliness(), transport_assoc_done(), and writer_activity().

◆ writers_lock_

ACE_RW_Thread_Mutex OpenDDS::DCPS::DataReaderImpl::writers_lock_

private

RW lock for reading/writing publications.

Definition at line 1025 of file DataReaderImpl.h.

Referenced by add_association(), data_received(), remove_all_associations(), remove_associations(), remove_associations_i(), signal_liveliness(), transport_assoc_done(), and writer_activity().

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

Classes

Public Types

Public Member Functions

Protected Types

Protected Member Functions

Static Protected Member Functions

Protected Attributes

Static Protected Attributes

Private Types

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ DRISporadicTask

◆ RCCompLess

◆ Reverse_Lock_t

◆ WriterStatePair

Constructor & Destructor Documentation

◆ DataReaderImpl()

◆ ~DataReaderImpl()

Member Function Documentation

◆ accept_coherent()

◆ accept_sample_processing()

◆ add_association()

◆ add_link()

◆ begin_access()

◆ cancel_all_deadlines()

◆ cancel_deadline()

◆ check_historic()

◆ check_transport_qos()

◆ cleanup()

◆ coherent_change_received() [1/2]

◆ coherent_change_received() [2/2]

◆ coherent_changes_completed()

◆ contains_sample()

◆ contains_sample_filtered()

◆ create_querycondition()

◆ create_readcondition()

◆ data_received()

◆ dds_demarshal()

◆ deadline_task()

◆ delete_contained_entities()

◆ delete_readcondition()

◆ deliver_historic()

◆ disable_transport()

◆ dispose_unregister()

◆ domain_id()

◆ enable()

◆ enable_filtering()

◆ enable_multi_topic()

◆ enable_specific()

◆ end_access()

◆ filter_sample()

◆ get_builtin_subscriber_proxy()

◆ get_cf_topic()

◆ get_crypto_handle()

◆ get_depth()

◆ get_dp_id()

◆ get_ext_listener()

◆ get_guid()

◆ get_handle_instance()

◆ get_ice_endpoint()

◆ get_instance_handle()

◆ get_instance_handles()

◆ get_latency_stats()

◆ get_listener()

◆ get_liveliness_changed_status()

◆ get_matched_publication_data()

◆ get_matched_publications()

◆ get_n_chunks()

◆ get_next_handle()

◆ get_ordered_data()

◆ get_priority_value()

◆ get_qos()

◆ get_reactor()

◆ get_requested_deadline_missed_status()

◆ get_requested_incompatible_qos_status()

◆ get_sample_lost_status()