Implements the OpenDDS::DCPS::DataWriterRemote interfaces and DDS::DataWriter interfaces. More...

#include <DataWriterImpl.h>

Inheritance diagram for OpenDDS::DCPS::DataWriterImpl:

[legend]

Collaboration diagram for OpenDDS::DCPS::DataWriterImpl:

[legend]

Classes
struct	AckCustomization

struct	AckToken

class	EncodingMode

struct	ReaderInfo

Public Types
typedef Dynamic_Cached_Allocator_With_Overflow< ACE_Thread_Mutex >	DataAllocator

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

typedef DDS::DataWriter ::_var_type	_var_type

Public Types inherited from CORBA::LocalObject
typedef LocalObject_ptr	_ptr_type

typedef LocalObject_var	_var_type

typedef LocalObject_out	_out_type

Public Types inherited from CORBA::Object
typedef Object_ptr	_ptr_type

typedef Object_var	_var_type

typedef Object_out	_out_type

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

typedef DDS::Entity ::_var_type	_var_type

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

Public Member Functions
typedef	OPENDDS_MAP_CMP (GUID_t, SequenceNumber, GUID_tKeyLessThan) RepoIdToSequenceMap

	DataWriterImpl ()

virtual	~DataWriterImpl ()

void	set_marshal_skip_serialize (bool value)

bool	get_marshal_skip_serialize () const

DataAllocator *	data_allocator () const

virtual DDS::InstanceHandle_t	get_instance_handle ()

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

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

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

virtual DDS::DataWriterListener_ptr	get_listener ()

virtual DDS::Topic_ptr	get_topic ()

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

virtual DDS::Publisher_ptr	get_publisher ()

virtual DDS::ReturnCode_t	get_liveliness_lost_status (DDS::LivelinessLostStatus &status)

virtual DDS::ReturnCode_t	get_offered_deadline_missed_status (DDS::OfferedDeadlineMissedStatus &status)

virtual DDS::ReturnCode_t	get_offered_incompatible_qos_status (DDS::OfferedIncompatibleQosStatus &status)

virtual DDS::ReturnCode_t	get_publication_matched_status (DDS::PublicationMatchedStatus &status)

TimeDuration	liveliness_check_interval (DDS::LivelinessQosPolicyKind kind)

bool	participant_liveliness_activity_after (const MonotonicTimePoint &tv)

virtual DDS::ReturnCode_t	assert_liveliness ()

DDS::ReturnCode_t	assert_liveliness_by_participant ()

typedef	OPENDDS_VECTOR (DDS::InstanceHandle_t) InstanceHandleVec

void	get_instance_handles (InstanceHandleVec &instance_handles)

void	get_readers (RepoIdSet &readers)

virtual DDS::ReturnCode_t	get_matched_subscriptions (DDS::InstanceHandleSeq &subscription_handles)

virtual DDS::ReturnCode_t	get_matched_subscription_data (DDS::SubscriptionBuiltinTopicData &subscription_data, DDS::InstanceHandle_t subscription_handle)

virtual DDS::ReturnCode_t	enable ()

virtual void	add_association (const GUID_t &yourId, const ReaderAssociation &reader, bool active)

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

virtual void	remove_associations (const ReaderIdSeq &readers, bool callback)

virtual void	replay_durable_data_for (const GUID_t &remote_sub_id)

virtual void	update_incompatible_qos (const IncompatibleQosStatus &status)

virtual void	update_subscription_params (const GUID_t &readerId, const DDS::StringSeq &params)

void	cleanup ()

void	init (TopicImpl topic_servant, const DDS::DataWriterQos &qos, DDS::DataWriterListener_ptr a_listener, const DDS::StatusMask &mask, WeakRcHandle< DomainParticipantImpl > participant_servant, PublisherImpl publisher_servant)

void	send_all_to_flush_control (ACE_Guard< ACE_Recursive_Thread_Mutex > &guard)

DDS::ReturnCode_t	register_instance_i (DDS::InstanceHandle_t &handle, Message_Block_Ptr data, const DDS::Time_t &source_timestamp)

DDS::ReturnCode_t	register_instance_from_durable_data (DDS::InstanceHandle_t &handle, Message_Block_Ptr data, const DDS::Time_t &source_timestamp)

DDS::ReturnCode_t	unregister_instance_i (DDS::InstanceHandle_t handle, const DDS::Time_t &source_timestamp)

void	unregister_instances (const DDS::Time_t &source_timestamp)

DDS::ReturnCode_t	write (Message_Block_Ptr sample, DDS::InstanceHandle_t handle, const DDS::Time_t &source_timestamp, GUIDSeq filter_out, const void real_data)

DDS::ReturnCode_t	write_sample (const Sample &sample, DDS::InstanceHandle_t handle, const DDS::Time_t &source_timestamp, GUIDSeq *filter_out)

DDS::ReturnCode_t	dispose (DDS::InstanceHandle_t handle, const DDS::Time_t &source_timestamp)

DDS::ReturnCode_t	num_samples (DDS::InstanceHandle_t handle, size_t &size)

ACE_UINT64	get_unsent_data (SendStateDataSampleList &list)

SendStateDataSampleList	get_resend_data ()

GUID_t	get_dp_id ()

void	unregister_all ()

void	data_delivered (const DataSampleElement *sample)

void	transport_discovery_change ()

void	control_delivered (const Message_Block_Ptr &sample)

bool	should_ack () const
	Does this writer have samples to be acknowledged? More...

AckToken	create_ack_token (DDS::Duration_t max_wait) const
	Create an AckToken for ack operations. More...

virtual void	retrieve_inline_qos_data (TransportSendListener::InlineQosData &qos_data) const

virtual bool	check_transport_qos (const TransportInst &inst)

bool	coherent_changes_pending ()
	Are coherent changes pending? More...

void	begin_coherent_changes ()
	Starts a coherent change set; should only be called once. More...

void	end_coherent_changes (const GroupCoherentSamples &group_samples)
	Ends a coherent change set; should only be called once. More...

char const *	get_type_name () const

void	data_dropped (const DataSampleElement *element, bool dropped_by_transport)

void	control_dropped (const Message_Block_Ptr &sample, bool dropped_by_transport)

ACE_Recursive_Thread_Mutex &	get_lock () const

DDS::DataWriterListener_ptr	listener_for (DDS::StatusKind kind)

virtual int	handle_timeout (const ACE_Time_Value &tv, const void *arg)
	Handle the assert liveliness timeout. More...

void	send_suspended_data ()

void	remove_all_associations ()

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

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

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

void	notify_publication_disconnected (const ReaderIdSeq &subids)

void	notify_publication_reconnected (const ReaderIdSeq &subids)

void	notify_publication_lost (const ReaderIdSeq &subids)

DDS::ReturnCode_t	create_sample_data_message (Message_Block_Ptr data, DDS::InstanceHandle_t instance_handle, DataSampleHeader &header_data, Message_Block_Ptr &message, const DDS::Time_t &source_timestamp, bool content_filter)

bool	persist_data ()

void	wait_pending ()
	Wait for pending data and control messages to drain. More...

void	set_wait_pending_deadline (const MonotonicTimePoint &deadline)

DDS::InstanceHandle_t	get_next_handle ()

virtual RcHandle< EntityImpl >	parent () const

bool	filter_out (const DataSampleElement &elt, const OPENDDS_STRING &filterClassName, const FilterEvaluator &evaluator, const DDS::StringSeq &expression_params) const

DataBlockLockPool::DataBlockLock *	get_db_lock ()

PublicationInstance_rch	get_handle_instance (DDS::InstanceHandle_t handle)

virtual WeakRcHandle< ICE::Endpoint >	get_ice_endpoint ()

GUID_t	get_guid () const

SequenceNumber	get_max_sn () const

const ValueDispatcher *	get_value_dispatcher () const

DDS::ReturnCode_t	get_key_value (Sample_rch &sample, DDS::InstanceHandle_t handle)

DDS::InstanceHandle_t	lookup_instance (const Sample &sample)

DDS::InstanceHandle_t	register_instance_w_timestamp (const Sample &sample, const DDS::Time_t &timestamp)

DDS::ReturnCode_t	unregister_instance_w_timestamp (const Sample &sample, DDS::InstanceHandle_t instance_handle, const DDS::Time_t &timestamp)

DDS::ReturnCode_t	dispose_w_timestamp (const Sample &sample, DDS::InstanceHandle_t instance_handle, const DDS::Time_t &source_timestamp)

Public Member Functions inherited from DDS::DataWriter
ReturnCode_t	set_qos (in DataWriterQos qos)

ReturnCode_t	get_qos (inout DataWriterQos qos)

ReturnCode_t	set_listener (in DataWriterListener a_listener, in StatusMask mask)

ReturnCode_t	wait_for_acknowledgments (in Duration_t max_wait)

ReturnCode_t	get_liveliness_lost_status (inout LivelinessLostStatus status)

ReturnCode_t	get_offered_deadline_missed_status (inout OfferedDeadlineMissedStatus status)

ReturnCode_t	get_offered_incompatible_qos_status (inout OfferedIncompatibleQosStatus status)

ReturnCode_t	get_publication_matched_status (inout PublicationMatchedStatus status)

ReturnCode_t	get_matched_subscriptions (inout InstanceHandleSeq subscription_handles)

ReturnCode_t	get_matched_subscription_data (inout SubscriptionBuiltinTopicData subscription_data, in InstanceHandle_t subscription_handle)

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

virtual void	_remove_ref ()

virtual CORBA::ULong	_refcount_value () const

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

virtual CORBA::Boolean	_non_existent (void)

virtual char *	_repository_id (void)

virtual CORBA::InterfaceDef_ptr	_get_interface (void)

virtual CORBA::Object_ptr	_get_component (void)

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

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

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

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

CORBA::Policy_ptr	_get_cached_policy (TAO_Cached_Policy_Type type)

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

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

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

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

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

virtual CORBA::ORB_ptr	_get_orb (void)

virtual TAO::ObjectKey *	_key (void)

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

virtual TAO_Abstract_ServantBase *	_servant (void) const

virtual CORBA::Boolean	_is_collocated (void) const

virtual CORBA::Boolean	_is_local (void) const

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

	Object (IOP::IOR ior, TAO_ORB_Core orb_core)

virtual TAO_Stub *	_stubobj (void) const

virtual TAO_Stub *	_stubobj (void)

virtual void	_proxy_broker (TAO::Object_Proxy_Broker *proxy_broker)

virtual CORBA::Boolean	marshal (TAO_OutputCDR &cdr)

CORBA::Boolean	is_evaluated (void) const

TAO_ORB_Core *	orb_core (void) const

IOP::IOR *	steal_ior (void)

const IOP::IOR &	ior (void) const

virtual bool	can_convert_to_ior (void) const

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

void	_decr_refcount (void)

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

virtual const char *	_interface_repository_id (void) const

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

CORBA::Policy_ptr	_get_cached_policy (TAO_Cached_Policy_Type type)

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

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

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

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

long	ref_count () const

WeakObject *	_get_weak_object () const

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

virtual	~DataWriterCallbacks ()

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

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

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::TransportSendListener
virtual	~TransportSendListener ()

virtual void	data_acked (const GUID_t &)

virtual SendControlStatus	send_control_customized (const DataLinkSet_rch &links, const DataSampleHeader &header, ACE_Message_Block msg, void extra)

Public Attributes
Atomic< int >	data_dropped_count_
	Statistics counter. More...

Atomic< int >	data_delivered_count_

MessageTracker	controlTracker

Protected Member Functions
void	check_and_set_repo_id (const GUID_t &id)

SequenceNumber	get_next_sn ()

SequenceNumber	get_next_sn_i ()

DataWriterListener_ptr	get_ext_listener ()

DDS::ReturnCode_t	wait_for_specific_ack (const AckToken &token)

void	prepare_to_delete ()

DDS::ReturnCode_t	setup_serialization ()

ACE_Message_Block *	serialize_sample (const Sample &sample)

typedef	OPENDDS_MAP_CMP (GUID_t, ReaderInfo, GUID_tKeyLessThan) RepoIdToReaderInfoMap

virtual SendControlStatus	send_control (const DataSampleHeader &header, Message_Block_Ptr msg)

TypeSupportImpl *	get_type_support () const

DDS::ReturnCode_t	instance_must_exist (const char *method_name, const Sample &sample, DDS::InstanceHandle_t &instance_handle, bool remove=false)

DDS::ReturnCode_t	get_or_create_instance_handle (DDS::InstanceHandle_t &handle, const Sample &sample, const DDS::Time_t &source_timestamp)

DDS::ReturnCode_t	write_w_timestamp (const Sample &sample, DDS::InstanceHandle_t handle, const DDS::Time_t &source_timestamp)

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

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

TAO::Object_Proxy_Broker *	proxy_broker () const

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

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

void	set_deleted (bool state)

bool	get_deleted () const

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

Protected Member Functions inherited from OpenDDS::DCPS::TransportClient
void	cdr_encapsulation (bool encap)

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

Protected Attributes
size_t	n_chunks_
	The number of chunks for the cached allocator. More...

size_t	association_chunk_multiplier_
	The multiplier for allocators affected by associations. More...

CORBA::String_var	type_name_
	The type name of associated topic. More...

DDS::DataWriterQos	qos_
	The qos policy list of this datawriter. More...

DDS::DataWriterQos	passed_qos_

WeakRcHandle< DomainParticipantImpl >	participant_servant_

ACE_Thread_Mutex	reader_info_lock_

RepoIdToReaderInfoMap	reader_info_

bool	skip_serialize_

class OpenDDS::DCPS::DataWriterImpl::EncodingMode	encoding_mode_

Security::SecurityConfig_rch	security_config_

DDS::Security::PermissionsHandle	participant_permissions_handle_

DDS::DynamicType_var	dynamic_type_

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

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

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

Private Member Functions
void	track_sequence_number (GUIDSeq *filter_out)

void	notify_publication_lost (const DDS::InstanceHandleSeq &handles)

DDS::ReturnCode_t	dispose_and_unregister (DDS::InstanceHandle_t handle, const DDS::Time_t &timestamp)

ACE_Message_Block *	create_control_message (MessageId message_id, DataSampleHeader &header, Message_Block_Ptr data, const DDS::Time_t &source_timestamp)

bool	send_liveliness (const MonotonicTimePoint &now)
	Send the liveliness message. More...

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

RcHandle< BitSubscriber >	get_builtin_subscriber_proxy () const

DDS::DomainId_t	domain_id () const

CORBA::Long	get_priority_value (const AssociationData &) const

DDS::Security::ParticipantCryptoHandle	get_crypto_handle () const

void	association_complete_i (const GUID_t &remote_id)

void	return_handle (DDS::InstanceHandle_t handle)

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

bool	need_sequence_repair ()

bool	need_sequence_repair_i () const

DDS::ReturnCode_t	send_end_historic_samples (const GUID_t &readerId)

DDS::ReturnCode_t	send_request_ack ()

typedef	OPENDDS_MAP (DDS::InstanceHandle_t, Sample_rch) InstanceHandlesToValues

typedef	OPENDDS_MAP_CMP (Sample_rch, DDS::InstanceHandle_t, SampleRchCmp) InstanceValuesToHandles

bool	insert_instance (DDS::InstanceHandle_t handle, Sample_rch &sample)

InstanceValuesToHandles::iterator	find_instance (const Sample &sample)

Private Attributes
unique_ptr< DataBlockLockPool >	db_lock_pool_

CORBA::String_var	topic_name_
	The name of associated topic. More...

GUID_t	topic_id_
	The associated topic repository id. More...

TopicDescriptionPtr< TopicImpl >	topic_servant_
	The topic servant. More...

TypeSupportImpl *	type_support_

ACE_Thread_Mutex	listener_mutex_
	Mutex to protect listener info. More...

DDS::StatusMask	listener_mask_

DDS::DataWriterListener_var	listener_
	Used to notify the entity for relevant events. More...

DDS::DomainId_t	domain_id_
	The domain id. More...

GUID_t	dp_id_

WeakRcHandle< PublisherImpl >	publisher_servant_
	The publisher servant which creates this datawriter. More...

GUID_t	publication_id_
	The repository id of this datawriter/publication. More...

SequenceNumber	sequence_number_
	The sequence number unique in DataWriter scope. More...

ACE_Thread_Mutex	sn_lock_
	Mutex for sequence_number_. More...

bool	coherent_

ACE_UINT32	coherent_samples_

RcHandle< WriteDataContainer >	data_container_
	The sample data container. More...

ACE_Recursive_Thread_Mutex	lock_

RepoIdToHandleMap	id_to_handle_map_

RepoIdSet	readers_

DDS::LivelinessLostStatus	liveliness_lost_status_
	Status conditions. More...

DDS::OfferedDeadlineMissedStatus	offered_deadline_missed_status_

DDS::OfferedIncompatibleQosStatus	offered_incompatible_qos_status_

DDS::PublicationMatchedStatus	publication_match_status_

bool	liveliness_lost_

unique_ptr< MessageBlockAllocator >	mb_allocator_
	The message block allocator. More...

unique_ptr< DataBlockAllocator >	db_allocator_
	The data block allocator. More...

unique_ptr< DataSampleHeaderAllocator >	header_allocator_
	The header data allocator. More...

unique_ptr< DataAllocator >	data_allocator_

ACE_Reactor_Timer_Interface *	reactor_

TimeDuration	liveliness_check_interval_
	The time interval for sending liveliness message. More...

MonotonicTimePoint	last_liveliness_activity_time_
	Timestamp of last write/dispose/assert_liveliness. More...

CORBA::Long	last_deadline_missed_total_count_

bool	is_bit_

ACE_UINT64	min_suspended_transaction_id_
	The cached available data while suspending and associated transaction ids. More...

ACE_UINT64	max_suspended_transaction_id_

SendStateDataSampleList	available_data_list_

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

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

bool	liveliness_asserted_

ACE_Thread_Mutex	sync_unreg_rem_assocs_lock_

RcHandle< LivenessTimer >	liveness_timer_

MonotonicTimePoint	wait_pending_deadline_

InstanceHandlesToValues	instance_handles_to_values_

InstanceValuesToHandles	instance_values_to_handles_

Friends
class	WriteDataContainer

class	PublisherImpl

class	::DDS_TEST

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

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

static LocalObject_ptr	_nil (void)

static LocalObject_ptr	_narrow (CORBA::Object_ptr obj)

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

static void	_tao_any_destructor (void *)

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

static void	tao_object_initialize (Object *)

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

static CORBA::Object_ptr	_nil (void)

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

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

Detailed Description

Implements the OpenDDS::DCPS::DataWriterRemote interfaces and DDS::DataWriter interfaces.

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

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

Note: : This class is responsible for allocating memory for the header message block (MessageBlock + DataBlock + DataSampleHeader) and the DataSampleElement. The data-type datawriter is responsible for allocating memory for the sample data message block. (e.g. MessageBlock + DataBlock + Foo data). But it gives up ownership to this WriteDataContainer.

Definition at line 80 of file DataWriterImpl.h.

Member Typedef Documentation

◆ DataAllocator

typedef Dynamic_Cached_Allocator_With_Overflow<ACE_Thread_Mutex> OpenDDS::DCPS::DataWriterImpl::DataAllocator

Definition at line 92 of file DataWriterImpl.h.

Constructor & Destructor Documentation

◆ DataWriterImpl()

OpenDDS::DCPS::DataWriterImpl::DataWriterImpl ( )

Definition at line 58 of file DataWriterImpl.cpp.

References DDS::PublicationMatchedStatus::current_count, DDS::PublicationMatchedStatus::current_count_change, DDS::HANDLE_NIL, DDS::OfferedDeadlineMissedStatus::last_instance_handle, DDS::OfferedIncompatibleQosStatus::last_policy_id, DDS::PublicationMatchedStatus::last_subscription_handle, liveliness_lost_status_, monitor_, offered_deadline_missed_status_, offered_incompatible_qos_status_, periodic_monitor_, DDS::OfferedIncompatibleQosStatus::policies, publication_match_status_, TheServiceParticipant, DDS::LivelinessLostStatus::total_count, DDS::OfferedDeadlineMissedStatus::total_count, DDS::OfferedIncompatibleQosStatus::total_count, DDS::PublicationMatchedStatus::total_count, DDS::LivelinessLostStatus::total_count_change, DDS::OfferedDeadlineMissedStatus::total_count_change, DDS::OfferedIncompatibleQosStatus::total_count_change, and DDS::PublicationMatchedStatus::total_count_change.

   : data_dropped_count_(0)
   , data_delivered_count_(0)
   , controlTracker("DataWriterImpl")
   , n_chunks_(TheServiceParticipant->n_chunks())
   , association_chunk_multiplier_(TheServiceParticipant->association_chunk_multiplier())
   , qos_(TheServiceParticipant->initial_DataWriterQos())
   , skip_serialize_(false)
   , db_lock_pool_(new DataBlockLockPool((unsigned long)TheServiceParticipant->n_chunks()))
   , topic_id_(GUID_UNKNOWN)
   , topic_servant_(0)
   , type_support_(0)
   , listener_mask_(DEFAULT_STATUS_MASK)
   , domain_id_(0)
   , publication_id_(GUID_UNKNOWN)
   , sequence_number_(SequenceNumber::SEQUENCENUMBER_UNKNOWN())
   , coherent_(false)
   , coherent_samples_(0)
   , liveliness_lost_(false)
   , reactor_(0)
   , liveliness_check_interval_(TimeDuration::max_value)
   , last_deadline_missed_total_count_(0)
   , is_bit_(false)
   , min_suspended_transaction_id_(0)
   , max_suspended_transaction_id_(0)
   , liveliness_asserted_(false)
   , liveness_timer_(make_rch<LivenessTimer>(ref(*this)))
 {
   liveliness_lost_status_.total_count = 0;
   liveliness_lost_status_.total_count_change = 0;
 
   offered_deadline_missed_status_.total_count = 0;
   offered_deadline_missed_status_.total_count_change = 0;
   offered_deadline_missed_status_.last_instance_handle = DDS::HANDLE_NIL;
 
   offered_incompatible_qos_status_.total_count = 0;
   offered_incompatible_qos_status_.total_count_change = 0;
   offered_incompatible_qos_status_.last_policy_id = 0;
   offered_incompatible_qos_status_.policies.length(0);
 
   publication_match_status_.total_count = 0;
   publication_match_status_.total_count_change = 0;
   publication_match_status_.current_count = 0;
   publication_match_status_.current_count_change = 0;
   publication_match_status_.last_subscription_handle = DDS::HANDLE_NIL;
 
   monitor_.reset(TheServiceParticipant->monitor_factory_->create_data_writer_monitor(this));
   periodic_monitor_.reset(TheServiceParticipant->monitor_factory_->create_data_writer_periodic_monitor(this));
 }

◆ ~DataWriterImpl()

OpenDDS::DCPS::DataWriterImpl::~DataWriterImpl ( )

virtual

Definition at line 110 of file DataWriterImpl.cpp.

References DBG_ENTRY_LVL, participant_servant_, and publication_id_.

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

Member Function Documentation

◆ add_association()

void OpenDDS::DCPS::DataWriterImpl::add_association	(	const GUID_t &	yourId,
		const ReaderAssociation &	reader,
		bool	active
	)

