OpenDDS::DCPS::DataReaderImpl_T< MessageType > Class Template Reference

#include <DataReaderImpl_T.h>

Inheritance diagram for OpenDDS::DCPS::DataReaderImpl_T< MessageType >:

Collaboration diagram for OpenDDS::DCPS::DataReaderImpl_T< MessageType >:

Classes
struct	FilterDelayedSample
struct	MessageTypeMemoryBlock
class	MessageTypeWithAllocator
class	SharedInstanceMap

Public Types
typedef DDSTraits< MessageType >	TraitsType
typedef MarshalTraits< MessageType >	MarshalTraitsType
typedef TraitsType::MessageSequenceType	MessageSequenceType
typedef RcHandle< SharedInstanceMap >	SharedInstanceMap_rch
typedef TraitsType::DataReaderType	Interface
typedef OpenDDS::DCPS::Cached_Allocator_With_Overflow< MessageTypeMemoryBlock, ACE_Thread_Mutex >	DataAllocator
Public Types inherited from OpenDDS::DCPS::LocalObject< DDSTraits< MessageType >::DataReaderType >
typedef DDSTraits< MessageType >::DataReaderType ::_ptr_type	_ptr_type
typedef DDSTraits< MessageType >::DataReaderType ::_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::DataReaderImpl
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 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_CMP_T (MessageType, DDS::InstanceHandle_t, typename TraitsType::LessThanType) InstanceMap
typedef	OPENDDS_MAP (DDS::InstanceHandle_t, typename InstanceMap::iterator) ReverseInstanceMap
CORBA::Boolean	_is_a (const char *type_id)
const char *	_interface_repository_id () const
CORBA::Boolean	marshal (TAO_OutputCDR &)
	DataReaderImpl_T ()
virtual	~DataReaderImpl_T ()
virtual DDS::ReturnCode_t	enable_specific ()
virtual DDS::ReturnCode_t	read (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)
virtual DDS::ReturnCode_t	take (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)
virtual DDS::ReturnCode_t	read_w_condition (MessageSequenceType &received_data, DDS::SampleInfoSeq &sample_info, ::CORBA::Long max_samples, DDS::ReadCondition_ptr a_condition)
virtual DDS::ReturnCode_t	take_w_condition (MessageSequenceType &received_data, DDS::SampleInfoSeq &sample_info, ::CORBA::Long max_samples, DDS::ReadCondition_ptr a_condition)
virtual DDS::ReturnCode_t	read_next_sample (MessageType &received_data, DDS::SampleInfo &sample_info_ref)
virtual DDS::ReturnCode_t	take_next_sample (MessageType &received_data, DDS::SampleInfo &sample_info_ref)
virtual DDS::ReturnCode_t	read_instance (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)
virtual DDS::ReturnCode_t	take_instance (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)
virtual DDS::ReturnCode_t	read_instance_w_condition (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::ReadCondition_ptr a_condition)
virtual DDS::ReturnCode_t	take_instance_w_condition (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::ReadCondition_ptr a_condition)
virtual DDS::ReturnCode_t	read_next_instance (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)
virtual DDS::ReturnCode_t	take_next_instance (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)
virtual DDS::ReturnCode_t	read_next_instance_w_condition (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::ReadCondition_ptr a_condition)
virtual DDS::ReturnCode_t	take_next_instance_w_condition (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::ReadCondition_ptr a_condition)
virtual DDS::ReturnCode_t	return_loan (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq)
virtual DDS::ReturnCode_t	get_key_value (MessageType &key_holder, DDS::InstanceHandle_t handle)
virtual DDS::InstanceHandle_t	lookup_instance (const MessageType &instance_data)
virtual DDS::ReturnCode_t	auto_return_loan (void *seq)
void	release_loan (MessageSequenceType &received_data)
bool	contains_sample_filtered (DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, const FilterEvaluator &evaluator, const DDS::StringSeq &params)
DDS::ReturnCode_t	read_generic (GenericBundle &gen, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, bool adjust_ref_count=false)
DDS::InstanceHandle_t	lookup_instance_generic (const void *data)
virtual DDS::ReturnCode_t	take (AbstractSamples &samples, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)
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)
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)
DDS::InstanceHandle_t	store_synthetic_data (const MessageType &sample, DDS::ViewStateKind view, const SystemTimePoint &timestamp=SystemTimePoint::now())
void	set_instance_state_i (DDS::InstanceHandle_t instance, DDS::InstanceHandle_t publication_handle, DDS::InstanceStateKind state, const SystemTimePoint &timestamp, const GUID_t &publication_id)
virtual void	lookup_instance (const OpenDDS::DCPS::ReceivedDataSample &sample, OpenDDS::DCPS::SubscriptionInstance_rch &instance)
virtual void	qos_change (const DDS::DataReaderQos &qos)
void	set_marshal_skip_serialize (bool value)
bool	get_marshal_skip_serialize () const
void	release_all_instances ()
	Release all instances held by the reader. More...
template<>
DDS::ReturnCode_t	read_generic (GenericBundle &, DDS::SampleStateMask, DDS::ViewStateMask, DDS::InstanceStateMask, bool)
template<>
DDS::ReturnCode_t	take (AbstractSamples &, DDS::SampleStateMask, DDS::ViewStateMask, DDS::InstanceStateMask)
template<>
OpenDDS_Dcps_Export DDS::ReturnCode_t	read_generic (GenericBundle &gen, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, bool adjust_ref_count)
template<>
OpenDDS_Dcps_Export DDS::ReturnCode_t	take (AbstractSamples &samples, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states)
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)
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)
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::DataReaderImpl
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)
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)
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 ()
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
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
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::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 Member Functions
virtual RcHandle< MessageHolder >	dds_demarshal (const OpenDDS::DCPS::ReceivedDataSample &sample, DDS::InstanceHandle_t publication_handle, OpenDDS::DCPS::SubscriptionInstance_rch &instance, bool &just_registered, bool &filtered, OpenDDS::DCPS::MarshalingType marshaling_type, bool full_copy)
virtual void	dispose_unregister (const OpenDDS::DCPS::ReceivedDataSample &sample, DDS::InstanceHandle_t publication_handle, OpenDDS::DCPS::SubscriptionInstance_rch &instance)
virtual void	purge_data (OpenDDS::DCPS::SubscriptionInstance_rch instance)
virtual void	release_instance_i (DDS::InstanceHandle_t handle)
virtual void	state_updated_i (DDS::InstanceHandle_t handle)
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::DataReaderImpl
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 ()
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)
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)
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 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 ()

Private Types
typedef DCPS::PmfSporadicTask< DataReaderImpl_T >	DRISporadicTask
typedef ACE_Strong_Bound_Ptr< const OpenDDS::DCPS::DataSampleHeader, ACE_Null_Mutex >	DataSampleHeader_ptr

Private Member Functions
void	dynamic_hook (MessageType &)
bool	store_instance_data_check (unique_ptr< MessageTypeWithAllocator > &instance_data, DDS::InstanceHandle_t publication_handle, const OpenDDS::DCPS::DataSampleHeader &header, OpenDDS::DCPS::SubscriptionInstance_rch &instance_ptr)
DDS::ReturnCode_t	read_i (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, CORBA::Long max_samples, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, DDS::QueryCondition_ptr a_condition)
DDS::ReturnCode_t	take_i (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, CORBA::Long max_samples, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, DDS::QueryCondition_ptr a_condition)
DDS::ReturnCode_t	read_instance_i (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, DDS::QueryCondition_ptr a_condition)
DDS::ReturnCode_t	take_instance_i (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, DDS::QueryCondition_ptr a_condition)
DDS::ReturnCode_t	read_next_instance_i (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, DDS::QueryCondition_ptr a_condition)
DDS::ReturnCode_t	take_next_instance_i (MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, CORBA::Long max_samples, DDS::InstanceHandle_t a_handle, DDS::SampleStateMask sample_states, DDS::ViewStateMask view_states, DDS::InstanceStateMask instance_states, DDS::QueryCondition_ptr a_condition)
void	store_instance_data (unique_ptr< MessageTypeWithAllocator > instance_data, DDS::InstanceHandle_t publication_handle, const OpenDDS::DCPS::DataSampleHeader &header, OpenDDS::DCPS::SubscriptionInstance_rch &instance_ptr, bool &just_registered, bool &filtered)
void	finish_store_instance_data (unique_ptr< MessageTypeWithAllocator > instance_data, const DataSampleHeader &header, SubscriptionInstance_rch instance_ptr, bool is_dispose_msg, bool is_unregister_msg)
void	notify_status_condition_no_sample_lock ()
DDS::ReturnCode_t	check_inputs (const char *method_name, MessageSequenceType &received_data, DDS::SampleInfoSeq &info_seq, ::CORBA::Long max_samples)
	Common input read* & take* input processing and precondition checks. More...
void	delay_sample (DDS::InstanceHandle_t handle, unique_ptr< MessageTypeWithAllocator > data, const OpenDDS::DCPS::DataSampleHeader &header, const bool just_registered, const MonotonicTimePoint &now, const MonotonicTimePoint &deadline)
void	clear_sample (DDS::InstanceHandle_t handle)
void	drop_sample (DDS::InstanceHandle_t handle)
void	filter_delayed (const MonotonicTimePoint &now)
unique_ptr< DataAllocator > &	data_allocator ()
typedef	OPENDDS_MAP (DDS::InstanceHandle_t, FilterDelayedSample) FilterDelayedSampleMap
typedef	OPENDDS_MULTIMAP (MonotonicTimePoint, DDS::InstanceHandle_t) FilterDelayedSampleQueue
template<>
void	dynamic_hook (XTypes::DynamicSample &sample)
template<>
void	dynamic_hook (XTypes::DynamicSample &sample)

Private Attributes
unique_ptr< DataAllocator >	data_allocator_
InstanceMap	instance_map_
ReverseInstanceMap	reverse_instance_map_
RcHandle< DRISporadicTask >	filter_delayed_sample_task_
FilterDelayedSampleMap	filter_delayed_sample_map_
FilterDelayedSampleQueue	filter_delayed_sample_queue_
bool	marshal_skip_serialize_

Additional Inherited Members
Static Public Member Functions inherited from OpenDDS::DCPS::LocalObject< DDSTraits< MessageType >::DataReaderType >
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< DataReaderEx >
static _ptr_type	_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...
Protected Types inherited from OpenDDS::DCPS::DataReaderImpl
typedef ACE_Reverse_Lock< ACE_Recursive_Thread_Mutex >	Reverse_Lock_t
Static Protected Member Functions inherited from OpenDDS::DCPS::DataReaderImpl
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 inherited from CORBA::Object
ACE_Atomic_Op< TAO_SYNCH_MUTEX, unsigned long >	refcount_
Protected Attributes inherited from OpenDDS::DCPS::DataReaderImpl
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 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 inherited from OpenDDS::DCPS::DataReaderImpl
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

Detailed Description

template<typename MessageType>
class OpenDDS::DCPS::DataReaderImpl_T< MessageType >

Servant for DataReader interface of Traits::MessageType data type.

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

Definition at line 41 of file DataReaderImpl_T.h.

Member Typedef Documentation

DataAllocator

template<typename MessageType>

typedef OpenDDS::DCPS::Cached_Allocator_With_Overflow<MessageTypeMemoryBlock, ACE_Thread_Mutex> OpenDDS::DCPS::DataReaderImpl_T< MessageType >::DataAllocator

Definition at line 118 of file DataReaderImpl_T.h.

DataSampleHeader_ptr

template<typename MessageType>

typedef ACE_Strong_Bound_Ptr<const OpenDDS::DCPS::DataSampleHeader, ACE_Null_Mutex> OpenDDS::DCPS::DataReaderImpl_T< MessageType >::DataSampleHeader_ptr

private

Definition at line 2426 of file DataReaderImpl_T.h.

DRISporadicTask

template<typename MessageType>

typedef DCPS::PmfSporadicTask<DataReaderImpl_T> OpenDDS::DCPS::DataReaderImpl_T< MessageType >::DRISporadicTask

private

Definition at line 2420 of file DataReaderImpl_T.h.

Interface

template<typename MessageType>

typedef TraitsType::DataReaderType OpenDDS::DCPS::DataReaderImpl_T< MessageType >::Interface

Definition at line 66 of file DataReaderImpl_T.h.

MarshalTraitsType

template<typename MessageType>

typedef MarshalTraits<MessageType> OpenDDS::DCPS::DataReaderImpl_T< MessageType >::MarshalTraitsType

Definition at line 51 of file DataReaderImpl_T.h.

MessageSequenceType

template<typename MessageType>

typedef TraitsType::MessageSequenceType OpenDDS::DCPS::DataReaderImpl_T< MessageType >::MessageSequenceType

Definition at line 52 of file DataReaderImpl_T.h.

SharedInstanceMap_rch

template<typename MessageType>