virtual

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 212 of file DataWriterImpl.cpp.

 {
   DBG_ENTRY_LVL("DataWriterImpl", "add_association", 6);
 
   if (DCPS_debug_level) {
     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) DataWriterImpl::add_association - ")
                ACE_TEXT("bit %d local %C remote %C\n"), is_bit_,
                LogGuid(yourId).c_str(),
                LogGuid(reader.readerId).c_str()));
   }
 
   if (get_deleted()) {
     if (DCPS_debug_level)
       ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) DataWriterImpl::add_association")
                  ACE_TEXT(" This is a deleted datawriter, ignoring add.\n")));
 
     return;
   }
 
   check_and_set_repo_id(yourId);
 
   {
     ACE_GUARD(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_);
     reader_info_.insert(std::make_pair(reader.readerId,
                                        ReaderInfo(reader.filterClassName,
                                                   TheServiceParticipant->publisher_content_filter() ? reader.filterExpression : "",
                                                   reader.exprParams, participant_servant_,
                                                   reader.readerQos.durability.kind > DDS::VOLATILE_DURABILITY_QOS)));
   }
 
   if (DCPS_debug_level > 4) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) DataWriterImpl::add_association(): ")
                ACE_TEXT("adding subscription to publication %C with priority %d.\n"),
                LogGuid(get_guid()).c_str(),
                qos_.transport_priority.value));
   }
 
   AssociationData data;
   data.remote_id_ = reader.readerId;
   data.remote_data_ = reader.readerTransInfo;
   data.discovery_locator_ = reader.readerDiscInfo;
   data.participant_discovered_at_ = reader.participantDiscoveredAt;
   data.remote_transport_context_ = reader.transportContext;
   data.remote_reliable_ =
     (reader.readerQos.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS);
   data.remote_durable_ =
     (reader.readerQos.durability.kind > DDS::VOLATILE_DURABILITY_QOS);
 
   if (associate(data, active)) {
     const Observer_rch observer = get_observer(Observer::e_ASSOCIATED);
     if (observer) {
       observer->on_associated(this, data.remote_id_);
     }
   } else {
     //FUTURE: inform inforepo and try again as passive peer
     if (DCPS_debug_level) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) DataWriterImpl::add_association: ")
                  ACE_TEXT("ERROR: transport layer failed to associate.\n")));
     }
   }
 }

◆ assert_liveliness()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::assert_liveliness ( )

virtual

Implements DDS::DataWriter.

Definition at line 1190 of file DataWriterImpl.cpp.

References DDS::AUTOMATIC_LIVELINESS_QOS, DDS::LivelinessQosPolicy::kind, DDS::DataWriterQos::liveliness, DDS::MANUAL_BY_PARTICIPANT_LIVELINESS_QOS, DDS::MANUAL_BY_TOPIC_LIVELINESS_QOS, OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), participant_servant_, qos_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, and send_liveliness().

 {
   switch (this->qos_.liveliness.kind) {
   case DDS::AUTOMATIC_LIVELINESS_QOS:
     // Do nothing.
     break;
   case DDS::MANUAL_BY_PARTICIPANT_LIVELINESS_QOS:
     {
       RcHandle<DomainParticipantImpl> participant = this->participant_servant_.lock();
       if (participant) {
         return participant->assert_liveliness();
       }
     }
     break;
   case DDS::MANUAL_BY_TOPIC_LIVELINESS_QOS:
     if (!send_liveliness(MonotonicTimePoint::now())) {
       return DDS::RETCODE_ERROR;
     }
     break;
   }
 
   return DDS::RETCODE_OK;
 }

◆ assert_liveliness_by_participant()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::assert_liveliness_by_participant ( )

Definition at line 1215 of file DataWriterImpl.cpp.

References DDS::LivelinessQosPolicy::kind, DDS::DataWriterQos::liveliness, liveliness_asserted_, DDS::MANUAL_BY_PARTICIPANT_LIVELINESS_QOS, qos_, and DDS::RETCODE_OK.

 {
   // This operation is called by participant.
   if (this->qos_.liveliness.kind == DDS::MANUAL_BY_PARTICIPANT_LIVELINESS_QOS) {
     // Set a flag indicating that we should send a liveliness message on the timer if necessary.
     liveliness_asserted_ = true;
   }
 
   return DDS::RETCODE_OK;
 }

◆ association_complete_i()

void OpenDDS::DCPS::DataWriterImpl::association_complete_i ( const GUID_t & remote_id )

private

Definition at line 370 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), available_data_list_, OpenDDS::DCPS::SendStateDataSampleList::begin(), OpenDDS::DCPS::bind(), controlTracker, create_control_message(), DDS::PublicationMatchedStatus::current_count, DDS::PublicationMatchedStatus::current_count_change, data_container_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::STUN::encoding(), OpenDDS::DCPS::SendStateDataSampleList::end(), OpenDDS::DCPS::END_HISTORIC_SAMPLES, OpenDDS::DCPS::SendStateDataSampleList::enqueue_tail(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), get_db_lock(), get_lock(), get_max_sn(), get_resend_data(), header, id_to_handle_map_, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::insert(), is_bit_, CORBA::is_nil(), OpenDDS::DCPS::Encoding::KIND_UNALIGNED_CDR, DDS::PublicationMatchedStatus::last_subscription_handle, DDS::DataWriterQos::lifespan, listener_for(), LM_DEBUG, LM_ERROR, LM_INFO, LM_WARNING, lock_, ACE_Message_Block::MB_DATA, OpenDDS::DCPS::MessageTracker::message_dropped(), OpenDDS::DCPS::MessageTracker::message_sent(), monitor_, OpenDDS::DCPS::move(), OpenDDS::DCPS::EntityImpl::notify_status_condition(), OpenDDS::DCPS::TimePoint_T< SystemClock >::now(), OPENDDS_STRING, participant_servant_, publication_id_, publication_match_status_, DDS::PUBLICATION_MATCHED_STATUS, publisher_servant_, qos_, reader_info_, reader_info_lock_, readers_, OpenDDS::DCPS::SEND_CONTROL_ERROR, OpenDDS::DCPS::TransportClient::send_w_control(), OpenDDS::DCPS::serialized_size(), OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), DDS::PublicationMatchedStatus::total_count, and DDS::PublicationMatchedStatus::total_count_change.

Referenced by transport_assoc_done().

 {
   DBG_ENTRY_LVL("DataWriterImpl", "association_complete_i", 6);
 
   bool reader_durable = false;
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
   OPENDDS_STRING filterClassName;
   RcHandle<FilterEvaluator> eval;
   DDS::StringSeq expression_params;
 #endif
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
     if (DCPS_debug_level >= 1) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) DataWriterImpl::association_complete_i - ")
                  ACE_TEXT("bit %d local %C remote %C\n"),
                  is_bit_,
                  LogGuid(this->publication_id_).c_str(),
                  LogGuid(remote_id).c_str()));
     }
 
     if (insert(readers_, remote_id) == -1) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::association_complete_i: ")
                  ACE_TEXT("insert %C from pending failed.\n"),
                  LogGuid(remote_id).c_str()));
     }
   }
   {
     ACE_GUARD(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_);
     RepoIdToReaderInfoMap::const_iterator it = reader_info_.find(remote_id);
 
     if (it != reader_info_.end()) {
       reader_durable = it->second.durable_;
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
       filterClassName = it->second.filter_class_name_;
       eval = it->second.eval_;
       expression_params = it->second.expression_params_;
 #endif
     }
   }
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 
   if (!is_bit_) {
 
     RcHandle<DomainParticipantImpl> participant = this->participant_servant_.lock();
 
     if (!participant)
       return;
 
     data_container_->add_reader_acks(remote_id, get_max_sn());
 
     const DDS::InstanceHandle_t handle = participant->assign_handle(remote_id);
 
     {
       // protect publication_match_status_ and status changed flags.
       ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
       if (DCPS::bind(id_to_handle_map_, remote_id, handle) != 0) {
         ACE_DEBUG((LM_WARNING,
                    ACE_TEXT("(%P|%t) WARNING: DataWriterImpl::association_complete_i: ")
                    ACE_TEXT("id_to_handle_map_%C = 0x%x failed.\n"),
                    LogGuid(remote_id).c_str(),
                    handle));
         return;
 
       } else if (DCPS_debug_level > 4) {
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) DataWriterImpl::association_complete_i: ")
                    ACE_TEXT("id_to_handle_map_%C = 0x%x.\n"),
                    LogGuid(remote_id).c_str(),
                    handle));
       }
 
       ++publication_match_status_.total_count;
       ++publication_match_status_.total_count_change;
       ++publication_match_status_.current_count;
       ++publication_match_status_.current_count_change;
       publication_match_status_.last_subscription_handle = handle;
       set_status_changed_flag(DDS::PUBLICATION_MATCHED_STATUS, true);
     }
 
     DDS::DataWriterListener_var listener =
       listener_for(DDS::PUBLICATION_MATCHED_STATUS);
 
     if (!CORBA::is_nil(listener.in())) {
 
       listener->on_publication_matched(this, publication_match_status_);
 
       // TBD - why does the spec say to change this but not
       // change the ChangeFlagStatus after a listener call?
       publication_match_status_.total_count_change = 0;
       publication_match_status_.current_count_change = 0;
     }
 
     notify_status_condition();
   } else {
     data_container_->add_reader_acks(remote_id, get_max_sn());
   }
 
   // Support DURABILITY QoS
   if (reader_durable) {
     // Tell the WriteDataContainer to resend all sending/sent
     // samples.
     this->data_container_->reenqueue_all(remote_id, this->qos_.lifespan
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
                                          , filterClassName, eval.in(), expression_params
 #endif
                                         );
 
     // Acquire the data writer container lock to avoid deadlock. The
     // thread calling association_complete() has to acquire lock in the
     // same order as the write()/register() operation.
 
     // Since the thread calling association_complete() is the ORB
     // thread, it may have some performance penalty. If the
     // performance is an issue, we may need a new thread to handle the
     // data_available() calls.
     ACE_GUARD(ACE_Recursive_Thread_Mutex,
               guard,
               this->get_lock());
 
     SendStateDataSampleList list = this->get_resend_data();
     {
       ACE_GUARD(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_);
       // Update the reader's expected sequence
       SequenceNumber& seq =
         reader_info_.find(remote_id)->second.expected_sequence_;
 
       for (SendStateDataSampleList::iterator list_el = list.begin();
            list_el != list.end(); ++list_el) {
         list_el->get_header().historic_sample_ = true;
 
         if (list_el->get_header().sequence_ > seq) {
           seq = list_el->get_header().sequence_;
         }
       }
     }
 
     RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();
     if (!publisher || publisher->is_suspended()) {
       this->available_data_list_.enqueue_tail(list);
 
     } else {
       if (DCPS_debug_level >= 4) {
         ACE_DEBUG((LM_INFO, "(%P|%t) Sending historic samples\n"));
       }
 
       const Encoding encoding(Encoding::KIND_UNALIGNED_CDR);
       size_t size = 0;
       serialized_size(encoding, size, remote_id);
       Message_Block_Ptr data(
         new ACE_Message_Block(size, ACE_Message_Block::MB_DATA, 0, 0, 0,
                               get_db_lock()));
       Serializer ser(data.get(), encoding);
       ser << remote_id;
 
       DataSampleHeader header;
       Message_Block_Ptr end_historic_samples(
         create_control_message(
           END_HISTORIC_SAMPLES, header, move(data),
           SystemTimePoint::now().to_dds_time()));
 
       this->controlTracker.message_sent();
       guard.release();
       ACE_Reverse_Lock<ACE_Recursive_Thread_Mutex> rev_lock(lock_);
       ACE_Guard<ACE_Reverse_Lock<ACE_Recursive_Thread_Mutex> > rev_guard(rev_lock);
       SendControlStatus ret = send_w_control(list, header, move(end_historic_samples), remote_id);
       if (ret == SEND_CONTROL_ERROR) {
         ACE_ERROR((LM_WARNING, ACE_TEXT("(%P|%t) WARNING: ")
                              ACE_TEXT("DataWriterImpl::association_complete_i: ")
                              ACE_TEXT("send_w_control failed.\n")));
         this->controlTracker.message_dropped();
       }
     }
   }
 }

◆ begin_coherent_changes()

void OpenDDS::DCPS::DataWriterImpl::begin_coherent_changes ( )

Starts a coherent change set; should only be called once.

Definition at line 2324 of file DataWriterImpl.cpp.

References ACE_GUARD, coherent_, and get_lock().

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

◆ check_and_set_repo_id()

void OpenDDS::DCPS::DataWriterImpl::check_and_set_repo_id ( const GUID_t & id )

inlineprotected

Definition at line 520 of file DataWriterImpl.h.

References OpenDDS::DCPS::GUID_UNKNOWN, and lock_.

Referenced by add_association().

   {
     ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
     if (GUID_UNKNOWN == publication_id_) {
       publication_id_ = id;
     }
   }

◆ check_transport_qos()

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

virtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 2303 of file DataWriterImpl.cpp.

 {
   // DataWriter does not impose any constraints on which transports
   // may be used based on QoS.
   return true;
 }

◆ cleanup()

void OpenDDS::DCPS::DataWriterImpl::cleanup ( void )

cleanup the DataWriter.

Definition at line 126 of file DataWriterImpl.cpp.

References OpenDDS::DCPS::NO_STATUS_MASK, set_listener(), topic_servant_, and type_support_.

Referenced by OpenDDS::DCPS::PublisherImpl::delete_datawriter().

 {
   // As first step set our listener to nill which will prevent us from calling
   // back onto the listener at the moment the related DDS entity has been
   // deleted
   set_listener(0, NO_STATUS_MASK);
   topic_servant_ = 0;
   type_support_ = 0;
 }

◆ coherent_changes_pending()

bool OpenDDS::DCPS::DataWriterImpl::coherent_changes_pending ( )

Are coherent changes pending?

Definition at line 2313 of file DataWriterImpl.cpp.

References ACE_GUARD_RETURN, coherent_, and get_lock().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    get_lock(),
                    false);
 
   return this->coherent_;
 }

◆ control_delivered()

void OpenDDS::DCPS::DataWriterImpl::control_delivered ( const Message_Block_Ptr & sample )

virtual

This is called by transport to notify that the control message is delivered.

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 2257 of file DataWriterImpl.cpp.

References controlTracker, DBG_ENTRY_LVL, and OpenDDS::DCPS::MessageTracker::message_delivered().

 {
   DBG_ENTRY_LVL("DataWriterImpl","control_delivered",6);
   controlTracker.message_delivered();
 }

◆ control_dropped()

void OpenDDS::DCPS::DataWriterImpl::control_dropped	(	const Message_Block_Ptr &	sample,
		bool	dropped_by_transport
	)

virtual

This is called by transport to notify that the control message is dropped.

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 2403 of file DataWriterImpl.cpp.

References controlTracker, DBG_ENTRY_LVL, and OpenDDS::DCPS::MessageTracker::message_dropped().

 {
   DBG_ENTRY_LVL("DataWriterImpl","control_dropped",6);
   controlTracker.message_dropped();
 }

◆ create_ack_token()

DataWriterImpl::AckToken OpenDDS::DCPS::DataWriterImpl::create_ack_token ( DDS::Duration_t max_wait ) const

Create an AckToken for ack operations.

Definition at line 1025 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, get_max_sn(), OpenDDS::DCPS::SequenceNumber::getValue(), and LM_DEBUG.

Referenced by wait_for_acknowledgments().

 {
   const SequenceNumber sn = get_max_sn();
   if (DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) DataWriterImpl::create_ack_token() - ")
                ACE_TEXT("for sequence %q\n"),
                sn.getValue()));
   }
   return AckToken(max_wait, sn);
 }

◆ create_control_message()

ACE_Message_Block * OpenDDS::DCPS::DataWriterImpl::create_control_message	(	MessageId	message_id,
		DataSampleHeader &	header,
		Message_Block_Ptr	data,
		const DDS::Time_t &	source_timestamp
	)

private

This method create a header message block and chain with the registered sample. The header contains the information needed. e.g. message id, length of whole message... The fast allocator is not used for the header.

Definition at line 2070 of file DataWriterImpl.cpp.

Referenced by association_complete_i(), dispose(), dispose_and_unregister(), end_coherent_changes(), register_instance_i(), replay_durable_data_for(), send_liveliness(), send_request_ack(), and unregister_instance_i().

 {
   header_data.message_id_ = message_id;
   header_data.byte_order_ =
     this->swap_bytes() ? !ACE_CDR_BYTE_ORDER : ACE_CDR_BYTE_ORDER;
   header_data.coherent_change_ = false;
 
   if (data) {
     header_data.message_length_ = static_cast<ACE_UINT32>(data->total_length());
   }
 
   header_data.sequence_ = SequenceNumber::SEQUENCENUMBER_UNKNOWN();
   header_data.sequence_repair_ = false; // set below
   header_data.source_timestamp_sec_ = source_timestamp.sec;
   header_data.source_timestamp_nanosec_ = source_timestamp.nanosec;
   header_data.publication_id_ = publication_id_;
 
   RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();
   if (!publisher) {
     return 0;
   }
 
   header_data.publisher_id_ = publisher->publisher_id_;
 
   ACE_Guard<ACE_Thread_Mutex> guard(sn_lock_);
   SequenceNumber sequence = sequence_number_;
   if (message_id == INSTANCE_REGISTRATION
       || message_id == DISPOSE_INSTANCE
       || message_id == UNREGISTER_INSTANCE
       || message_id == DISPOSE_UNREGISTER_INSTANCE
       || message_id == REQUEST_ACK) {
 
     header_data.sequence_repair_ = need_sequence_repair();
     header_data.sequence_ = get_next_sn_i();
     header_data.key_fields_only_ = true;
     sequence = sequence_number_;
   }
   guard.release();
 
   ACE_Message_Block* message = 0;
   ACE_NEW_MALLOC_RETURN(message,
                         static_cast<ACE_Message_Block*>(
                           mb_allocator_->malloc(sizeof(ACE_Message_Block))),
                         ACE_Message_Block(
                           DataSampleHeader::get_max_serialized_size(),
                           ACE_Message_Block::MB_DATA,
                           header_data.message_length_ ? data.release() : 0, //cont
                           0, //data
                           0, //allocator_strategy
                           get_db_lock(), //locking_strategy
                           ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                           ACE_Time_Value::zero,
                           ACE_Time_Value::max_time,
                           db_allocator_.get(),
                           mb_allocator_.get()),
                         0);
 
   *message << header_data;
 
   // If we incremented sequence number for this control message
   if (header_data.sequence_ != SequenceNumber::SEQUENCENUMBER_UNKNOWN()) {
     ACE_GUARD_RETURN(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_, 0);
     // Update the expected sequence number for all readers
     RepoIdToReaderInfoMap::iterator reader;
 
     for (reader = reader_info_.begin(); reader != reader_info_.end(); ++reader) {
       reader->second.expected_sequence_ = sequence;
     }
   }
   if (DCPS_debug_level >= 4) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) DataWriterImpl::create_control_message: ")
                ACE_TEXT("from publication %C sending control sample: %C .\n"),
                LogGuid(publication_id_).c_str(),
                to_string(header_data).c_str()));
   }
   return message;
 }

◆ create_sample_data_message()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::create_sample_data_message	(	Message_Block_Ptr	data,
		DDS::InstanceHandle_t	instance_handle,
		DataSampleHeader &	header_data,
		Message_Block_Ptr &	message,
		const DDS::Time_t &	source_timestamp,
		bool	content_filter
	)

This method create a header message block and chain with the sample data. The header contains the information needed. e.g. message id, length of whole message... The fast allocator is used to allocate the message block, data block and header.

Definition at line 2153 of file DataWriterImpl.cpp.

Referenced by write().

 {
   PublicationInstance_rch instance =
     data_container_->get_handle_instance(instance_handle);
 
   if (0 == instance) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) DataWriterImpl::create_sample_data_message ")
                       ACE_TEXT("failed to find instance for handle %d\n"),
                       instance_handle),
                      DDS::RETCODE_ERROR);
   }
 
   header_data.message_id_ = SAMPLE_DATA;
   header_data.byte_order_ =
     this->swap_bytes() ? !ACE_CDR_BYTE_ORDER : ACE_CDR_BYTE_ORDER;
   header_data.coherent_change_ = this->coherent_;
 
   RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();
 
   if (!publisher) {
     return DDS::RETCODE_ERROR;
   }
 
 #ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
   header_data.group_coherent_ =
     publisher->qos_.presentation.access_scope
     == DDS::GROUP_PRESENTATION_QOS;
 #endif
   header_data.content_filter_ = content_filter;
   header_data.cdr_encapsulation_ = this->cdr_encapsulation();
   header_data.message_length_ = static_cast<ACE_UINT32>(data->total_length());
   {
     ACE_Guard<ACE_Thread_Mutex> guard(sn_lock_);
     header_data.sequence_repair_ = need_sequence_repair();
     header_data.sequence_ = get_next_sn_i();
   }
   header_data.source_timestamp_sec_ = source_timestamp.sec;
   header_data.source_timestamp_nanosec_ = source_timestamp.nanosec;
 
   if (qos_.lifespan.duration.sec != DDS::DURATION_INFINITE_SEC
       || qos_.lifespan.duration.nanosec != DDS::DURATION_INFINITE_NSEC) {
     header_data.lifespan_duration_ = true;
     header_data.lifespan_duration_sec_ = qos_.lifespan.duration.sec;
     header_data.lifespan_duration_nanosec_ = qos_.lifespan.duration.nanosec;
   }
 
   header_data.publication_id_ = publication_id_;
   header_data.publisher_id_ = publisher->publisher_id_;
 
   ACE_Message_Block* tmp_message;
   ACE_NEW_MALLOC_RETURN(tmp_message,
                         static_cast<ACE_Message_Block*>(
                           mb_allocator_->malloc(sizeof(ACE_Message_Block))),
                         ACE_Message_Block(DataSampleHeader::get_max_serialized_size(),
                                           ACE_Message_Block::MB_DATA,
                                           data.release(), //cont
                                           0, //data
                                           header_allocator_.get(), //alloc_strategy
                                           get_db_lock(), //locking_strategy
                                           ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                                           ACE_Time_Value::zero,
                                           ACE_Time_Value::max_time,
                                           db_allocator_.get(),
                                           mb_allocator_.get()),
                         DDS::RETCODE_ERROR);
   message.reset(tmp_message);
   *message << header_data;
   if (DCPS_debug_level >= 4) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) DataWriterImpl::create_sample_data_message: ")
                ACE_TEXT("from publication %C sending data sample: %C .\n"),
                LogGuid(publication_id_).c_str(),
                to_string(header_data).c_str()));
   }
   return DDS::RETCODE_OK;
 }

◆ data_allocator()

DataAllocator* OpenDDS::DCPS::DataWriterImpl::data_allocator ( ) const

inline

Definition at line 134 of file DataWriterImpl.h.

References dispose(), init(), OpenDDS::DCPS::OPENDDS_VECTOR(), and write().

   {
     return data_allocator_.get();
   }

◆ data_delivered()

void OpenDDS::DCPS::DataWriterImpl::data_delivered ( const DataSampleElement * sample )

virtual

This is called by transport to notify that the sample is delivered and it is delegated to WriteDataContainer to adjust the internal data sample threads.

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 2237 of file DataWriterImpl.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), data_container_, data_delivered_count_, DBG_ENTRY_LVL, OpenDDS::DCPS::DataSampleElement::get_pub_id(), LM_ERROR, and publication_id_.

 {
   DBG_ENTRY_LVL("DataWriterImpl","data_delivered",6);
 
   if (!(sample->get_pub_id() == this->publication_id_)) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::data_delivered: ")
                ACE_TEXT("The publication id %C from delivered element ")
                ACE_TEXT("does not match the datawriter's id %C\n"),
                LogGuid(sample->get_pub_id()).c_str(),
                LogGuid(publication_id_).c_str()));
     return;
   }
   //provided for statistics tracking in tests
   ++data_delivered_count_;
 
   this->data_container_->data_delivered(sample);
 }

◆ data_dropped()

void OpenDDS::DCPS::DataWriterImpl::data_dropped	(	const DataSampleElement *	element,
		bool	dropped_by_transport
	)

virtual

This mothod is called by transport to notify the instance sample is dropped and it delegates to WriteDataContainer to update the internal list.

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 2391 of file DataWriterImpl.cpp.

References data_container_, data_dropped_count_, and DBG_ENTRY_LVL.

 {
   DBG_ENTRY_LVL("DataWriterImpl","data_dropped",6);
 
   //provided for statistics tracking in tests
   ++data_dropped_count_;
 
   this->data_container_->data_dropped(element, dropped_by_transport);
 }

◆ dispose()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::dispose	(	DDS::InstanceHandle_t	handle,
		const DDS::Time_t &	source_timestamp
	)

Delegate to the WriteDataContainer to dispose all data samples for a given instance and tell the transport to broadcast the disposed instance.

Definition at line 1983 of file DataWriterImpl.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), create_control_message(), data_container_, DBG_ENTRY_LVL, OpenDDS::DCPS::DISPOSE_INSTANCE, OpenDDS::DCPS::EntityImpl::enabled_, OpenDDS::DCPS::DataSampleElement::get_header(), get_lock(), LM_ERROR, OpenDDS::DCPS::move(), DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, send_all_to_flush_control(), and OpenDDS::DCPS::DataSampleElement::set_sample().

Referenced by dispose_w_timestamp().

 {
   DBG_ENTRY_LVL("DataWriterImpl","dispose",6);
 
   if (!enabled_) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::dispose: ")
                       ACE_TEXT("Entity is not enabled.\n")),
                      DDS::RETCODE_NOT_ENABLED);
   }
 
   DDS::ReturnCode_t ret = DDS::RETCODE_ERROR;
 
   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, guard, get_lock(), ret);
 
   Message_Block_Ptr registered_sample_data;
   ret = this->data_container_->dispose(handle, registered_sample_data);
 
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::dispose: ")
                       ACE_TEXT("dispose failed.\n")),
                      ret);
   }
 
   DataSampleElement* element = 0;
   ret = this->data_container_->obtain_buffer_for_control(element);
 
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::dispose: ")
                       ACE_TEXT("obtain_buffer_for_control returned %d.\n"),
                       ret),
                      ret);
   }
 
   Message_Block_Ptr sample(create_control_message(DISPOSE_INSTANCE,
                                                   element->get_header(),
                                                   move(registered_sample_data),
                                                   source_timestamp));
   element->set_sample(move(sample));
 
   ret = this->data_container_->enqueue_control(element);
 
   if (ret != DDS::RETCODE_OK) {
     data_container_->release_buffer(element);
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::dispose: ")
                       ACE_TEXT("enqueue_control failed.\n")),
                      ret);
   }
 
   send_all_to_flush_control(guard);
 
   return DDS::RETCODE_OK;
 }

◆ dispose_and_unregister()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::dispose_and_unregister	(	DDS::InstanceHandle_t	handle,
		const DDS::Time_t &	timestamp
	)

private

Definition at line 1750 of file DataWriterImpl.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), create_control_message(), data_container_, DBG_ENTRY_LVL, OpenDDS::DCPS::DISPOSE_UNREGISTER_INSTANCE, OpenDDS::DCPS::DataSampleElement::get_header(), get_lock(), LM_ERROR, OpenDDS::DCPS::move(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, send_all_to_flush_control(), and OpenDDS::DCPS::DataSampleElement::set_sample().

Referenced by unregister_instance_i().

 {
   DBG_ENTRY_LVL("DataWriterImpl", "dispose_and_unregister", 6);
 
   DDS::ReturnCode_t ret = DDS::RETCODE_ERROR;
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, get_lock(), ret);
 
   Message_Block_Ptr data_sample;
   ret = this->data_container_->dispose(handle, data_sample);
 
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::dispose_and_unregister: ")
                       ACE_TEXT("dispose on container failed.\n")),
                      ret);
   }
 
   ret = this->data_container_->unregister(handle, data_sample, false);
 
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::dispose_and_unregister: ")
                       ACE_TEXT("unregister with container failed.\n")),
                      ret);
   }
 
   DataSampleElement* element = 0;
   ret = this->data_container_->obtain_buffer_for_control(element);
 
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::dispose_and_unregister: ")
                       ACE_TEXT("obtain_buffer_for_control returned %d.\n"),
                       ret),
                      ret);
   }
 
   Message_Block_Ptr sample(create_control_message(DISPOSE_UNREGISTER_INSTANCE,
                                                   element->get_header(),
                                                   move(data_sample),
                                                   source_timestamp));
   element->set_sample(move(sample));
 
   ret = this->data_container_->enqueue_control(element);
 
   if (ret != DDS::RETCODE_OK) {
     data_container_->release_buffer(element);
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::dispose_and_unregister: ")
                       ACE_TEXT("enqueue_control failed.\n")),
                      ret);
   }
 
   send_all_to_flush_control(guard);
   return DDS::RETCODE_OK;
 }

◆ dispose_w_timestamp()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::dispose_w_timestamp	(	const Sample &	sample,
		DDS::InstanceHandle_t	instance_handle,
		const DDS::Time_t &	source_timestamp
	)

Definition at line 2942 of file DataWriterImpl.cpp.

References ACE_ERROR, DDS::Security::SecurityException::code, dispose(), dynamic_type_, OpenDDS::DCPS::Sample::get_dynamic_data(), DDS::HANDLE_NIL, instance_must_exist(), LM_NOTICE, OpenDDS::DCPS::log_level, DDS::Security::SecurityException::message, DDS::Security::SecurityException::minor_code, OpenDDS::DCPS::LogLevel::Notice, participant_permissions_handle_, DDS::Security::RETCODE_NOT_ALLOWED_BY_SECURITY, DDS::RETCODE_OK, and security_config_.

Referenced by OpenDDS::XTypes::DynamicDataWriterImpl::dispose_w_timestamp(), and OpenDDS::DCPS::DataWriterImpl_T< MessageType >::dispose_w_timestamp().

 {
 #if defined(OPENDDS_SECURITY) && OPENDDS_HAS_DYNAMIC_DATA_ADAPTER
   DDS::DynamicData_var dynamic_data = sample.get_dynamic_data(dynamic_type_);
   DDS::Security::SecurityException ex;
   if (dynamic_data && security_config_ &&
       participant_permissions_handle_ != DDS::HANDLE_NIL &&
       !security_config_->get_access_control()->check_local_datawriter_dispose_instance(participant_permissions_handle_, this, dynamic_data, ex)) {
     if (log_level >= LogLevel::Notice) {
       ACE_ERROR((LM_NOTICE,
                  "(%P|%t) NOTICE: DataWriterImpl::dispose_w_timestamp: unable to dispose instance SecurityException[%d.%d]: %C\n",
                  ex.code, ex.minor_code, ex.message.in()));
     }
     return DDS::Security::RETCODE_NOT_ALLOWED_BY_SECURITY;
   }
 #endif
 
   const DDS::ReturnCode_t rc = instance_must_exist(
     "dispose_w_timestamp", sample, instance_handle);
   if (rc != DDS::RETCODE_OK) {
     return rc;
   }
   return dispose(instance_handle, source_timestamp);
 }

◆ domain_id()

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

inlineprivatevirtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 716 of file DataWriterImpl.h.

Referenced by transport_discovery_change().

   {
     return this->domain_id_;
   }

◆ enable()

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

virtual

Implements DDS::Entity.

Definition at line 1314 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::TypeSupportImpl::add_types(), association_chunk_multiplier_, OpenDDS::DCPS::TransportClient::cdr_encapsulation(), OpenDDS::DCPS::TransportClient::connection_info(), data_container_, db_allocator_, OpenDDS::DCPS::DCPS_debug_level, DDS::DataWriterQos::deadline, DDS::HistoryQosPolicy::depth, domain_id_, dp_id_, DDS::DataWriterQos::durability, DDS::DataWriterQos::durability_service, DDS::DURATION_INFINITE_NSEC, DDS::DURATION_INFINITE_SEC, dynamic_type_, OpenDDS::DCPS::Observer::e_ENABLED, OpenDDS::DCPS::TransportClient::enable_transport(), OpenDDS::DCPS::EntityImpl::get_observer(), OpenDDS::DCPS::TypeSupportImpl::get_type(), get_type_name(), OpenDDS::DCPS::GUID_UNKNOWN, header_allocator_, DDS::DataWriterQos::history, if(), TAO::String_var< charT >::in(), OpenDDS::DCPS::EntityImpl::is_enabled(), DDS::KEEP_ALL_HISTORY_QOS, DDS::DurabilityQosPolicy::kind, DDS::ReliabilityQosPolicy::kind, DDS::HistoryQosPolicy::kind, last_deadline_missed_total_count_, DDS::LivelinessQosPolicy::lease_duration, DDS::LENGTH_UNLIMITED, DDS::DataWriterQos::lifespan, DDS::DataWriterQos::liveliness, liveliness_check_interval_, liveness_timer_, LM_DEBUG, LM_ERROR, LM_WARNING, lock_, DDS::ReliabilityQosPolicy::max_blocking_time, DDS::ResourceLimitsQosPolicy::max_instances, DDS::ResourceLimitsQosPolicy::max_samples, DDS::ResourceLimitsQosPolicy::max_samples_per_instance, mb_allocator_, monitor_, n_chunks_, DDS::Duration_t::nanosec, offered_deadline_missed_status_, OpenDDS::DCPS::Observer::on_enabled(), participant_permissions_handle_, participant_servant_, DDS::DeadlineQosPolicy::period, publication_id_, publisher_servant_, qos_, OpenDDS::DCPS::rchandle_from(), reactor_, ACE_Guard< ACE_LOCK >::release(), DDS::DataWriterQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, DDS::DataWriterQos::representation, DDS::DataWriterQos::resource_limits, DDS::RETCODE_ERROR, DDS::RETCODE_OK, DDS::RETCODE_PRECONDITION_NOT_MET, ACE_Reactor_Timer_Interface::schedule_timer(), DDS::Duration_t::sec, security_config_, OpenDDS::DCPS::EntityImpl::set_enabled(), OpenDDS::DCPS::set_writer_effective_data_rep_qos(), setup_serialization(), TheServiceParticipant, OpenDDS::DCPS::TypeSupportImpl::to_type_info(), topic_name_, topic_servant_, type_support_, OpenDDS::DCPS::TimeDuration::value(), DDS::DataRepresentationQosPolicy::value, DDS::VOLATILE_DURABILITY_QOS, and WriteDataContainer.

Referenced by OpenDDS::DCPS::PublisherImpl::create_datawriter().

 {
   //According spec:
   // - Calling enable on an already enabled Entity returns OK and has no
   // effect.
   // - Calling enable on an Entity whose factory is not enabled will fail
   // and return PRECONDITION_NOT_MET.
 
   if (this->is_enabled()) {
     return DDS::RETCODE_OK;
   }
 
   RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();
   if (!publisher || !publisher->is_enabled()) {
     return DDS::RETCODE_PRECONDITION_NOT_MET;
   }
 
   if (!topic_servant_->is_enabled()) {
     return DDS::RETCODE_PRECONDITION_NOT_MET;
   }
 
   RcHandle<DomainParticipantImpl> participant = participant_servant_.lock();
   if (participant) {
     dp_id_ = participant->get_id();
   }
 
   // Note: do configuration based on QoS in enable() because
   //       before enable is called the QoS can be changed -- even
   //       for Changeable=NO
 
   // Configure WriteDataContainer constructor parameters from qos.
 
   const bool reliable = qos_.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS;
 
   CORBA::Long const max_samples_per_instance =
     (qos_.resource_limits.max_samples_per_instance == DDS::LENGTH_UNLIMITED)
     ? 0x7fffffff : qos_.resource_limits.max_samples_per_instance;
 
   CORBA::Long max_instances = 0, max_total_samples = 0;
 
   if (qos_.resource_limits.max_samples != DDS::LENGTH_UNLIMITED) {
     n_chunks_ = qos_.resource_limits.max_samples;
 
     if (qos_.resource_limits.max_instances == DDS::LENGTH_UNLIMITED ||
         (qos_.resource_limits.max_samples < qos_.resource_limits.max_instances)
         || (qos_.resource_limits.max_samples <
             (qos_.resource_limits.max_instances * max_samples_per_instance))) {
       max_total_samples = reliable ? qos_.resource_limits.max_samples : 0;
     }
   }
 
   if (reliable && qos_.resource_limits.max_instances != DDS::LENGTH_UNLIMITED)
     max_instances = qos_.resource_limits.max_instances;
 
   const CORBA::Long history_depth =
     (qos_.history.kind == DDS::KEEP_ALL_HISTORY_QOS ||
      qos_.history.depth == DDS::LENGTH_UNLIMITED) ? 0x7fffffff : qos_.history.depth;
 
   const CORBA::Long max_durable_per_instance =
     qos_.durability.kind == DDS::VOLATILE_DURABILITY_QOS ? 0 : history_depth;
 
 #ifndef OPENDDS_NO_PERSISTENCE_PROFILE
   // Get data durability cache if DataWriter QoS requires durable
   // samples.  Publisher servant retains ownership of the cache.
   DataDurabilityCache* const durability_cache =
     TheServiceParticipant->get_data_durability_cache(qos_.durability);
 #endif
 
   //Note: the QoS used to set n_chunks_ is Changeable=No so
   // it is OK that we cannot change the size of our allocators.
   data_container_ = RcHandle<WriteDataContainer>(
     new WriteDataContainer(
       this,
       max_samples_per_instance,
       history_depth,
       max_durable_per_instance,
       qos_.reliability.max_blocking_time,
       n_chunks_,
       domain_id_,
       topic_name_,
       get_type_name(),
 #ifndef OPENDDS_NO_PERSISTENCE_PROFILE
       durability_cache,
       qos_.durability_service,
 #endif
       max_instances,
       max_total_samples,
       lock_,
       offered_deadline_missed_status_,
       last_deadline_missed_total_count_),
      keep_count());
 
   // +1 because we might allocate one before releasing another
   // TBD - see if this +1 can be removed.
   mb_allocator_.reset(new MessageBlockAllocator(n_chunks_ * association_chunk_multiplier_));
   db_allocator_.reset(new DataBlockAllocator(n_chunks_+1));
   header_allocator_.reset(new DataSampleHeaderAllocator(n_chunks_+1));
 
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) DataWriterImpl::enable-mb"
                " Cached_Allocator_With_Overflow %x with %B chunks\n",
                mb_allocator_.get(),
                n_chunks_));
 
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) DataWriterImpl::enable-db"
                " Cached_Allocator_With_Overflow %x with %B chunks\n",
                db_allocator_.get(),
                n_chunks_));
 
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) DataWriterImpl::enable-header"
                " Cached_Allocator_With_Overflow %x with %B chunks\n",
                header_allocator_.get(),
                n_chunks_));
   }
 
   if (qos_.liveliness.lease_duration.sec != DDS::DURATION_INFINITE_SEC &&
       qos_.liveliness.lease_duration.nanosec != DDS::DURATION_INFINITE_NSEC) {
     // Must be at least 1 micro second.
     liveliness_check_interval_ = std::max(
       TimeDuration(qos_.liveliness.lease_duration) * (TheServiceParticipant->liveliness_factor() / 100.0),
       TimeDuration(0, 1));
 
     if (reactor_->schedule_timer(liveness_timer_.in(),
                                  0,
                                  liveliness_check_interval_.value(),
                                  liveliness_check_interval_.value()) == -1) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::enable: %p.\n"),
                  ACE_TEXT("schedule_timer")));
 
     }
   }
 
   if (!participant) {
     return DDS::RETCODE_ERROR;
   }
 
   participant->add_adjust_liveliness_timers(this);
 
   data_container_->set_deadline_period(TimeDuration(qos_.deadline.period));
 
   Discovery_rch disco = TheServiceParticipant->get_discovery(this->domain_id_);
   disco->pre_writer(this);
 
   this->set_enabled();
 
   try {
     this->enable_transport(reliable,
                            this->qos_.durability.kind > DDS::VOLATILE_DURABILITY_QOS);
 
   } catch (const Transport::Exception&) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::enable, ")
                ACE_TEXT("Transport Exception.\n")));
     data_container_->shutdown_ = true;
     return DDS::RETCODE_ERROR;
   }
 
   // Must be done after transport enabled.
   set_writer_effective_data_rep_qos(qos_.representation.value, cdr_encapsulation());
   if (!topic_servant_->check_data_representation(qos_.representation.value, true)) {
     data_container_->shutdown_ = true;
     return DDS::RETCODE_ERROR;
   }
 
   // Done after enable_transport so we know its swap_bytes.
   const DDS::ReturnCode_t setup_serialization_result = setup_serialization();
   if (setup_serialization_result != DDS::RETCODE_OK) {
     data_container_->shutdown_ = true;
     return setup_serialization_result;
   }
 
   const TransportLocatorSeq& trans_conf_info = connection_info();
   DDS::PublisherQos pub_qos;
   publisher->get_qos(pub_qos);
 
   XTypes::TypeInformation type_info;
   type_support_->to_type_info(type_info);
 
   XTypes::TypeLookupService_rch type_lookup_service = participant->get_type_lookup_service();
   type_support_->add_types(type_lookup_service);
 
   const GUID_t publication_id =
     disco->add_publication(this->domain_id_,
                            this->dp_id_,
                            this->topic_servant_->get_id(),
                            rchandle_from(this),
                            this->qos_,
                            trans_conf_info,
                            pub_qos,
                            type_info);
 
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
   publication_id_ = publication_id;
 
   if (publication_id_ == GUID_UNKNOWN) {
     if (DCPS_debug_level >= 1) {
       ACE_DEBUG((LM_WARNING, "(%P|%t) WARNING: DataWriterImpl::enable: "
         "add_publication failed\n"));
     }
     data_container_->shutdown_ = true;
     return DDS::RETCODE_ERROR;
   }
 
 #if defined(OPENDDS_SECURITY)
   security_config_ = participant->get_security_config();
   participant_permissions_handle_ = participant->permissions_handle();
   dynamic_type_ = type_support_->get_type();
 #endif
 
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) DataWriterImpl::enable: "
       "got GUID %C, publishing to topic name \"%C\" type \"%C\"\n",
       LogGuid(publication_id_).c_str(),
       topic_servant_->topic_name(), topic_servant_->type_name()));
   }
 
   this->data_container_->publication_id_ = this->publication_id_;
 
   guard.release();
 
   const DDS::ReturnCode_t writer_enabled_result =
     publisher->writer_enabled(topic_name_.in(), this);
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 
 #ifndef OPENDDS_NO_PERSISTENCE_PROFILE
 
   // Move cached data from the durability cache to the unsent data
   // queue.
   if (durability_cache != 0) {
 
     if (!durability_cache->get_data(this->domain_id_,
                                     this->topic_name_,
                                     get_type_name(),
                                     this,
                                     this->mb_allocator_.get(),
                                     this->db_allocator_.get(),
                                     this->qos_.lifespan)) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::enable: ")
                  ACE_TEXT("unable to retrieve durable data\n")));
     }
   }
 
 #endif
 
   if (writer_enabled_result == DDS::RETCODE_OK) {
     const Observer_rch observer = get_observer(Observer::e_ENABLED);
     if (observer) {
       observer->on_enabled(this);
     }
   }
 
   return writer_enabled_result;
 }

◆ end_coherent_changes()

void OpenDDS::DCPS::DataWriterImpl::end_coherent_changes ( const GroupCoherentSamples & group_samples )

Ends a coherent change set; should only be called once.

Definition at line 2334 of file DataWriterImpl.cpp.

References ACE_ERROR, ACE_GUARD, ACE_TEXT(), coherent_, OpenDDS::DCPS::CoherentChangeControl::coherent_samples_, coherent_samples_, create_control_message(), OpenDDS::DCPS::END_COHERENT_CHANGES, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), get_db_lock(), get_lock(), OpenDDS::DCPS::CoherentChangeControl::get_max_serialized_size(), get_max_sn(), OpenDDS::DCPS::CoherentChangeControl::group_coherent_, OpenDDS::DCPS::CoherentChangeControl::group_coherent_samples_, DDS::GROUP_PRESENTATION_QOS, header, OpenDDS::DCPS::Encoding::KIND_UNALIGNED_CDR, OpenDDS::DCPS::WriterCoherentSample::last_sample_, LM_ERROR, ACE_Message_Block::MB_DATA, OpenDDS::DCPS::move(), OpenDDS::DCPS::TimePoint_T< SystemClock >::now(), OpenDDS::DCPS::WriterCoherentSample::num_samples_, OpenDDS::DCPS::CoherentChangeControl::publisher_id_, publisher_servant_, send_control(), OpenDDS::DCPS::SEND_CONTROL_ERROR, and OpenDDS::DCPS::TransportClient::swap_bytes().

 {
   // PublisherImpl::pi_lock_ should be held.
   ACE_GUARD(ACE_Recursive_Thread_Mutex,
             guard,
             get_lock());
 
   CoherentChangeControl end_msg;
   end_msg.coherent_samples_.num_samples_ = this->coherent_samples_;
   end_msg.coherent_samples_.last_sample_ = get_max_sn();
 
   RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();
 
   if (publisher) {
     end_msg.group_coherent_
       = publisher->qos_.presentation.access_scope == DDS::GROUP_PRESENTATION_QOS;
   }
 
   if (publisher && end_msg.group_coherent_) {
     end_msg.publisher_id_ = publisher->publisher_id_;
     end_msg.group_coherent_samples_ = group_samples;
   }
 
   Message_Block_Ptr data(
     new ACE_Message_Block(
       end_msg.get_max_serialized_size(),
       ACE_Message_Block::MB_DATA,
       0, // cont
       0, // data
       0, // alloc_strategy
       get_db_lock()));
 
   Serializer serializer(data.get(), Encoding::KIND_UNALIGNED_CDR,
     this->swap_bytes());
 
   serializer << end_msg;
 
   DataSampleHeader header;
   Message_Block_Ptr control(
     create_control_message(
       END_COHERENT_CHANGES, header, move(data),
       SystemTimePoint::now().to_dds_time()));
 
   this->coherent_ = false;
   this->coherent_samples_ = 0;
 
   guard.release();
   if (this->send_control(header, move(control)) == SEND_CONTROL_ERROR) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::end_coherent_changes:")
                ACE_TEXT(" unable to send END_COHERENT_CHANGES control message!\n")));
   }
 }

◆ filter_out()

bool OpenDDS::DCPS::DataWriterImpl::filter_out	(	const DataSampleElement &	elt,
		const OPENDDS_STRING &	filterClassName,
		const FilterEvaluator &	evaluator,
		const DDS::StringSeq &	expression_params
	)		const

Definition at line 2271 of file DataWriterImpl.cpp.

References ACE_ERROR, ACE_Message_Block::cont(), OpenDDS::DCPS::DataWriterImpl::EncodingMode::encoding(), encoding_mode_, OpenDDS::DCPS::LogLevel::Error, OpenDDS::DCPS::FilterEvaluator::eval(), OpenDDS::DCPS::DataSampleElement::get_header(), OpenDDS::DCPS::DataSampleElement::get_sample(), OpenDDS::DCPS::FilterEvaluator::has_non_key_fields(), LM_ERROR, OpenDDS::DCPS::log_level, type_support_, and OpenDDS::DCPS::DataSampleHeader::valid_data().

Referenced by OpenDDS::DCPS::WriteDataContainer::copy_and_prepend(), and write_w_timestamp().

 {
   if (!type_support_) {
     if (log_level >= LogLevel::Error) {
       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: DataWriterImpl::filter_out: Could not cast type support, not filtering\n"));
     }
     return false;
   }
 
   if (filterClassName == "DDSSQL" ||
       filterClassName == "OPENDDSSQL") {
     if (!elt.get_header().valid_data() && evaluator.has_non_key_fields(*type_support_)) {
       return true;
     }
     try {
       return !evaluator.eval(elt.get_sample()->cont(), encoding_mode_.encoding(),
                              *type_support_, expression_params);
     } catch (const std::runtime_error&) {
       // if the eval fails, the throws will do the logging
       // return false here so that the sample is not filtered
       return false;
     }
   } else {
     return false;
   }
 }

◆ find_instance()

DataWriterImpl::InstanceValuesToHandles::iterator OpenDDS::DCPS::DataWriterImpl::find_instance ( const Sample & sample )

private

Definition at line 3068 of file DataWriterImpl.cpp.

References OpenDDS::DCPS::RcHandle< T >::_retn(), and instance_values_to_handles_.