typedef RcHandle<SharedInstanceMap> OpenDDS::DCPS::DataReaderImpl_T< MessageType >::SharedInstanceMap_rch

Definition at line 64 of file DataReaderImpl_T.h.

TraitsType

template<typename MessageType>

typedef DDSTraits<MessageType> OpenDDS::DCPS::DataReaderImpl_T< MessageType >::TraitsType

Definition at line 50 of file DataReaderImpl_T.h.

Constructor & Destructor Documentation

DataReaderImpl_T()

template<typename MessageType>

OpenDDS::DCPS::DataReaderImpl_T< MessageType >::DataReaderImpl_T ( )

inline

Definition at line 120 of file DataReaderImpl_T.h.

      : filter_delayed_sample_task_(make_rch<DRISporadicTask>(TheServiceParticipant->time_source(), TheServiceParticipant->interceptor(), rchandle_from(this), &DataReaderImpl_T::filter_delayed))
      , marshal_skip_serialize_(false)
    {
      initialize_lookup_maps();
    }

~DataReaderImpl_T()

template<typename MessageType>

virtual OpenDDS::DCPS::DataReaderImpl_T< MessageType >::~DataReaderImpl_T ( )

inlinevirtual

Definition at line 127 of file DataReaderImpl_T.h.

    {
      filter_delayed_sample_task_->cancel();

      for (typename InstanceMap::iterator it = instance_map_.begin();
           it != instance_map_.end(); ++it)
        {
          OpenDDS::DCPS::SubscriptionInstance_rch ptr = get_handle_instance(it->second);
          if (!ptr) continue;
          purge_data(ptr);
        }
      //X SHH release the data samples in the instance_map_.
    }

Member Function Documentation

_interface_repository_id()

template<typename MessageType>

const char* OpenDDS::DCPS::DataReaderImpl_T< MessageType >::_interface_repository_id ( void  ) const

inlinevirtual

Reimplemented from CORBA::Object.

Definition at line 73 of file DataReaderImpl_T.h.

    {
      return Interface::_interface_repository_id();
    }

_is_a()

template<typename MessageType>

CORBA::Boolean OpenDDS::DCPS::DataReaderImpl_T< MessageType >::_is_a ( const char *  type_id )

inlinevirtual

Reimplemented from CORBA::Object.

Definition at line 68 of file DataReaderImpl_T.h.

    {
      return Interface::_is_a(type_id);
    }

auto_return_loan()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::auto_return_loan ( void *  seq )

inlinevirtual

Definition at line 682 of file DataReaderImpl_T.h.

  {
    MessageSequenceType& received_data =
      *static_cast< MessageSequenceType*> (seq);

    if (!received_data.release())
      {
        // release_loan(received_data);
        received_data.length(0);
      }
    return DDS::RETCODE_OK;
  }

check_inputs()

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::check_inputs ( const char *  method_name, MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples  )

inlineprivate

Common input read* & take* input processing and precondition checks.

Definition at line 2180 of file DataReaderImpl_T.h.

{
  typename DDSTraits<MessageType>::MessageSequenceAdapterType received_data_p(received_data);

  // ---- start of preconditions common to read and take -----
  // SPEC ref v1.2 7.1.2.5.3.8 #1
  // NOTE: We can't check maximum() or release() here since those are
  //       implementation details of the sequences.  In general, the
  //       info_seq will have release() == true and maximum() == 0.
  //       If we're in zero-copy mode, the received_data will have
  //       release() == false and maximum() == 0.  If it's not
  //       zero-copy then received_data will have release == true()
  //       and maximum() == anything.
  if (received_data.length() != info_seq.length())
    {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) %CDataReaderImpl::%C ")
                 ACE_TEXT("PRECONDITION_NOT_MET sample and info input ")
                 ACE_TEXT("sequences do not match.\n"),
                 TraitsType::type_name(),
                 method_name ));
      return DDS::RETCODE_PRECONDITION_NOT_MET;
    }

  //SPEC ref v1.2 7.1.2.5.3.8 #4
  if ((received_data.maximum() > 0) && (received_data.release() == false))
    {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) %CDataReaderImpl::%C ")
                 ACE_TEXT("PRECONDITION_NOT_MET mismatch of ")
                 ACE_TEXT("maximum %d and owns %d\n"),
                 TraitsType::type_name(),
                 method_name,
                 received_data.maximum(),
                 received_data.release() ));

      return DDS::RETCODE_PRECONDITION_NOT_MET;
    }

  if (received_data.maximum() == 0)
    {
      // not in SPEC but needed.
      if (max_samples == DDS::LENGTH_UNLIMITED)
        {
          max_samples =
            static_cast< ::CORBA::Long> (received_data_p.max_slots());
        }
    }
  else
    {
      if (max_samples == DDS::LENGTH_UNLIMITED)
        {
          //SPEC ref v1.2 7.1.2.5.3.8 #5a
          max_samples = received_data.maximum();
        }
      else if (
               max_samples > static_cast< ::CORBA::Long> (received_data.maximum()))
        {
          //SPEC ref v1.2 7.1.2.5.3.8 #5c
          ACE_DEBUG((LM_DEBUG,
                     ACE_TEXT("(%P|%t) %CDataReaderImpl::%C ")
                     ACE_TEXT("PRECONDITION_NOT_MET max_samples %d > maximum %d\n"),
                     TraitsType::type_name(),
                     method_name,
                     max_samples,
                     received_data.maximum()));
          return DDS::RETCODE_PRECONDITION_NOT_MET;
        }
      //else
      //SPEC ref v1.2 7.1.2.5.3.8 #5b - is true by impl below.
    }

  // The spec does not say what to do in this case but it appears to be a good thing.
  // Note: max_slots is the greater of the sequence's maximum and init_size.
  if (static_cast< ::CORBA::Long> (received_data_p.max_slots()) < max_samples)
    {
      max_samples = static_cast< ::CORBA::Long> (received_data_p.max_slots());
    }
  //---- end of preconditions common to read and take -----

  return DDS::RETCODE_OK;
}

clear_sample()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::clear_sample ( DDS::InstanceHandle_t  handle )

inlineprivate

Definition at line 2310 of file DataReaderImpl_T.h.

{
  // sample_lock_ should already be held

  typename FilterDelayedSampleMap::iterator sample = filter_delayed_sample_map_.find(handle);
  if (sample != filter_delayed_sample_map_.end()) {
    // leave the entry in the container, so that the key remains valid if the reactor is waiting on this lock while this is occurring
    sample->second.message.reset();
  }
}

contains_sample_filtered()

template<typename MessageType>

bool OpenDDS::DCPS::DataReaderImpl_T< MessageType >::contains_sample_filtered ( DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states, const FilterEvaluator &  evaluator, const DDS::StringSeq &  params  )

inlinevirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 701 of file DataReaderImpl_T.h.

  {
    ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, sample_lock_, false);
    ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, instance_guard, instances_lock_, false);

    TopicDescriptionPtr<TopicImpl> topic(topic_servant_);
    TypeSupport* const ts = topic->get_type_support();
    TypeSupportImpl* const type_support = dynamic_cast<TypeSupportImpl*>(ts);
    const bool filter_has_non_key_fields = type_support ? evaluator.has_non_key_fields(*type_support) : true;

    const HandleSet& matches = lookup_matching_instances(sample_states, view_states, instance_states);
    for (HandleSet::const_iterator it = matches.begin(), next = it; it != matches.end(); it = next) {
      ++next; // pre-increment iterator, in case updates cause changes to match set
      const DDS::InstanceHandle_t handle = *it;
      const SubscriptionInstance_rch inst = get_handle_instance(handle);
      if (!inst) continue;

      for (ReceivedDataElement* item = inst->rcvd_samples_.get_next_match(sample_states, 0); item;
           item = inst->rcvd_samples_.get_next_match(sample_states, item)) {
        if (!item->registered_data_ || (!item->valid_data_ && filter_has_non_key_fields)) {
          continue;
        }
        if (evaluator.eval(*static_cast<MessageType*>(item->registered_data_), params)) {
          return true;
        }
      }
    }

    return false;
  }

data_allocator()

template<typename MessageType>

unique_ptr<DataAllocator>& OpenDDS::DCPS::DataReaderImpl_T< MessageType >::data_allocator ( )

inlineprivate

Definition at line 2413 of file DataReaderImpl_T.h.

{ return data_allocator_; }

dds_demarshal()

template<typename MessageType>

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

inlineprotectedvirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 1067 of file DataReaderImpl_T.h.

  {
    unique_ptr<MessageTypeWithAllocator> data(new (*data_allocator()) MessageTypeWithAllocator);
    dynamic_hook(*data);
    RcHandle<MessageHolder> message_holder;

    Message_Block_Ptr payload(sample.data(&mb_alloc_));
    if (marshal_skip_serialize_) {
      if (!MarshalTraitsType::from_message_block(*data, *payload)) {
        if (DCPS_debug_level > 0) {
          ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: DataReaderImpl::dds_demarshal: ")
                    ACE_TEXT("attempting to skip serialize but bad from_message_block. Returning from demarshal.\n")));
        }
        return message_holder;
      }
      store_instance_data(move(data), publication_handle, sample.header_, instance, just_registered, filtered);
      return message_holder;
    }
    const bool encapsulated = sample.header_.cdr_encapsulation_;

    OpenDDS::DCPS::Serializer ser(
      payload.get(),
      encapsulated ? Encoding::KIND_XCDR1 : Encoding::KIND_UNALIGNED_CDR,
      static_cast<Endianness>(sample.header_.byte_order_));

    if (encapsulated) {
      EncapsulationHeader encap;
      if (!(ser >> encap)) {
        if (DCPS_debug_level > 0) {
          ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR ")
            ACE_TEXT("%CDataReaderImpl::dds_demarshal: ")
            ACE_TEXT("deserialization of encapsulation header failed.\n"),
            TraitsType::type_name()));
        }
        return message_holder;
      }
      Encoding encoding;
      if (!encap.to_encoding(encoding, type_support_->base_extensibility())) {
        return message_holder;
      }

      if (decoding_modes_.find(encoding.kind()) == decoding_modes_.end()) {
        if (DCPS_debug_level >= 1) {
          ACE_DEBUG((LM_WARNING, ACE_TEXT("(%P|%t) WARNING ")
            ACE_TEXT("%CDataReaderImpl::dds_demarshal: ")
            ACE_TEXT("Encoding kind %C of the received sample does not ")
            ACE_TEXT("match the ones specified by DataReader.\n"),
            TraitsType::type_name(),
            Encoding::kind_to_string(encoding.kind()).c_str()));
        }
        return message_holder;
      }
      if (DCPS_debug_level >= 8) {
        ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ")
          ACE_TEXT("%CDataReaderImpl::dds_demarshal: ")
          ACE_TEXT("Deserializing with encoding kind %C.\n"),
          TraitsType::type_name(),
          Encoding::kind_to_string(encoding.kind()).c_str()));
      }

      ser.encoding(encoding);
    }

    const bool key_only_marshaling =
      marshaling_type == OpenDDS::DCPS::KEY_ONLY_MARSHALING;

    bool ser_ret = true;
    if (key_only_marshaling) {
      ser_ret = ser >> OpenDDS::DCPS::KeyOnly<MessageType>(*data);
    } else {
      ser_ret = ser >> *data;
      if (full_copy) {
        message_holder = make_rch<MessageHolder_T<MessageType> >(*data);
      }
    }
    if (!ser_ret) {
      if (ser.get_construction_status() != Serializer::ConstructionSuccessful) {
        if (DCPS_debug_level > 1) {
          ACE_DEBUG((LM_WARNING, ACE_TEXT("(%P|%t) %CDataReaderImpl::dds_demarshal ")
                     ACE_TEXT("object construction failure, dropping sample.\n"),
                     TraitsType::type_name()));
        }
      } else {
        if (DCPS_debug_level > 0) {
          ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR %CDataReaderImpl::dds_demarshal ")
                    ACE_TEXT("deserialization failed, dropping sample.\n"),
                    TraitsType::type_name()));
        }
      }
      return message_holder;
    }

#ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
    /*
     * If sample.header_.content_filter_ is true, the writer has already
     * filtered.
     */
    if (!sample.header_.content_filter_) {
      ACE_Guard<ACE_Thread_Mutex> guard(content_filtered_topic_mutex_);
      if (content_filtered_topic_) {
        const bool sample_only_has_key_fields = !sample.header_.valid_data();
        if (key_only_marshaling != sample_only_has_key_fields) {
          if (DCPS_debug_level > 0) {
            ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR ")
              ACE_TEXT("%CDataReaderImpl::dds_demarshal: ")
              ACE_TEXT("Mismatch between the key only and valid data properties ")
              ACE_TEXT("of a %C message of a content filtered topic!\n"),
              TraitsType::type_name(),
              to_string(static_cast<MessageId>(sample.header_.message_id_))));
          }
          filtered = true;
          message_holder.reset();
          return message_holder;
        }
        const MessageType& type = static_cast<MessageType&>(*data);
        if (!content_filtered_topic_->filter(type, sample_only_has_key_fields)) {
          filtered = true;
          message_holder.reset();
          return message_holder;
        }
      }
    }