Referenced by instance_must_exist(), and lookup_instance().

 {
   Sample_rch dummy_rch(const_cast<Sample*>(&sample), keep_count());
   InstanceValuesToHandles::iterator pos = instance_values_to_handles_.find(dummy_rch);
   dummy_rch._retn();
   return pos;
 }

◆ get_builtin_subscriber_proxy()

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

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 201 of file DataWriterImpl.cpp.

References participant_servant_.

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

◆ get_crypto_handle()

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

privatevirtual

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 2739 of file DataWriterImpl.cpp.

References DDS::HANDLE_NIL, and participant_servant_.

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

◆ get_db_lock()

DataBlockLockPool::DataBlockLock* OpenDDS::DCPS::DataWriterImpl::get_db_lock ( )

inline

Definition at line 475 of file DataWriterImpl.h.

Referenced by association_complete_i(), create_control_message(), create_sample_data_message(), end_coherent_changes(), OpenDDS::DCPS::DataDurabilityCache::get_data(), replay_durable_data_for(), and serialize_sample().

   {
     return db_lock_pool_->get_lock();
   }

◆ get_dp_id()

GUID_t OpenDDS::DCPS::DataWriterImpl::get_dp_id ( )

Accessor of the repository id of the domain participant.

Definition at line 2058 of file DataWriterImpl.cpp.

References dp_id_.

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

 {
   return dp_id_;
 }

◆ get_ext_listener()

DataWriterListener_ptr OpenDDS::DCPS::DataWriterImpl::get_ext_listener ( )

protected

Definition at line 1003 of file DataWriterImpl.cpp.

References listener_, and listener_mutex_.

Referenced by notify_publication_disconnected(), notify_publication_lost(), and notify_publication_reconnected().

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

◆ get_guid()

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

inlinevirtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 488 of file DataWriterImpl.h.

References lock_.

Referenced by add_association(), OpenDDS::DCPS::PublisherImpl::delete_datawriter(), OpenDDS::Federator::ManagerImpl::initialize(), OpenDDS::DCPS::DWPeriodicMonitorImpl::report(), and OpenDDS::DCPS::DWMonitorImpl::report().

   {
     ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
     return publication_id_;
   }

◆ get_handle_instance()

PublicationInstance_rch OpenDDS::DCPS::DataWriterImpl::get_handle_instance ( DDS::InstanceHandle_t handle )

Attempt to locate an existing instance for the given handle.

Definition at line 2562 of file DataWriterImpl.cpp.

References data_container_.

Referenced by get_or_create_instance_handle().

 {
 
   if (0 != data_container_) {
     return data_container_->get_handle_instance(handle);
   }
 
   return PublicationInstance_rch();
 }

◆ get_ice_endpoint()

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

virtual

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 2782 of file DataWriterImpl.cpp.

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

 {
   return TransportClient::get_ice_endpoint();
 }

◆ get_instance_handle()

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

virtual

Implements OpenDDS::DCPS::EntityImpl.

Definition at line 176 of file DataWriterImpl.cpp.

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

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

◆ get_instance_handles()

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

Definition at line 2715 of file DataWriterImpl.cpp.

References data_container_.

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

 {
   this->data_container_->get_instance_handles(instance_handles);
 }

◆ get_key_value()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::get_key_value	(	Sample_rch &	sample,
		DDS::InstanceHandle_t	handle
	)

Definition at line 2898 of file DataWriterImpl.cpp.

References ACE_GUARD_RETURN, get_lock(), instance_handles_to_values_, OpenDDS::DCPS::Sample::Mutable, DDS::RETCODE_BAD_PARAMETER, DDS::RETCODE_ERROR, and DDS::RETCODE_OK.

Referenced by OpenDDS::XTypes::DynamicDataWriterImpl::get_key_value(), and OpenDDS::DCPS::DataWriterImpl_T< MessageType >::get_key_value().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, get_lock(), DDS::RETCODE_ERROR);
   const InstanceHandlesToValues::iterator it = instance_handles_to_values_.find(handle);
   if (it == instance_handles_to_values_.end()) {
     return DDS::RETCODE_BAD_PARAMETER;
   }
   sample = it->second->copy(Sample::Mutable);
   return DDS::RETCODE_OK;
 }

◆ get_listener()

DDS::DataWriterListener_ptr OpenDDS::DCPS::DataWriterImpl::get_listener ( )

virtual

Implements DDS::DataWriter.

Definition at line 996 of file DataWriterImpl.cpp.

References listener_, and listener_mutex_.

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

◆ get_liveliness_lost_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::get_liveliness_lost_status ( DDS::LivelinessLostStatus & status )

virtual

Definition at line 1121 of file DataWriterImpl.cpp.

References ACE_GUARD_RETURN, DDS::LIVELINESS_LOST_STATUS, liveliness_lost_status_, lock_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), and DDS::LivelinessLostStatus::total_count_change.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    this->lock_,
                    DDS::RETCODE_ERROR);
   set_status_changed_flag(DDS::LIVELINESS_LOST_STATUS, false);
   status = liveliness_lost_status_;
   liveliness_lost_status_.total_count_change = 0;
   return DDS::RETCODE_OK;
 }

◆ get_lock()

ACE_Recursive_Thread_Mutex& OpenDDS::DCPS::DataWriterImpl::get_lock ( ) const

inline

Accessor of the WriterDataContainer's lock.

Definition at line 382 of file DataWriterImpl.h.

References ACE_Recursive_Thread_Mutex::lock_.

Referenced by association_complete_i(), begin_coherent_changes(), coherent_changes_pending(), dispose(), dispose_and_unregister(), end_coherent_changes(), get_key_value(), get_or_create_instance_handle(), instance_must_exist(), lookup_instance(), register_instance_from_durable_data(), replay_durable_data_for(), send_request_ack(), unregister_instance_i(), and write().

   {
     return data_container_->lock_;
   }

◆ get_marshal_skip_serialize()

bool OpenDDS::DCPS::DataWriterImpl::get_marshal_skip_serialize ( ) const

inline

Definition at line 129 of file DataWriterImpl.h.

   {
     return skip_serialize_;
   }

◆ get_matched_subscription_data()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::get_matched_subscription_data	(	DDS::SubscriptionBuiltinTopicData &	subscription_data,
		DDS::InstanceHandle_t	subscription_handle
	)

virtual

Definition at line 1281 of file DataWriterImpl.cpp.

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

 {
   if (!enabled_) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::")
                       ACE_TEXT("get_matched_subscription_data: ")
                       ACE_TEXT("Entity is not enabled.\n")),
                      DDS::RETCODE_NOT_ENABLED);
   }
   RcHandle<DomainParticipantImpl> participant = this->participant_servant_.lock();
 
   DDS::ReturnCode_t ret = DDS::RETCODE_ERROR;
   DDS::SubscriptionBuiltinTopicDataSeq data;
 
   if (participant) {
     ret = instance_handle_to_bit_data<DDS::SubscriptionBuiltinTopicDataDataReader_var>(
             participant.in(),
             BUILT_IN_SUBSCRIPTION_TOPIC,
             subscription_handle,
             data);
   }
 
   if (ret == DDS::RETCODE_OK) {
     subscription_data = data[0];
   }
 
   return ret;
 }

◆ get_matched_subscriptions()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::get_matched_subscriptions ( DDS::InstanceHandleSeq & subscription_handles )

virtual

Definition at line 1248 of file DataWriterImpl.cpp.

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

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

◆ get_max_sn()

SequenceNumber OpenDDS::DCPS::DataWriterImpl::get_max_sn ( ) const

inlinevirtual

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 494 of file DataWriterImpl.h.

Referenced by association_complete_i(), create_ack_token(), end_coherent_changes(), and track_sequence_number().

   {
     ACE_Guard<ACE_Thread_Mutex> guard(sn_lock_);
     return sequence_number_;
   }

◆ get_next_handle()

DDS::InstanceHandle_t OpenDDS::DCPS::DataWriterImpl::get_next_handle ( )

Get an instance handle for a new instance.

Definition at line 183 of file DataWriterImpl.cpp.

References DDS::HANDLE_NIL, and participant_servant_.

Referenced by OpenDDS::DCPS::WriteDataContainer::register_instance().

 {
   RcHandle<DomainParticipantImpl> participant = this->participant_servant_.lock();
   if (participant) {
     return participant->assign_handle();
   }
   return DDS::HANDLE_NIL;
 }

◆ get_next_sn()

SequenceNumber OpenDDS::DCPS::DataWriterImpl::get_next_sn ( )

inlineprotected

Definition at line 528 of file DataWriterImpl.h.

   {
     ACE_Guard<ACE_Thread_Mutex> guard(sn_lock_);
     return get_next_sn_i();
   }

◆ get_next_sn_i()

SequenceNumber OpenDDS::DCPS::DataWriterImpl::get_next_sn_i ( )

inlineprotected

Definition at line 534 of file DataWriterImpl.h.

References OpenDDS::DCPS::SequenceNumber::SEQUENCENUMBER_UNKNOWN().

Referenced by create_control_message(), and create_sample_data_message().

   {
     if (sequence_number_ == SequenceNumber::SEQUENCENUMBER_UNKNOWN()) {
       sequence_number_ = SequenceNumber();
     } else {
       ++sequence_number_;
     }
     return sequence_number_;
   }

◆ get_offered_deadline_missed_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::get_offered_deadline_missed_status ( DDS::OfferedDeadlineMissedStatus & status )

virtual

Definition at line 1135 of file DataWriterImpl.cpp.

References ACE_GUARD_RETURN, last_deadline_missed_total_count_, lock_, DDS::OFFERED_DEADLINE_MISSED_STATUS, offered_deadline_missed_status_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), DDS::OfferedDeadlineMissedStatus::total_count, and DDS::OfferedDeadlineMissedStatus::total_count_change.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    this->lock_,
                    DDS::RETCODE_ERROR);
 
   set_status_changed_flag(DDS::OFFERED_DEADLINE_MISSED_STATUS, false);
 
   this->offered_deadline_missed_status_.total_count_change =
     this->offered_deadline_missed_status_.total_count
     - this->last_deadline_missed_total_count_;
 
   // Update for next status check.
   this->last_deadline_missed_total_count_ =
     this->offered_deadline_missed_status_.total_count;
 
   status = offered_deadline_missed_status_;
 
   this->offered_deadline_missed_status_.total_count_change = 0;
 
   return DDS::RETCODE_OK;
 }

◆ get_offered_incompatible_qos_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::get_offered_incompatible_qos_status ( DDS::OfferedIncompatibleQosStatus & status )

virtual

Definition at line 1161 of file DataWriterImpl.cpp.

References ACE_GUARD_RETURN, lock_, DDS::OFFERED_INCOMPATIBLE_QOS_STATUS, offered_incompatible_qos_status_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), and DDS::OfferedIncompatibleQosStatus::total_count_change.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    this->lock_,
                    DDS::RETCODE_ERROR);
   set_status_changed_flag(DDS::OFFERED_INCOMPATIBLE_QOS_STATUS, false);
   status = offered_incompatible_qos_status_;
   offered_incompatible_qos_status_.total_count_change = 0;
   return DDS::RETCODE_OK;
 }

◆ get_or_create_instance_handle()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::get_or_create_instance_handle	(	DDS::InstanceHandle_t &	handle,
		const Sample &	sample,
		const DDS::Time_t &	source_timestamp
	)

protected

Definition at line 3076 of file DataWriterImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, DDS::Security::SecurityException::code, OpenDDS::DCPS::Sample::copy(), OpenDDS::XTypes::copy(), dynamic_type_, get_handle_instance(), get_lock(), get_type_support(), DDS::HANDLE_NIL, insert_instance(), OpenDDS::DCPS::Sample::KeyOnly, LM_NOTICE, OpenDDS::DCPS::log_level, lookup_instance(), DDS::Security::SecurityException::message, DDS::Security::SecurityException::minor_code, OpenDDS::DCPS::move(), OpenDDS::DCPS::TypeSupportImpl::name(), OpenDDS::DCPS::LogLevel::Notice, participant_permissions_handle_, OpenDDS::DCPS::Sample::ReadOnly, register_instance_i(), DDS::RETCODE_ERROR, DDS::Security::RETCODE_NOT_ALLOWED_BY_SECURITY, DDS::RETCODE_OK, security_config_, send_all_to_flush_control(), and serialize_sample().

Referenced by register_instance_w_timestamp(), and write_w_timestamp().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, get_lock(), DDS::RETCODE_ERROR);
 
   handle = lookup_instance(sample);
   if (handle == DDS::HANDLE_NIL || !get_handle_instance(handle)) {
     Sample_rch copy = sample.copy(Sample::ReadOnly, Sample::KeyOnly);
 #if defined(OPENDDS_SECURITY) && OPENDDS_HAS_DYNAMIC_DATA_ADAPTER
     DDS::DynamicData_var dynamic_data = copy->get_dynamic_data(dynamic_type_);
     DDS::Security::SecurityException ex;
     if (dynamic_data && security_config_ &&
         participant_permissions_handle_ != DDS::HANDLE_NIL &&
         !security_config_->get_access_control()->check_local_datawriter_register_instance(participant_permissions_handle_, this, dynamic_data, ex)) {
       if (log_level >= LogLevel::Notice) {
         ACE_ERROR((LM_NOTICE,
                    "(%P|%t) NOTICE: DataWriterImpl::get_or_create_instance_handle: unable to register instance SecurityException[%d.%d]: %C\n",
                    ex.code, ex.minor_code, ex.message.in()));
       }
       return DDS::Security::RETCODE_NOT_ALLOWED_BY_SECURITY;
     }
 #endif
 
     // don't use fast allocator for registration.
     const TypeSupportImpl* const ts = get_type_support();
     Message_Block_Ptr serialized(serialize_sample(*copy));
     if (!serialized) {
       if (log_level >= LogLevel::Notice) {
         ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: %CDataWriterImpl::get_or_create_instance_handle: "
           "failed to serialize sample\n", ts->name()));
       }
       return DDS::RETCODE_ERROR;
     }
 
     // tell DataWriterLocal and Publisher about the instance.
     const DDS::ReturnCode_t ret = register_instance_i(handle, move(serialized), source_timestamp);
     // note: the WriteDataContainer/PublicationInstance maintains ownership
     // of the marshalled sample.
     if (ret != DDS::RETCODE_OK) {
       handle = DDS::HANDLE_NIL;
       return ret;
     }
 
     if (!insert_instance(handle, copy)) {
       handle = DDS::HANDLE_NIL;
       if (log_level >= LogLevel::Notice) {
         ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: %CDataWriterImpl::get_or_create_instance_handle: "
            "insert instance failed\n", ts->name()));
       }
       return DDS::RETCODE_ERROR;
     }
 
     send_all_to_flush_control(guard);
   }
 
   return DDS::RETCODE_OK;
 }

◆ get_priority_value()

CORBA::Long OpenDDS::DCPS::DataWriterImpl::get_priority_value ( const AssociationData & ) const

inlineprivatevirtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 721 of file DataWriterImpl.h.

References DDS::DataWriterQos::transport_priority, and DDS::TransportPriorityQosPolicy::value.

   {
     return this->qos_.transport_priority.value;
   }

◆ get_publication_matched_status()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::get_publication_matched_status ( DDS::PublicationMatchedStatus & status )

virtual

Definition at line 1175 of file DataWriterImpl.cpp.

References ACE_GUARD_RETURN, DDS::PublicationMatchedStatus::current_count_change, lock_, publication_match_status_, DDS::PUBLICATION_MATCHED_STATUS, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), and DDS::PublicationMatchedStatus::total_count_change.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    this->lock_,
                    DDS::RETCODE_ERROR);
   set_status_changed_flag(DDS::PUBLICATION_MATCHED_STATUS, false);
   status = publication_match_status_;
   publication_match_status_.total_count_change = 0;
   publication_match_status_.current_count_change = 0;
   return DDS::RETCODE_OK;
 }

◆ get_publisher()

DDS::Publisher_ptr OpenDDS::DCPS::DataWriterImpl::get_publisher ( )

virtual

Implements DDS::DataWriter.

Definition at line 1115 of file DataWriterImpl.cpp.

References publisher_servant_.

Referenced by OpenDDS::DCPS::PublisherImpl::delete_datawriter(), OpenDDS::DCPS::StaticDiscovery::pre_writer(), and OpenDDS::DCPS::DWMonitorImpl::report().

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

◆ get_qos()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::get_qos ( DDS::DataWriterQos & qos )

virtual

Definition at line 978 of file DataWriterImpl.cpp.

References passed_qos_, and DDS::RETCODE_OK.

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

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

◆ get_readers()

void OpenDDS::DCPS::DataWriterImpl::get_readers ( RepoIdSet & readers )

Definition at line 2721 of file DataWriterImpl.cpp.

References ACE_GUARD, lock_, and readers_.

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

 {
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
   readers = this->readers_;
 }

◆ get_resend_data()

SendStateDataSampleList OpenDDS::DCPS::DataWriterImpl::get_resend_data ( )

inline

Definition at line 306 of file DataWriterImpl.h.

Referenced by association_complete_i(), and replay_durable_data_for().

   {
     return data_container_->get_resend_data();
   }

◆ get_topic()

DDS::Topic_ptr OpenDDS::DCPS::DataWriterImpl::get_topic ( )

virtual

Implements DDS::DataWriter.

Definition at line 1010 of file DataWriterImpl.cpp.

References topic_servant_.

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

 {
   return DDS::Topic::_duplicate(topic_servant_.get());
 }

◆ get_type_name()

char const * OpenDDS::DCPS::DataWriterImpl::get_type_name ( ) const

Get associated topic type name.

Definition at line 2064 of file DataWriterImpl.cpp.

References TAO::String_var< charT >::in(), and type_name_.

Referenced by enable().

 {
   return type_name_.in();
 }

◆ get_type_support()

TypeSupportImpl* OpenDDS::DCPS::DataWriterImpl::get_type_support ( ) const

inlineprotected

Definition at line 665 of file DataWriterImpl.h.

References write_w_timestamp().

Referenced by get_or_create_instance_handle(), and write_w_timestamp().

   {
     return type_support_;
   }

◆ get_unsent_data()

ACE_UINT64 OpenDDS::DCPS::DataWriterImpl::get_unsent_data ( SendStateDataSampleList & list )

inline

Retrieve the unsent data from the WriteDataContainer.

Definition at line 301 of file DataWriterImpl.h.

Referenced by send_all_to_flush_control(), and write().

   {
     return data_container_->get_unsent_data(list);
   }

◆ get_value_dispatcher()

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

inline

Definition at line 500 of file DataWriterImpl.h.

References dispose_w_timestamp(), get_key_value(), lookup_instance(), register_instance_w_timestamp(), timestamp(), and unregister_instance_w_timestamp().

Referenced by write().

   {
     return dynamic_cast<const ValueDispatcher*>(type_support_);
   }

◆ handle_timeout()

int OpenDDS::DCPS::DataWriterImpl::handle_timeout	(	const ACE_Time_Value &	tv,
		const void *	arg
	)

virtual

Handle the assert liveliness timeout.

Definition at line 2431 of file DataWriterImpl.cpp.

References ACE_ERROR, ACE_TEXT(), DDS::AUTOMATIC_LIVELINESS_QOS, ACE_Reactor_Timer_Interface::cancel_timer(), CORBA::is_nil(), DDS::LivelinessQosPolicy::kind, last_liveliness_activity_time_, DDS::LivelinessQosPolicy::lease_duration, listener_for(), DDS::DataWriterQos::liveliness, liveliness_asserted_, liveliness_check_interval_, liveliness_lost_, DDS::LIVELINESS_LOST_STATUS, liveliness_lost_status_, liveness_timer_, LM_ERROR, lock_, DDS::MANUAL_BY_PARTICIPANT_LIVELINESS_QOS, DDS::MANUAL_BY_TOPIC_LIVELINESS_QOS, qos_, reactor_, ACE_Reactor_Timer_Interface::schedule_timer(), send_liveliness(), TheServiceParticipant, DDS::LivelinessLostStatus::total_count, DDS::LivelinessLostStatus::total_count_change, value, and OpenDDS::DCPS::TimeDuration::value().

 {
   ThreadStatusManager::Event ev(TheServiceParticipant->get_thread_status_manager());
 
   const MonotonicTimePoint now(tv);
   bool liveliness_lost = false;
 
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
 
   TimeDuration elapsed = now - last_liveliness_activity_time_;
 
   // Do we need to send a liveliness message?
   if (elapsed >= liveliness_check_interval_) {
     switch (this->qos_.liveliness.kind) {
     case DDS::AUTOMATIC_LIVELINESS_QOS:
       if (!send_liveliness(now)) {
         liveliness_lost = true;
       }
       break;
 
     case DDS::MANUAL_BY_PARTICIPANT_LIVELINESS_QOS:
       if (liveliness_asserted_) {
         if (!send_liveliness(now)) {
           liveliness_lost = true;
         }
       }
       break;
 
     case DDS::MANUAL_BY_TOPIC_LIVELINESS_QOS:
       // Do nothing.
       break;
     }
   }
   else {
     // Reschedule.
     if (reactor_->cancel_timer(liveness_timer_.in()) == -1) {
       ACE_ERROR((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::handle_timeout: %p.\n"),
         ACE_TEXT("cancel_timer")));
     }
     if (reactor_->schedule_timer(liveness_timer_.in(), 0,
       (liveliness_check_interval_ - elapsed).value(),
       liveliness_check_interval_.value()) == -1)
     {
       ACE_ERROR((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::handle_timeout: %p.\n"),
         ACE_TEXT("schedule_timer")));
     }
     return 0;
   }
 
   liveliness_asserted_ = false;
   elapsed = now - last_liveliness_activity_time_;
 
   // Have we lost liveliness?
   if (elapsed >= TimeDuration(qos_.liveliness.lease_duration)) {
     liveliness_lost = true;
   }
 
   if (!this->liveliness_lost_ && liveliness_lost) {
     ++ this->liveliness_lost_status_.total_count;
     ++ this->liveliness_lost_status_.total_count_change;
 
     DDS::DataWriterListener_var listener =
       listener_for(DDS::LIVELINESS_LOST_STATUS);
 
     if (!CORBA::is_nil(listener.in())) {
       {
         ACE_Reverse_Lock<ACE_Recursive_Thread_Mutex> rev_lock(lock_);
         ACE_Guard<ACE_Reverse_Lock<ACE_Recursive_Thread_Mutex> > rev_guard(rev_lock);
         listener->on_liveliness_lost(this, this->liveliness_lost_status_);
       }
       this->liveliness_lost_status_.total_count_change = 0;
     }
   }
 
   this->liveliness_lost_ = liveliness_lost;
   return 0;
 }

◆ init()

void OpenDDS::DCPS::DataWriterImpl::init	(	TopicImpl *	topic_servant,
		const DDS::DataWriterQos &	qos,
		DDS::DataWriterListener_ptr	a_listener,
		const DDS::StatusMask &	mask,
		WeakRcHandle< DomainParticipantImpl >	participant_servant,
		PublisherImpl *	publisher_servant
	)

Initialize the data members.

Definition at line 137 of file DataWriterImpl.cpp.

References DBG_ENTRY_LVL, domain_id_, OpenDDS::DCPS::TopicDescriptionImpl::get_type_support(), TAO::String_var< charT >::in(), is_bit_, OpenDDS::DCPS::WeakRcHandle< T >::lock(), participant_servant_, passed_qos_, publisher_servant_, qos_, reactor_, set_listener(), TheServiceParticipant, topic_id_, topic_name_, topic_servant_, OpenDDS::DCPS::topicIsBIT(), type_name_, and type_support_.

Referenced by OpenDDS::DCPS::PublisherImpl::create_datawriter().

 {
   DBG_ENTRY_LVL("DataWriterImpl", "init", 6);
   topic_servant_ = topic_servant;
   type_support_ = dynamic_cast<TypeSupportImpl*>(topic_servant->get_type_support());
   topic_name_ = topic_servant_->get_name();
   topic_id_ = topic_servant_->get_id();
   type_name_ = topic_servant_->get_type_name();
 
 #if !defined (DDS_HAS_MINIMUM_BIT)
   is_bit_ = topicIsBIT(topic_name_.in(), type_name_.in());
 #endif // !defined (DDS_HAS_MINIMUM_BIT)
 
   qos_ = qos;
   passed_qos_ = qos;
 
   set_listener(a_listener, mask);
 
   // Only store the participant pointer, since it is our "grand"
   // parent, we will exist as long as it does.
   participant_servant_ = participant_servant;
 
   RcHandle<DomainParticipantImpl> participant = participant_servant.lock();
   domain_id_ = participant->get_domain_id();
 
   // Only store the publisher pointer, since it is our parent, we will
   // exist as long as it does.
   publisher_servant_ = *publisher_servant;
 
   this->reactor_ = TheServiceParticipant->timer();
 }

◆ insert_instance()

bool OpenDDS::DCPS::DataWriterImpl::insert_instance	(	DDS::InstanceHandle_t	handle,
		Sample_rch &	sample
	)

private

Definition at line 3052 of file DataWriterImpl.cpp.

References instance_handles_to_values_, instance_values_to_handles_, and OPENDDS_ASSERT.

Referenced by get_or_create_instance_handle().

 {
   OPENDDS_ASSERT(sample->key_only());
   if (!instance_handles_to_values_.insert(
         InstanceHandlesToValues::value_type(handle, sample)).second) {
     return false;
   }
   if (!instance_values_to_handles_.insert(
         InstanceValuesToHandles::value_type(sample, handle)).second) {
     instance_handles_to_values_.erase(handle);
     return false;
   }
   return true;
 }

◆ instance_must_exist()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::instance_must_exist	(	const char *	method_name,
		const Sample &	sample,
		DDS::InstanceHandle_t &	instance_handle,
		bool	remove = `false`
	)

protected

Definition at line 3136 of file DataWriterImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, find_instance(), get_lock(), DDS::HANDLE_NIL, instance_handles_to_values_, instance_values_to_handles_, OpenDDS::DCPS::Sample::key_only(), LM_NOTICE, OpenDDS::DCPS::log_level, OpenDDS::DCPS::LogLevel::Notice, OPENDDS_ASSERT, DDS::RETCODE_ERROR, DDS::RETCODE_OK, and DDS::RETCODE_PRECONDITION_NOT_MET.

Referenced by dispose_w_timestamp(), and unregister_instance_w_timestamp().

 {
   OPENDDS_ASSERT(sample.key_only());
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, get_lock(), DDS::RETCODE_ERROR);
 
   const InstanceValuesToHandles::iterator pos = find_instance(sample);
   if (pos == instance_values_to_handles_.end()) {
     if (log_level >= LogLevel::Notice) {
       ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: DataWriterImpl::%C: "
         "The instance sample is not registered\n",
         method_name));
     }
     return DDS::RETCODE_ERROR;
   }
 
   if (instance_handle != DDS::HANDLE_NIL && instance_handle != pos->second) {
     return DDS::RETCODE_PRECONDITION_NOT_MET;
   }
 
   instance_handle = pos->second;
 
   if (remove) {
     instance_values_to_handles_.erase(pos);
     instance_handles_to_values_.erase(instance_handle);
   }
 
   return DDS::RETCODE_OK;
 }

◆ listener_for()

DDS::DataWriterListener_ptr OpenDDS::DCPS::DataWriterImpl::listener_for ( DDS::StatusKind kind )

This is used to retrieve the listener for a certain status change.

If this datawriter has a registered listener and the status kind is in the listener mask then the listener is returned. Otherwise, the query for the listener is propagated up to the factory/publisher.

Definition at line 2411 of file DataWriterImpl.cpp.

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

Referenced by association_complete_i(), handle_timeout(), OpenDDS::DCPS::WriteDataContainer::process_deadlines(), remove_associations(), and update_incompatible_qos().

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

◆ liveliness_check_interval()

TimeDuration OpenDDS::DCPS::DataWriterImpl::liveliness_check_interval ( DDS::LivelinessQosPolicyKind kind )

Definition at line 1227 of file DataWriterImpl.cpp.

References DDS::LivelinessQosPolicy::kind, DDS::DataWriterQos::liveliness, liveliness_check_interval_, OpenDDS::DCPS::TimeDuration::max_value, and qos_.

Referenced by OpenDDS::DCPS::DomainParticipantImpl::LivelinessTimer::add_adjust().

 {
   if (this->qos_.liveliness.kind == kind) {
     return liveliness_check_interval_;
   } else {
     return TimeDuration::max_value;
   }
 }

◆ lookup_instance()

DDS::InstanceHandle_t OpenDDS::DCPS::DataWriterImpl::lookup_instance ( const Sample & sample )

Definition at line 2909 of file DataWriterImpl.cpp.

References ACE_GUARD_RETURN, find_instance(), get_lock(), DDS::HANDLE_NIL, instance_values_to_handles_, and DDS::RETCODE_ERROR.

Referenced by get_or_create_instance_handle(), OpenDDS::XTypes::DynamicDataWriterImpl::lookup_instance(), and OpenDDS::DCPS::DataWriterImpl_T< MessageType >::lookup_instance().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, get_lock(), DDS::RETCODE_ERROR);
   const InstanceValuesToHandles::iterator it = find_instance(sample);
   return it == instance_values_to_handles_.end() ? DDS::HANDLE_NIL : it->second;
 }

◆ lookup_instance_handles()

void OpenDDS::DCPS::DataWriterImpl::lookup_instance_handles	(	const ReaderIdSeq &	ids,
		DDS::InstanceHandleSeq &	hdls
	)

private

Lookup the instance handles by the subscription repo ids.

Definition at line 2667 of file DataWriterImpl.cpp.

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

Referenced by notify_publication_disconnected(), notify_publication_lost(), notify_publication_reconnected(), and remove_associations().

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

◆ need_sequence_repair()

bool OpenDDS::DCPS::DataWriterImpl::need_sequence_repair ( )

private

Definition at line 2747 of file DataWriterImpl.cpp.

References ACE_GUARD_RETURN, need_sequence_repair_i(), and reader_info_lock_.

Referenced by create_control_message(), and create_sample_data_message().

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_, false);
   return need_sequence_repair_i();
 }

◆ need_sequence_repair_i()

bool OpenDDS::DCPS::DataWriterImpl::need_sequence_repair_i ( ) const

private

Definition at line 2754 of file DataWriterImpl.cpp.

References reader_info_, and sequence_number_.

Referenced by need_sequence_repair().

 {
   for (RepoIdToReaderInfoMap::const_iterator it = reader_info_.begin(),
        end = reader_info_.end(); it != end; ++it) {
     if (it->second.expected_sequence_ != sequence_number_) {
       return true;
     }
   }
 
   return false;
 }

◆ notify_publication_disconnected()

void OpenDDS::DCPS::DataWriterImpl::notify_publication_disconnected ( const ReaderIdSeq & subids )

virtual

Implements OpenDDS::DCPS::TransportSendListener.

Definition at line 2573 of file DataWriterImpl.cpp.

References DBG_ENTRY_LVL, get_ext_listener(), is_bit_, CORBA::is_nil(), lookup_instance_handles(), and OpenDDS::DCPS::PublicationLostStatus::subscription_handles.

 {
   DBG_ENTRY_LVL("DataWriterImpl","notify_publication_disconnected",6);
 
   if (!is_bit_) {
     // Narrow to DDS::DCPS::DataWriterListener. If a DDS::DataWriterListener
     // is given to this DataWriter then narrow() fails.
     DataWriterListener_var the_listener = get_ext_listener();
 
     if (!CORBA::is_nil(the_listener.in())) {
       PublicationDisconnectedStatus status;
       // Since this callback may come after remove_association which
       // removes the reader from id_to_handle map, we can ignore this
       // error.
       this->lookup_instance_handles(subids,
                                     status.subscription_handles);
       the_listener->on_publication_disconnected(this, status);
     }
   }
 }

◆ notify_publication_lost() [1/2]

void OpenDDS::DCPS::DataWriterImpl::notify_publication_lost ( const ReaderIdSeq & subids )

virtual

Implements OpenDDS::DCPS::TransportSendListener.

Definition at line 2617 of file DataWriterImpl.cpp.

References DBG_ENTRY_LVL, get_ext_listener(), is_bit_, CORBA::is_nil(), lookup_instance_handles(), and OpenDDS::DCPS::PublicationLostStatus::subscription_handles.

Referenced by remove_associations().

 {
   DBG_ENTRY_LVL("DataWriterImpl","notify_publication_lost",6);
 
   if (!is_bit_) {
     // Narrow to DDS::DCPS::DataWriterListener. If a
     // DDS::DataWriterListener is given to this DataWriter then
     // narrow() fails.
     DataWriterListener_var the_listener = get_ext_listener();
 
     if (!CORBA::is_nil(the_listener.in())) {
       PublicationLostStatus status;
 
       // Since this callback may come after remove_association which removes
       // the reader from id_to_handle map, we can ignore this error.
       this->lookup_instance_handles(subids,
                                     status.subscription_handles);
       the_listener->on_publication_lost(this, status);
     }
   }
 }

◆ notify_publication_lost() [2/2]

void OpenDDS::DCPS::DataWriterImpl::notify_publication_lost ( const DDS::InstanceHandleSeq & handles )

private

Definition at line 2640 of file DataWriterImpl.cpp.

References DBG_ENTRY_LVL, get_ext_listener(), is_bit_, CORBA::is_nil(), and OpenDDS::DCPS::PublicationLostStatus::subscription_handles.

 {
   DBG_ENTRY_LVL("DataWriterImpl","notify_publication_lost",6);
 
   if (!is_bit_) {
     // Narrow to DDS::DCPS::DataWriterListener. If a
     // DDS::DataWriterListener is given to this DataWriter then
     // narrow() fails.
     DataWriterListener_var the_listener = get_ext_listener();
 
     if (!CORBA::is_nil(the_listener.in())) {
       PublicationLostStatus status;
 
       CORBA::ULong len = handles.length();
       status.subscription_handles.length(len);
 
       for (CORBA::ULong i = 0; i < len; ++ i) {
         status.subscription_handles[i] = handles[i];
       }
 
       the_listener->on_publication_lost(this, status);
     }
   }
 }

◆ notify_publication_reconnected()

void OpenDDS::DCPS::DataWriterImpl::notify_publication_reconnected ( const ReaderIdSeq & subids )

virtual

Implements OpenDDS::DCPS::TransportSendListener.

Definition at line 2595 of file DataWriterImpl.cpp.

References DBG_ENTRY_LVL, get_ext_listener(), is_bit_, CORBA::is_nil(), lookup_instance_handles(), and OpenDDS::DCPS::PublicationLostStatus::subscription_handles.

 {
   DBG_ENTRY_LVL("DataWriterImpl","notify_publication_reconnected",6);
 
   if (!is_bit_) {
     // Narrow to DDS::DCPS::DataWriterListener. If a
     // DDS::DataWriterListener is given to this DataWriter then
     // narrow() fails.
     DataWriterListener_var the_listener = get_ext_listener();
 
     if (!CORBA::is_nil(the_listener.in())) {
       PublicationDisconnectedStatus status;
 
       // If it's reconnected then the reader should be in id_to_handle
       this->lookup_instance_handles(subids, status.subscription_handles);
 
       the_listener->on_publication_reconnected(this, status);
     }
   }
 }

◆ num_samples()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::num_samples	(	DDS::InstanceHandle_t	handle,
		size_t &	size
	)

Return the number of samples for a given instance.

Definition at line 2045 of file DataWriterImpl.cpp.

References data_container_.

 {
   return data_container_->num_samples(handle, size);
 }

◆ OPENDDS_MAP()

typedef OpenDDS::DCPS::DataWriterImpl::OPENDDS_MAP	(	DDS::InstanceHandle_t	,
		Sample_rch
	)

private

◆ OPENDDS_MAP_CMP() [1/4]

typedef OpenDDS::DCPS::DataWriterImpl::OPENDDS_MAP_CMP	(	GUID_t	,
		SequenceNumber	,
		GUID_tKeyLessThan
	)

◆ OPENDDS_MAP_CMP() [2/4]

typedef OpenDDS::DCPS::DataWriterImpl::OPENDDS_MAP_CMP	(	GUID_t	,
		ReaderInfo	,
		GUID_tKeyLessThan
	)

protected

◆ OPENDDS_MAP_CMP() [3/4]

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

private

◆ OPENDDS_MAP_CMP() [4/4]

typedef OpenDDS::DCPS::DataWriterImpl::OPENDDS_MAP_CMP	(	Sample_rch	,
		DDS::InstanceHandle_t	,
		SampleRchCmp
	)

private

◆ OPENDDS_VECTOR()

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

◆ parent()

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

virtual

Reimplemented from OpenDDS::DCPS::EntityImpl.

Definition at line 2264 of file DataWriterImpl.cpp.

References publisher_servant_.

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

◆ participant_liveliness_activity_after()

bool OpenDDS::DCPS::DataWriterImpl::participant_liveliness_activity_after ( const MonotonicTimePoint & tv )

Definition at line 1237 of file DataWriterImpl.cpp.

References DDS::LivelinessQosPolicy::kind, last_liveliness_activity_time_, DDS::DataWriterQos::liveliness, lock_, DDS::MANUAL_BY_PARTICIPANT_LIVELINESS_QOS, and qos_.

 {
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
   if (this->qos_.liveliness.kind == DDS::MANUAL_BY_PARTICIPANT_LIVELINESS_QOS) {
     return last_liveliness_activity_time_ > tv;
   } else {
     return false;
   }
 }

◆ persist_data()

bool OpenDDS::DCPS::DataWriterImpl::persist_data ( )

Make sent data available beyond the lifetime of this DataWriter.

Definition at line 2700 of file DataWriterImpl.cpp.

References data_container_.

Referenced by prepare_to_delete().

 {
   return this->data_container_->persist_data();
 }

◆ prepare_to_delete()

void OpenDDS::DCPS::DataWriterImpl::prepare_to_delete ( )

protected

Definition at line 2536 of file DataWriterImpl.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::Observer::e_DELETED, OpenDDS::DCPS::EntityImpl::get_observer(), LM_ERROR, OpenDDS::DCPS::TimePoint_T< SystemClock >::now(), OpenDDS::DCPS::Observer::on_deleted(), persist_data(), OpenDDS::DCPS::EntityImpl::set_deleted(), OpenDDS::DCPS::TransportClient::stop_associating(), OpenDDS::DCPS::TransportClient::terminate_send_if_suspended(), and unregister_instances().

Referenced by OpenDDS::DCPS::PublisherImpl::delete_datawriter(), and OpenDDS::DCPS::PublisherImpl::prepare_to_delete_datawriters().

 {
   const Observer_rch observer = get_observer(Observer::e_DELETED);
   if (observer) {
     observer->on_deleted(this);
   }
 
   this->set_deleted(true);
   this->stop_associating();
   this->terminate_send_if_suspended();
 
 #ifndef OPENDDS_NO_PERSISTENCE_PROFILE
   // Trigger data to be persisted, i.e. made durable, if so
   // configured. This needs be called before unregister_instances
   // because unregister_instances may cause instance dispose.
   if (!persist_data() && DCPS_debug_level >= 2) {
     ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::prepare_to_delete: ")
       ACE_TEXT("failed to make data durable.\n")));
   }
 #endif
 
   // Unregister all registered instances prior to deletion.
   unregister_instances(SystemTimePoint::now().to_dds_time());
 }

◆ register_for_reader()

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

virtual

Reimplemented from OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 822 of file DataWriterImpl.cpp.

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

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

◆ register_instance_from_durable_data()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::register_instance_from_durable_data	(	DDS::InstanceHandle_t &	handle,
		Message_Block_Ptr	data,
		const DDS::Time_t &	source_timestamp
	)

Delegate to the WriteDataContainer to register and tell the transport to broadcast the registered instance.

Definition at line 1658 of file DataWriterImpl.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), DBG_ENTRY_LVL, get_lock(), LM_ERROR, OpenDDS::DCPS::move(), register_instance_i(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::retcode_to_string(), and send_all_to_flush_control().

Referenced by OpenDDS::DCPS::DataDurabilityCache::get_data().

 {
   DBG_ENTRY_LVL("DataWriterImpl","register_instance_from_durable_data",6);
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    get_lock(),
                    DDS::RETCODE_ERROR);
 
   const DDS::ReturnCode_t ret = register_instance_i(handle, move(data), source_timestamp);
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::register_instance_from_durable_data: ")
                       ACE_TEXT("register instance with container failed, returned <%C>.\n"),
                       retcode_to_string(ret)),
                      ret);
   }
 
   send_all_to_flush_control(guard);
 
   return ret;
 }

◆ register_instance_i()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::register_instance_i	(	DDS::InstanceHandle_t &	handle,
		Message_Block_Ptr	data,
		const DDS::Time_t &	source_timestamp
	)

Delegate to the WriteDataContainer to register Must tell the transport to broadcast the registered instance upon returning.

Definition at line 1594 of file DataWriterImpl.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), create_control_message(), data_container_, DBG_ENTRY_LVL, OpenDDS::DCPS::EntityImpl::enabled_, OpenDDS::DCPS::DataSampleElement::get_header(), OpenDDS::DCPS::INSTANCE_REGISTRATION, LM_ERROR, monitor_, OpenDDS::DCPS::move(), DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, OpenDDS::DCPS::retcode_to_string(), and OpenDDS::DCPS::DataSampleElement::set_sample().

Referenced by get_or_create_instance_handle(), and register_instance_from_durable_data().

 {
   DBG_ENTRY_LVL("DataWriterImpl","register_instance_i",6);
 
   if (!enabled_) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::register_instance_i: ")
                       ACE_TEXT("Entity is not enabled.\n")),
                      DDS::RETCODE_NOT_ENABLED);
   }
 
   DDS::ReturnCode_t ret = data_container_->register_instance(handle, data);
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::register_instance_i: ")
                       ACE_TEXT("register instance with container failed, returned <%C>.\n"),
                       retcode_to_string(ret)),
                      ret);
   }
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 
   DataSampleElement* element = 0;
   ret = this->data_container_->obtain_buffer_for_control(element);
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::register_instance_i: ")
                       ACE_TEXT("obtain_buffer_for_control failed, returned <%C>.\n"),
                       retcode_to_string(ret)),
                      ret);
   }
 
   // Add header with the registration sample data.
   Message_Block_Ptr sample(
     create_control_message(
      INSTANCE_REGISTRATION,
      element->get_header(),
      move(data),
      source_timestamp));
 
   element->set_sample(move(sample));
 
   ret = this->data_container_->enqueue_control(element);
 
   if (ret != DDS::RETCODE_OK) {
     data_container_->release_buffer(element);
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::register_instance_i: ")
                       ACE_TEXT("enqueue_control failed, returned <%C>\n"),
                       retcode_to_string(ret)),
                      ret);
   }
 
   return ret;
 }

◆ register_instance_w_timestamp()

DDS::InstanceHandle_t OpenDDS::DCPS::DataWriterImpl::register_instance_w_timestamp	(	const Sample &	sample,
		const DDS::Time_t &	timestamp
	)

Definition at line 2916 of file DataWriterImpl.cpp.

References ACE_ERROR, ACE_TEXT(), get_or_create_instance_handle(), DDS::HANDLE_NIL, LM_NOTICE, OpenDDS::DCPS::log_level, OpenDDS::DCPS::LogLevel::Notice, DDS::RETCODE_OK, and OpenDDS::DCPS::retcode_to_string().

Referenced by OpenDDS::XTypes::DynamicDataWriterImpl::register_instance_w_timestamp(), and OpenDDS::DCPS::DataWriterImpl_T< MessageType >::register_instance_w_timestamp().

 {
   DDS::InstanceHandle_t registered_handle = DDS::HANDLE_NIL;
   const DDS::ReturnCode_t ret = get_or_create_instance_handle(registered_handle, sample, timestamp);
   if (ret != DDS::RETCODE_OK && log_level >= LogLevel::Notice) {
     ACE_ERROR((LM_NOTICE, ACE_TEXT("(%P|%t) NOTICE: DataWriterImpl::register_instance_w_timestamp: ")
                ACE_TEXT("register failed: %C\n"),
                retcode_to_string(ret)));
   }
   return registered_handle;
 }

◆ remove_all_associations()

void OpenDDS::DCPS::DataWriterImpl::remove_all_associations ( )

Definition at line 783 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), DBG_ENTRY_LVL, LM_WARNING, lock_, readers_, remove_associations(), OpenDDS::DCPS::TransportClient::stop_associating(), and OpenDDS::DCPS::TransportClient::transport_stop().

Referenced by OpenDDS::DCPS::PublisherImpl::delete_datawriter().

 {
   DBG_ENTRY_LVL("DataWriterImpl", "remove_all_associations", 6);
   // stop pending associations
   this->stop_associating();
 
   ReaderIdSeq readers;
   CORBA::ULong size;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, lock_);
 
     size = static_cast<CORBA::ULong>(readers_.size());
     readers.length(size);
 
     RepoIdSet::iterator itEnd = readers_.end();
     int i = 0;
 
     for (RepoIdSet::iterator it = readers_.begin(); it != itEnd; ++it) {
       readers[i ++] = *it;
     }
   }
 
   try {
     if (0 < size) {
       CORBA::Boolean dont_notify_lost = false;
 
       this->remove_associations(readers, dont_notify_lost);
     }
 
   } catch (const CORBA::Exception&) {
       ACE_DEBUG((LM_WARNING,
                  ACE_TEXT("(%P|%t) WARNING: DataWriterImpl::remove_all_associations() - ")
                  ACE_TEXT("caught exception from remove_associations.\n")));
   }
 
   transport_stop();
 }

◆ remove_associations()

void OpenDDS::DCPS::DataWriterImpl::remove_associations	(	const ReaderIdSeq &	readers,
		bool	callback
	)

virtual

Section 7.1.4.1: total_count will not decrement.

: Reconcile this with the verbiage in section 7.1.4.1

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 553 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), DDS::PublicationMatchedStatus::current_count, DDS::PublicationMatchedStatus::current_count_change, data_container_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TransportClient::disassociate(), OpenDDS::DCPS::Observer::e_DISASSOCIATED, OpenDDS::DCPS::EntityImpl::get_observer(), id_to_handle_map_, is_bit_, CORBA::is_nil(), DDS::PublicationMatchedStatus::last_subscription_handle, listener_for(), LM_DEBUG, lock_, lookup_instance_handles(), notify_publication_lost(), OpenDDS::DCPS::EntityImpl::notify_status_condition(), OpenDDS::DCPS::Observer::on_disassociated(), participant_servant_, publication_id_, publication_match_status_, DDS::PUBLICATION_MATCHED_STATUS, reader_info_, reader_info_lock_, readers_, OpenDDS::DCPS::EntityImpl::set_status_changed_flag(), OpenDDS::DCPS::TransportClient::stop_associating(), sync_unreg_rem_assocs_lock_, and DDS::PublicationMatchedStatus::total_count_change.