#endif

    store_instance_data(move(data), publication_handle, sample.header_, instance, just_registered, filtered);
    return message_holder;
  }

delay_sample()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::delay_sample ( DDS::InstanceHandle_t  handle, unique_ptr< MessageTypeWithAllocator >  data, const OpenDDS::DCPS::DataSampleHeader &  header, const bool  just_registered, const MonotonicTimePoint &  now, const MonotonicTimePoint &  deadline  )

inlineprivate

Definition at line 2266 of file DataReaderImpl_T.h.

{
  // sample_lock_ should already be held
  DataSampleHeader_ptr hdr(new OpenDDS::DCPS::DataSampleHeader(header));

  typename FilterDelayedSampleMap::iterator i = filter_delayed_sample_map_.find(handle);
  if (i == filter_delayed_sample_map_.end()) {

    // emplace()/insert() only if the sample is going to be
    // new (otherwise we call move(data) twice).
    std::pair<typename FilterDelayedSampleMap::iterator, bool> result =
#ifdef ACE_HAS_CPP11
      filter_delayed_sample_map_.emplace(std::piecewise_construct,
                                         std::forward_as_tuple(handle),
                                         std::forward_as_tuple(move(data), hdr, just_registered));
#else
      filter_delayed_sample_map_.insert(std::make_pair(handle, FilterDelayedSample(move(data), hdr, just_registered)));
#endif
    FilterDelayedSample& sample = result.first->second;
    sample.expiration_time = deadline;
    const bool schedule = filter_delayed_sample_queue_.empty();
    filter_delayed_sample_queue_.insert(std::make_pair(deadline, handle));
    if (schedule) {
      filter_delayed_sample_task_->schedule(now - deadline);
    } else if (filter_delayed_sample_queue_.begin()->second == handle) {
      filter_delayed_sample_task_->cancel();
      filter_delayed_sample_task_->schedule(now - deadline);
    }
  } else {
    FilterDelayedSample& sample = i->second;
    // we only care about the most recently filtered sample, so clean up the last one

    sample.message = move(data);
    sample.header = hdr;
    sample.new_instance = just_registered;
    // already scheduled for timeout at the desired time
  }
}

dispose_unregister()

template<typename MessageType>

virtual void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::dispose_unregister ( const OpenDDS::DCPS::ReceivedDataSample &  sample, DDS::InstanceHandle_t  publication_handle, OpenDDS::DCPS::SubscriptionInstance_rch &  instance  )

inlineprotectedvirtual

!! caller should already have the sample_lock_

Reimplemented from OpenDDS::DCPS::DataReaderImpl.

Definition at line 1201 of file DataReaderImpl_T.h.

  {
    //!!! caller should already have the sample_lock_

    // The data sample in this dispose message does not contain any valid data.
    // What it needs here is the key value to identify the instance to dispose.
    // The demarshal push this "sample" to received sample list so the user
    // can be notified the dispose event.
    bool just_registered = false;
    bool filtered = false;
    OpenDDS::DCPS::MarshalingType marshaling = OpenDDS::DCPS::FULL_MARSHALING;
    if (sample.header_.key_fields_only_) {
      marshaling = OpenDDS::DCPS::KEY_ONLY_MARSHALING;
    }
    dds_demarshal(sample, publication_handle, instance, just_registered, filtered, marshaling, false);
  }

drop_sample()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::drop_sample ( DDS::InstanceHandle_t  handle )

inlineprivate

Definition at line 2321 of file DataReaderImpl_T.h.

{
  // sample_lock_ should already be held

  typename FilterDelayedSampleMap::iterator sample = filter_delayed_sample_map_.find(handle);
  if (sample != filter_delayed_sample_map_.end()) {
    for (FilterDelayedSampleQueue::iterator pos = filter_delayed_sample_queue_.lower_bound(sample->second.expiration_time), limit = filter_delayed_sample_queue_.upper_bound(sample->second.expiration_time); pos != limit; ++pos) {
      if (pos->second == handle) {
        filter_delayed_sample_queue_.erase(pos);
        break;
      }
    }

    // use the handle to erase, since the sample lock was released
    filter_delayed_sample_map_.erase(handle);
  }
}

dynamic_hook() [1/3]

template<>

void OpenDDS::DCPS::DataReaderImpl_T< XTypes::DynamicSample >::dynamic_hook ( XTypes::DynamicSample &  sample )

private

dynamic_hook() [2/3]

template<>

void OpenDDS::DCPS::DataReaderImpl_T< XTypes::DynamicSample >::dynamic_hook ( XTypes::DynamicSample &  sample )

inlineprivate

Definition at line 114 of file DynamicDataReaderImpl.h.

  {
    XTypes::DynamicDataReaderImpl* const self = dynamic_cast<XTypes::DynamicDataReaderImpl*>(this);
    if (self) {
      self->imbue_type(sample);
    }
  }

dynamic_hook() [3/3]

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::dynamic_hook ( MessageType &  )

inlineprivate

Available for specialization so that some types of MessageType can observe and change the sample before dds_demarshal deserializes into it

Definition at line 1257 of file DataReaderImpl_T.h.

{}

enable_specific()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::enable_specific ( )

inlinevirtual

Do parts of enable specific to the datatype. Called by DataReaderImpl::enable().

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 145 of file DataReaderImpl_T.h.

    {
      data_allocator().reset(new DataAllocator(get_n_chunks ()));
      if (OpenDDS::DCPS::DCPS_debug_level >= 2)
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) %CDataReaderImpl::")
                   ACE_TEXT("enable_specific-data")
                   ACE_TEXT(" Cached_Allocator_With_Overflow ")
                   ACE_TEXT("%x with %d chunks\n"),
                   TraitsType::type_name(),
                   data_allocator().get(),
                   get_n_chunks ()));

      return DDS::RETCODE_OK;
    }

filter_delayed()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::filter_delayed ( const MonotonicTimePoint &  now )

inlineprivate

Definition at line 2339 of file DataReaderImpl_T.h.

{
  ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());

  // Make a copy because finish_store_instance_data will release the sample lock.
  typedef OPENDDS_VECTOR(DDS::InstanceHandle_t) Handles;
  Handles handles;

  ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);

  for (FilterDelayedSampleQueue::iterator pos = filter_delayed_sample_queue_.begin(), limit = filter_delayed_sample_queue_.end(); pos != limit && pos->first <= now;) {
    handles.push_back(pos->second);
    filter_delayed_sample_queue_.erase(pos++);
  }

  const TimeDuration interval(qos_.time_based_filter.minimum_separation);

  for (Handles::const_iterator pos = handles.begin(), limit = handles.end(); pos != limit; ++pos) {
    const DDS::InstanceHandle_t handle = *pos;

    SubscriptionInstance_rch instance = get_handle_instance(handle);
    if (!instance) {
      continue;
    }

    typename FilterDelayedSampleMap::iterator data = filter_delayed_sample_map_.find(handle);
    if (data == filter_delayed_sample_map_.end()) {
      continue;
    }

    if (data->second.message) {
      const bool NOT_DISPOSE_MSG = false;
      const bool NOT_UNREGISTER_MSG = false;
      // clear the message, since ownership is being transferred to finish_store_instance_data.

      instance->last_accepted_.set_to_now();
      const DataSampleHeader_ptr header = data->second.header;
      const bool new_instance = data->second.new_instance;

      // should not use data iterator anymore, since finish_store_instance_data releases sample_lock_
      finish_store_instance_data(move(data->second.message),
                                 *header,
                                 instance,
                                 NOT_DISPOSE_MSG,
                                 NOT_UNREGISTER_MSG);

      accept_sample_processing(instance, *header, new_instance);

      // Refresh the iterator.
      data = filter_delayed_sample_map_.find(handle);
      if (data == filter_delayed_sample_map_.end()) {
        continue;
      }

      // Reschedule.
      data->second.expiration_time = now + interval;
      filter_delayed_sample_queue_.insert(std::make_pair(data->second.expiration_time, handle));

    } else {
      // this check is performed to handle the corner case where
      // store_instance_data received and delivered a sample, while this
      // method was waiting for the lock
      if (MonotonicTimePoint::now() - instance->last_sample_tv_ >= interval) {
        // no new data to process, so remove from container
        filter_delayed_sample_map_.erase(data);
      }
    }
  }

  if (!filter_delayed_sample_queue_.empty()) {
    filter_delayed_sample_task_->schedule(filter_delayed_sample_queue_.begin()->first - now);
  }
}

finish_store_instance_data()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::finish_store_instance_data ( unique_ptr< MessageTypeWithAllocator >  instance_data, const DataSampleHeader &  header, SubscriptionInstance_rch  instance_ptr, bool  is_dispose_msg, bool  is_unregister_msg  )

inlineprivate

Definition at line 1945 of file DataReaderImpl_T.h.

{
  if ((qos_.resource_limits.max_samples_per_instance !=
        DDS::LENGTH_UNLIMITED) &&
      (instance_ptr->rcvd_samples_.size() >=
        static_cast<size_t>(qos_.resource_limits.max_samples_per_instance))) {

    // According to spec 1.2, Samples that contain no data do not
    // count towards the limits imposed by the RESOURCE_LIMITS QoS policy
    // so do not remove the oldest sample when unregister/dispose
    // message arrives.

    if (!is_dispose_msg && !is_unregister_msg
      && !instance_ptr->rcvd_samples_.matches(DDS::READ_SAMPLE_STATE))
    {
      DDS::DataReaderListener_var listener
        = listener_for(DDS::SAMPLE_REJECTED_STATUS);

      set_status_changed_flag(DDS::SAMPLE_REJECTED_STATUS, true);

      sample_rejected_status_.last_reason =
        DDS::REJECTED_BY_SAMPLES_PER_INSTANCE_LIMIT;
      ++sample_rejected_status_.total_count;
      ++sample_rejected_status_.total_count_change;
      sample_rejected_status_.last_instance_handle = instance_ptr->instance_handle_;

      if (!CORBA::is_nil(listener.in()))
      {
        ACE_GUARD(typename DataReaderImpl::Reverse_Lock_t, unlock_guard, reverse_sample_lock_);

        listener->on_sample_rejected(this, sample_rejected_status_);
        sample_rejected_status_.total_count_change = 0;
      }  // do we want to do something if listener is nil???
      notify_status_condition_no_sample_lock();
      return;
    }
    else if (!is_dispose_msg && !is_unregister_msg)
    {
      // Discard the oldest previously-read sample
      OpenDDS::DCPS::ReceivedDataElement* item =
        instance_ptr->rcvd_samples_.remove_head();
      item->dec_ref();
    }
  }
  else if (qos_.resource_limits.max_samples != DDS::LENGTH_UNLIMITED)
  {
    CORBA::Long total_samples = 0;
    {
      ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, instances_lock_);
      for (OpenDDS::DCPS::DataReaderImpl::SubscriptionInstanceMapType::iterator iter = instances_.begin();
        iter != instances_.end();
        ++iter) {
        OpenDDS::DCPS::SubscriptionInstance_rch ptr = iter->second;

        total_samples += (CORBA::Long) ptr->rcvd_samples_.size();
      }
    }

    if (total_samples >= qos_.resource_limits.max_samples)
    {
      // According to spec 1.2, Samples that contain no data do not
      // count towards the limits imposed by the RESOURCE_LIMITS QoS policy
      // so do not remove the oldest sample when unregister/dispose
      // message arrives.

      if (!is_dispose_msg && !is_unregister_msg
        && !instance_ptr->rcvd_samples_.matches(DDS::READ_SAMPLE_STATE))
      {
        DDS::DataReaderListener_var listener
          = listener_for(DDS::SAMPLE_REJECTED_STATUS);

        set_status_changed_flag(DDS::SAMPLE_REJECTED_STATUS, true);

        sample_rejected_status_.last_reason =
          DDS::REJECTED_BY_SAMPLES_LIMIT;
        ++sample_rejected_status_.total_count;
        ++sample_rejected_status_.total_count_change;
        sample_rejected_status_.last_instance_handle = instance_ptr->instance_handle_;
        if (!CORBA::is_nil(listener.in()))
        {
          ACE_GUARD(typename DataReaderImpl::Reverse_Lock_t, unlock_guard, reverse_sample_lock_);

          listener->on_sample_rejected(this, sample_rejected_status_);
          sample_rejected_status_.total_count_change = 0;
        }  // do we want to do something if listener is nil???
        notify_status_condition_no_sample_lock();

        return;
      }
      else if (!is_dispose_msg && !is_unregister_msg)
      {
        // Discard the oldest previously-read sample
        OpenDDS::DCPS::ReceivedDataElement *item =
          instance_ptr->rcvd_samples_.remove_head();
        item->dec_ref();
      }
    }
  }

  bool event_notify = false;

  if (is_dispose_msg) {
    event_notify = instance_ptr->instance_state_->dispose_was_received(header.publication_id_);
  }

  if (is_unregister_msg) {
    if (instance_ptr->instance_state_->unregister_was_received(header.publication_id_)) {
      event_notify = true;
    }
  }

  if (!is_dispose_msg && !is_unregister_msg) {
    event_notify = true;
    instance_ptr->instance_state_->data_was_received(header.publication_id_);
  }

  if (!event_notify) {
    return;
  }

  ReceivedDataElement* const ptr =
    new (*rd_allocator_.get()) ReceivedDataElementWithType<MessageTypeWithAllocator>(
      header, instance_data.release(), &sample_lock_);

  ptr->disposed_generation_count_ =
    instance_ptr->instance_state_->disposed_generation_count();
  ptr->no_writers_generation_count_ =
    instance_ptr->instance_state_->no_writers_generation_count();

  instance_ptr->last_sequence_ = header.sequence_;

  instance_ptr->rcvd_strategy_->add(ptr);

  if (! is_dispose_msg  && ! is_unregister_msg
      && instance_ptr->rcvd_samples_.size() > get_depth())
    {
      OpenDDS::DCPS::ReceivedDataElement* head_ptr =
        instance_ptr->rcvd_samples_.remove_head();

      if (head_ptr->sample_state_ == DDS::NOT_READ_SAMPLE_STATE)
        {
          DDS::DataReaderListener_var listener
            = listener_for (DDS::SAMPLE_LOST_STATUS);

          ++sample_lost_status_.total_count;
          ++sample_lost_status_.total_count_change;

          set_status_changed_flag(DDS::SAMPLE_LOST_STATUS, true);

          if (!CORBA::is_nil(listener.in()))
            {
              ACE_GUARD(typename DataReaderImpl::Reverse_Lock_t, unlock_guard, reverse_sample_lock_);

              listener->on_sample_lost(this, sample_lost_status_);

              sample_lost_status_.total_count_change = 0;
            }

          notify_status_condition_no_sample_lock();
        }

      head_ptr->dec_ref();
    }