Referenced by remove_all_associations().

 {
   if (readers.length() == 0) {
     return;
   }
 
   const Observer_rch observer = get_observer(Observer::e_DISASSOCIATED);
   if (observer) {
     for (CORBA::ULong i = 0; i < readers.length(); ++i) {
       observer->on_disassociated(this, readers[i]);
     }
   }
 
   if (DCPS_debug_level >= 1) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) DataWriterImpl::remove_associations: ")
                ACE_TEXT("bit %d local %C remote %C num remotes %d\n"),
                is_bit_,
                LogGuid(publication_id_).c_str(),
                LogGuid(readers[0]).c_str(),
                readers.length()));
   }
 
   // stop pending associations for these reader ids
   this->stop_associating(readers.get_buffer(), readers.length());
 
   ReaderIdSeq fully_associated_readers;
   CORBA::ULong fully_associated_len = 0;
   ReaderIdSeq rds;
   CORBA::ULong rds_len = 0;
   DDS::InstanceHandleSeq handles;
 
   ACE_GUARD(ACE_Thread_Mutex, wait_guard, sync_unreg_rem_assocs_lock_);
   {
     // Ensure the same acquisition order as in wait_for_acknowledgments().
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
     //Remove the readers from fully associated reader list.
     //If the supplied reader is not in the cached reader list then it is
     //already removed. We just need remove the readers in the list that have
     //not been removed.
 
     CORBA::ULong len = readers.length();
 
     for (CORBA::ULong i = 0; i < len; ++i) {
       //Remove the readers from fully associated reader list. If it's not
       //in there, the association_complete() is not called yet and remove it
       //from pending list.
 
       if (remove(readers_, readers[i]) == 0) {
         ++ fully_associated_len;
         fully_associated_readers.length(fully_associated_len);
         fully_associated_readers [fully_associated_len - 1] = readers[i];
 
         ++ rds_len;
         rds.length(rds_len);
         rds [rds_len - 1] = readers[i];
       }
 
       data_container_->remove_reader_acks(readers[i]);
 
       ACE_GUARD(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_);
       reader_info_.erase(readers[i]);
       //else reader is already removed which indicates remove_association()
       //is called multiple times.
     }
 
     if (fully_associated_len > 0 && !is_bit_) {
       // The reader should be in the id_to_handle map at this time
       this->lookup_instance_handles(fully_associated_readers, handles);
 
       for (CORBA::ULong i = 0; i < fully_associated_len; ++i) {
         id_to_handle_map_.erase(fully_associated_readers[i]);
       }
     }
 
     // Mirror the PUBLICATION_MATCHED_STATUS processing from
     // association_complete() here.
     if (!this->is_bit_) {
 
       // Derive the change in the number of subscriptions reading this writer.
       int matchedSubscriptions =
         static_cast<int>(this->id_to_handle_map_.size());
       this->publication_match_status_.current_count_change =
         matchedSubscriptions - this->publication_match_status_.current_count;
 
       // Only process status if the number of subscriptions has changed.
       if (this->publication_match_status_.current_count_change != 0) {
         this->publication_match_status_.current_count = matchedSubscriptions;
 
         /// Section 7.1.4.1: total_count will not decrement.
 
         /// @TODO: Reconcile this with the verbiage in section 7.1.4.1
         this->publication_match_status_.last_subscription_handle =
           handles[fully_associated_len - 1];
 
         set_status_changed_flag(DDS::PUBLICATION_MATCHED_STATUS, true);
 
         DDS::DataWriterListener_var listener =
           this->listener_for(DDS::PUBLICATION_MATCHED_STATUS);
 
         if (!CORBA::is_nil(listener.in())) {
           listener->on_publication_matched(this, this->publication_match_status_);
 
           // Listener consumes the change.
           this->publication_match_status_.total_count_change = 0;
           this->publication_match_status_.current_count_change = 0;
         }
 
         this->notify_status_condition();
       }
     }
   }
 
   for (CORBA::ULong i = 0; i < rds.length(); ++i) {
     this->disassociate(rds[i]);
   }
 
   // If this remove_association is invoked when the InfoRepo
   // detects a lost reader then make a callback to notify
   // subscription lost.
   if (notify_lost && handles.length() > 0) {
     this->notify_publication_lost(handles);
   }
 
   const RcHandle<DomainParticipantImpl> participant = participant_servant_.lock();
   for (unsigned int i = 0; i < handles.length(); ++i) {
     participant->return_handle(handles[i]);
   }
 }

◆ replay_durable_data_for()

void OpenDDS::DCPS::DataWriterImpl::replay_durable_data_for ( const GUID_t & remote_sub_id )

virtual

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 684 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), available_data_list_, OpenDDS::DCPS::SendStateDataSampleList::begin(), controlTracker, create_control_message(), data_container_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::STUN::encoding(), OpenDDS::DCPS::SendStateDataSampleList::end(), OpenDDS::DCPS::END_HISTORIC_SAMPLES, OpenDDS::DCPS::SendStateDataSampleList::enqueue_tail(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), get_db_lock(), get_lock(), get_resend_data(), header, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::Encoding::KIND_UNALIGNED_CDR, DDS::DataWriterQos::lifespan, LM_ERROR, LM_INFO, ACE_Message_Block::MB_DATA, OpenDDS::DCPS::MessageTracker::message_dropped(), OpenDDS::DCPS::MessageTracker::message_sent(), OpenDDS::DCPS::move(), OpenDDS::DCPS::TimePoint_T< SystemClock >::now(), OPENDDS_STRING, publisher_servant_, qos_, reader_info_, reader_info_lock_, OpenDDS::DCPS::SEND_CONTROL_ERROR, OpenDDS::DCPS::TransportClient::send_w_control(), and OpenDDS::DCPS::serialized_size().

 {
   DBG_ENTRY_LVL("DataWriterImpl", "replay_durable_data_for", 6);
 
   bool reader_durable = false;
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
   OPENDDS_STRING filterClassName;
   RcHandle<FilterEvaluator> eval;
   DDS::StringSeq expression_params;
 #endif
 
   {
     ACE_GUARD(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_);
     RepoIdToReaderInfoMap::const_iterator it = reader_info_.find(remote_id);
 
     if (it != reader_info_.end()) {
       reader_durable = it->second.durable_;
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
       filterClassName = it->second.filter_class_name_;
       eval = it->second.eval_;
       expression_params = it->second.expression_params_;
 #endif
     }
   }
 
   // Support DURABILITY QoS
   if (reader_durable) {
     // Tell the WriteDataContainer to resend all sending/sent
     // samples.
     this->data_container_->reenqueue_all(remote_id, this->qos_.lifespan
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
                                          , filterClassName, eval.in(), expression_params
 #endif
                                          );
 
     // Acquire the data writer container lock to avoid deadlock. The
     // thread calling association_complete() has to acquire lock in the
     // same order as the write()/register() operation.
 
     // Since the thread calling association_complete() is the ORB
     // thread, it may have some performance penalty. If the
     // performance is an issue, we may need a new thread to handle the
     // data_available() calls.
     ACE_GUARD(ACE_Recursive_Thread_Mutex,
               guard,
               this->get_lock());
 
     SendStateDataSampleList list = this->get_resend_data();
     {
       ACE_GUARD(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_);
       // Update the reader's expected sequence
       SequenceNumber& seq =
         reader_info_.find(remote_id)->second.expected_sequence_;
 
       for (SendStateDataSampleList::iterator list_el = list.begin();
            list_el != list.end(); ++list_el) {
         list_el->get_header().historic_sample_ = true;
 
         if (list_el->get_header().sequence_ > seq) {
           seq = list_el->get_header().sequence_;
         }
       }
     }
 
     RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();
     if (!publisher || publisher->is_suspended()) {
       this->available_data_list_.enqueue_tail(list);
 
     } else {
       if (DCPS_debug_level >= 4) {
         ACE_DEBUG((LM_INFO, ACE_TEXT("(%P|%t) DataWriterImpl::replay_durable_data_for: Sending historic samples\n")));
       }
 
       const Encoding encoding(Encoding::KIND_UNALIGNED_CDR);
       size_t size = 0;
       serialized_size(encoding, size, remote_id);
       Message_Block_Ptr data(
         new ACE_Message_Block(size, ACE_Message_Block::MB_DATA, 0, 0, 0,
                               get_db_lock()));
       Serializer ser(data.get(), encoding);
       ser << remote_id;
 
       DataSampleHeader header;
       Message_Block_Ptr end_historic_samples(create_control_message(END_HISTORIC_SAMPLES, header, move(data),
                                                                     SystemTimePoint::now().to_dds_time()));
 
       this->controlTracker.message_sent();
       guard.release();
       const SendControlStatus ret = send_w_control(list, header, move(end_historic_samples), remote_id);
       if (ret == SEND_CONTROL_ERROR) {
         ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: ")
                    ACE_TEXT("DataWriterImpl::replay_durable_data_for: ")
                    ACE_TEXT("send_w_control failed.\n")));
         this->controlTracker.message_dropped();
       }
     }
   }
 }

◆ retrieve_inline_qos_data()

void OpenDDS::DCPS::DataWriterImpl::retrieve_inline_qos_data ( TransportSendListener::InlineQosData & qos_data ) const

virtual

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 2728 of file DataWriterImpl.cpp.

References OpenDDS::DCPS::TransportSendListener::InlineQosData::dw_qos, TAO::String_var< charT >::in(), OpenDDS::DCPS::TransportSendListener::InlineQosData::pub_qos, publisher_servant_, qos_, OpenDDS::DCPS::TransportSendListener::InlineQosData::topic_name, and topic_name_.

 {
   RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();
   if (publisher) {
     publisher->get_qos(qos_data.pub_qos);
   }
   qos_data.dw_qos = this->qos_;
   qos_data.topic_name = this->topic_name_.in();
 }

◆ return_handle()

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

private

Definition at line 192 of file DataWriterImpl.cpp.

References participant_servant_.

Referenced by OpenDDS::DCPS::WriteDataContainer::unregister_all().

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

◆ send_all_to_flush_control()

void OpenDDS::DCPS::DataWriterImpl::send_all_to_flush_control ( ACE_Guard< ACE_Recursive_Thread_Mutex > & guard )

Definition at line 1577 of file DataWriterImpl.cpp.

References controlTracker, DBG_ENTRY_LVL, get_unsent_data(), OpenDDS::DCPS::MessageTracker::message_sent(), ACE_Guard< ACE_LOCK >::release(), and OpenDDS::DCPS::TransportClient::send().

Referenced by dispose(), dispose_and_unregister(), get_or_create_instance_handle(), register_instance_from_durable_data(), send_request_ack(), and unregister_instance_i().

 {
   DBG_ENTRY_LVL("DataWriterImpl","send_all_to_flush_control",6);
 
   SendStateDataSampleList list;
 
   ACE_UINT64 transaction_id = this->get_unsent_data(list);
 
   controlTracker.message_sent();
 
   //need to release guard to call down to transport
   guard.release();
 
   this->send(list, transaction_id);
 }

◆ send_control()

SendControlStatus OpenDDS::DCPS::DataWriterImpl::send_control	(	const DataSampleHeader &	header,
		Message_Block_Ptr	msg
	)

protectedvirtual

Definition at line 2767 of file DataWriterImpl.cpp.

References controlTracker, OpenDDS::DCPS::MessageTracker::message_dropped(), OpenDDS::DCPS::MessageTracker::message_sent(), OpenDDS::DCPS::move(), OpenDDS::DCPS::TransportClient::send_control(), and OpenDDS::DCPS::SEND_CONTROL_OK.

Referenced by end_coherent_changes(), and send_liveliness().

 {
   controlTracker.message_sent();
 
   SendControlStatus status = TransportClient::send_control(header, move(msg));
 
   if (status != SEND_CONTROL_OK) {
     controlTracker.message_dropped();
   }
 
   return status;
 }

◆ send_end_historic_samples()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::send_end_historic_samples ( const GUID_t & readerId )

private

◆ send_liveliness()

bool OpenDDS::DCPS::DataWriterImpl::send_liveliness ( const MonotonicTimePoint & now )

private

Send the liveliness message.

Definition at line 2513 of file DataWriterImpl.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), create_control_message(), OpenDDS::DCPS::DATAWRITER_LIVELINESS, domain_id_, header, DDS::LivelinessQosPolicy::kind, last_liveliness_activity_time_, DDS::DataWriterQos::liveliness, LM_ERROR, DDS::MANUAL_BY_TOPIC_LIVELINESS_QOS, OpenDDS::DCPS::move(), OpenDDS::DCPS::TimePoint_T< SystemClock >::now(), qos_, send_control(), OpenDDS::DCPS::SEND_CONTROL_ERROR, and TheServiceParticipant.

Referenced by assert_liveliness(), and handle_timeout().

 {
   if (this->qos_.liveliness.kind == DDS::MANUAL_BY_TOPIC_LIVELINESS_QOS ||
       !TheServiceParticipant->get_discovery(domain_id_)->supports_liveliness()) {
     DataSampleHeader header;
     Message_Block_Ptr empty;
     Message_Block_Ptr liveliness_msg(
       create_control_message(
         DATAWRITER_LIVELINESS, header, move(empty),
         SystemTimePoint::now().to_dds_time()));
 
     if (this->send_control(header, move(liveliness_msg)) == SEND_CONTROL_ERROR) {
       ACE_ERROR_RETURN((LM_ERROR,
                         ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::send_liveliness: ")
                         ACE_TEXT("send_control failed.\n")),
                        false);
     }
   }
   last_liveliness_activity_time_ = now;
   return true;
 }

◆ send_request_ack()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::send_request_ack ( )

private

Definition at line 1040 of file DataWriterImpl.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), create_control_message(), data_container_, OpenDDS::DCPS::DataSampleElement::get_header(), get_lock(), LM_ERROR, OpenDDS::DCPS::move(), OpenDDS::DCPS::TimePoint_T< SystemClock >::now(), OpenDDS::DCPS::REQUEST_ACK, DDS::RETCODE_ERROR, DDS::RETCODE_OK, send_all_to_flush_control(), OpenDDS::DCPS::DataSampleElement::set_sample(), and OpenDDS::DCPS::TimePoint_T< AceClock >::to_dds_time().

Referenced by wait_for_acknowledgments().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    get_lock(),
                    DDS::RETCODE_ERROR);
 
 
   DataSampleElement* element = 0;
   DDS::ReturnCode_t ret = this->data_container_->obtain_buffer_for_control(element);
 
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::send_request_ack: ")
                       ACE_TEXT("obtain_buffer_for_control returned %d.\n"),
                       ret),
                      ret);
   }
 
   Message_Block_Ptr blk;
   // Add header with the registration sample data.
   Message_Block_Ptr sample(
     create_control_message(
       REQUEST_ACK,
       element->get_header(),
       move(blk),
       SystemTimePoint::now().to_dds_time()));
 
   element->set_sample(move(sample));
 
   ret = this->data_container_->enqueue_control(element);
 
   if (ret != DDS::RETCODE_OK) {
     data_container_->release_buffer(element);
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::send_request_ack: ")
                       ACE_TEXT("enqueue_control failed.\n")),
                      ret);
   }
 
 
   send_all_to_flush_control(guard);
 
   return DDS::RETCODE_OK;
 }

◆ send_suspended_data()

void OpenDDS::DCPS::DataWriterImpl::send_suspended_data ( )

Called by the PublisherImpl to indicate that the Publisher is now resumed and any data collected while it was suspended should now be sent.

Definition at line 1964 of file DataWriterImpl.cpp.

References available_data_list_, max_suspended_transaction_id_, min_suspended_transaction_id_, OpenDDS::DCPS::SendStateDataSampleList::reset(), and OpenDDS::DCPS::TransportClient::send().

 {
   //this serves to get TransportClient's max_transaction_id_seen_
   //to the correct value for this list of transactions
   if (max_suspended_transaction_id_ != 0) {
     this->send(this->available_data_list_, max_suspended_transaction_id_);
     max_suspended_transaction_id_ = 0;
   }
 
   //this serves to actually have the send proceed in
   //sending the samples to the datalinks by passing it
   //the min_suspended_transaction_id_ which should be the
   //TransportClient's expected_transaction_id_
   this->send(this->available_data_list_, min_suspended_transaction_id_);
   min_suspended_transaction_id_ = 0;
   this->available_data_list_.reset();
 }

◆ serialize_sample()

ACE_Message_Block * OpenDDS::DCPS::DataWriterImpl::serialize_sample ( const Sample & sample )

protected

Definition at line 2970 of file DataWriterImpl.cpp.

References ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, ACE_ERROR, ACE_NEW_MALLOC_RETURN, ACE_NEW_RETURN, OpenDDS::DCPS::TypeSupportImpl::base_extensibility(), OpenDDS::DCPS::DataWriterImpl::EncodingMode::buffer_size(), OpenDDS::DCPS::TransportClient::cdr_encapsulation(), data_allocator_, db_allocator_, OpenDDS::STUN::encoding(), OpenDDS::DCPS::DataWriterImpl::EncodingMode::encoding(), encoding_mode_, OpenDDS::DCPS::LogLevel::Error, OpenDDS::DCPS::EncapsulationHeader::from_encoding(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), get_db_lock(), OpenDDS::DCPS::Sample::key_only(), LM_ERROR, OpenDDS::DCPS::log_level, ACE_Time_Value::max_time, mb_allocator_, ACE_Message_Block::MB_DATA, OpenDDS::DCPS::unique_ptr< T, Deleter >::release(), OpenDDS::DCPS::unique_ptr< T, Deleter >::reset(), OpenDDS::DCPS::Sample::serialize(), OpenDDS::DCPS::EncapsulationHeader::set_encapsulation_options(), skip_serialize_, OpenDDS::DCPS::Sample::to_message_block(), type_support_, and ACE_Time_Value::zero.

Referenced by get_or_create_instance_handle(), and write_sample().

 {
   const bool encapsulated = cdr_encapsulation();
   const Encoding& encoding = encoding_mode_.encoding();
   Message_Block_Ptr mb;
   ACE_Message_Block* tmp_mb;
 
   // Don't use the cached allocator for the registered sample message
   // block.
   if (sample.key_only() && !skip_serialize_) {
     ACE_NEW_RETURN(tmp_mb,
       ACE_Message_Block(
         encoding_mode_.buffer_size(sample),
         ACE_Message_Block::MB_DATA,
         0, // cont
         0, // data
         0, // alloc_strategy
         get_db_lock()),
       0);
   } else {
     ACE_NEW_MALLOC_RETURN(tmp_mb,
       static_cast<ACE_Message_Block*>(
         mb_allocator_->malloc(sizeof(ACE_Message_Block))),
       ACE_Message_Block(
         encoding_mode_.buffer_size(sample),
         ACE_Message_Block::MB_DATA,
         0, // cont
         0, // data
         data_allocator_.get(), // allocator_strategy
         get_db_lock(), // data block locking_strategy
         ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
         ACE_Time_Value::zero,
         ACE_Time_Value::max_time,
         db_allocator_.get(),
         mb_allocator_.get()),
       0);
   }
   mb.reset(tmp_mb);
 
   if (skip_serialize_) {
     if (!sample.to_message_block(*mb)) {
       if (log_level >= LogLevel::Error) {
         ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: DataWriterImpl::serialize_sample: "
                    "to_message_block failed\n"));
       }
       return 0;
     }
   } else {
     Serializer serializer(mb.get(), encoding);
     if (encapsulated) {
       EncapsulationHeader encap;
       if (!encap.from_encoding(encoding, type_support_->base_extensibility())) {
         // from_encoding logged the error
         return 0;
       }
       if (!(serializer << encap)) {
         if (log_level >= LogLevel::Error) {
           ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: DataWriterImpl::serialize_sample: "
             "failed to serialize data encapsulation header\n"));
         }
         return 0;
       }
     }
     if (!sample.serialize(serializer)) {
       if (log_level >= LogLevel::Error) {
         ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: DataWriterImpl::serialize_sample: "
           "failed to serialize sample data\n"));
       }
       return 0;
     }
     if (encapsulated && !EncapsulationHeader::set_encapsulation_options(mb)) {
       if (log_level >= LogLevel::Error) {
         ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: DataWriterImpl::serialize_sample: "
           "set_encapsulation_options failed\n"));
       }
       return 0;
     }
   }
 
   return mb.release();
 }

◆ set_listener()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::set_listener	(	DDS::DataWriterListener_ptr	a_listener,
		DDS::StatusMask	mask
	)

virtual

Definition at line 985 of file DataWriterImpl.cpp.

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

Referenced by cleanup(), and init().

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

◆ set_marshal_skip_serialize()

void OpenDDS::DCPS::DataWriterImpl::set_marshal_skip_serialize ( bool value )

inline

Definition at line 124 of file DataWriterImpl.h.

References value.

   {
     skip_serialize_ = value;
   }

◆ set_qos()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::set_qos ( const DDS::DataWriterQos & qos )

virtual

Definition at line 916 of file DataWriterImpl.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), OpenDDS::DCPS::Qos_Helper::changeable(), OpenDDS::DCPS::Qos_Helper::consistent(), data_container_, DDS::DataWriterQos::deadline, domain_id_, dp_id_, OpenDDS::DCPS::Observer::e_QOS_CHANGED, OpenDDS::DCPS::EntityImpl::enabled_, OpenDDS::DCPS::EntityImpl::get_observer(), LM_ERROR, OpenDDS::DCPS::Observer::on_qos_changed(), OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK, OPENDDS_NO_DURABILITY_SERVICE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_STRENGTH_COMPATIBILITY_CHECK, passed_qos_, DDS::DeadlineQosPolicy::period, publication_id_, publisher_servant_, qos_, DDS::DataWriterQos::representation, DDS::RETCODE_ERROR, DDS::RETCODE_IMMUTABLE_POLICY, DDS::RETCODE_INCONSISTENT_POLICY, DDS::RETCODE_OK, DDS::RETCODE_UNSUPPORTED, TheServiceParticipant, OpenDDS::DCPS::Qos_Helper::valid(), and DDS::DataRepresentationQosPolicy::value.

 {
   OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
   OPENDDS_NO_OWNERSHIP_STRENGTH_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
   OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
   OPENDDS_NO_DURABILITY_SERVICE_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
   OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
 
   DDS::DataWriterQos new_qos = qos;
   new_qos.representation.value = qos_.representation.value;
   if (Qos_Helper::valid(new_qos) && Qos_Helper::consistent(new_qos)) {
     if (qos_ == new_qos)
       return DDS::RETCODE_OK;
 
     if (enabled_) {
       if (!Qos_Helper::changeable(qos_, new_qos)) {
         return DDS::RETCODE_IMMUTABLE_POLICY;
       }
 
       Discovery_rch disco = TheServiceParticipant->get_discovery(domain_id_);
       DDS::PublisherQos publisherQos;
       RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();
 
       bool status = false;
       if (publisher) {
         publisher->get_qos(publisherQos);
         status
           = disco->update_publication_qos(domain_id_,
                                           dp_id_,
                                           this->publication_id_,
                                           new_qos,
                                           publisherQos);
       }
       if (!status) {
         ACE_ERROR_RETURN((LM_ERROR,
                           ACE_TEXT("(%P|%t) DataWriterImpl::set_qos, ")
                           ACE_TEXT("qos not updated.\n")),
                          DDS::RETCODE_ERROR);
       }
 
       if (!(qos_ == new_qos)) {
         data_container_->set_deadline_period(TimeDuration(qos.deadline.period));
         qos_ = new_qos;
       }
     }
 
     qos_ = new_qos;
     passed_qos_ = qos;
 
     const Observer_rch observer = get_observer(Observer::e_QOS_CHANGED);
     if (observer) {
       observer->on_qos_changed(this);
     }
 
     return DDS::RETCODE_OK;
 
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ set_wait_pending_deadline()

void OpenDDS::DCPS::DataWriterImpl::set_wait_pending_deadline ( const MonotonicTimePoint & deadline )

Set deadline to complete wait_pending by. If 0, then wait_pending will wait indefinitely if needed.

Definition at line 2787 of file DataWriterImpl.cpp.

References wait_pending_deadline_.

Referenced by OpenDDS::DCPS::PublisherImpl::delete_datawriter(), and OpenDDS::DCPS::PublisherImpl::set_wait_pending_deadline().

 {
   wait_pending_deadline_ = deadline;
 }

◆ setup_serialization()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::setup_serialization ( )

protected

Setup CDR serialization options.

Definition at line 2823 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_ERROR, OpenDDS::DCPS::DataWriterImpl::EncodingMode::buffer_size_bound(), OpenDDS::DCPS::TransportClient::cdr_encapsulation(), data_allocator_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::DataWriterImpl::EncodingMode::encoding(), encoding_mode_, OpenDDS::DCPS::SerializedSizeBound::get(), OpenDDS::DCPS::Encoding::kind(), OpenDDS::DCPS::Encoding::kind_to_string(), OpenDDS::DCPS::Encoding::KIND_UNALIGNED_CDR, OpenDDS::DCPS::Encoding::KIND_XCDR1, LM_DEBUG, LM_NOTICE, LM_WARNING, OpenDDS::DCPS::log_level, OpenDDS::DCPS::TypeSupportImpl::max_extensibility(), OpenDDS::DCPS::MUTABLE, n_chunks_, OpenDDS::DCPS::TypeSupportImpl::name(), OpenDDS::DCPS::LogLevel::Notice, qos_, OpenDDS::DCPS::repr_to_encoding_kind(), OpenDDS::DCPS::repr_to_string(), DDS::DataWriterQos::representation, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::TransportClient::swap_bytes(), type_support_, OpenDDS::DCPS::UNALIGNED_CDR_DATA_REPRESENTATION, OpenDDS::DCPS::DataWriterImpl::EncodingMode::valid(), DDS::DataRepresentationQosPolicy::value, and OpenDDS::DCPS::LogLevel::Warning.

Referenced by enable().

 {
   if (qos_.representation.value.length() > 0 &&
       qos_.representation.value[0] != UNALIGNED_CDR_DATA_REPRESENTATION) {
     // If the QoS explicitly sets XCDR, XCDR2, or XML, force encapsulation
     cdr_encapsulation(true);
   }
 
   if (cdr_encapsulation()) {
     Encoding::Kind encoding_kind;
     // There should only be one data representation in a DataWriter, so
     // simply use qos_.representation.value[0].
     if (repr_to_encoding_kind(qos_.representation.value[0], encoding_kind)) {
       encoding_mode_ = EncodingMode(type_support_, encoding_kind, swap_bytes());
       if (encoding_kind == Encoding::KIND_XCDR1 &&
           type_support_->max_extensibility() == MUTABLE) {
         if (log_level >= LogLevel::Notice) {
           ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: DataWriterImpl::setup_serialization: "
             "Encountered unsupported combination of XCDR1 encoding and mutable extensibility "
             "for writer of type %C\n",
             type_support_->name()));
         }
         return DDS::RETCODE_ERROR;
       } else if (encoding_kind == Encoding::KIND_UNALIGNED_CDR) {
         if (log_level >= LogLevel::Notice) {
           ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: DataWriterImpl::setup_serialization: "
             "Unaligned CDR is not supported by transport types that require encapsulation\n"));
         }
         return DDS::RETCODE_ERROR;
       }
     } else if (log_level >= LogLevel::Warning) {
       ACE_ERROR((LM_WARNING, "(%P|%t) WARNING: DataWriterImpl::setup_serialization: "
                  "Encountered unsupported or unknown data representation: %C ",
                  "for writer of type %C\n",
                  repr_to_string(qos_.representation.value[0]).c_str(),
                  type_support_->name()));
     }
   } else {
     // Pick unaligned CDR as it is the implicit representation for non-encapsulated
     encoding_mode_ = EncodingMode(type_support_, Encoding::KIND_UNALIGNED_CDR, swap_bytes());
   }
   if (!encoding_mode_.valid()) {
     if (log_level >= LogLevel::Notice) {
       ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: DataWriterImpl::setup_serialization: "
                  "Could not find a valid data representation\n"));
     }
     return DDS::RETCODE_ERROR;
   }
 
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) WriterImpl::setup_serialization: "
       "Setup successfully with %C data representation.\n",
       Encoding::kind_to_string(encoding_mode_.encoding().kind()).c_str()));
   }
 
   // Set up allocator with reserved space for data if it is bounded
   const SerializedSizeBound buffer_size_bound = encoding_mode_.buffer_size_bound();
   if (buffer_size_bound) {
     const size_t chunk_size = buffer_size_bound.get();
     data_allocator_.reset(new DataAllocator(n_chunks_, chunk_size));
     if (DCPS_debug_level >= 2) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) DataWriterImpl::setup_serialization: "
         "using data allocator at %x with %B %B byte chunks\n",
         data_allocator_.get(),
         n_chunks_,
         chunk_size));
     }
   } else if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) DataWriterImpl::setup_serialization: "
       "sample size is unbounded, not using data allocator, "
       "always allocating from heap\n"));
   }
   return DDS::RETCODE_OK;
 }

◆ should_ack()

bool OpenDDS::DCPS::DataWriterImpl::should_ack ( ) const

Does this writer have samples to be acknowledged?

Definition at line 1016 of file DataWriterImpl.cpp.

References readers_.

 {
   // N.B. It may be worthwhile to investigate a more efficient
   // heuristic for determining if a writer should send SAMPLE_ACK
   // control samples. Perhaps based on a sequence number delta?
   return this->readers_.size() != 0;
 }

◆ track_sequence_number()

void OpenDDS::DCPS::DataWriterImpl::track_sequence_number ( GUIDSeq * filter_out )

private

Definition at line 1930 of file DataWriterImpl.cpp.

References ACE_GUARD, get_max_sn(), reader_info_, and reader_info_lock_.

Referenced by write().

 {
   const SequenceNumber sn = get_max_sn();
   ACE_GUARD(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_);
 
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
   // Track individual expected sequence numbers in ReaderInfo
   RepoIdSet excluded;
 
   if (filter_out && !reader_info_.empty()) {
     const GUID_t* buf = filter_out->get_buffer();
     excluded.insert(buf, buf + filter_out->length());
   }
 
   for (RepoIdToReaderInfoMap::iterator iter = reader_info_.begin(),
        end = reader_info_.end(); iter != end; ++iter) {
     // If not excluding this reader, update expected sequence
     if (excluded.count(iter->first) == 0) {
       iter->second.expected_sequence_ = sn;
     }
   }
 
 #else
   ACE_UNUSED_ARG(filter_out);
   for (RepoIdToReaderInfoMap::iterator iter = reader_info_.begin(),
        end = reader_info_.end(); iter != end; ++iter) {
     iter->second.expected_sequence_ = sn;
   }
 
 #endif // OPENDDS_NO_CONTENT_FILTERED_TOPIC
 
 }

◆ transport_assoc_done()

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

virtual

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 279 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::TransportClient::ASSOC_ACTIVE, OpenDDS::DCPS::TransportClient::ASSOC_OK, association_complete_i(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, LM_DEBUG, LM_ERROR, LM_INFO, lock_, and publication_id_.

 {
   DBG_ENTRY_LVL("DataWriterImpl", "transport_assoc_done", 6);
 
   if (!(flags & ASSOC_OK)) {
     if (DCPS_debug_level) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) DataWriterImpl::transport_assoc_done: ")
                  ACE_TEXT("ERROR: transport layer failed to associate %C\n"),
                  LogGuid(remote_id).c_str()));
     }
 
     return;
   }
 
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
 
   if (DCPS_debug_level) {
     ACE_DEBUG((LM_INFO,
                ACE_TEXT("(%P|%t) DataWriterImpl::transport_assoc_done: ")
                ACE_TEXT("writer %C succeeded in associating with reader %C\n"),
                LogGuid(publication_id_).c_str(),
                LogGuid(remote_id).c_str()));
   }
 
   if (flags & ASSOC_ACTIVE) {
 
     // Have we already received an association_complete() callback?
     if (DCPS_debug_level) {
       ACE_DEBUG((LM_DEBUG,
                  ACE_TEXT("(%P|%t) DataWriterImpl::transport_assoc_done: ")
                  ACE_TEXT("writer %C reader %C calling association_complete_i\n"),
                  LogGuid(publication_id_).c_str(),
                  LogGuid(remote_id).c_str()));
     }
     association_complete_i(remote_id);
 
   } else {
     // In the current implementation, DataWriter is always active, so this
     // code will not be applicable.
     if (DCPS_debug_level) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) DataWriterImpl::transport_assoc_done: ")
                  ACE_TEXT("ERROR: DataWriter (%C) should always be active in current implementation\n"),
                  LogGuid(publication_id_).c_str()));
     }
   }
 }

◆ transport_discovery_change()

void OpenDDS::DCPS::DataWriterImpl::transport_discovery_change ( )

virtual

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 2805 of file DataWriterImpl.cpp.

References OpenDDS::DCPS::TransportClient::connection_info(), domain_id(), domain_id_, dp_id_, lock_, OpenDDS::DCPS::TransportClient::populate_connection_info(), publication_id_, ACE_Guard< ACE_LOCK >::release(), and TheServiceParticipant.

 {
   populate_connection_info();
   const TransportLocatorSeq& trans_conf_info = connection_info();
 
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
   const GUID_t dp_id_copy = dp_id_;
   const GUID_t publication_id_copy = publication_id_;
   const int domain_id = domain_id_;
   guard.release();
 
   Discovery_rch disco = TheServiceParticipant->get_discovery(domain_id);
   disco->update_publication_locators(domain_id,
                                      dp_id_copy,
                                      publication_id_copy,
                                      trans_conf_info);
 }

◆ unregister_all()

void OpenDDS::DCPS::DataWriterImpl::unregister_all ( )

Delegate to WriteDataContainer to unregister all instances.

Definition at line 2052 of file DataWriterImpl.cpp.

References data_container_.

Referenced by OpenDDS::DCPS::PublisherImpl::delete_datawriter().

 {
   data_container_->unregister_all();
 }

◆ unregister_for_reader()

void OpenDDS::DCPS::DataWriterImpl::unregister_for_reader	(	const GUID_t &	participant,
		const GUID_t &	writerid,
		const GUID_t &	readerid
	)

virtual

Reimplemented from OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 832 of file DataWriterImpl.cpp.

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

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

◆ unregister_instance_i()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::unregister_instance_i	(	DDS::InstanceHandle_t	handle,
		const DDS::Time_t &	source_timestamp
	)

Delegate to the WriteDataContainer to unregister and tell the transport to broadcast the unregistered instance.

Definition at line 1685 of file DataWriterImpl.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), DDS::WriterDataLifecycleQosPolicy::autodispose_unregistered_instances, create_control_message(), data_container_, DBG_ENTRY_LVL, dispose_and_unregister(), OpenDDS::DCPS::EntityImpl::enabled_, OpenDDS::DCPS::DataSampleElement::get_header(), get_lock(), LM_ERROR, OpenDDS::DCPS::move(), qos_, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, send_all_to_flush_control(), OpenDDS::DCPS::DataSampleElement::set_sample(), OpenDDS::DCPS::UNREGISTER_INSTANCE, and DDS::DataWriterQos::writer_data_lifecycle.

Referenced by unregister_instance_w_timestamp(), and unregister_instances().

 {
   DBG_ENTRY_LVL("DataWriterImpl","unregister_instance_i",6);
 
   if (!enabled_) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::unregister_instance_i: ")
                       ACE_TEXT("Entity is not enabled.\n")),
                      DDS::RETCODE_NOT_ENABLED);
   }
 
   // According to spec 1.2, autodispose_unregistered_instances true causes
   // dispose on the instance prior to calling unregister operation.
   if (this->qos_.writer_data_lifecycle.autodispose_unregistered_instances) {
     return this->dispose_and_unregister(handle, source_timestamp);
   }
 
   DDS::ReturnCode_t ret = DDS::RETCODE_ERROR;
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, get_lock(), ret);
   Message_Block_Ptr unregistered_sample_data;
   ret = this->data_container_->unregister(handle, unregistered_sample_data);
 
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::unregister_instance_i: ")
                       ACE_TEXT("unregister with container failed.\n")),
                      ret);
   }
 
   DataSampleElement* element = 0;
   ret = this->data_container_->obtain_buffer_for_control(element);
 
   if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::unregister_instance_i: ")
                       ACE_TEXT("obtain_buffer_for_control returned %d.\n"),
                       ret),
                      ret);
   }
 
   Message_Block_Ptr sample(create_control_message(UNREGISTER_INSTANCE,
                                                   element->get_header(),
                                                   move(unregistered_sample_data),
                                                   source_timestamp));
   element->set_sample(move(sample));
 
   ret = this->data_container_->enqueue_control(element);
 
   if (ret != DDS::RETCODE_OK) {
     data_container_->release_buffer(element);
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::unregister_instance_i: ")
                       ACE_TEXT("enqueue_control failed.\n")),
                      ret);
   }
 
   send_all_to_flush_control(guard);
   return DDS::RETCODE_OK;
 }

◆ unregister_instance_w_timestamp()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::unregister_instance_w_timestamp	(	const Sample &	sample,
		DDS::InstanceHandle_t	instance_handle,
		const DDS::Time_t &	timestamp
	)

Definition at line 2929 of file DataWriterImpl.cpp.

References instance_must_exist(), DDS::RETCODE_OK, and unregister_instance_i().

Referenced by OpenDDS::XTypes::DynamicDataWriterImpl::unregister_instance_w_timestamp(), and OpenDDS::DCPS::DataWriterImpl_T< MessageType >::unregister_instance_w_timestamp().

 {
   const DDS::ReturnCode_t rc = instance_must_exist(
     "unregister_instance_w_timestamp", sample, instance_handle, /* remove = */ true);
   if (rc != DDS::RETCODE_OK) {
     return rc;
   }
   return unregister_instance_i(instance_handle, timestamp);
 }

◆ unregister_instances()

void OpenDDS::DCPS::DataWriterImpl::unregister_instances ( const DDS::Time_t & source_timestamp )

Unregister all registered instances and tell the transport to broadcast the unregistered instances.

Definition at line 1813 of file DataWriterImpl.cpp.

References ACE_GUARD, data_container_, sync_unreg_rem_assocs_lock_, and unregister_instance_i().

Referenced by prepare_to_delete().

 {
   ACE_GUARD(ACE_Thread_Mutex, guard, sync_unreg_rem_assocs_lock_);
 
   while (!this->data_container_->instances_.empty()) {
     this->unregister_instance_i(this->data_container_->instances_.begin()->first, source_timestamp);
   }
 }

◆ update_incompatible_qos()

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

virtual

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 854 of file DataWriterImpl.cpp.

 {
   DDS::DataWriterListener_var listener =
     listener_for(DDS::OFFERED_INCOMPATIBLE_QOS_STATUS);
 
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
 #if 0
 
   if (this->offered_incompatible_qos_status_.total_count == status.total_count) {
     // This test should make the method idempotent.
     return;
   }
 
 #endif
 
   set_status_changed_flag(DDS::OFFERED_INCOMPATIBLE_QOS_STATUS, true);
 
   // copy status and increment change
   offered_incompatible_qos_status_.total_count = status.total_count;
   offered_incompatible_qos_status_.total_count_change +=
     status.count_since_last_send;
   offered_incompatible_qos_status_.last_policy_id = status.last_policy_id;
   offered_incompatible_qos_status_.policies = status.policies;
 
   if (!CORBA::is_nil(listener.in())) {
     listener->on_offered_incompatible_qos(this, offered_incompatible_qos_status_);
 
     // TBD - Why does the spec say to change this but not change the
     //       ChangeFlagStatus after a listener call?
     offered_incompatible_qos_status_.total_count_change = 0;
   }
 
   notify_status_condition();
 }

◆ update_locators()

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

virtual

Reimplemented from OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 840 of file DataWriterImpl.cpp.

References ACE_GUARD, reader_info_, reader_info_lock_, and OpenDDS::DCPS::TransportClient::update_locators().

 {
   {
     ACE_GUARD(ACE_Thread_Mutex, reader_info_guard, reader_info_lock_);
     RepoIdToReaderInfoMap::const_iterator iter = reader_info_.find(readerId);
     if (iter == reader_info_.end()) {
       return;
     }
   }
   TransportClient::update_locators(readerId, locators);
 }

◆ update_subscription_params()

void OpenDDS::DCPS::DataWriterImpl::update_subscription_params	(	const GUID_t &	readerId,
		const DDS::StringSeq &	params
	)

virtual

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 891 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, LM_WARNING, lock_, publication_id_, reader_info_, reader_info_lock_, and TheServiceParticipant.

 {
 #ifdef OPENDDS_NO_CONTENT_FILTERED_TOPIC
   ACE_UNUSED_ARG(readerId);
   ACE_UNUSED_ARG(params);
 #else
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
   ACE_GUARD(ACE_Thread_Mutex, reader_info_guard, this->reader_info_lock_);
   RepoIdToReaderInfoMap::iterator iter = reader_info_.find(readerId);
 
   if (iter != reader_info_.end()) {
     iter->second.expression_params_ = params;
 
   } else if (DCPS_debug_level > 4 &&
              TheServiceParticipant->publisher_content_filter()) {
     ACE_DEBUG((LM_WARNING,
                ACE_TEXT("(%P|%t) WARNING: DataWriterImpl::update_subscription_params()")
                ACE_TEXT(" - writer: %C has no info about reader: %C\n"),
                LogGuid(this->publication_id_).c_str(), LogGuid(readerId).c_str()));
   }
 
 #endif
 }

◆ wait_for_acknowledgments()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::wait_for_acknowledgments ( const DDS::Duration_t & max_wait )

virtual

Definition at line 1089 of file DataWriterImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), create_ack_token(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::SequenceNumber::getValue(), DDS::ReliabilityQosPolicy::kind, LM_DEBUG, qos_, DDS::DataWriterQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, DDS::RETCODE_OK, send_request_ack(), OpenDDS::DCPS::DataWriterImpl::AckToken::sequence_, and wait_for_specific_ack().

 {
   if (this->qos_.reliability.kind != DDS::RELIABLE_RELIABILITY_QOS)
     return DDS::RETCODE_OK;
 
   DDS::ReturnCode_t ret = send_request_ack();
 
   if (ret != DDS::RETCODE_OK)
     return ret;
 
   DataWriterImpl::AckToken token = create_ack_token(max_wait);
   if (DCPS_debug_level) {
     ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) DataWriterImpl::wait_for_acknowledgments")
                           ACE_TEXT(" waiting for acknowledgment of sequence %q at %T\n"),
                           token.sequence_.getValue()));
   }
   return wait_for_specific_ack(token);
 }

◆ wait_for_specific_ack()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::wait_for_specific_ack ( const AckToken & token )

protected

Definition at line 1109 of file DataWriterImpl.cpp.

References data_container_, OpenDDS::DCPS::DataWriterImpl::AckToken::deadline(), OpenDDS::DCPS::DataWriterImpl::AckToken::deadline_is_infinite(), and OpenDDS::DCPS::DataWriterImpl::AckToken::sequence_.

Referenced by wait_for_acknowledgments().

 {
   return this->data_container_->wait_ack_of_seq(token.deadline(), token.deadline_is_infinite(), token.sequence_);
 }

◆ wait_pending()

void OpenDDS::DCPS::DataWriterImpl::wait_pending ( )

Wait for pending data and control messages to drain.

Definition at line 2706 of file DataWriterImpl.cpp.

References controlTracker, data_container_, OpenDDS::DCPS::TransportRegistry::instance(), OpenDDS::DCPS::MessageTracker::wait_messages_pending(), and wait_pending_deadline_.

Referenced by OpenDDS::DCPS::PublisherImpl::delete_datawriter().

 {
   if (!TransportRegistry::instance()->released()) {
     data_container_->wait_pending(wait_pending_deadline_);
     controlTracker.wait_messages_pending("DataWriterImpl::wait_pending", wait_pending_deadline_);
   }
 }

◆ write()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::write	(	Message_Block_Ptr	sample,
		DDS::InstanceHandle_t	handle,
		const DDS::Time_t &	source_timestamp,
		GUIDSeq *	filter_out,
		const void *	real_data
	)

Delegate to the WriteDataContainer to queue the instance sample and finally tell the transport to send the sample.

Parameters

filter_out can either be null (if the writer can't or won't evaluate the filters), or a list of associated reader GUID_ts that should NOT get the data sample due to content filtering.

Definition at line 1823 of file DataWriterImpl.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), available_data_list_, coherent_, coherent_samples_, create_sample_data_message(), data_container_, DBG_ENTRY_LVL, OpenDDS::DCPS::Observer::e_SAMPLE_SENT, OpenDDS::DCPS::EntityImpl::enabled_, OpenDDS::DCPS::SendStateDataSampleList::enqueue_tail(), OpenDDS::DCPS::DataSampleElement::get_header(), get_lock(), OpenDDS::DCPS::EntityImpl::get_observer(), get_unsent_data(), get_value_dispatcher(), OpenDDS::DCPS::DataSampleHeader::instance_state(), last_liveliness_activity_time_, LM_ERROR, lock_, max_suspended_transaction_id_, min_suspended_transaction_id_, OpenDDS::DCPS::move(), OpenDDS::DCPS::Observer::on_sample_sent(), publisher_servant_, ACE_Guard< ACE_LOCK >::release(), DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, DDS::RETCODE_TIMEOUT, OpenDDS::DCPS::TransportClient::send(), OpenDDS::DCPS::DataSampleHeader::sequence_, OpenDDS::DCPS::DataSampleElement::set_filter_out(), OpenDDS::DCPS::DataSampleElement::set_sample(), OpenDDS::DCPS::TimePoint_T< AceClock >::set_to_now(), and track_sequence_number().

Referenced by OpenDDS::DCPS::DataDurabilityCache::get_data(), and write_sample().

 {
   DBG_ENTRY_LVL("DataWriterImpl","write",6);
 
   ACE_Guard<ACE_Recursive_Thread_Mutex> guard(lock_);
 
   // take ownership of sequence allocated in FooDWImpl::write_w_timestamp()
   GUIDSeq_var filter_out_var(filter_out);
 
   if (!enabled_) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::write: ")
                       ACE_TEXT("Entity is not enabled.\n")),
                      DDS::RETCODE_NOT_ENABLED);
   }
 
   ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex,
                     dc_guard,
                     get_lock(),
                     DDS::RETCODE_ERROR);
 
   DataSampleElement* element = 0;
   DDS::ReturnCode_t ret = this->data_container_->obtain_buffer(element, handle);
 
   if (ret == DDS::RETCODE_TIMEOUT) {
     return ret; // silent for timeout
 
   } else if (ret != DDS::RETCODE_OK) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::write: ")
                       ACE_TEXT("obtain_buffer returned %d.\n"),
                       ret),
                      ret);
   }
 
   Message_Block_Ptr temp;
   ret = create_sample_data_message(move(data),
                                    handle,
                                    element->get_header(),
                                    temp,
                                    source_timestamp,
                                    (filter_out != 0));
   element->set_sample(move(temp));
 
   if (ret != DDS::RETCODE_OK) {
     data_container_->release_buffer(element);
     return ret;
   }
 
   element->set_filter_out(filter_out_var._retn()); // ownership passed to element
 
   ret = this->data_container_->enqueue(element, handle);
 
   if (ret != DDS::RETCODE_OK) {
     data_container_->release_buffer(element);
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("DataWriterImpl::write: ")
                       ACE_TEXT("enqueue failed.\n")),
                      ret);
   }
   last_liveliness_activity_time_.set_to_now();
 
   track_sequence_number(filter_out);
 
   if (this->coherent_) {
     ++this->coherent_samples_;
   }
   SendStateDataSampleList list;
 
   ACE_UINT64 transaction_id = this->get_unsent_data(list);
 
   RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();
   if (!publisher || publisher->is_suspended()) {
     if (min_suspended_transaction_id_ == 0) {
       //provides transaction id for lower bound of suspended transactions
       //or transaction id for single suspended write transaction
       min_suspended_transaction_id_ = transaction_id;
     } else {
       //when multiple write transactions have suspended, provides the upper bound
       //for suspended transactions.
       max_suspended_transaction_id_ = transaction_id;
     }
     this->available_data_list_.enqueue_tail(list);
 
   } else {
     dc_guard.release();
     guard.release();
     this->send(list, transaction_id);
   }
 
   const ValueDispatcher* vd = get_value_dispatcher();
   const Observer_rch observer = get_observer(Observer::e_SAMPLE_SENT);
   if (observer && real_data && vd) {
     Observer::Sample s(handle, element->get_header().instance_state(), source_timestamp, element->get_header().sequence_, real_data, *vd);
     observer->on_sample_sent(this, s);
   }
 
   return DDS::RETCODE_OK;
 }

◆ write_sample()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::write_sample	(	const Sample &	sample,
		DDS::InstanceHandle_t	handle,
		const DDS::Time_t &	source_timestamp,
		GUIDSeq *	filter_out
	)