#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
  if (! ptr->coherent_change_) {
#endif
    RcHandle<OpenDDS::DCPS::SubscriberImpl> sub = get_subscriber_servant();
    if (!sub || get_deleted())
      return;

    sub->set_status_changed_flag(DDS::DATA_ON_READERS_STATUS, true);

    set_status_changed_flag(DDS::DATA_AVAILABLE_STATUS, true);

    DDS::SubscriberListener_var sub_listener =
        sub->listener_for(DDS::DATA_ON_READERS_STATUS);
    if (!CORBA::is_nil(sub_listener.in()) && !coherent_) {
      if (!is_bit()) {
        sub->set_status_changed_flag(DDS::DATA_ON_READERS_STATUS, false);
        ACE_GUARD(typename DataReaderImpl::Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
        sub_listener->on_data_on_readers(sub.in());
      } else {
        TheServiceParticipant->job_queue()->enqueue(make_rch<OnDataOnReaders>(sub, sub_listener, rchandle_from(static_cast<DataReaderImpl*>(this)), true, false));
      }
    } else {
      sub->notify_status_condition();

      DDS::DataReaderListener_var listener =
        listener_for (DDS::DATA_AVAILABLE_STATUS);

      if (!CORBA::is_nil(listener.in())) {
        if (!is_bit()) {
          set_status_changed_flag(DDS::DATA_AVAILABLE_STATUS, false);
          sub->set_status_changed_flag(DDS::DATA_ON_READERS_STATUS, false);
          sub.reset();
          ACE_GUARD(typename DataReaderImpl::Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
          listener->on_data_available(this);
        } else {
          TheServiceParticipant->job_queue()->enqueue(make_rch<OnDataAvailable>(listener, rchandle_from(static_cast<DataReaderImpl*>(this)), true, true, true));
        }
      } else {
        notify_status_condition_no_sample_lock();
      }
    }
#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
  }
#endif
}

get_key_value()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::get_key_value ( MessageType &  key_holder, DDS::InstanceHandle_t  handle  )

inlinevirtual

Definition at line 657 of file DataReaderImpl_T.h.

  {
    ACE_Guard<ACE_Recursive_Thread_Mutex> guard(sample_lock_);

    const typename ReverseInstanceMap::const_iterator pos = reverse_instance_map_.find(handle);
    if (pos != reverse_instance_map_.end()) {
      key_holder = pos->second->first;
      return DDS::RETCODE_OK;
    }

    return DDS::RETCODE_BAD_PARAMETER;
  }

get_marshal_skip_serialize()

template<typename MessageType>

bool OpenDDS::DCPS::DataReaderImpl_T< MessageType >::get_marshal_skip_serialize ( ) const

inline

Definition at line 1047 of file DataReaderImpl_T.h.

  {
    return marshal_skip_serialize_;
  }

lookup_instance() [1/2]

template<typename MessageType>

virtual DDS::InstanceHandle_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::lookup_instance ( const MessageType &  instance_data )

inlinevirtual

Definition at line 671 of file DataReaderImpl_T.h.

Referenced by OpenDDS::DCPS::BitSubscriber::remove_connection_record(), and OpenDDS::DCPS::BitSubscriber::remove_thread_status().

  {
    ACE_Guard<ACE_Recursive_Thread_Mutex> guard(sample_lock_);

    const typename InstanceMap::const_iterator it = instance_map_.find(instance_data);
    if (it != instance_map_.end()) {
      return it->second;
    }
    return DDS::HANDLE_NIL;
  }

lookup_instance() [2/2]

template<typename MessageType>

virtual void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::lookup_instance ( const OpenDDS::DCPS::ReceivedDataSample &  sample, OpenDDS::DCPS::SubscriptionInstance_rch &  instance  )

inlinevirtual

!! caller should already have the sample_lock_

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 916 of file DataReaderImpl_T.h.

  {
    //!!! caller should already have the sample_lock_
    const bool encapsulated = sample.header_.cdr_encapsulation_;
    Message_Block_Ptr payload(sample.data(&mb_alloc_));
    OpenDDS::DCPS::Serializer ser(
      payload.get(),
      encapsulated ? Encoding::KIND_XCDR1 : Encoding::KIND_UNALIGNED_CDR,
      static_cast<Endianness>(sample.header_.byte_order_));

    if (encapsulated) {
      EncapsulationHeader encap;
      if (!(ser >> encap)) {
        if (DCPS_debug_level > 0) {
          ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR ")
            ACE_TEXT("%CDataReaderImpl::lookup_instance: ")
            ACE_TEXT("deserialization of encapsulation header failed.\n"),
            TraitsType::type_name()));
        }
        return;
      }
      Encoding encoding;
      if (!encap.to_encoding(encoding, type_support_->base_extensibility())) {
        return;
      }

      if (decoding_modes_.find(encoding.kind()) == decoding_modes_.end()) {
        if (DCPS_debug_level >= 1) {
          ACE_DEBUG((LM_WARNING, ACE_TEXT("(%P|%t) WARNING ")
            ACE_TEXT("%CDataReaderImpl::lookup_instance: ")
            ACE_TEXT("Encoding kind of the received sample (%C) does not ")
            ACE_TEXT("match the ones specified by DataReader.\n"),
            TraitsType::type_name(),
            Encoding::kind_to_string(encoding.kind()).c_str()));
        }
        return;
      }
      if (DCPS_debug_level >= 8) {
        ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) ")
          ACE_TEXT("%CDataReaderImpl::lookup_instance: ")
          ACE_TEXT("Deserializing with encoding kind %C.\n"),
          TraitsType::type_name(),
          Encoding::kind_to_string(encoding.kind()).c_str()));
      }

      ser.encoding(encoding);
    }

    bool ser_ret = true;
    MessageType data;
    if (sample.header_.key_fields_only_) {
      ser_ret = ser >> OpenDDS::DCPS::KeyOnly<MessageType>(data);
    } else {
      ser_ret = ser >> data;
    }
    if (!ser_ret) {
      if (ser.get_construction_status() != Serializer::ConstructionSuccessful) {
        if (DCPS_debug_level > 1) {
          ACE_DEBUG((LM_WARNING, ACE_TEXT("(%P|%t) %CDataReaderImpl::lookup_instance ")
                     ACE_TEXT("object construction failure, dropping sample.\n"),
                     TraitsType::type_name()));
        }
      } else {
        if (DCPS_debug_level > 0) {
          ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) %CDataReaderImpl::lookup_instance ")
                    ACE_TEXT("deserialization failed.\n"),
                    TraitsType::type_name()));
        }
      }
      return;
    }

    DDS::InstanceHandle_t handle(DDS::HANDLE_NIL);
    typename InstanceMap::const_iterator const it = instance_map_.find(data);
    if (it != instance_map_.end()) {
      handle = it->second;
    }

    if (handle == DDS::HANDLE_NIL) {
      instance.reset();
    } else {
      instance = get_handle_instance(handle);
    }
  }

lookup_instance_generic()

template<typename MessageType>

DDS::InstanceHandle_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::lookup_instance_generic ( const void *  data )

inlinevirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 760 of file DataReaderImpl_T.h.

  {
    return lookup_instance(*static_cast<const MessageType*>(data));
  }

marshal()

template<typename MessageType>

CORBA::Boolean OpenDDS::DCPS::DataReaderImpl_T< MessageType >::marshal ( TAO_OutputCDR &  )

inlinevirtual

Reimplemented from CORBA::Object.

Definition at line 78 of file DataReaderImpl_T.h.

    {
      return false;
    }

notify_status_condition_no_sample_lock()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::notify_status_condition_no_sample_lock ( )

inlineprivate

Release sample_lock_ during status notifications in store_instance_data() as the lock is not needed and could cause deadlock condition. See comments in member function implementation for details.

Definition at line 2159 of file DataReaderImpl_T.h.

{
  // This member function avoids a deadlock condition which otherwise
  // could occur as follows:
  // Thread 1: Call to WaitSet::wait() causes WaitSet::lock_ to lock and
  // eventually DataReaderImpl::sample_lock_ to lock in call to
  // DataReaderImpl::contains_samples().
  // Thread2: Call to DataReaderImpl::data_received()
  // causes DataReaderImpl::sample_lock_ to lock and eventually
  // during notify of status condition a call to WaitSet::signal()
  // causes WaitSet::lock_ to lock.
  // Because the DataReaderImpl::sample_lock_ is not needed during
  // status notification this member function is used in
  // store_instance_data() to release sample_lock_ before making
  // the notification.
  ACE_GUARD(typename DataReaderImpl::Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
  notify_status_condition();
}

OPENDDS_MAP() [1/2]

template<typename MessageType>

typedef OpenDDS::DCPS::DataReaderImpl_T< MessageType >::OPENDDS_MAP	(	DDS::InstanceHandle_t	,
		typename InstanceMap::iterator
	)

OPENDDS_MAP() [2/2]

template<typename MessageType>

typedef OpenDDS::DCPS::DataReaderImpl_T< MessageType >::OPENDDS_MAP ( DDS::InstanceHandle_t  , FilterDelayedSample    )

private

OPENDDS_MAP_CMP_T()

template<typename MessageType>

typedef OpenDDS::DCPS::DataReaderImpl_T< MessageType >::OPENDDS_MAP_CMP_T	(	MessageType	,
		DDS::InstanceHandle_t	,
		typename TraitsType::LessThanType
	)

OPENDDS_MULTIMAP()

template<typename MessageType>

typedef OpenDDS::DCPS::DataReaderImpl_T< MessageType >::OPENDDS_MULTIMAP ( MonotonicTimePoint  , DDS::InstanceHandle_t    )

private

purge_data()

template<typename MessageType>

virtual void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::purge_data ( OpenDDS::DCPS::SubscriptionInstance_rch  instance )

inlineprotectedvirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 1220 of file DataReaderImpl_T.h.

  {
    drop_sample(instance->instance_handle_);


    instance->instance_state_->cancel_release();

    while (instance->rcvd_samples_.size() > 0)
      {
        OpenDDS::DCPS::ReceivedDataElement* head =
          instance->rcvd_samples_.remove_head();
        head->dec_ref();
      }
  }

qos_change()

template<typename MessageType>

virtual void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::qos_change ( const DDS::DataReaderQos &  qos )

inlinevirtual

Reimplemented from OpenDDS::DCPS::DataReaderImpl.

Definition at line 1002 of file DataReaderImpl_T.h.

  {
    // reliability is not changeable, just time_based_filter
    if (qos.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS) {
      if (qos.time_based_filter.minimum_separation != qos_.time_based_filter.minimum_separation) {
        const DDS::Duration_t zero = { DDS::DURATION_ZERO_SEC, DDS::DURATION_ZERO_NSEC };
        if (qos_.time_based_filter.minimum_separation != zero) {
          if (qos.time_based_filter.minimum_separation != zero) {
            const MonotonicTimePoint now = MonotonicTimePoint::now();
            const TimeDuration interval(qos_.time_based_filter.minimum_separation);
            FilterDelayedSampleQueue queue;

            ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
            for (typename FilterDelayedSampleMap::iterator pos = filter_delayed_sample_map_.begin(), limit = filter_delayed_sample_map_.end(); pos != limit; ++pos) {
              FilterDelayedSample& sample = pos->second;
              sample.expiration_time = now + (interval - (sample.expiration_time - now));
              queue.insert(std::make_pair(sample.expiration_time, pos->first));
            }
            std::swap(queue, filter_delayed_sample_queue_);

            if (!filter_delayed_sample_queue_.empty()) {
              filter_delayed_sample_task_->cancel();
              filter_delayed_sample_task_->schedule(interval);
            }

          } else {
            filter_delayed_sample_task_->cancel();
            ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
            filter_delayed_sample_map_.clear();
            filter_delayed_sample_queue_.clear();
          }
        }
        // else no existing timers to change/cancel
      }
      // else no qos change so nothing to change
    }

    DataReaderImpl::qos_change(qos);
  }

read()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states  )

inlinevirtual

Definition at line 161 of file DataReaderImpl_T.h.

    {
      DDS::ReturnCode_t const precond =
        check_inputs("read", received_data, info_seq, max_samples);
      if (DDS::RETCODE_OK != precond)
        {
          return precond;
        }

      ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex,
                        guard,
                        sample_lock_,
                        DDS::RETCODE_ERROR);

      return read_i(received_data, info_seq, max_samples, sample_states,
                    view_states, instance_states, 0);
    }

read_generic() [1/3]

template<>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< XTypes::DynamicSample >::read_generic ( GenericBundle &  , DDS::SampleStateMask  , DDS::ViewStateMask  , DDS::InstanceStateMask  , bool    )

virtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 21 of file DynamicDataReaderImpl.cpp.

  {
    return DDS::RETCODE_UNSUPPORTED;
  }

read_generic() [2/3]

template<>

OpenDDS_Dcps_Export DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< XTypes::DynamicSample >::read_generic ( GenericBundle &  gen, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states, bool  adjust_ref_count  )

virtual

Implements OpenDDS::DCPS::DataReaderImpl.

read_generic() [3/3]

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_generic ( GenericBundle &  gen, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states, bool  adjust_ref_count = false  )

inlinevirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 736 of file DataReaderImpl_T.h.

  {
    MessageSequenceType data;
    DDS::ReturnCode_t rc;
    {
      ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, sample_lock_, DDS::RETCODE_ERROR);
      rc = read_i(data, gen.info_, DDS::LENGTH_UNLIMITED,
                  sample_states, view_states, instance_states, 0);
      if (adjust_ref_count) {
        typename DDSTraits<MessageType>::MessageSequenceAdapterType received_data_p(data);
        received_data_p.increment_references();
      }
    }
    gen.samples_.reserve(data.length());
    for (CORBA::ULong i = 0; i < data.length(); ++i) {
      gen.samples_.push_back(&data[i]);
    }
    return rc;
  }

read_i()

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_i ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, CORBA::Long  max_samples, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states, DDS::QueryCondition_ptr  a_condition  )

inlineprivate

Definition at line 1328 of file DataReaderImpl_T.h.

{

  typename DDSTraits<MessageType>::MessageSequenceAdapterType received_data_p(received_data);

#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
  if (subqos_.presentation.access_scope == DDS::GROUP_PRESENTATION_QOS && !coherent_) {
    return DDS::RETCODE_PRECONDITION_NOT_MET;
  }

  const bool group_coherent_ordered =
    subqos_.presentation.access_scope == DDS::GROUP_PRESENTATION_QOS
    && subqos_.presentation.coherent_access
    && subqos_.presentation.ordered_access;

  if (group_coherent_ordered && coherent_) {
    max_samples = 1;
  }
#endif

  RakeResults<MessageType> results(this, received_data, info_seq, max_samples, subqos_.presentation,
#ifndef OPENDDS_NO_QUERY_CONDITION
                                   a_condition,
#endif
                                   DDS_OPERATION_READ);

  const Observer_rch observer = get_observer(Observer::e_SAMPLE_READ);

#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
  if (!group_coherent_ordered) {
#endif
    const HandleSet& matches = lookup_matching_instances(sample_states, view_states, instance_states);
    for (HandleSet::const_iterator it = matches.begin(), next = it; it != matches.end(); it = next) {
      ++next; // pre-increment iterator, in case updates cause changes to match set
      const DDS::InstanceHandle_t handle = *it;
      const SubscriptionInstance_rch inst = get_handle_instance(handle);
      if (!inst) continue;

      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)) {
        results.insert_sample(item, &inst->rcvd_samples_, inst, ++i);

        const ValueDispatcher* vd = get_value_dispatcher();
        if (observer && item->registered_data_ && vd) {
          Observer::Sample s(handle, inst->instance_state_->instance_state(), *item, *vd);
          observer->on_sample_read(this, s);
        }
      }
    }
#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
  } else {
    const RakeData item = group_coherent_ordered_data_.get_data();
    results.insert_sample(item.rde_, item.rdel_, item.si_, item.index_in_instance_);
    const ValueDispatcher* vd = get_value_dispatcher();
    if (observer && item.rde_->registered_data_ && vd) {
      typename InstanceMap::iterator i = instance_map_.begin();
      const DDS::InstanceHandle_t handle = (i != instance_map_.end()) ? i->second : DDS::HANDLE_NIL;
      Observer::Sample s(handle, item.si_->instance_state_->instance_state(), *item.rde_, *vd);
      observer->on_sample_read(this, s);
    }
  }
#endif

  results.copy_to_user();

  DDS::ReturnCode_t ret = DDS::RETCODE_NO_DATA;
  if (received_data.length()) {
    ret = DDS::RETCODE_OK;
    if (received_data.maximum() == 0) { // using ZeroCopy
      received_data_p.set_loaner(this);
    }
  }

  post_read_or_take();
  return ret;
}

read_instance()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_instance ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states  )

inlinevirtual

Definition at line 385 of file DataReaderImpl_T.h.

  {
    DDS::ReturnCode_t const precond =
      check_inputs("read_instance", received_data, info_seq, max_samples);
    if (DDS::RETCODE_OK != precond)
      {
        return precond;
      }

    ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex,
                      guard,
                      sample_lock_,
                      DDS::RETCODE_ERROR);
    return read_instance_i(received_data, info_seq, max_samples, a_handle,
                           sample_states, view_states, instance_states, 0);
  }

read_instance_generic()

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_instance_generic ( void *&  data, DDS::SampleInfo &  info, DDS::InstanceHandle_t  instance, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states  )

inlinevirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 788 of file DataReaderImpl_T.h.

  {
    MessageSequenceType dataseq;
    DDS::SampleInfoSeq infoseq;
    const DDS::ReturnCode_t rc = read_instance_i(dataseq, infoseq,
                                                 DDS::LENGTH_UNLIMITED, instance, sample_states, view_states,
                                                 instance_states, 0);
    if (rc != DDS::RETCODE_NO_DATA)
      {
        const CORBA::ULong last = dataseq.length() - 1;
        data = new MessageType(dataseq[last]);
        info = infoseq[last];
      }
    return rc;
  }

read_instance_i()

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_instance_i ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states, DDS::QueryCondition_ptr  a_condition  )

inlineprivate

Definition at line 1496 of file DataReaderImpl_T.h.

{
  const SubscriptionInstance_rch inst = get_handle_instance(a_handle);
  if (!inst) return DDS::RETCODE_BAD_PARAMETER;

  typename DDSTraits<MessageType>::MessageSequenceAdapterType received_data_p(received_data);

  RakeResults<MessageType> results(this, received_data, info_seq, max_samples, subqos_.presentation,
#ifndef OPENDDS_NO_QUERY_CONDITION
                                   a_condition,
#endif
                                   DDS_OPERATION_READ);

  const InstanceState_rch state_obj = inst->instance_state_;
  if (state_obj->match(view_states, instance_states)) {
    const Observer_rch observer = get_observer(Observer::e_SAMPLE_READ);
    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)) {
      results.insert_sample(item, &inst->rcvd_samples_, inst, ++i);
      const ValueDispatcher* vd = get_value_dispatcher();
      if (observer && item->registered_data_ && vd) {
        Observer::Sample s(a_handle, inst->instance_state_->instance_state(), *item, *vd);
        observer->on_sample_read(this, s);
      }
    }
  } else if (DCPS_debug_level >= 8) {
    OPENDDS_STRING msg;
    if ((state_obj->view_state() & view_states) == 0) {
      msg = "view state is not valid";
    }
    if ((state_obj->instance_state() & instance_states) == 0) {
      if (!msg.empty()) msg += " and ";
      msg += "instance state is ";
      msg += state_obj->instance_state_string();
      msg += " while the validity mask is " + InstanceState::instance_state_mask_string(instance_states);
    }
    ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) DataReaderImpl_T::read_instance_i: ")
               ACE_TEXT("will return no data reading sub %C because:\n  %C\n"),
               LogGuid(get_guid()).c_str(), msg.c_str()));
  }

  results.copy_to_user();

  DDS::ReturnCode_t ret = DDS::RETCODE_NO_DATA;
  if (received_data.length()) {
    ret = DDS::RETCODE_OK;
    if (received_data.maximum() == 0) { // using ZeroCopy
      received_data_p.set_loaner(this);
    }
  }

  post_read_or_take();
  return ret;
}

read_instance_w_condition()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_instance_w_condition ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::ReadCondition_ptr  a_condition  )

inlinevirtual

Definition at line 433 of file DataReaderImpl_T.h.

  {
    DDS::ReturnCode_t const precond =
      check_inputs("read_instance_w_condition", received_data, info_seq,
                   max_samples);
    if (DDS::RETCODE_OK != precond)
      {
        return precond;
      }

    ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, guard, sample_lock_,
                      DDS::RETCODE_ERROR);

    if (!has_readcondition(a_condition))
      {
        return DDS::RETCODE_PRECONDITION_NOT_MET;
      }

#ifndef OPENDDS_NO_QUERY_CONDITION
    DDS::QueryCondition_ptr query_condition =
        dynamic_cast< DDS::QueryCondition_ptr >(a_condition);
#endif

    return read_instance_i(received_data, info_seq, max_samples, a_handle,
                           a_condition->get_sample_state_mask(),
                           a_condition->get_view_state_mask(),
                           a_condition->get_instance_state_mask(),
#ifndef OPENDDS_NO_QUERY_CONDITION
                           query_condition
#else
                           0
#endif
                           );
  }

read_next_instance()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_next_instance ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states  )

inlinevirtual

Definition at line 513 of file DataReaderImpl_T.h.

  {
    DDS::ReturnCode_t const precond =
      check_inputs("read_next_instance", received_data, info_seq, max_samples);
    if (DDS::RETCODE_OK != precond)
      {
        return precond;
      }

    return read_next_instance_i(received_data, info_seq, max_samples, a_handle,
                                sample_states, view_states, instance_states, 0);
  }

read_next_instance_generic()

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::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  )

inlinevirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 807 of file DataReaderImpl_T.h.

  {
    MessageSequenceType dataseq;
    DDS::SampleInfoSeq infoseq;
    const DDS::ReturnCode_t rc = read_next_instance_i(dataseq, infoseq,
                                                      DDS::LENGTH_UNLIMITED, previous_instance, sample_states, view_states,
                                                      instance_states, 0);
    if (rc != DDS::RETCODE_NO_DATA)
      {
        const CORBA::ULong last = dataseq.length() - 1;
        data = new MessageType(dataseq[last]);
        info = infoseq[last];
      }
    return rc;
  }

read_next_instance_i()

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_next_instance_i ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states, DDS::QueryCondition_ptr  a_condition  )

inlineprivate

Definition at line 1616 of file DataReaderImpl_T.h.

{
  ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, sample_lock_, DDS::RETCODE_ERROR);

  typename InstanceMap::iterator it = instance_map_.begin();
  const typename InstanceMap::iterator the_end = instance_map_.end();
  if (a_handle != DDS::HANDLE_NIL) {
    const typename ReverseInstanceMap::const_iterator pos = reverse_instance_map_.find(a_handle);
    if (pos != reverse_instance_map_.end()) {
      it = pos->second;
      ++it;
    } else {
      it = the_end;
    }
  }

  DDS::InstanceHandle_t handle(DDS::HANDLE_NIL);
  for (; it != the_end; ++it) {
    handle = it->second;
    const DDS::ReturnCode_t status =
      read_instance_i(received_data, info_seq, max_samples, handle,
                      sample_states, view_states, instance_states,
#ifndef OPENDDS_NO_QUERY_CONDITION
                      a_condition);
#else
      0);