Definition at line 3218 of file DataWriterImpl.cpp.

References ACE_ERROR, LM_NOTICE, OpenDDS::DCPS::log_level, OpenDDS::DCPS::move(), OpenDDS::DCPS::Sample::native_data(), OpenDDS::DCPS::LogLevel::Notice, OPENDDS_END_VERSIONED_NAMESPACE_DECL, DDS::RETCODE_ERROR, serialize_sample(), and write().

Referenced by write_w_timestamp().

 {
   Message_Block_Ptr serialized(serialize_sample(sample));
   if (!serialized) {
     if (log_level >= LogLevel::Notice) {
       ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: DataWriterImpl::write_sample: "
         "failed to serialize sample\n"));
     }
     return DDS::RETCODE_ERROR;
   }
 
   return write(move(serialized), handle, source_timestamp, filter_out, sample.native_data());
 }

◆ write_w_timestamp()

DDS::ReturnCode_t OpenDDS::DCPS::DataWriterImpl::write_w_timestamp	(	const Sample &	sample,
		DDS::InstanceHandle_t	handle,
		const DDS::Time_t &	source_timestamp
	)

protected

Definition at line 3170 of file DataWriterImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, OpenDDS::DCPS::Sample::eval(), OpenDDS::DCPS::DataWriterImpl::ReaderInfo::eval_, OpenDDS::DCPS::DataWriterImpl::ReaderInfo::expression_params_, filter_out(), get_or_create_instance_handle(), get_type_support(), DDS::HANDLE_NIL, LM_NOTICE, OpenDDS::DCPS::log_level, name, OpenDDS::DCPS::LogLevel::Notice, OpenDDS::DCPS::push_back(), reader_info_, reader_info_lock_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::retcode_to_string(), TheServiceParticipant, and write_sample().

Referenced by OpenDDS::XTypes::DynamicDataWriterImpl::write_w_timestamp(), and OpenDDS::DCPS::DataWriterImpl_T< MessageType >::write_w_timestamp().

 {
   // This operation assumes the provided handle is valid. The handle provided
   // will not be verified.
 
   if (handle == DDS::HANDLE_NIL) {
     DDS::InstanceHandle_t registered_handle = DDS::HANDLE_NIL;
     const DDS::ReturnCode_t ret =
       get_or_create_instance_handle(registered_handle, sample, source_timestamp);
     if (ret != DDS::RETCODE_OK) {
       if (log_level >= LogLevel::Notice) {
         ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: %CDataWriterImpl::write_w_timestamp: "
                    "register failed: %C\n",
                    get_type_support()->name(),
                    retcode_to_string(ret)));
       }
       return ret;
     }
 
     handle = registered_handle;
   }
 
   // list of reader GUID_ts that should not get data
   GUIDSeq_var filter_out;
 #ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC
   if (TheServiceParticipant->publisher_content_filter()) {
     ACE_GUARD_RETURN(ACE_Thread_Mutex, reader_info_guard, reader_info_lock_, DDS::RETCODE_ERROR);
     for (RepoIdToReaderInfoMap::iterator iter = reader_info_.begin(),
          end = reader_info_.end(); iter != end; ++iter) {
       const ReaderInfo& ri = iter->second;
       if (!ri.eval_.is_nil()) {
         if (!filter_out.ptr()) {
           filter_out = new OpenDDS::DCPS::GUIDSeq;
         }
         if (!sample.eval(*ri.eval_, ri.expression_params_)) {
           push_back(filter_out.inout(), iter->first);
         }
       }
     }
   }
 #endif
 
   return write_sample(sample, handle, source_timestamp, filter_out._retn());
 }

Friends And Related Function Documentation

◆ ::DDS_TEST

friend class ::DDS_TEST

friend

Definition at line 734 of file DataWriterImpl.h.

◆ PublisherImpl

friend class PublisherImpl

friend

Definition at line 89 of file DataWriterImpl.h.

◆ WriteDataContainer

friend class WriteDataContainer

friend

Definition at line 88 of file DataWriterImpl.h.

Referenced by enable().

Member Data Documentation

◆ association_chunk_multiplier_

size_t OpenDDS::DCPS::DataWriterImpl::association_chunk_multiplier_

protected

The multiplier for allocators affected by associations.

Definition at line 562 of file DataWriterImpl.h.

Referenced by enable().

◆ available_data_list_

SendStateDataSampleList OpenDDS::DCPS::DataWriterImpl::available_data_list_

private

Definition at line 831 of file DataWriterImpl.h.

Referenced by association_complete_i(), replay_durable_data_for(), send_suspended_data(), and write().

◆ coherent_

bool OpenDDS::DCPS::DataWriterImpl::coherent_

private

Flag indicating DataWriter current belongs to a coherent change set.

Definition at line 768 of file DataWriterImpl.h.

Referenced by begin_coherent_changes(), coherent_changes_pending(), create_sample_data_message(), end_coherent_changes(), and write().

◆ coherent_samples_

ACE_UINT32 OpenDDS::DCPS::DataWriterImpl::coherent_samples_

private

The number of samples belonging to the current coherent change set.

Definition at line 771 of file DataWriterImpl.h.

Referenced by end_coherent_changes(), and write().

◆ controlTracker

MessageTracker OpenDDS::DCPS::DataWriterImpl::controlTracker

Definition at line 429 of file DataWriterImpl.h.

Referenced by association_complete_i(), control_delivered(), control_dropped(), OpenDDS::DCPS::WriteDataContainer::data_delivered(), OpenDDS::DCPS::WriteDataContainer::data_dropped(), replay_durable_data_for(), send_all_to_flush_control(), send_control(), and wait_pending().

◆ data_allocator_

unique_ptr<DataAllocator> OpenDDS::DCPS::DataWriterImpl::data_allocator_

private

Definition at line 811 of file DataWriterImpl.h.

Referenced by serialize_sample(), and setup_serialization().

◆ data_container_

RcHandle<WriteDataContainer> OpenDDS::DCPS::DataWriterImpl::data_container_

private

The sample data container.

Definition at line 773 of file DataWriterImpl.h.

Referenced by association_complete_i(), create_sample_data_message(), data_delivered(), data_dropped(), dispose(), dispose_and_unregister(), enable(), get_handle_instance(), get_instance_handles(), num_samples(), persist_data(), register_instance_i(), remove_associations(), replay_durable_data_for(), send_request_ack(), set_qos(), unregister_all(), unregister_instance_i(), unregister_instances(), wait_for_specific_ack(), wait_pending(), and write().

◆ data_delivered_count_

Atomic<int> OpenDDS::DCPS::DataWriterImpl::data_delivered_count_

Definition at line 427 of file DataWriterImpl.h.

Referenced by data_delivered().

◆ data_dropped_count_

Atomic<int> OpenDDS::DCPS::DataWriterImpl::data_dropped_count_

Statistics counter.

Definition at line 426 of file DataWriterImpl.h.

Referenced by data_dropped().

◆ db_allocator_

unique_ptr<DataBlockAllocator> OpenDDS::DCPS::DataWriterImpl::db_allocator_

private

The data block allocator.

Definition at line 808 of file DataWriterImpl.h.

Referenced by create_control_message(), create_sample_data_message(), enable(), and serialize_sample().

◆ db_lock_pool_

unique_ptr<DataBlockLockPool> OpenDDS::DCPS::DataWriterImpl::db_lock_pool_

private

Definition at line 738 of file DataWriterImpl.h.

◆ domain_id_

DDS::DomainId_t OpenDDS::DCPS::DataWriterImpl::domain_id_

private

The domain id.

Definition at line 756 of file DataWriterImpl.h.

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

◆ dp_id_

GUID_t OpenDDS::DCPS::DataWriterImpl::dp_id_

private

Definition at line 757 of file DataWriterImpl.h.

Referenced by enable(), get_dp_id(), set_qos(), and transport_discovery_change().

◆ dynamic_type_

DDS::DynamicType_var OpenDDS::DCPS::DataWriterImpl::dynamic_type_

protected

Definition at line 869 of file DataWriterImpl.h.

Referenced by dispose_w_timestamp(), enable(), and get_or_create_instance_handle().

◆ encoding_mode_

class OpenDDS::DCPS::DataWriterImpl::EncodingMode OpenDDS::DCPS::DataWriterImpl::encoding_mode_

protected

Referenced by filter_out(), serialize_sample(), and setup_serialization().

◆ header_allocator_

unique_ptr<DataSampleHeaderAllocator> OpenDDS::DCPS::DataWriterImpl::header_allocator_

private

The header data allocator.

Definition at line 810 of file DataWriterImpl.h.

Referenced by create_sample_data_message(), and enable().

◆ id_to_handle_map_

RepoIdToHandleMap OpenDDS::DCPS::DataWriterImpl::id_to_handle_map_

private

Definition at line 779 of file DataWriterImpl.h.

Referenced by association_complete_i(), get_matched_subscriptions(), and remove_associations().

◆ instance_handles_to_values_

InstanceHandlesToValues OpenDDS::DCPS::DataWriterImpl::instance_handles_to_values_

private

Definition at line 858 of file DataWriterImpl.h.

Referenced by get_key_value(), insert_instance(), and instance_must_exist().

◆ instance_values_to_handles_

InstanceValuesToHandles OpenDDS::DCPS::DataWriterImpl::instance_values_to_handles_

private

Definition at line 860 of file DataWriterImpl.h.

Referenced by find_instance(), insert_instance(), instance_must_exist(), and lookup_instance().

◆ is_bit_

bool OpenDDS::DCPS::DataWriterImpl::is_bit_

private

Flag indicates that this datawriter is a builtin topic datawriter.

Definition at line 826 of file DataWriterImpl.h.

Referenced by add_association(), association_complete_i(), init(), notify_publication_disconnected(), notify_publication_lost(), notify_publication_reconnected(), and remove_associations().

◆ last_deadline_missed_total_count_

CORBA::Long OpenDDS::DCPS::DataWriterImpl::last_deadline_missed_total_count_

private

Total number of offered deadlines missed during last offered deadline status check.

Definition at line 822 of file DataWriterImpl.h.

Referenced by enable(), and get_offered_deadline_missed_status().

◆ last_liveliness_activity_time_

MonotonicTimePoint OpenDDS::DCPS::DataWriterImpl::last_liveliness_activity_time_

private

Timestamp of last write/dispose/assert_liveliness.

Definition at line 819 of file DataWriterImpl.h.

Referenced by handle_timeout(), participant_liveliness_activity_after(), send_liveliness(), and write().

◆ listener_

DDS::DataWriterListener_var OpenDDS::DCPS::DataWriterImpl::listener_

private

Used to notify the entity for relevant events.

Definition at line 754 of file DataWriterImpl.h.

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

◆ listener_mask_

DDS::StatusMask OpenDDS::DCPS::DataWriterImpl::listener_mask_

private

The StatusKind bit mask indicates which status condition change can be notified by the listener of this entity.

Definition at line 752 of file DataWriterImpl.h.

Referenced by listener_for(), and set_listener().

◆ listener_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::DataWriterImpl::listener_mutex_

private

Mutex to protect listener info.

Definition at line 749 of file DataWriterImpl.h.

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

◆ liveliness_asserted_

bool OpenDDS::DCPS::DataWriterImpl::liveliness_asserted_

private

Definition at line 848 of file DataWriterImpl.h.

Referenced by assert_liveliness_by_participant(), and handle_timeout().

◆ liveliness_check_interval_

TimeDuration OpenDDS::DCPS::DataWriterImpl::liveliness_check_interval_

private

The time interval for sending liveliness message.

Definition at line 817 of file DataWriterImpl.h.

Referenced by enable(), handle_timeout(), and liveliness_check_interval().

◆ liveliness_lost_

bool OpenDDS::DCPS::DataWriterImpl::liveliness_lost_

private

True if the writer failed to actively signal its liveliness within its offered liveliness period.

Definition at line 791 of file DataWriterImpl.h.

Referenced by handle_timeout().

◆ liveliness_lost_status_

DDS::LivelinessLostStatus OpenDDS::DCPS::DataWriterImpl::liveliness_lost_status_

private

Status conditions.

Definition at line 784 of file DataWriterImpl.h.

Referenced by DataWriterImpl(), get_liveliness_lost_status(), and handle_timeout().

◆ liveness_timer_

RcHandle<LivenessTimer> OpenDDS::DCPS::DataWriterImpl::liveness_timer_

private

Definition at line 853 of file DataWriterImpl.h.

Referenced by enable(), and handle_timeout().

◆ lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::DataWriterImpl::lock_

mutableprivate

The lock to protect the activate subscriptions and status changes.

Definition at line 776 of file DataWriterImpl.h.

Referenced by association_complete_i(), enable(), get_liveliness_lost_status(), get_matched_subscriptions(), get_offered_deadline_missed_status(), get_offered_incompatible_qos_status(), get_publication_matched_status(), get_readers(), handle_timeout(), participant_liveliness_activity_after(), remove_all_associations(), remove_associations(), transport_assoc_done(), transport_discovery_change(), update_incompatible_qos(), update_subscription_params(), and write().

◆ max_suspended_transaction_id_

ACE_UINT64 OpenDDS::DCPS::DataWriterImpl::max_suspended_transaction_id_

private

Definition at line 830 of file DataWriterImpl.h.

Referenced by send_suspended_data(), and write().

◆ mb_allocator_

unique_ptr<MessageBlockAllocator> OpenDDS::DCPS::DataWriterImpl::mb_allocator_

private

The message block allocator.

Todo:: The publication_lost_status_ and publication_reconnecting_status_ are left here for future use when we add get_publication_lost_status() and get_publication_reconnecting_status() methods.

Definition at line 806 of file DataWriterImpl.h.

Referenced by create_control_message(), create_sample_data_message(), enable(), and serialize_sample().

◆ min_suspended_transaction_id_

ACE_UINT64 OpenDDS::DCPS::DataWriterImpl::min_suspended_transaction_id_

private

The cached available data while suspending and associated transaction ids.

Definition at line 829 of file DataWriterImpl.h.

Referenced by send_suspended_data(), and write().

◆ monitor_

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

private

Monitor object for this entity.

Definition at line 834 of file DataWriterImpl.h.

Referenced by association_complete_i(), DataWriterImpl(), enable(), and register_instance_i().

◆ n_chunks_

size_t OpenDDS::DCPS::DataWriterImpl::n_chunks_

protected

The number of chunks for the cached allocator.

Definition at line 559 of file DataWriterImpl.h.

Referenced by enable(), and setup_serialization().

◆ offered_deadline_missed_status_

DDS::OfferedDeadlineMissedStatus OpenDDS::DCPS::DataWriterImpl::offered_deadline_missed_status_

private

Definition at line 785 of file DataWriterImpl.h.

Referenced by DataWriterImpl(), enable(), and get_offered_deadline_missed_status().

◆ offered_incompatible_qos_status_

DDS::OfferedIncompatibleQosStatus OpenDDS::DCPS::DataWriterImpl::offered_incompatible_qos_status_

private

Definition at line 786 of file DataWriterImpl.h.

Referenced by DataWriterImpl(), get_offered_incompatible_qos_status(), and update_incompatible_qos().

◆ participant_permissions_handle_

DDS::Security::PermissionsHandle OpenDDS::DCPS::DataWriterImpl::participant_permissions_handle_

protected

Definition at line 868 of file DataWriterImpl.h.

Referenced by dispose_w_timestamp(), enable(), and get_or_create_instance_handle().

◆ participant_servant_

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

protected

The participant servant which creats the publisher that creates this datawriter.

Definition at line 576 of file DataWriterImpl.h.

Referenced by add_association(), assert_liveliness(), association_complete_i(), enable(), get_builtin_subscriber_proxy(), get_crypto_handle(), get_instance_handle(), get_matched_subscription_data(), get_next_handle(), init(), lookup_instance_handles(), remove_associations(), return_handle(), and ~DataWriterImpl().

◆ passed_qos_

DDS::DataWriterQos OpenDDS::DCPS::DataWriterImpl::passed_qos_

protected

The qos policy passed in by the user. Differs from qos_ because representation has been interpreted.

Definition at line 572 of file DataWriterImpl.h.

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

◆ periodic_monitor_

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

private

Periodic Monitor object for this entity.

Definition at line 837 of file DataWriterImpl.h.

Referenced by DataWriterImpl().

◆ publication_id_

GUID_t OpenDDS::DCPS::DataWriterImpl::publication_id_

private

The repository id of this datawriter/publication.

Definition at line 761 of file DataWriterImpl.h.

Referenced by association_complete_i(), create_control_message(), create_sample_data_message(), data_delivered(), enable(), get_instance_handle(), remove_associations(), set_qos(), transport_assoc_done(), transport_discovery_change(), update_subscription_params(), and ~DataWriterImpl().

◆ publication_match_status_

DDS::PublicationMatchedStatus OpenDDS::DCPS::DataWriterImpl::publication_match_status_

private

Definition at line 787 of file DataWriterImpl.h.

Referenced by association_complete_i(), DataWriterImpl(), get_publication_matched_status(), and remove_associations().

◆ publisher_servant_

WeakRcHandle<PublisherImpl> OpenDDS::DCPS::DataWriterImpl::publisher_servant_

private

The publisher servant which creates this datawriter.

Definition at line 759 of file DataWriterImpl.h.

Referenced by association_complete_i(), create_control_message(), create_sample_data_message(), enable(), end_coherent_changes(), get_publisher(), init(), listener_for(), parent(), replay_durable_data_for(), retrieve_inline_qos_data(), set_qos(), and write().

◆ qos_

DDS::DataWriterQos OpenDDS::DCPS::DataWriterImpl::qos_

protected

The qos policy list of this datawriter.

Definition at line 569 of file DataWriterImpl.h.

Referenced by add_association(), assert_liveliness(), assert_liveliness_by_participant(), association_complete_i(), create_sample_data_message(), enable(), handle_timeout(), init(), liveliness_check_interval(), OpenDDS::DCPS::WriteDataContainer::obtain_buffer(), participant_liveliness_activity_after(), replay_durable_data_for(), retrieve_inline_qos_data(), send_liveliness(), set_qos(), setup_serialization(), unregister_instance_i(), and wait_for_acknowledgments().

◆ reactor_

ACE_Reactor_Timer_Interface* OpenDDS::DCPS::DataWriterImpl::reactor_

private

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

Definition at line 815 of file DataWriterImpl.h.

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

◆ reader_info_

RepoIdToReaderInfoMap OpenDDS::DCPS::DataWriterImpl::reader_info_

protected

Definition at line 597 of file DataWriterImpl.h.

Referenced by add_association(), association_complete_i(), create_control_message(), need_sequence_repair_i(), remove_associations(), replay_durable_data_for(), track_sequence_number(), update_locators(), update_subscription_params(), and write_w_timestamp().

◆ reader_info_lock_

ACE_Thread_Mutex OpenDDS::DCPS::DataWriterImpl::reader_info_lock_

protected

Definition at line 579 of file DataWriterImpl.h.

Referenced by add_association(), association_complete_i(), create_control_message(), need_sequence_repair(), remove_associations(), replay_durable_data_for(), track_sequence_number(), update_locators(), update_subscription_params(), and write_w_timestamp().

◆ readers_

RepoIdSet OpenDDS::DCPS::DataWriterImpl::readers_

private

Definition at line 781 of file DataWriterImpl.h.

Referenced by association_complete_i(), get_readers(), remove_all_associations(), remove_associations(), and should_ack().

◆ security_config_

Security::SecurityConfig_rch OpenDDS::DCPS::DataWriterImpl::security_config_

protected

Definition at line 867 of file DataWriterImpl.h.

Referenced by dispose_w_timestamp(), enable(), and get_or_create_instance_handle().

◆ sequence_number_

SequenceNumber OpenDDS::DCPS::DataWriterImpl::sequence_number_

private

The sequence number unique in DataWriter scope.

Definition at line 763 of file DataWriterImpl.h.

Referenced by create_control_message(), and need_sequence_repair_i().

◆ skip_serialize_

bool OpenDDS::DCPS::DataWriterImpl::skip_serialize_

protected

Definition at line 608 of file DataWriterImpl.h.

Referenced by serialize_sample().

◆ sn_lock_

ACE_Thread_Mutex OpenDDS::DCPS::DataWriterImpl::sn_lock_

mutableprivate

Mutex for sequence_number_.

Definition at line 765 of file DataWriterImpl.h.

Referenced by create_control_message(), and create_sample_data_message().

◆ sync_unreg_rem_assocs_lock_

ACE_Thread_Mutex OpenDDS::DCPS::DataWriterImpl::sync_unreg_rem_assocs_lock_

private

Definition at line 852 of file DataWriterImpl.h.

Referenced by remove_associations(), and unregister_instances().

◆ topic_id_

GUID_t OpenDDS::DCPS::DataWriterImpl::topic_id_

private

The associated topic repository id.

Definition at line 743 of file DataWriterImpl.h.

Referenced by init().

◆ topic_name_

CORBA::String_var OpenDDS::DCPS::DataWriterImpl::topic_name_

private

The name of associated topic.

Definition at line 741 of file DataWriterImpl.h.

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

◆ topic_servant_

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

private

The topic servant.

Definition at line 745 of file DataWriterImpl.h.

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

◆ type_name_

CORBA::String_var OpenDDS::DCPS::DataWriterImpl::type_name_

protected

The type name of associated topic.

Definition at line 566 of file DataWriterImpl.h.

Referenced by get_type_name(), and init().

◆ type_support_

TypeSupportImpl* OpenDDS::DCPS::DataWriterImpl::type_support_

private

Definition at line 746 of file DataWriterImpl.h.

Referenced by cleanup(), enable(), filter_out(), init(), serialize_sample(), and setup_serialization().

◆ wait_pending_deadline_

MonotonicTimePoint OpenDDS::DCPS::DataWriterImpl::wait_pending_deadline_

private

Definition at line 855 of file DataWriterImpl.h.

Referenced by set_wait_pending_deadline(), and wait_pending().

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

Classes

Public Types

Public Member Functions

Public Attributes

Protected Member Functions

Protected Attributes

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ DataAllocator

Constructor & Destructor Documentation

◆ DataWriterImpl()

◆ ~DataWriterImpl()

Member Function Documentation

◆ add_association()

◆ assert_liveliness()

◆ assert_liveliness_by_participant()

◆ association_complete_i()

◆ begin_coherent_changes()

◆ check_and_set_repo_id()

◆ check_transport_qos()

◆ cleanup()

◆ coherent_changes_pending()

◆ control_delivered()

◆ control_dropped()

◆ create_ack_token()

◆ create_control_message()

◆ create_sample_data_message()

◆ data_allocator()

◆ data_delivered()

◆ data_dropped()

◆ dispose()

◆ dispose_and_unregister()

◆ dispose_w_timestamp()

◆ domain_id()

◆ enable()

◆ end_coherent_changes()

◆ filter_out()

◆ find_instance()

◆ get_builtin_subscriber_proxy()

◆ get_crypto_handle()

◆ get_db_lock()

◆ get_dp_id()

◆ get_ext_listener()

◆ get_guid()

◆ get_handle_instance()

◆ get_ice_endpoint()

◆ get_instance_handle()

◆ get_instance_handles()

◆ get_key_value()

◆ get_listener()

◆ get_liveliness_lost_status()

◆ get_lock()

◆ get_marshal_skip_serialize()

◆ get_matched_subscription_data()

◆ get_matched_subscriptions()

◆ get_max_sn()

◆ get_next_handle()

◆ get_next_sn()

◆ get_next_sn_i()

◆ get_offered_deadline_missed_status()

◆ get_offered_incompatible_qos_status()

◆ get_or_create_instance_handle()

◆ get_priority_value()

◆ get_publication_matched_status()

◆ get_publisher()

◆ get_qos()

◆ get_readers()

◆ get_resend_data()

◆ get_topic()

◆ get_type_name()

◆ get_type_support()

◆ get_unsent_data()

◆ get_value_dispatcher()

◆ handle_timeout()

◆ init()

◆ insert_instance()