#endif
    if (status != DDS::RETCODE_NO_DATA) {
      post_read_or_take();
      return status;
    }
  }

  post_read_or_take();
  return DDS::RETCODE_NO_DATA;
}

read_next_instance_w_condition()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_next_instance_w_condition ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::ReadCondition_ptr  a_condition  )

inlinevirtual

Definition at line 553 of file DataReaderImpl_T.h.

  {
    DDS::ReturnCode_t const precond =
      check_inputs("read_next_instance_w_condition", received_data, info_seq,
                   max_samples);
    if (DDS::RETCODE_OK != precond)
      {
        return precond;
      }

    ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, guard, sample_lock_,
                      DDS::RETCODE_ERROR);

    if (!has_readcondition(a_condition))
      {
        return DDS::RETCODE_PRECONDITION_NOT_MET;
      }

#ifndef OPENDDS_NO_QUERY_CONDITION
    DDS::QueryCondition_ptr query_condition =
        dynamic_cast< DDS::QueryCondition_ptr >(a_condition);
#endif

    return read_next_instance_i(received_data, info_seq, max_samples, a_handle,
                                a_condition->get_sample_state_mask(),
                                a_condition->get_view_state_mask(),
                                a_condition->get_instance_state_mask(),
#ifndef OPENDDS_NO_QUERY_CONDITION
                                query_condition
#else
                                0
#endif
                                );
  }

read_next_sample()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_next_sample ( MessageType &  received_data, DDS::SampleInfo &  sample_info_ref  )

inlinevirtual

Definition at line 274 of file DataReaderImpl_T.h.

  {
    bool found_data = false;
    ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, sample_lock_, DDS::RETCODE_ERROR);

    const Observer_rch observer = get_observer(Observer::e_SAMPLE_READ);

    const CORBA::ULong sample_states = DDS::NOT_READ_SAMPLE_STATE;
    const HandleSet& matches = lookup_matching_instances(sample_states, DDS::ANY_VIEW_STATE, DDS::ANY_INSTANCE_STATE);
    for (HandleSet::const_iterator it = matches.begin(), next = it; it != matches.end(); it = next) {
      ++next; // pre-increment iterator, in case updates cause changes to match set
      const DDS::InstanceHandle_t handle = *it;
      const SubscriptionInstance_rch inst = get_handle_instance(handle);
      if (!inst) continue;

      bool most_recent_generation = false;
      for (ReceivedDataElement* item = inst->rcvd_samples_.get_next_match(sample_states, 0);
           !found_data && item; item = inst->rcvd_samples_.get_next_match(sample_states, item)) {
        if (item->registered_data_) {
          received_data = *static_cast<MessageType*>(item->registered_data_);
        }
        inst->instance_state_->sample_info(sample_info_ref, item);
        inst->rcvd_samples_.mark_read(item);

        const ValueDispatcher* vd = get_value_dispatcher();
        if (observer && item->registered_data_ && vd) {
          Observer::Sample s(sample_info_ref.instance_handle, sample_info_ref.instance_state, *item, *vd);
          observer->on_sample_read(this, s);
        }

        if (!most_recent_generation) {
          most_recent_generation = inst->instance_state_->most_recent_generation(item);
        }

        found_data = true;
      }

      if (found_data) {
        if (most_recent_generation) {
          inst->instance_state_->accessed();
        }
        // Get the sample_ranks, generation_ranks, and
        // absolute_generation_ranks for this info_seq
        sample_info(sample_info_ref, inst->rcvd_samples_.peek_tail());

        break;
      }
    }

    post_read_or_take();
    return found_data ? DDS::RETCODE_OK : DDS::RETCODE_NO_DATA;
  }

read_w_condition()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::read_w_condition ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  sample_info, ::CORBA::Long  max_samples, DDS::ReadCondition_ptr  a_condition  )

inlinevirtual

Definition at line 209 of file DataReaderImpl_T.h.

    {
      DDS::ReturnCode_t const precond =
        check_inputs("read_w_condition", received_data, sample_info, max_samples);
      if (DDS::RETCODE_OK != precond)
        {
          return precond;
        }

      ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, guard, sample_lock_,
                        DDS::RETCODE_ERROR);

      if (!has_readcondition(a_condition))
        {
          return DDS::RETCODE_PRECONDITION_NOT_MET;
        }

      return read_i(received_data, sample_info, max_samples,
                    a_condition->get_sample_state_mask(),
                    a_condition->get_view_state_mask(),
                    a_condition->get_instance_state_mask(),
#ifndef OPENDDS_NO_QUERY_CONDITION
                    dynamic_cast< DDS::QueryCondition_ptr >(a_condition));
#else
      0);
#endif
  }

release_all_instances()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::release_all_instances ( )

inlinevirtual

Release all instances held by the reader.

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 1052 of file DataReaderImpl_T.h.

  {
    ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);

    const typename InstanceMap::iterator end = instance_map_.end();
    typename InstanceMap::iterator it = instance_map_.begin();
    while (it != end) {
      const DDS::InstanceHandle_t handle = it->second;
      ++it; // it will be invalid, so iterate now.
      release_instance(handle);
    }
  }

release_instance_i()

template<typename MessageType>

virtual void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::release_instance_i ( DDS::InstanceHandle_t  handle )

inlineprotectedvirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 1235 of file DataReaderImpl_T.h.

  {
    const typename ReverseInstanceMap::iterator pos = reverse_instance_map_.find(handle);
    if (pos != reverse_instance_map_.end()) {
      remove_from_lookup_maps(handle);
      instance_map_.erase(pos->second);
      reverse_instance_map_.erase(pos);
    }
  }

release_loan()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::release_loan ( MessageSequenceType &  received_data )

inline

Definition at line 695 of file DataReaderImpl_T.h.

  {
    received_data.length(0);
  }

return_loan()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::return_loan ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq  )

inlinevirtual

Definition at line 633 of file DataReaderImpl_T.h.

  {
    // Some incomplete tests to see that the data and info are from the
    // same read.
    if (received_data.length() != info_seq.length())
      {
        return DDS::RETCODE_PRECONDITION_NOT_MET;
      }

    if (received_data.release())
      {
        // nothing to do because this is not zero-copy data
        return DDS::RETCODE_OK;
      }
    else
      {
        info_seq.length(0);
        received_data.length(0);
      }
    return DDS::RETCODE_OK;
  }

set_instance_state_i()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::set_instance_state_i ( DDS::InstanceHandle_t  instance, DDS::InstanceHandle_t  publication_handle, DDS::InstanceStateKind  state, const SystemTimePoint &  timestamp, const GUID_t &  publication_id  )

inlinevirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 887 of file DataReaderImpl_T.h.

  {
    // sample_lock_ must be held.
    using namespace OpenDDS::DCPS;

    SubscriptionInstance_rch si = get_handle_instance(instance);
    if (si && state != DDS::ALIVE_INSTANCE_STATE) {
      const DDS::Time_t now = timestamp.to_dds_time();
      DataSampleHeader header;
      header.publication_id_ = publication_id;
      header.source_timestamp_sec_ = now.sec;
      header.source_timestamp_nanosec_ = now.nanosec;
      const int msg = (state == DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE)
        ? DISPOSE_INSTANCE : UNREGISTER_INSTANCE;
      header.message_id_ = static_cast<char>(msg);
      bool just_registered, filtered;
      unique_ptr<MessageTypeWithAllocator> data(new (*data_allocator()) MessageTypeWithAllocator);
      get_key_value(*data, instance);
      store_instance_data(move(data), publication_handle, header, si, just_registered, filtered);
      if (!filtered) {
        notify_read_conditions();
      }
    }
  }

set_marshal_skip_serialize()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::set_marshal_skip_serialize ( bool  value )

inline

Definition at line 1042 of file DataReaderImpl_T.h.

  {
    marshal_skip_serialize_ = value;
  }

state_updated_i()

template<typename MessageType>

virtual void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::state_updated_i ( DDS::InstanceHandle_t  handle )

inlineprotectedvirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 1245 of file DataReaderImpl_T.h.

  {
    const typename SubscriptionInstanceMapType::iterator pos = instances_.find(handle);
    if (pos != instances_.end()) {
      update_lookup_maps(pos);
    }
  }

store_instance_data()

template<typename MessageType>

void OpenDDS::DCPS::DataReaderImpl_T< MessageType >::store_instance_data ( unique_ptr< MessageTypeWithAllocator >  instance_data, DDS::InstanceHandle_t  publication_handle, const OpenDDS::DCPS::DataSampleHeader &  header, OpenDDS::DCPS::SubscriptionInstance_rch &  instance_ptr, bool &  just_registered, bool &  filtered  )

inlineprivate

!! caller should already have the sample_lock_

Definition at line 1713 of file DataReaderImpl_T.h.

{
  ACE_UNUSED_ARG(publication_handle);

  const bool is_dispose_msg =
    header.message_id_ == OpenDDS::DCPS::DISPOSE_INSTANCE ||
    header.message_id_ == OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE;
  const bool is_unregister_msg =
    header.message_id_ == OpenDDS::DCPS::UNREGISTER_INSTANCE ||
    header.message_id_ == OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE;

  if (!store_instance_data_check(instance_data, publication_handle, header, instance_ptr)) {
    return;
  }

  // not filtering any data, except what is specifically identified as filtered below
  filtered = false;

  DDS::InstanceHandle_t handle(DDS::HANDLE_NIL);

  //!!! caller should already have the sample_lock_
  //We will unlock it before calling into listeners

  typename InstanceMap::const_iterator const it = instance_map_.find(*instance_data);

  if (it == instance_map_.end()) {
    if (is_dispose_msg || is_unregister_msg) {
      return;
    }

    std::size_t instances_size = 0;
    {
      ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, instances_lock_);
      instances_size = instances_.size();
    }
    if ((qos_.resource_limits.max_instances != DDS::LENGTH_UNLIMITED) &&
      ((::CORBA::Long) instances_size >= qos_.resource_limits.max_instances))
    {
      DDS::DataReaderListener_var listener
        = listener_for (DDS::SAMPLE_REJECTED_STATUS);

      set_status_changed_flag (DDS::SAMPLE_REJECTED_STATUS, true);

      sample_rejected_status_.last_reason = DDS::REJECTED_BY_INSTANCES_LIMIT;
      ++sample_rejected_status_.total_count;
      ++sample_rejected_status_.total_count_change;
      sample_rejected_status_.last_instance_handle = handle;

      if (!CORBA::is_nil(listener.in()))
      {
        ACE_GUARD(typename DataReaderImpl::Reverse_Lock_t, unlock_guard, reverse_sample_lock_);

        listener->on_sample_rejected(this, sample_rejected_status_);
        sample_rejected_status_.total_count_change = 0;
      }  // do we want to do something if listener is nil???
      notify_status_condition_no_sample_lock();

      return;
    }

    {
      ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, instances_lock_);

#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
      SharedInstanceMap_rch inst;
      OwnershipManagerScopedAccess ownership_scoped_access;
      OwnershipManagerPtr owner_manager = ownership_manager();

      bool new_handle = true;
      if (is_exclusive_ownership_) {
        OwnershipManagerScopedAccess temp(owner_manager);
        temp.swap(ownership_scoped_access);
        if (!owner_manager || ownership_scoped_access.lock_result_ != 0) {
          if (DCPS_debug_level > 0) {
            ACE_ERROR ((LM_ERROR,
                        ACE_TEXT("(%P|%t) ")
                        ACE_TEXT("%CDataReaderImpl::")
                        ACE_TEXT("store_instance_data, ")
                        ACE_TEXT("acquire instance_lock failed.\n"), TraitsType::type_name()));
          }
          return;
        }

        inst = dynamic_rchandle_cast<SharedInstanceMap>(
          owner_manager->get_instance_map(topic_servant_->type_name(), this));
        if (inst != 0) {
          typename InstanceMap::const_iterator const iter = inst->find(*instance_data);
          if (iter != inst->end ()) {
            handle = iter->second;
            new_handle = false;
          }
        }
      }
#endif

      just_registered = true;
      DDS::BuiltinTopicKey_t key = OpenDDS::DCPS::keyFromSample(static_cast<MessageType*>(instance_data.get()));
      bool owns_handle = false;
      if (handle == DDS::HANDLE_NIL) {
        handle = get_next_handle(key);
        owns_handle = true;
      }
      OpenDDS::DCPS::SubscriptionInstance_rch instance =
        OpenDDS::DCPS::make_rch<OpenDDS::DCPS::SubscriptionInstance>(
          rchandle_from(this),
          qos_,
          ref(instances_lock_),
          handle, owns_handle);

      const std::pair<typename SubscriptionInstanceMapType::iterator, bool> bpair =
        instances_.insert(typename SubscriptionInstanceMapType::value_type(handle, instance));

      if (bpair.second == false) {
        if (DCPS_debug_level > 0) {
          ACE_ERROR((LM_ERROR,
                     ACE_TEXT("(%P|%t) ")
                     ACE_TEXT("%CDataReaderImpl::")
                     ACE_TEXT("store_instance_data, ")
                     ACE_TEXT("insert handle failed.\n"), TraitsType::type_name()));
        }
        return;
      }
      update_lookup_maps(bpair.first);

#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
      if (owner_manager) {
        if (!inst) {
          inst = make_rch<SharedInstanceMap>();
          owner_manager->set_instance_map(
            topic_servant_->type_name(),
            inst,
            this);
        }

        if (new_handle) {
          const std::pair<typename InstanceMap::iterator, bool> bpair =
            inst->insert(typename InstanceMap::value_type(*instance_data, handle));
          if (!bpair.second) {
            if (DCPS_debug_level > 0) {
              ACE_ERROR ((LM_ERROR,
                          ACE_TEXT("(%P|%t) ")
                          ACE_TEXT("%CDataReaderImpl::")
                          ACE_TEXT("store_instance_data, ")
                          ACE_TEXT("insert to participant scope %C failed.\n"), TraitsType::type_name(), TraitsType::type_name()));
            }
            return;
          }
        }

        OwnershipManagerScopedAccess temp;
        temp.swap(ownership_scoped_access);
        if (temp.release() != 0) {
          if (DCPS_debug_level > 0) {
            ACE_ERROR ((LM_ERROR,
                        ACE_TEXT("(%P|%t) ")
                        ACE_TEXT("%CDataReaderImpl::")
                        ACE_TEXT("store_instance_data, ")
                        ACE_TEXT("release instance_lock failed.\n"), TraitsType::type_name()));
          }
          return;
        }
      }
#endif
    } // scope for instances_lock_

    std::pair<typename InstanceMap::iterator, bool> bpair =
      instance_map_.insert(typename InstanceMap::value_type(*instance_data,
        handle));
    if (bpair.second == false)
    {
      if (DCPS_debug_level > 0) {
        ACE_ERROR ((LM_ERROR,
                    ACE_TEXT("(%P|%t) ")
                    ACE_TEXT("%CDataReaderImpl::")
                    ACE_TEXT("store_instance_data, ")
                    ACE_TEXT("insert %C failed.\n"), TraitsType::type_name(), TraitsType::type_name()));
      }
      return;
    }
    reverse_instance_map_[handle] = bpair.first;
  }
  else
  {
    just_registered = false;
    handle = it->second;
  }

  if (header.message_id_ != OpenDDS::DCPS::INSTANCE_REGISTRATION)
  {
    instance_ptr = get_handle_instance(handle);
    OPENDDS_ASSERT(instance_ptr);

    if (header.message_id_ == OpenDDS::DCPS::SAMPLE_DATA)
    {
      {
        ACE_GUARD(ACE_Recursive_Thread_Mutex, instance_guard, instances_lock_);
        filtered = ownership_filter_instance(instance_ptr, header.publication_id_);
      }

      MonotonicTimePoint now;
      MonotonicTimePoint deadline;
      if (!filtered && time_based_filter_instance(instance_ptr, now, deadline)) {
        filtered = true;
        if (qos_.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS) {
          delay_sample(handle, move(instance_data), header, just_registered, now, deadline);
        }
      } else {
        // nothing time based filtered now
        clear_sample(handle);

      }

      if (filtered) {
        return;
      }
    }

    finish_store_instance_data(move(instance_data), header, instance_ptr, is_dispose_msg, is_unregister_msg);
  }
  else
  {
    instance_ptr = get_handle_instance(handle);
    OPENDDS_ASSERT(instance_ptr);
    instance_ptr->instance_state_->lively(header.publication_id_);
  }
}

store_instance_data_check()

template<typename MessageType>

bool OpenDDS::DCPS::DataReaderImpl_T< MessageType >::store_instance_data_check ( unique_ptr< MessageTypeWithAllocator > &  instance_data, DDS::InstanceHandle_t  publication_handle, const OpenDDS::DCPS::DataSampleHeader &  header, OpenDDS::DCPS::SubscriptionInstance_rch &  instance_ptr  )

inlineprivate

Definition at line 1259 of file DataReaderImpl_T.h.

  {
#if defined(OPENDDS_SECURITY) && OPENDDS_HAS_DYNAMIC_DATA_ADAPTER
    const bool is_dispose_msg =
      header.message_id_ == OpenDDS::DCPS::DISPOSE_INSTANCE ||
      header.message_id_ == OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE;

    if (!is_bit() && security_config_) {
      if (header.message_id_ == SAMPLE_DATA ||
          header.message_id_ == INSTANCE_REGISTRATION) {

        // Pubulisher has already gone through the check.
        if (instance_ptr &&
            instance_ptr->instance_state_ &&
            instance_ptr->instance_state_->writes_instance(header.publication_id_)) {
          return true;
        }

        DDS::Security::SecurityException ex;
        const GUID_t local_participant = make_part_guid(get_guid());
        const GUID_t remote_participant = make_part_guid(header.publication_id_);
        const DDS::Security::ParticipantCryptoHandle remote_participant_permissions_handle = security_config_->get_handle_registry(local_participant)->get_remote_participant_permissions_handle(remote_participant);
        // Construct a DynamicData around the deserialized sample.
        DDS::DynamicData_var dda =
          XTypes::get_dynamic_data_adapter(dynamic_type_, *instance_data->message());
        // The remote participant might not be using security.
        if (remote_participant_permissions_handle != DDS::HANDLE_NIL &&
            !security_config_->get_access_control()->check_remote_datawriter_register_instance(remote_participant_permissions_handle, this, publication_handle, dda, ex)) {
          if (log_level >= LogLevel::Warning) {
            ACE_ERROR((LM_WARNING,
                       "(%P|%t) WARNING: DataReaderImpl_T::store_instance_data_check: unable to register instance SecurityException[%d.%d]: %C\n",
                       ex.code, ex.minor_code, ex.message.in()));
          }
          return false;
        }
      } else if (is_dispose_msg) {

        DDS::Security::SecurityException ex;
        const GUID_t local_participant = make_part_guid(get_guid());
        const GUID_t remote_participant = make_part_guid(header.publication_id_);
        const DDS::Security::ParticipantCryptoHandle remote_participant_permissions_handle = security_config_->get_handle_registry(local_participant)->get_remote_participant_permissions_handle(remote_participant);
        // Construct a DynamicData around the deserialized sample.
        DDS::DynamicData_var dda =
          XTypes::get_dynamic_data_adapter(dynamic_type_, *instance_data->message());
        // The remote participant might not be using security.
        if (remote_participant_permissions_handle != DDS::HANDLE_NIL &&
            !security_config_->get_access_control()->check_remote_datawriter_dispose_instance(remote_participant_permissions_handle, this, publication_handle, dda, ex)) {
          if (log_level >= LogLevel::Warning) {
            ACE_ERROR((LM_WARNING,
                       "(%P|%t) WARNING: DataReaderImpl_T::store_instance_data_check: unable to dispose instance SecurityException[%d.%d]: %C\n",
                       ex.code, ex.minor_code, ex.message.in()));
          }
          return false;
        }
      }
    }
#else
    ACE_UNUSED_ARG(instance_data);
    ACE_UNUSED_ARG(publication_handle);
    ACE_UNUSED_ARG(header);
    ACE_UNUSED_ARG(instance_ptr);
#endif

    return true;
  }

store_synthetic_data()

template<typename MessageType>

DDS::InstanceHandle_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::store_synthetic_data ( const MessageType &  sample, DDS::ViewStateKind  view, const SystemTimePoint &  timestamp = SystemTimePoint::now()  )

inline

Definition at line 828 of file DataReaderImpl_T.h.

Referenced by OpenDDS::DCPS::BitSubscriber::add_thread_status(), and OpenDDS::DCPS::StaticEndpointManager::init_bit().

  {
    using namespace OpenDDS::DCPS;
    ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, sample_lock_,
                     DDS::HANDLE_NIL);
#ifndef OPENDDS_NO_MULTI_TOPIC
    DDS::TopicDescription_var descr = get_topicdescription();
    if (MultiTopicImpl* mt = dynamic_cast<MultiTopicImpl*>(descr.in())) {
      if (!mt->filter(sample)) {
        return DDS::HANDLE_NIL;
      }
    }
#endif

    get_subscriber_servant()->data_received(this);

    DDS::InstanceHandle_t inst = lookup_instance(sample);
    bool filtered = false;
    SubscriptionInstance_rch instance;

    const DDS::Time_t now = timestamp.to_dds_time();
    DataSampleHeader header;
    header.source_timestamp_sec_ = now.sec;
    header.source_timestamp_nanosec_ = now.nanosec;

    // Call store_instance_data() once or twice, depending on if we need to
    // process the INSTANCE_REGISTRATION.  In either case, store_instance_data()
    // owns the memory for the sample and it must come from the correct allocator.
    for (int i = 0; i < 2; ++i) {
      if (i == 0 && inst != DDS::HANDLE_NIL) continue;

      const int msg = i ? SAMPLE_DATA : INSTANCE_REGISTRATION;
      header.message_id_ = static_cast<char>(msg);

      bool just_registered;
      unique_ptr<MessageTypeWithAllocator> data(new (*data_allocator()) MessageTypeWithAllocator(sample));
      store_instance_data(move(data), DDS::HANDLE_NIL, header, instance, just_registered, filtered);
      if (instance) inst = instance->instance_handle_;
    }

    if (!filtered) {
      if (view == DDS::NOT_NEW_VIEW_STATE) {
        if (instance) instance->instance_state_->accessed();
      }
      notify_read_conditions();
    }

    const ValueDispatcher* vd = get_value_dispatcher();
    const Observer_rch observer = get_observer(Observer::e_SAMPLE_RECEIVED);
    if (observer && vd) {
      Observer::Sample s(instance ? instance->instance_handle_ : DDS::HANDLE_NIL, header.instance_state(), now, header.sequence_, &sample, *vd);
      observer->on_sample_received(this, s);
    }

    return inst;
  }

take() [1/4]

template<>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< XTypes::DynamicSample >::take ( AbstractSamples &  , DDS::SampleStateMask  , DDS::ViewStateMask  , DDS::InstanceStateMask    )

virtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 32 of file DynamicDataReaderImpl.cpp.

  {
    return DDS::RETCODE_UNSUPPORTED;
  }

take() [2/4]

template<>

OpenDDS_Dcps_Export DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< XTypes::DynamicSample >::take ( AbstractSamples &  samples, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states  )

virtual

Implements OpenDDS::DCPS::DataReaderImpl.

take() [3/4]

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states  )

inlinevirtual

Definition at line 185 of file DataReaderImpl_T.h.

    {
      DDS::ReturnCode_t const precond =
        check_inputs("take", received_data, info_seq, max_samples);
      if (DDS::RETCODE_OK != precond)
        {
          return precond;
        }

      ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex,
                        guard,
                        sample_lock_,
                        DDS::RETCODE_ERROR);

      return take_i(received_data, info_seq, max_samples, sample_states,
                    view_states, instance_states, 0);
    }

take() [4/4]

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take ( AbstractSamples &  samples, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states  )

inlinevirtual

Implements OpenDDS::DCPS::DataReaderImpl.

Definition at line 765 of file DataReaderImpl_T.h.

  {
    ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex,
                      guard,
                      sample_lock_,
                      DDS::RETCODE_ERROR);

    MessageSequenceType data;
    DDS::SampleInfoSeq infos;
    const DDS::ReturnCode_t rc = take_i(data, infos, DDS::LENGTH_UNLIMITED,
                                        sample_states, view_states, instance_states, 0);

    samples.reserve(data.length());

    for (CORBA::ULong i = 0; i < data.length(); ++i) {
      samples.push_back(infos[i], &data[i]);
    }

    return rc;
  }

take_i()

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take_i ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, CORBA::Long  max_samples, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states, DDS::QueryCondition_ptr  a_condition  )

inlineprivate

Definition at line 1416 of file DataReaderImpl_T.h.

{
  typename DDSTraits<MessageType>::MessageSequenceAdapterType received_data_p(received_data);

#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
  if (subqos_.presentation.access_scope == DDS::GROUP_PRESENTATION_QOS && !coherent_) {
    return DDS::RETCODE_PRECONDITION_NOT_MET;
  }

  const bool group_coherent_ordered =
    subqos_.presentation.access_scope == DDS::GROUP_PRESENTATION_QOS
    && subqos_.presentation.coherent_access
    && subqos_.presentation.ordered_access;

  if (group_coherent_ordered && coherent_) {
    max_samples = 1;
  }
#endif

  RakeResults<MessageType> results(this, received_data, info_seq, max_samples, subqos_.presentation,
#ifndef OPENDDS_NO_QUERY_CONDITION
                                   a_condition,
#endif
                                   DDS_OPERATION_TAKE);

  const Observer_rch observer = get_observer(Observer::e_SAMPLE_TAKEN);

#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
  if (!group_coherent_ordered) {
#endif
    const HandleSet& matches = lookup_matching_instances(sample_states, view_states, instance_states);
    for (HandleSet::const_iterator it = matches.begin(), next = it; it != matches.end(); it = next) {
      ++next; // pre-increment iterator, in case updates cause changes to match set
      const DDS::InstanceHandle_t handle = *it;
      const SubscriptionInstance_rch inst = get_handle_instance(handle);
      if (!inst) continue;

      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)) {
        results.insert_sample(item, &inst->rcvd_samples_, inst, ++i);

        const ValueDispatcher* vd = get_value_dispatcher();
        if (observer && item->registered_data_ && vd) {
          Observer::Sample s(handle, inst->instance_state_->instance_state(), *item, *vd);
          observer->on_sample_taken(this, s);
        }
      }
    }
#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
  } else {
    const RakeData item = group_coherent_ordered_data_.get_data();
    results.insert_sample(item.rde_, item.rdel_, item.si_, item.index_in_instance_);
  }
#endif

  results.copy_to_user();

  DDS::ReturnCode_t ret = DDS::RETCODE_NO_DATA;
  if (received_data.length()) {
    ret = DDS::RETCODE_OK;
    if (received_data.maximum() == 0) { // using ZeroCopy
      received_data_p.set_loaner(this);
    }
  }

  post_read_or_take();
  return ret;
}

take_instance()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take_instance ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states  )

inlinevirtual

Definition at line 409 of file DataReaderImpl_T.h.

  {
    DDS::ReturnCode_t const precond =
      check_inputs("take_instance", received_data, info_seq, max_samples);
    if (DDS::RETCODE_OK != precond)
      {
        return precond;
      }

    ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex,
                      guard,
                      sample_lock_,
                      DDS::RETCODE_ERROR);
    return take_instance_i(received_data, info_seq, max_samples, a_handle,
                           sample_states, view_states, instance_states, 0);
  }

take_instance_i()

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take_instance_i ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states, DDS::QueryCondition_ptr  a_condition  )

inlineprivate

Definition at line 1563 of file DataReaderImpl_T.h.

{
  const SubscriptionInstance_rch inst = get_handle_instance(a_handle);
  if (!inst) return DDS::RETCODE_BAD_PARAMETER;

  typename DDSTraits<MessageType>::MessageSequenceAdapterType received_data_p(received_data);

  RakeResults<MessageType> results(this, received_data, info_seq, max_samples, subqos_.presentation,
#ifndef OPENDDS_NO_QUERY_CONDITION
                                   a_condition,
#endif
                                   DDS_OPERATION_TAKE);

  const InstanceState_rch state_obj = inst->instance_state_;
  if (state_obj->match(view_states, instance_states)) {
    const Observer_rch observer = get_observer(Observer::e_SAMPLE_TAKEN);
    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)) {
      results.insert_sample(item, &inst->rcvd_samples_, inst, ++i);
      const ValueDispatcher* vd = get_value_dispatcher();
      if (observer && item->registered_data_ && vd) {
        Observer::Sample s(a_handle, inst->instance_state_->instance_state(), *item, *vd);
        observer->on_sample_taken(this, s);
      }
    }
  }

  results.copy_to_user();

  DDS::ReturnCode_t ret = DDS::RETCODE_NO_DATA;
  if (received_data.length()) {
    ret = DDS::RETCODE_OK;
    if (received_data.maximum() == 0) { // using ZeroCopy
      received_data_p.set_loaner(this);
    }
  }

  post_read_or_take();
  return ret;
}

take_instance_w_condition()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take_instance_w_condition ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::ReadCondition_ptr  a_condition  )

inlinevirtual

Definition at line 473 of file DataReaderImpl_T.h.

  {
    DDS::ReturnCode_t const precond =
      check_inputs("take_instance_w_condition", received_data, info_seq,
                   max_samples);
    if (DDS::RETCODE_OK != precond)
      {
        return precond;
      }

    ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, guard, sample_lock_,
                      DDS::RETCODE_ERROR);

    if (!has_readcondition(a_condition))
      {
        return DDS::RETCODE_PRECONDITION_NOT_MET;
      }

#ifndef OPENDDS_NO_QUERY_CONDITION
    DDS::QueryCondition_ptr query_condition =
        dynamic_cast< DDS::QueryCondition_ptr >(a_condition);
#endif

    return take_instance_i(received_data, info_seq, max_samples, a_handle,
                           a_condition->get_sample_state_mask(),
                           a_condition->get_view_state_mask(),
                           a_condition->get_instance_state_mask(),
#ifndef OPENDDS_NO_QUERY_CONDITION
                           query_condition
#else
                           0
#endif
                           );
  }

take_next_instance()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take_next_instance ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states  )

inlinevirtual

Definition at line 533 of file DataReaderImpl_T.h.

  {
    DDS::ReturnCode_t const precond =
      check_inputs("take_next_instance", received_data, info_seq, max_samples);
    if (DDS::RETCODE_OK != precond)
      {
        return precond;
      }

    return take_next_instance_i(received_data, info_seq, max_samples, a_handle,
                                sample_states, view_states, instance_states, 0);
  }

take_next_instance_i()

template<typename MessageType>

DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take_next_instance_i ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::SampleStateMask  sample_states, DDS::ViewStateMask  view_states, DDS::InstanceStateMask  instance_states, DDS::QueryCondition_ptr  a_condition  )

inlineprivate

Definition at line 1664 of file DataReaderImpl_T.h.

{
  ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, sample_lock_, DDS::RETCODE_ERROR);

  typename InstanceMap::iterator it = instance_map_.begin();
  const typename InstanceMap::iterator the_end = instance_map_.end();
  if (a_handle != DDS::HANDLE_NIL) {
    const typename ReverseInstanceMap::const_iterator pos = reverse_instance_map_.find(a_handle);
    if (pos != reverse_instance_map_.end()) {
      it = pos->second;
      ++it;
    } else {
      it = the_end;
    }
  }

  DDS::InstanceHandle_t handle(DDS::HANDLE_NIL);
  for (; it != the_end; ++it) {
    handle = it->second;
    const DDS::ReturnCode_t status =
      take_instance_i(received_data, info_seq, max_samples, handle,
                      sample_states, view_states, instance_states,
#ifndef OPENDDS_NO_QUERY_CONDITION
                      a_condition);
#else
      0);
#endif
    if (status != DDS::RETCODE_NO_DATA) {
      total_samples();  // see if we are empty
      post_read_or_take();
      return status;
    }
  }

  post_read_or_take();
  return DDS::RETCODE_NO_DATA;
}

take_next_instance_w_condition()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take_next_instance_w_condition ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  info_seq, ::CORBA::Long  max_samples, DDS::InstanceHandle_t  a_handle, DDS::ReadCondition_ptr  a_condition  )

inlinevirtual

Definition at line 593 of file DataReaderImpl_T.h.

  {
    DDS::ReturnCode_t const precond =
      check_inputs("take_next_instance_w_condition", received_data, info_seq,
                   max_samples);
    if (DDS::RETCODE_OK != precond)
      {
        return precond;
      }

    ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, guard, sample_lock_,
                      DDS::RETCODE_ERROR);

    if (!has_readcondition(a_condition))
      {
        return DDS::RETCODE_PRECONDITION_NOT_MET;
      }

#ifndef OPENDDS_NO_QUERY_CONDITION
    DDS::QueryCondition_ptr query_condition =
        dynamic_cast< DDS::QueryCondition_ptr >(a_condition);
#endif

    return take_next_instance_i(received_data, info_seq, max_samples, a_handle,
                                a_condition->get_sample_state_mask(),
                                a_condition->get_view_state_mask(),
                                a_condition->get_instance_state_mask(),
#ifndef OPENDDS_NO_QUERY_CONDITION
                                query_condition
#else
                                0
#endif
                                );
  }

take_next_sample()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take_next_sample ( MessageType &  received_data, DDS::SampleInfo &  sample_info_ref  )

inlinevirtual

Definition at line 328 of file DataReaderImpl_T.h.

  {
    bool found_data = false;
    ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, sample_lock_, DDS::RETCODE_ERROR);

    const Observer_rch observer = get_observer(Observer::e_SAMPLE_TAKEN);

    const CORBA::ULong sample_states = DDS::NOT_READ_SAMPLE_STATE;
    const HandleSet& matches = lookup_matching_instances(sample_states, DDS::ANY_VIEW_STATE, DDS::ANY_INSTANCE_STATE);
    for (HandleSet::const_iterator it = matches.begin(), next = it; it != matches.end(); it = next) {
      ++next; // pre-increment iterator, in case updates cause changes to match set
      const DDS::InstanceHandle_t handle = *it;
      const SubscriptionInstance_rch inst = get_handle_instance(handle);
      if (!inst) continue;

      bool most_recent_generation = false;
      ReceivedDataElement* item = inst->rcvd_samples_.get_next_match(sample_states, 0);
      if (item) {
        if (item->registered_data_) {
          received_data = *static_cast<MessageType*>(item->registered_data_);
        }
        inst->instance_state_->sample_info(sample_info_ref, item);
        inst->rcvd_samples_.mark_read(item);

        const ValueDispatcher* vd = get_value_dispatcher();
        if (observer && item->registered_data_ && vd) {
          Observer::Sample s(sample_info_ref.instance_handle, sample_info_ref.instance_state, *item, *vd);
          observer->on_sample_taken(this, s);
        }

        if (!most_recent_generation) {
          most_recent_generation = inst->instance_state_->most_recent_generation(item);
        }

        if (most_recent_generation) {
          inst->instance_state_->accessed();
        }

        // Get the sample_ranks, generation_ranks, and
        // absolute_generation_ranks for this info_seq
        sample_info(sample_info_ref, inst->rcvd_samples_.peek_tail());

        inst->rcvd_samples_.remove(item);
        item->dec_ref();
        item = 0;

        found_data = true;

        break;
      }
    }

    post_read_or_take();
    return found_data ? DDS::RETCODE_OK : DDS::RETCODE_NO_DATA;
  }

take_w_condition()

template<typename MessageType>

virtual DDS::ReturnCode_t OpenDDS::DCPS::DataReaderImpl_T< MessageType >::take_w_condition ( MessageSequenceType &  received_data, DDS::SampleInfoSeq &  sample_info, ::CORBA::Long  max_samples, DDS::ReadCondition_ptr  a_condition  )

inlinevirtual

Definition at line 241 of file DataReaderImpl_T.h.

    {
      DDS::ReturnCode_t const precond =
        check_inputs("take_w_condition", received_data, sample_info, max_samples);
      if (DDS::RETCODE_OK != precond)
        {
          return precond;
        }

      ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, guard, sample_lock_,
                        DDS::RETCODE_ERROR);

      if (!has_readcondition(a_condition))
        {
          return DDS::RETCODE_PRECONDITION_NOT_MET;
        }

      return take_i(received_data, sample_info, max_samples,
                    a_condition->get_sample_state_mask(),
                    a_condition->get_view_state_mask(),
                    a_condition->get_instance_state_mask(),
#ifndef OPENDDS_NO_QUERY_CONDITION
                    dynamic_cast< DDS::QueryCondition_ptr >(a_condition)
#else
                    0
#endif
                    );
    }

Member Data Documentation

data_allocator_

template<typename MessageType>

unique_ptr<DataAllocator> OpenDDS::DCPS::DataReaderImpl_T< MessageType >::data_allocator_

private

Definition at line 2415 of file DataReaderImpl_T.h.

filter_delayed_sample_map_

template<typename MessageType>

FilterDelayedSampleMap OpenDDS::DCPS::DataReaderImpl_T< MessageType >::filter_delayed_sample_map_

private

Definition at line 2440 of file DataReaderImpl_T.h.

filter_delayed_sample_queue_

template<typename MessageType>

FilterDelayedSampleQueue OpenDDS::DCPS::DataReaderImpl_T< MessageType >::filter_delayed_sample_queue_

private

Definition at line 2442 of file DataReaderImpl_T.h.

filter_delayed_sample_task_

template<typename MessageType>

RcHandle<DRISporadicTask> OpenDDS::DCPS::DataReaderImpl_T< MessageType >::filter_delayed_sample_task_

private

Definition at line 2422 of file DataReaderImpl_T.h.

instance_map_

template<typename MessageType>

InstanceMap OpenDDS::DCPS::DataReaderImpl_T< MessageType >::instance_map_

private

Definition at line 2417 of file DataReaderImpl_T.h.

marshal_skip_serialize_

template<typename MessageType>

bool OpenDDS::DCPS::DataReaderImpl_T< MessageType >::marshal_skip_serialize_

private

Definition at line 2444 of file DataReaderImpl_T.h.

reverse_instance_map_

template<typename MessageType>

ReverseInstanceMap OpenDDS::DCPS::DataReaderImpl_T< MessageType >::reverse_instance_map_

private

Definition at line 2418 of file DataReaderImpl_T.h.

---

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

• DataReaderImpl_T.h