Implementation of Replayer functionality. More...

#include <ReplayerImpl.h>

Inheritance diagram for OpenDDS::DCPS::ReplayerImpl:

[legend]

Collaboration diagram for OpenDDS::DCPS::ReplayerImpl:

[legend]

Classes
struct	ReaderInfo

Public Member Functions
	ReplayerImpl ()

	~ReplayerImpl ()

DDS::ReturnCode_t	cleanup ()

virtual void	init (DDS::Topic_ptr topic, TopicImpl topic_servant, const DDS::DataWriterQos &qos, ReplayerListener_rch a_listener, const DDS::StatusMask &mask, OpenDDS::DCPS::DomainParticipantImpl participant_servant, const DDS::PublisherQos &publisher_qos)

virtual DDS::ReturnCode_t	write (const RawDataSample &sample)

virtual DDS::ReturnCode_t	write_to_reader (DDS::InstanceHandle_t subscription, const RawDataSample &sample)

virtual DDS::ReturnCode_t	write_to_reader (DDS::InstanceHandle_t subscription, const RawDataSampleList &samples)

virtual DDS::ReturnCode_t	set_qos (const DDS::PublisherQos &publisher_qos, const DDS::DataWriterQos &datawriter_qos)

virtual DDS::ReturnCode_t	get_qos (DDS::PublisherQos &publisher_qos, DDS::DataWriterQos &datawriter_qos)

virtual DDS::ReturnCode_t	set_listener (const ReplayerListener_rch &a_listener, DDS::StatusMask mask)

virtual ReplayerListener_rch	get_listener ()

virtual bool	check_transport_qos (const TransportInst &inst)

virtual GUID_t	get_guid () const

DDS::DomainId_t	domain_id () const

virtual CORBA::Long	get_priority_value (const AssociationData &data) const

SequenceNumber	get_max_sn () const

virtual void	data_delivered (const DataSampleElement *sample)

virtual void	data_dropped (const DataSampleElement *sample, bool dropped_by_transport)

virtual void	control_delivered (const Message_Block_Ptr &sample)

virtual void	control_dropped (const Message_Block_Ptr &sample, bool dropped_by_transport)

virtual void	notify_publication_disconnected (const ReaderIdSeq &subids)

virtual void	notify_publication_reconnected (const ReaderIdSeq &subids)

virtual void	notify_publication_lost (const ReaderIdSeq &subids)

virtual void	retrieve_inline_qos_data (InlineQosData &qos_data) const

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

virtual void	remove_associations (const ReaderIdSeq &readers, CORBA::Boolean callback)

virtual void	replay_durable_data_for (const GUID_t &)

virtual void	update_incompatible_qos (const IncompatibleQosStatus &status)

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

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 DCPS::WeakRcHandle< ICE::Endpoint >	get_ice_endpoint ()

DDS::ReturnCode_t	enable ()

DomainParticipantImpl *	participant ()

virtual DDS::InstanceHandle_t	get_instance_handle ()

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

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

virtual RcHandle< BitSubscriber >	get_builtin_subscriber_proxy () const

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)

virtual void	transport_discovery_change ()

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

virtual	~DataWriterCallbacks ()

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

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

virtual RcHandle< EntityImpl >	parent () const

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

Observer_rch	get_observer (Observer::Event e)

Public Attributes
int	data_dropped_count_
	Statistics counter. More...

int	data_delivered_count_

Private Member Functions
void	notify_publication_lost (const DDS::InstanceHandleSeq &handles)

DDS::ReturnCode_t	write (const RawDataSample sample_array, int array_size, DDS::InstanceHandle_t reader)

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

bool	need_sequence_repair () const

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

typedef	OPENDDS_MAP_CMP (GUID_t, ReaderInfo, GUID_tKeyLessThan) RepoIdToReaderInfoMap

void	association_complete_i (const GUID_t &remote_id)

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

typedef	OPENDDS_MAP_CMP (GUID_t, SequenceNumber, GUID_tKeyLessThan) RepoIdToSequenceMap

Private 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_

DomainParticipantImpl *	participant_servant_

RepoIdToReaderInfoMap	reader_info_

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

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

DDS::Topic_var	topic_objref_
	The object reference of the associated topic. More...

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

DDS::StatusMask	listener_mask_

ReplayerListener_rch	listener_
	Used to notify the entity for relevant events. More...

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

PublisherImpl *	publisher_servant_
	The publisher servant which creates this datawriter. More...

DDS::PublisherQos	publisher_qos_

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

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

ACE_Recursive_Thread_Mutex	lock_
	The sample data container. More...

RepoIdToHandleMap	id_to_handle_map_

RepoIdSet	readers_

DDS::OfferedIncompatibleQosStatus	offered_incompatible_qos_status_
	Status conditions. More...

DDS::PublicationMatchedStatus	publication_match_status_

unique_ptr< MessageBlockAllocator >	mb_allocator_

unique_ptr< DataBlockAllocator >	db_allocator_

unique_ptr< DataSampleHeaderAllocator >	header_allocator_

unique_ptr< DataSampleElementAllocator >	sample_list_element_allocator_

bool	is_bit_
	The time interval for sending liveliness message. More...

RepoIdToSequenceMap	idToSequence_

ConditionVariable< ACE_Recursive_Thread_Mutex >	empty_condition_

int	pending_write_count_

Friends
class	::DDS_TEST

Additional Inherited Members
Public Types inherited from OpenDDS::DCPS::Replayer
typedef Replayer_ptr	_ptr_type

typedef Replayer_var	_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::TransportClient
enum	{ ASSOC_OK = 1, ASSOC_ACTIVE = 2 }

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

typedef DDS::Entity ::_var_type	_var_type

Static Public Member Functions inherited from OpenDDS::DCPS::Replayer
static Replayer_ptr	_duplicate (Replayer_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)

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::TransportClient
void	cdr_encapsulation (bool encap)

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

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

void	set_deleted (bool state)

bool	get_deleted () const

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

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

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

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

Detailed Description

Implementation of Replayer functionality.

This class is the implementation of the Replayer. Inheritance is used to limit the applications access to underlying system methods.

Definition at line 61 of file ReplayerImpl.h.

Constructor & Destructor Documentation

◆ ReplayerImpl()

OpenDDS::DCPS::ReplayerImpl::ReplayerImpl ( )

Definition at line 52 of file ReplayerImpl.cpp.

References DDS::PublicationMatchedStatus::current_count, DDS::PublicationMatchedStatus::current_count_change, DDS::HANDLE_NIL, DDS::OfferedIncompatibleQosStatus::last_policy_id, DDS::PublicationMatchedStatus::last_subscription_handle, offered_incompatible_qos_status_, DDS::OfferedIncompatibleQosStatus::policies, publication_match_status_, DDS::OfferedIncompatibleQosStatus::total_count, DDS::PublicationMatchedStatus::total_count, DDS::OfferedIncompatibleQosStatus::total_count_change, and DDS::PublicationMatchedStatus::total_count_change.

   : data_dropped_count_(0),
   data_delivered_count_(0),
   n_chunks_(TheServiceParticipant->n_chunks()),
   association_chunk_multiplier_(TheServiceParticipant->association_chunk_multiplier()),
   qos_(TheServiceParticipant->initial_DataWriterQos()),
   participant_servant_(0),
   topic_id_(GUID_UNKNOWN),
   topic_servant_(0),
   listener_mask_(DEFAULT_STATUS_MASK),
   domain_id_(0),
   publisher_servant_(0),
   publication_id_(GUID_UNKNOWN),
   sequence_number_(SequenceNumber::SEQUENCENUMBER_UNKNOWN()),
   // data_container_(0),
   // liveliness_lost_(false),
   // last_deadline_missed_total_count_(0),
   is_bit_(false),
   empty_condition_(lock_),
   pending_write_count_(0)
 {
   // 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;
 
 }

◆ ~ReplayerImpl()

OpenDDS::DCPS::ReplayerImpl::~ReplayerImpl ( )

Definition at line 95 of file ReplayerImpl.cpp.

References DBG_ENTRY_LVL.

 {
   DBG_ENTRY_LVL("ReplayerImpl","~ReplayerImpl",6);
 }

Member Function Documentation

◆ add_association()

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

virtual

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 401 of file ReplayerImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::TransportClient::associate(), association_complete_i(), OpenDDS::DCPS::LogGuid::c_str(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::AssociationData::discovery_locator_, DDS::DataReaderQos::durability, OpenDDS::DCPS::ReaderAssociation::exprParams, OpenDDS::DCPS::ReaderAssociation::filterExpression, OpenDDS::DCPS::GUID_UNKNOWN, is_bit_, DDS::DurabilityQosPolicy::kind, DDS::ReliabilityQosPolicy::kind, LM_DEBUG, LM_ERROR, lock_, participant_servant_, publication_id_, qos_, reader_info_, OpenDDS::DCPS::ReaderAssociation::readerDiscInfo, OpenDDS::DCPS::ReaderAssociation::readerId, OpenDDS::DCPS::ReaderAssociation::readerQos, OpenDDS::DCPS::ReaderAssociation::readerTransInfo, DDS::DataReaderQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, OpenDDS::DCPS::AssociationData::remote_data_, OpenDDS::DCPS::AssociationData::remote_durable_, OpenDDS::DCPS::AssociationData::remote_id_, OpenDDS::DCPS::AssociationData::remote_reliable_, OpenDDS::DCPS::AssociationData::remote_transport_context_, TheServiceParticipant, DDS::DataWriterQos::transport_priority, OpenDDS::DCPS::ReaderAssociation::transportContext, DDS::TransportPriorityQosPolicy::value, and DDS::VOLATILE_DURABILITY_QOS.

 {
   DBG_ENTRY_LVL("ReplayerImpl", "add_association", 6);
 
   if (DCPS_debug_level >= 1) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) ReplayerImpl::add_association - ")
                ACE_TEXT("bit %d local %C remote %C\n"),
                is_bit_,
                LogGuid(yourId).c_str(),
                LogGuid(reader.readerId).c_str()));
   }
 
   // if (entity_deleted_) {
   //   if (DCPS_debug_level >= 1)
   //     ACE_DEBUG((LM_DEBUG,
   //                ACE_TEXT("(%P|%t) ReplayerImpl::add_association")
   //                ACE_TEXT(" This is a deleted datawriter, ignoring add.\n")));
   //
   //   return;
   // }
 
   if (GUID_UNKNOWN == publication_id_) {
     publication_id_ = yourId;
   }
 
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
     reader_info_.insert(std::make_pair(reader.readerId,
                                        ReaderInfo(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) ReplayerImpl::add_association(): ")
                ACE_TEXT("adding subscription to publication %C with priority %d.\n"),
                LogGuid(publication_id_).c_str(),
                qos_.transport_priority.value));
   }
 
   AssociationData data;
   data.remote_id_ = reader.readerId;
   data.remote_data_ = reader.readerTransInfo;
   data.discovery_locator_ = reader.readerDiscInfo;
   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 (!this->associate(data, active)) {
     //FUTURE: inform inforepo and try again as passive peer
     if (DCPS_debug_level) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ReplayerImpl::add_association: ")
                  ACE_TEXT("ERROR: transport layer failed to associate.\n")));
     }
     return;
   }
 
   if (active) {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
     association_complete_i(reader.readerId);
   }
 }

◆ association_complete_i()

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

private

Definition at line 491 of file ReplayerImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DomainParticipantImpl::assign_handle(), OpenDDS::DCPS::bind(), DDS::PublicationMatchedStatus::current_count, DDS::PublicationMatchedStatus::current_count_change, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, id_to_handle_map_, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::insert(), is_bit_, DDS::PublicationMatchedStatus::last_subscription_handle, listener_, LM_DEBUG, LM_ERROR, LM_WARNING, lock_, OpenDDS::DCPS::ReplayerListener::on_replayer_matched(), participant_servant_, publication_match_status_, readers_, DDS::PublicationMatchedStatus::total_count, and DDS::PublicationMatchedStatus::total_count_change.

Referenced by add_association().

 {
   DBG_ENTRY_LVL("ReplayerImpl", "association_complete_i", 6);
   // bool reader_durable = false;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
     if (OpenDDS::DCPS::insert(readers_, remote_id) == -1) {
       ACE_ERROR((LM_ERROR,
                  ACE_TEXT("(%P|%t) ERROR: ReplayerImpl::association_complete_i: ")
                  ACE_TEXT("insert %C from pending failed.\n"),
                  LogGuid(remote_id).c_str()));
     }
     // RepoIdToReaderInfoMap::const_iterator it = reader_info_.find(remote_id);
     // if (it != reader_info_.end()) {
     //   reader_durable = it->second.durable_;
     // }
   }
 
   if (!is_bit_) {
 
     const DDS::InstanceHandle_t handle = participant_servant_->assign_handle(remote_id);
 
     {
       // protect publication_match_status_ and status changed flags.
       ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
       // update the publication_match_status_
       ++publication_match_status_.total_count;
       ++publication_match_status_.total_count_change;
       ++publication_match_status_.current_count;
       ++publication_match_status_.current_count_change;
 
       if (OpenDDS::DCPS::bind(id_to_handle_map_, remote_id, handle) != 0) {
         ACE_DEBUG((LM_WARNING,
                    ACE_TEXT("(%P|%t) ERROR: ReplayerImpl::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) ReplayerImpl::association_complete_i: ")
                    ACE_TEXT("id_to_handle_map_%C = 0x%x.\n"),
                    LogGuid(remote_id).c_str(),
                    handle));
       }
 
       publication_match_status_.last_subscription_handle = handle;
 
     }
 
 
     if (listener_.in()) {
       listener_->on_replayer_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;
     }
 
   }
 
 }

◆ check_transport_qos()

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

virtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 763 of file ReplayerImpl.cpp.

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

◆ cleanup()

DDS::ReturnCode_t OpenDDS::DCPS::ReplayerImpl::cleanup ( void )

cleanup the DataWriter.

Definition at line 102 of file ReplayerImpl.cpp.

References ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_TEXT(), domain_id_, empty_condition_, LM_ERROR, lock_, pending_write_count_, publication_id_, remove_all_associations(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, TheServiceParticipant, topic_objref_, topic_servant_, and OpenDDS::DCPS::ConditionVariable< Mutex >::wait().

Referenced by OpenDDS::DCPS::Service_Participant::delete_replayer(), and OpenDDS::DCPS::DomainParticipantImpl::handle_exception().

 {
 
   //     // Unregister all registered instances prior to deletion.
   //     // this->unregister_instances(SystemTimePoint::now().to_dds_time());
   //
   //     // CORBA::String_var topic_name = this->get_Atopic_name();
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, DDS::RETCODE_ERROR);
 
     // Wait for pending samples to drain prior to removing associations
     // and unregistering the publication.
     ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
     while (this->pending_write_count_) {
       this->empty_condition_.wait(thread_status_manager);
     }
 
     // Call remove association before unregistering the datawriter
     // with the transport, otherwise some callbacks resulted from
     // remove_association may lost.
     this->remove_all_associations();
 
     // release our Topic_var
     topic_objref_ = DDS::Topic::_nil();
     topic_servant_ = 0;
 
   }
 
   // not just unregister but remove any pending writes/sends.
   // this->unregister_all();
 
   Discovery_rch disco = TheServiceParticipant->get_discovery(this->domain_id_);
   if (!disco->remove_publication(
         this->domain_id_,
         this->participant_servant_->get_id(),
         this->publication_id_)) {
     ACE_ERROR_RETURN((LM_ERROR,
                       ACE_TEXT("(%P|%t) ERROR: ")
                       ACE_TEXT("PublisherImpl::delete_datawriter, ")
                       ACE_TEXT("publication not removed from discovery.\n")),
                      DDS::RETCODE_ERROR);
   }
   return DDS::RETCODE_OK;
 }

◆ control_delivered()

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

virtual

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 808 of file ReplayerImpl.cpp.

 {
   ACE_UNUSED_ARG(sample);
 }

◆ control_dropped()

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

virtual

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 832 of file ReplayerImpl.cpp.

 {
   ACE_UNUSED_ARG(sample);
 }

◆ create_sample_data_message()

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

private

Definition at line 952 of file ReplayerImpl.cpp.

Referenced by write().

 {
   header_data.message_id_ = SAMPLE_DATA;
   header_data.coherent_change_ = content_filter;
 
   header_data.content_filter_ = false;
   header_data.cdr_encapsulation_ = this->cdr_encapsulation();
   header_data.message_length_ = static_cast<ACE_UINT32>(data->total_length());
   header_data.sequence_repair_ = need_sequence_repair();
   if (this->sequence_number_ == SequenceNumber::SEQUENCENUMBER_UNKNOWN()) {
     this->sequence_number_ = SequenceNumber();
   } else {
     ++this->sequence_number_;
   }
   header_data.sequence_ = this->sequence_number_;
   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_ = this->publisher_servant_->publisher_id_;
   ACE_Message_Block* tmp;
   ACE_NEW_MALLOC_RETURN(tmp,
                         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
                                           0,   //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 << header_data;
   return DDS::RETCODE_OK;
 }

◆ data_delivered()

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

virtual

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 782 of file ReplayerImpl.cpp.

References ACE_DES_FREE, ACE_ERROR, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), data_delivered_count_, DBG_ENTRY_LVL, empty_condition_, OpenDDS::DCPS::DataSampleElement::get_pub_id(), LM_ERROR, lock_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), pending_write_count_, publication_id_, and sample_list_element_allocator_.

 {
   DBG_ENTRY_LVL("ReplayerImpl","data_delivered",6);
   if (!(sample->get_pub_id() == this->publication_id_)) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: ReplayerImpl::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;
   }
   DataSampleElement* elem = const_cast<DataSampleElement*>(sample);
   // this->data_container_->data_delivered(sample);
   ACE_DES_FREE(elem, sample_list_element_allocator_->free, DataSampleElement);
   ++data_delivered_count_;
 
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
     if (--pending_write_count_ == 0) {
       empty_condition_.notify_all();
     }
   }
 }

◆ data_dropped()

void OpenDDS::DCPS::ReplayerImpl::data_dropped	(	const DataSampleElement *	sample,
		bool	dropped_by_transport
	)

virtual

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 814 of file ReplayerImpl.cpp.

References ACE_DES_FREE, ACE_GUARD, data_dropped_count_, DBG_ENTRY_LVL, empty_condition_, lock_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), pending_write_count_, and sample_list_element_allocator_.

 {
   DBG_ENTRY_LVL("ReplayerImpl","data_dropped",6);
   // this->data_container_->data_dropped(element, dropped_by_transport);
   ACE_UNUSED_ARG(dropped_by_transport);
   DataSampleElement* elem = const_cast<DataSampleElement*>(sample);
   ACE_DES_FREE(elem, sample_list_element_allocator_->free, DataSampleElement);
   ++data_dropped_count_;
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
     if ((--pending_write_count_) == 0) {
       empty_condition_.notify_all();
     }
   }
 }

◆ domain_id()

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

inlinevirtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 105 of file ReplayerImpl.h.

105 { return this->domain_id_; }

OpenDDS::DCPS::ReplayerImpl::domain_id_

DDS::DomainId_t domain_id_

The domain id.

Definition: ReplayerImpl.h:232

◆ enable()

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

Implements DDS::Entity.

Definition at line 298 of file ReplayerImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), association_chunk_multiplier_, OpenDDS::DCPS::TransportClient::cdr_encapsulation(), OpenDDS::XTypes::TypeInformation::complete, OpenDDS::DCPS::TransportClient::connection_info(), db_allocator_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::XTypes::TypeIdentifierWithDependencies::dependent_typeid_count, domain_id_, DDS::DataWriterQos::durability, OpenDDS::DCPS::TransportClient::enable_transport(), OpenDDS::DCPS::DomainParticipantImpl::get_id(), OpenDDS::DCPS::GUID_UNKNOWN, header_allocator_, OpenDDS::DCPS::EntityImpl::is_enabled(), DDS::DurabilityQosPolicy::kind, DDS::ReliabilityQosPolicy::kind, DDS::LENGTH_UNLIMITED, LM_DEBUG, LM_ERROR, DDS::ResourceLimitsQosPolicy::max_samples, mb_allocator_, OpenDDS::XTypes::TypeInformation::minimal, n_chunks_, participant_servant_, publication_id_, publisher_qos_, qos_, OpenDDS::DCPS::rchandle_from(), DDS::DataWriterQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, DDS::DataWriterQos::representation, DDS::DataWriterQos::resource_limits, DDS::RETCODE_ERROR, DDS::RETCODE_OK, sample_list_element_allocator_, OpenDDS::DCPS::EntityImpl::set_enabled(), OpenDDS::DCPS::set_writer_effective_data_rep_qos(), TheServiceParticipant, topic_servant_, OpenDDS::XTypes::TypeIdentifierWithDependencies::typeid_with_size, OpenDDS::XTypes::TypeIdentifierWithSize::typeobject_serialized_size, DDS::DataRepresentationQosPolicy::value, and DDS::VOLATILE_DURABILITY_QOS.

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

 {
   //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;
   }
 
   // if (!this->publisher_servant_->is_enabled()) {
   //   return DDS::RETCODE_PRECONDITION_NOT_MET;
   // }
   //
   const bool reliable = qos_.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS;
 
   if (qos_.resource_limits.max_samples != DDS::LENGTH_UNLIMITED) {
     n_chunks_ = qos_.resource_limits.max_samples;
   }
   // +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));
 
   sample_list_element_allocator_.reset(new DataSampleElementAllocator(2 * n_chunks_));
 
 
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) ReplayerImpl::enable-mb"
                " Cached_Allocator_With_Overflow %x with %d chunks\n",
                mb_allocator_.get(),
                n_chunks_));
 
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) ReplayerImpl::enable-db"
                " Cached_Allocator_With_Overflow %x with %d chunks\n",
                db_allocator_.get(),
                n_chunks_));
 
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) ReplayerImpl::enable-header"
                " Cached_Allocator_With_Overflow %x with %d chunks\n",
                header_allocator_.get(),
                n_chunks_));
   }
 
   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: ReplayerImpl::enable, ")
                ACE_TEXT("Transport Exception.\n")));
     return DDS::RETCODE_ERROR;
 
   }
 
   const TransportLocatorSeq& trans_conf_info = connection_info();
 
 
   Discovery_rch disco = TheServiceParticipant->get_discovery(this->domain_id_);
 
   set_writer_effective_data_rep_qos(qos_.representation.value, cdr_encapsulation());
   if (!topic_servant_->check_data_representation(qos_.representation.value, true)) {
     return DDS::RETCODE_ERROR;
   }
 
   XTypes::TypeInformation type_info;
   type_info.minimal.typeid_with_size.typeobject_serialized_size = 0;
   type_info.minimal.dependent_typeid_count = 0;
   type_info.complete.typeid_with_size.typeobject_serialized_size = 0;
   type_info.complete.dependent_typeid_count = 0;
 
   this->publication_id_ =
     disco->add_publication(this->domain_id_,
                            this->participant_servant_->get_id(),
                            this->topic_servant_->get_id(),
                            rchandle_from(this),
                            this->qos_,
                            trans_conf_info,
                            this->publisher_qos_,
                            type_info);
 
   if (this->publication_id_ == GUID_UNKNOWN) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: ReplayerImpl::enable, ")
                ACE_TEXT("add_publication returned invalid id.\n")));
     return DDS::RETCODE_ERROR;
   }
 
   return DDS::RETCODE_OK;
 }

◆ get_guid()

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

virtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 770 of file ReplayerImpl.cpp.

References publication_id_.

 {
   return this->publication_id_;
 }

◆ get_ice_endpoint()

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

inlinevirtual

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 157 of file ReplayerImpl.h.

157 { return DCPS::WeakRcHandle<ICE::Endpoint>(); }

◆ get_instance_handle()

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

virtual

Implements OpenDDS::DCPS::EntityImpl.

Definition at line 1045 of file ReplayerImpl.cpp.

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

 {
   return get_entity_instance_handle(publication_id_, rchandle_from(participant_servant_));
 }

◆ get_listener()

ReplayerListener_rch OpenDDS::DCPS::ReplayerImpl::get_listener ( )

virtual

Get the listener for this Replayer.

Implements OpenDDS::DCPS::Replayer.

Definition at line 292 of file ReplayerImpl.cpp.

References listener_.

 {
   return listener_;
 }

◆ get_max_sn()

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

inlinevirtual

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 107 of file ReplayerImpl.h.

107 { return sequence_number_; }

OpenDDS::DCPS::ReplayerImpl::sequence_number_

SequenceNumber sequence_number_

The sequence number unique in DataWriter scope.

Definition: ReplayerImpl.h:240

◆ get_priority_value()

CORBA::Long OpenDDS::DCPS::ReplayerImpl::get_priority_value ( const AssociationData & data ) const

virtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 776 of file ReplayerImpl.cpp.

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

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

◆ get_qos()

DDS::ReturnCode_t OpenDDS::DCPS::ReplayerImpl::get_qos	(	DDS::PublisherQos &	publisher_qos,
		DDS::DataWriterQos &	datawriter_qos
	)

virtual

Get the Quality of Service settings for the Replayer.

Implements OpenDDS::DCPS::Replayer.

Definition at line 275 of file ReplayerImpl.cpp.

References passed_qos_, publisher_qos_, and DDS::RETCODE_OK.

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

◆ init()

void OpenDDS::DCPS::ReplayerImpl::init	(	DDS::Topic_ptr	topic,
		TopicImpl *	topic_servant,
		const DDS::DataWriterQos &	qos,
		ReplayerListener_rch	a_listener,
		const DDS::StatusMask &	mask,
		OpenDDS::DCPS::DomainParticipantImpl *	participant_servant,
		const DDS::PublisherQos &	publisher_qos
	)

virtual

Initialize the data members.

Definition at line 148 of file ReplayerImpl.cpp.

References DBG_ENTRY_LVL, domain_id_, OpenDDS::DCPS::DomainParticipantImpl::get_domain_id(), TAO::String_var< charT >::in(), is_bit_, listener_, listener_mask_, participant_servant_, passed_qos_, publisher_qos_, qos_, topic_id_, topic_name_, topic_objref_, topic_servant_, OpenDDS::DCPS::topicIsBIT(), and type_name_.

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

 {
   DBG_ENTRY_LVL("ReplayerImpl","init",6);
   topic_objref_ = DDS::Topic::_duplicate(topic);
   topic_servant_ = topic_servant;
   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;
 
   //Note: OK to _duplicate(nil).
   listener_ = a_listener;
   listener_mask_ = mask;
 
   // Only store the participant pointer, since it is our "grand"
   // parent, we will exist as long as it does.
   participant_servant_ = participant_servant;
   domain_id_ = participant_servant_->get_domain_id();
 
   publisher_qos_ = publisher_qos;
 }

◆ lookup_instance_handles()

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

private

Lookup the instance handles by the subscription repo ids.

Definition at line 1005 of file ReplayerImpl.cpp.

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

Referenced by remove_associations().

 {
   CORBA::ULong const num_rds = ids.length();
 
   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) ReplayerImpl::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_servant_->lookup_handle(ids[i]);
   }
 }

◆ need_sequence_repair()

bool OpenDDS::DCPS::ReplayerImpl::need_sequence_repair ( ) const

private

Definition at line 1033 of file ReplayerImpl.cpp.

References reader_info_, and sequence_number_.

Referenced by create_sample_data_message().

 {
   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::ReplayerImpl::notify_publication_disconnected ( const ReaderIdSeq & subids )

virtual

Implements OpenDDS::DCPS::TransportSendListener.

Definition at line 839 of file ReplayerImpl.cpp.

 {
   ACE_UNUSED_ARG(subids);
 }

◆ notify_publication_lost() [1/2]

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

virtual

Implements OpenDDS::DCPS::TransportSendListener.

Definition at line 851 of file ReplayerImpl.cpp.

Referenced by remove_associations().

 {
   ACE_UNUSED_ARG(subids);
 }

◆ notify_publication_lost() [2/2]

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

private

Definition at line 857 of file ReplayerImpl.cpp.

 {
   ACE_UNUSED_ARG(handles);
 }

◆ notify_publication_reconnected()

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

virtual

Implements OpenDDS::DCPS::TransportSendListener.

Definition at line 845 of file ReplayerImpl.cpp.

 {
   ACE_UNUSED_ARG(subids);
 }

◆ OPENDDS_MAP_CMP() [1/3]

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

private

◆ OPENDDS_MAP_CMP() [2/3]

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

private

◆ OPENDDS_MAP_CMP() [3/3]

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

private

◆ participant()

DomainParticipantImpl* OpenDDS::DCPS::ReplayerImpl::participant ( )

inline

Definition at line 161 of file ReplayerImpl.h.

References write().

Referenced by OpenDDS::DCPS::Service_Participant::delete_replayer().

                                                 {
     return participant_servant_;
   }

◆ register_for_reader()

void OpenDDS::DCPS::ReplayerImpl::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 721 of file ReplayerImpl.cpp.

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

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

◆ remove_all_associations()

void OpenDDS::DCPS::ReplayerImpl::remove_all_associations ( )

Definition at line 688 of file ReplayerImpl.cpp.

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

Referenced by cleanup().

 {
   this->stop_associating();
 
   OpenDDS::DCPS::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&) {
   }
 
   transport_stop();
 }

◆ remove_associations()

void OpenDDS::DCPS::ReplayerImpl::remove_associations	(	const ReaderIdSeq &	readers,
		CORBA::Boolean	callback
	)

virtual

Section 7.1.4.1: total_count will not decrement.

: Reconcile this with the verbiage in section 7.1.4.1 TODO: Should rds_len really be fully_associated_len here??

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 559 of file ReplayerImpl.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), DDS::PublicationMatchedStatus::current_count, DDS::PublicationMatchedStatus::current_count_change, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TransportClient::disassociate(), id_to_handle_map_, idToSequence_, OpenDDS::DCPS::RcHandle< T >::in(), is_bit_, DDS::PublicationMatchedStatus::last_subscription_handle, listener_, LM_DEBUG, lock_, lookup_instance_handles(), notify_publication_lost(), OpenDDS::DCPS::ReplayerListener::on_replayer_matched(), participant_servant_, publication_id_, publication_match_status_, reader_info_, readers_, OpenDDS::DCPS::remove(), OpenDDS::DCPS::DomainParticipantImpl::return_handle(), OpenDDS::DCPS::TransportClient::stop_associating(), and DDS::PublicationMatchedStatus::total_count_change.

Referenced by remove_all_associations().

 {
   if (DCPS_debug_level >= 1) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) ReplayerImpl::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()));
   }
 
   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;
 
   {
     // Ensure the same acquisition order as in wait_for_acknowledgments().
     // ACE_GUARD(ACE_SYNCH_MUTEX, wfaGuard, this->wfaLock_);
     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 (OpenDDS::DCPS::remove(readers_, readers[i]) == 0) {
         ++fully_associated_len;
         fully_associated_readers.length(fully_associated_len);
         fully_associated_readers [fully_associated_len - 1] = readers[i];
 
         // Remove this reader from the ACK sequence map if its there.
         // This is where we need to be holding the wfaLock_ obtained
         // above.
         RepoIdToSequenceMap::iterator where
           = this->idToSequence_.find(readers[i]);
 
         if (where != this->idToSequence_.end()) {
           this->idToSequence_.erase(where);
 
           // It is possible that this subscription was causing the wait
           // to continue, so give the opportunity to find out.
           // this->wfaCondition_.broadcast();
         }
 
         ++rds_len;
         rds.length(rds_len);
         rds [rds_len - 1] = readers[i];
       }
       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]);
       }
     }
 
     // wfaGuard.release();
 
     // 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
         /// TODO: Should rds_len really be fully_associated_len here??
         this->publication_match_status_.last_subscription_handle =
           handles[rds_len - 1];
 
 
         if (listener_.in()) {
           listener_->on_replayer_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;
         }
 
       }
     }
   }
 
   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);
   }
 
   for (unsigned int i = 0; i < handles.length(); ++i) {
     participant_servant_->return_handle(handles[i]);
   }
 }

◆ replay_durable_data_for()

virtual void OpenDDS::DCPS::ReplayerImpl::replay_durable_data_for ( const GUID_t & )

inlinevirtual

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 138 of file ReplayerImpl.h.

138 {}

◆ retrieve_inline_qos_data()

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

virtual

Reimplemented from OpenDDS::DCPS::TransportSendListener.

Definition at line 864 of file ReplayerImpl.cpp.

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

 {
   qos_data.pub_qos = this->publisher_qos_;
   qos_data.dw_qos = this->qos_;
   qos_data.topic_name = this->topic_name_.in();
 }

◆ set_listener()

DDS::ReturnCode_t OpenDDS::DCPS::ReplayerImpl::set_listener	(	const ReplayerListener_rch &	a_listener,
		DDS::StatusMask	mask
	)

virtual

Change the listener for this Replayer.

Implements OpenDDS::DCPS::Replayer.

Definition at line 284 of file ReplayerImpl.cpp.

References listener_, listener_mask_, and DDS::RETCODE_OK.

 {
   listener_ = a_listener;
   listener_mask_ = mask;
   return DDS::RETCODE_OK;
 }

◆ set_qos()

DDS::ReturnCode_t OpenDDS::DCPS::ReplayerImpl::set_qos	(	const DDS::PublisherQos &	publisher_qos,
		const DDS::DataWriterQos &	datawriter_qos
	)

virtual

Set the Quality of Service settings for the Replayer.

Implements OpenDDS::DCPS::Replayer.

Definition at line 184 of file ReplayerImpl.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), OpenDDS::DCPS::Qos_Helper::changeable(), OpenDDS::DCPS::Qos_Helper::consistent(), domain_id_, OpenDDS::DCPS::EntityImpl::enabled_, OpenDDS::DCPS::DomainParticipantImpl::get_domain_id(), OpenDDS::DCPS::DomainParticipantImpl::get_id(), LM_ERROR, OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK, OPENDDS_NO_DURABILITY_SERVICE_COMPATIBILITY_CHECK, OPENDDS_NO_OBJECT_MODEL_PROFILE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_STRENGTH_COMPATIBILITY_CHECK, participant_servant_, publication_id_, publisher_qos_, qos_, DDS::RETCODE_ERROR, DDS::RETCODE_IMMUTABLE_POLICY, DDS::RETCODE_INCONSISTENT_POLICY, DDS::RETCODE_OK, DDS::RETCODE_UNSUPPORTED, TheServiceParticipant, and OpenDDS::DCPS::Qos_Helper::valid().

 {
 
   OPENDDS_NO_OBJECT_MODEL_PROFILE_COMPATIBILITY_CHECK(publisher_qos, DDS::RETCODE_UNSUPPORTED);
 
   if (Qos_Helper::valid(publisher_qos) && Qos_Helper::consistent(publisher_qos)) {
     if (publisher_qos_ == publisher_qos)
       return DDS::RETCODE_OK;
 
     // for the not changeable qos, it can be changed before enable
     if (!Qos_Helper::changeable(publisher_qos_, publisher_qos) && enabled_) {
       return DDS::RETCODE_IMMUTABLE_POLICY;
 
     } else {
       publisher_qos_ = publisher_qos;
     }
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 
   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);
 
   if (Qos_Helper::valid(qos) && Qos_Helper::consistent(qos)) {
     if (qos_ == qos)
       return DDS::RETCODE_OK;
 
     if (!Qos_Helper::changeable(qos_, qos) && enabled_) {
       return DDS::RETCODE_IMMUTABLE_POLICY;
 
     } else {
       Discovery_rch disco = TheServiceParticipant->get_discovery(domain_id_);
       // DDS::PublisherQos publisherQos;
       // this->publisher_servant_->get_qos(publisherQos);
       DDS::PublisherQos publisherQos = this->publisher_qos_;
       const bool status
         = disco->update_publication_qos(this->participant_servant_->get_domain_id(),
                                         this->participant_servant_->get_id(),
                                         this->publication_id_,
                                         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_ == qos)) {
       // Reset the deadline timer if the period has changed.
       // if (qos_.deadline.period.sec != qos.deadline.period.sec
       //     || qos_.deadline.period.nanosec != qos.deadline.period.nanosec) {
       //   if (qos_.deadline.period.sec == DDS::DURATION_INFINITE_SEC
       //       && qos_.deadline.period.nanosec == DDS::DURATION_INFINITE_NSEC) {
       //     ACE_auto_ptr_reset(this->watchdog_,
       //                        new OfferedDeadlineWatchdog(
       //                          this->reactor_,
       //                          this->lock_,
       //                          qos.deadline,
       //                          this,
       //                          this,
       //                          this->offered_deadline_missed_status_,
       //                          this->last_deadline_missed_total_count_));
       //
       //   } else if (qos.deadline.period.sec == DDS::DURATION_INFINITE_SEC
       //              && qos.deadline.period.nanosec == DDS::DURATION_INFINITE_NSEC) {
       //     this->watchdog_->cancel_all();
       //     this->watchdog_.reset();
       //
       //   } else {
       //     this->watchdog_->reset_interval(
       //       duration_to_time_value(qos.deadline.period));
       //   }
       // }
 
       qos_ = qos;
     }
 
     return DDS::RETCODE_OK;
 
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ unregister_for_reader()

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

virtual

Reimplemented from OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 731 of file ReplayerImpl.cpp.

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

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

◆ update_incompatible_qos()

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

virtual

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 739 of file ReplayerImpl.cpp.

References ACE_GUARD, OpenDDS::DCPS::IncompatibleQosStatus::count_since_last_send, OpenDDS::DCPS::IncompatibleQosStatus::last_policy_id, DDS::OfferedIncompatibleQosStatus::last_policy_id, lock_, offered_incompatible_qos_status_, OpenDDS::DCPS::IncompatibleQosStatus::policies, DDS::OfferedIncompatibleQosStatus::policies, OpenDDS::DCPS::IncompatibleQosStatus::total_count, DDS::OfferedIncompatibleQosStatus::total_count, and DDS::OfferedIncompatibleQosStatus::total_count_change.

 {
 
 
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->lock_);
 
   // 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;
 
 }

◆ update_subscription_params()

void OpenDDS::DCPS::ReplayerImpl::update_subscription_params	(	const GUID_t &	readerId,
		const DDS::StringSeq &	exprParams
	)

virtual

Implements OpenDDS::DCPS::DataWriterCallbacks.

Definition at line 755 of file ReplayerImpl.cpp.

 {
   ACE_UNUSED_ARG(readerId);
   ACE_UNUSED_ARG(params);
 }

◆ write() [1/2]

DDS::ReturnCode_t OpenDDS::DCPS::ReplayerImpl::write ( const RawDataSample & sample )

virtual

Send the sample to all associated DataReaders.

Note: Only samples of type SAMPLE_DATA should be sent.

Implements OpenDDS::DCPS::Replayer.

Definition at line 946 of file ReplayerImpl.cpp.

Referenced by write_to_reader().

 {
   return this->write(&sample, 1, 0);
 }

◆ write() [2/2]

DDS::ReturnCode_t OpenDDS::DCPS::ReplayerImpl::write	(	const RawDataSample *	sample_array,
		int	array_size,
		DDS::InstanceHandle_t *	reader
	)

private

Definition at line 872 of file ReplayerImpl.cpp.

References ACE_DES_FREE, ACE_ERROR_RETURN, ACE_GUARD_RETURN, ACE_NEW_MALLOC_RETURN, ACE_TEXT(), OpenDDS::DCPS::DataSampleHeader::byte_order_, create_sample_data_message(), DBG_ENTRY_LVL, OpenDDS::DCPS::SendStateDataSampleList::dequeue(), OpenDDS::DCPS::SendStateDataSampleList::enqueue_tail(), OpenDDS::DCPS::DataSampleElement::get_header(), OpenDDS::DCPS::DomainParticipantImpl::get_repoid(), OpenDDS::DCPS::GUID_UNKNOWN, LM_ERROR, lock_, OpenDDS::DCPS::move(), participant_servant_, pending_write_count_, OpenDDS::DCPS::DataSampleHeader::publication_id_, publication_id_, reader_info_, OpenDDS::DCPS::TransportClient::repo_id(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::RawDataSample::sample_byte_order_, sample_list_element_allocator_, OpenDDS::DCPS::TransportClient::send(), sequence_number_, OpenDDS::DCPS::DataSampleElement::set_num_subs(), OpenDDS::DCPS::DataSampleElement::set_sample(), OpenDDS::DCPS::DataSampleElement::set_sub_id(), and OpenDDS::DCPS::RawDataSample::source_timestamp_.

 {
   DBG_ENTRY_LVL("ReplayerImpl","write",6);
 
   OpenDDS::DCPS::GUID_t repo_id;
   if (reader_ih_ptr) {
     repo_id = this->participant_servant_->get_repoid(*reader_ih_ptr);
     if (repo_id == GUID_UNKNOWN) {
       ACE_ERROR_RETURN((LM_ERROR,
                         ACE_TEXT("(%P|%t) ERROR: ReplayerImpl::write: ")
                         ACE_TEXT("Invalid reader instance handle (%d)\n"), *reader_ih_ptr),
                        DDS::RETCODE_ERROR);
     }
   }
 
   SendStateDataSampleList list;
 
   for (int i = 0; i < num_samples; ++i) {
     DataSampleElement* element = 0;
 
     ACE_NEW_MALLOC_RETURN(
       element,
       static_cast<DataSampleElement*>(
         sample_list_element_allocator_->malloc(
           sizeof(DataSampleElement))),
       DataSampleElement(publication_id_,
                             this,
                             PublicationInstance_rch()),
       DDS::RETCODE_ERROR);
 
     element->get_header().byte_order_ = samples[i].sample_byte_order_;
     element->get_header().publication_id_ = this->publication_id_;
     list.enqueue_tail(element);
     Message_Block_Ptr temp;
     Message_Block_Ptr sample(samples[i].sample_->duplicate());
     DDS::ReturnCode_t ret = create_sample_data_message(move(sample),
                                                        element->get_header(),
                                                        temp,
                                                        samples[i].source_timestamp_,
                                                        false);
     element->set_sample(move(temp));
     if (reader_ih_ptr) {
       element->set_num_subs(1);
       element->set_sub_id(0, repo_id);
     }
 
     if (ret != DDS::RETCODE_OK) {
       // we need to free the list
       while (list.dequeue(element)) {
         ACE_DES_FREE(element, sample_list_element_allocator_->free, DataSampleElement);
       }
 
       return ret;
     }
   }
 
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, this->lock_, DDS::RETCODE_ERROR);
     ++pending_write_count_;
   }
 
   this->send(list);
 
   for (RepoIdToReaderInfoMap::iterator iter = reader_info_.begin(),
        end = reader_info_.end(); iter != end; ++iter) {
     iter->second.expected_sequence_ = sequence_number_;
   }
 
   return DDS::RETCODE_OK;
 }

◆ write_to_reader() [1/2]

DDS::ReturnCode_t OpenDDS::DCPS::ReplayerImpl::write_to_reader	(	DDS::InstanceHandle_t	subscription,
		const RawDataSample &	sample
	)

virtual

Send the sample to the specified DataReader.

Note: Only samples of type SAMPLE_DATA should be sent.

Implements OpenDDS::DCPS::Replayer.

Definition at line 1051 of file ReplayerImpl.cpp.

References write().

 {
   return write(&sample, 1, &subscription);
 }

◆ write_to_reader() [2/2]

DDS::ReturnCode_t OpenDDS::DCPS::ReplayerImpl::write_to_reader	(	DDS::InstanceHandle_t	subscription,
		const RawDataSampleList &	samples
	)

virtual

Send the samples to the specified DataReader.

Note: Only samples of type SAMPLE_DATA should be sent.

Implements OpenDDS::DCPS::Replayer.

Definition at line 1058 of file ReplayerImpl.cpp.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL, DDS::RETCODE_ERROR, and write().

 {
   if (!samples.empty())
     return write(&samples[0], static_cast<int>(samples.size()), &subscription);
   return DDS::RETCODE_ERROR;
 }

Friends And Related Function Documentation

◆ ::DDS_TEST

friend class ::DDS_TEST

friend

Definition at line 215 of file ReplayerImpl.h.

Member Data Documentation

◆ association_chunk_multiplier_

size_t OpenDDS::DCPS::ReplayerImpl::association_chunk_multiplier_

private

The multiplier for allocators affected by associations.

Definition at line 187 of file ReplayerImpl.h.

Referenced by enable().

◆ data_delivered_count_

int OpenDDS::DCPS::ReplayerImpl::data_delivered_count_

Definition at line 125 of file ReplayerImpl.h.

Referenced by data_delivered().

◆ data_dropped_count_

int OpenDDS::DCPS::ReplayerImpl::data_dropped_count_

Statistics counter.

Definition at line 124 of file ReplayerImpl.h.

Referenced by data_dropped().

◆ db_allocator_

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

private

Definition at line 279 of file ReplayerImpl.h.

Referenced by create_sample_data_message(), and enable().

◆ domain_id_

DDS::DomainId_t OpenDDS::DCPS::ReplayerImpl::domain_id_

private

The domain id.

Definition at line 232 of file ReplayerImpl.h.

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

◆ empty_condition_

ConditionVariable<ACE_Recursive_Thread_Mutex> OpenDDS::DCPS::ReplayerImpl::empty_condition_

private

Definition at line 313 of file ReplayerImpl.h.

Referenced by cleanup(), data_delivered(), and data_dropped().

◆ header_allocator_

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

private

Definition at line 281 of file ReplayerImpl.h.

Referenced by create_sample_data_message(), and enable().

◆ id_to_handle_map_

RepoIdToHandleMap OpenDDS::DCPS::ReplayerImpl::id_to_handle_map_

private

Definition at line 250 of file ReplayerImpl.h.

Referenced by association_complete_i(), and remove_associations().

◆ idToSequence_

RepoIdToSequenceMap OpenDDS::DCPS::ReplayerImpl::idToSequence_

private

Definition at line 311 of file ReplayerImpl.h.

Referenced by remove_associations().

◆ is_bit_

bool OpenDDS::DCPS::ReplayerImpl::is_bit_

private

The time interval for sending liveliness message.

The orb's reactor to be used to register the liveliness timer.Timestamp of last write/dispose/assert_liveliness. Total number of offered deadlines missed during last offered deadline status check. Watchdog responsible for reporting missed offered deadlines. The flag indicates whether the liveliness timer is scheduled and needs be cancelled. Flag indicates that this datawriter is a builtin topic datawriter.

Definition at line 306 of file ReplayerImpl.h.

Referenced by add_association(), association_complete_i(), init(), and remove_associations().

◆ listener_

ReplayerListener_rch OpenDDS::DCPS::ReplayerImpl::listener_

private

Used to notify the entity for relevant events.

Definition at line 230 of file ReplayerImpl.h.

Referenced by association_complete_i(), get_listener(), init(), remove_associations(), and set_listener().

◆ listener_mask_

DDS::StatusMask OpenDDS::DCPS::ReplayerImpl::listener_mask_

private

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

Definition at line 228 of file ReplayerImpl.h.

Referenced by init(), and set_listener().

◆ lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::ReplayerImpl::lock_

private

The sample data container.

The lock to protect the activate subscriptions and status changes.

Definition at line 246 of file ReplayerImpl.h.

Referenced by add_association(), association_complete_i(), cleanup(), data_delivered(), data_dropped(), remove_all_associations(), remove_associations(), update_incompatible_qos(), and write().

◆ mb_allocator_

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

private

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

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 277 of file ReplayerImpl.h.

Referenced by create_sample_data_message(), and enable().

◆ n_chunks_

size_t OpenDDS::DCPS::ReplayerImpl::n_chunks_

private

The number of chunks for the cached allocator.

Definition at line 184 of file ReplayerImpl.h.

Referenced by enable().

◆ offered_incompatible_qos_status_

DDS::OfferedIncompatibleQosStatus OpenDDS::DCPS::ReplayerImpl::offered_incompatible_qos_status_

private

Status conditions.

Definition at line 257 of file ReplayerImpl.h.

Referenced by ReplayerImpl(), and update_incompatible_qos().

◆ participant_servant_

DomainParticipantImpl* OpenDDS::DCPS::ReplayerImpl::participant_servant_

private

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

Definition at line 200 of file ReplayerImpl.h.

Referenced by add_association(), association_complete_i(), enable(), get_instance_handle(), init(), lookup_instance_handles(), remove_associations(), set_qos(), and write().

◆ passed_qos_

DDS::DataWriterQos OpenDDS::DCPS::ReplayerImpl::passed_qos_

private

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

Definition at line 196 of file ReplayerImpl.h.

Referenced by get_qos(), and init().

◆ pending_write_count_

int OpenDDS::DCPS::ReplayerImpl::pending_write_count_

private

Definition at line 314 of file ReplayerImpl.h.

Referenced by cleanup(), data_delivered(), data_dropped(), and write().

◆ publication_id_

GUID_t OpenDDS::DCPS::ReplayerImpl::publication_id_

private

The repository id of this datawriter/publication.

Definition at line 238 of file ReplayerImpl.h.

Referenced by add_association(), cleanup(), data_delivered(), enable(), get_guid(), get_instance_handle(), remove_associations(), set_qos(), and write().

◆ publication_match_status_

DDS::PublicationMatchedStatus OpenDDS::DCPS::ReplayerImpl::publication_match_status_

private

Definition at line 258 of file ReplayerImpl.h.

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

◆ publisher_qos_

DDS::PublisherQos OpenDDS::DCPS::ReplayerImpl::publisher_qos_

private

Definition at line 235 of file ReplayerImpl.h.

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

◆ publisher_servant_

PublisherImpl* OpenDDS::DCPS::ReplayerImpl::publisher_servant_

private

The publisher servant which creates this datawriter.

Definition at line 234 of file ReplayerImpl.h.

◆ qos_

DDS::DataWriterQos OpenDDS::DCPS::ReplayerImpl::qos_

private

The qos policy list of this datawriter.

Definition at line 193 of file ReplayerImpl.h.

Referenced by add_association(), create_sample_data_message(), enable(), get_priority_value(), init(), retrieve_inline_qos_data(), and set_qos().

◆ reader_info_

RepoIdToReaderInfoMap OpenDDS::DCPS::ReplayerImpl::reader_info_

private

Definition at line 211 of file ReplayerImpl.h.

Referenced by add_association(), need_sequence_repair(), remove_associations(), and write().

◆ readers_

RepoIdSet OpenDDS::DCPS::ReplayerImpl::readers_

private

Definition at line 252 of file ReplayerImpl.h.

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

◆ sample_list_element_allocator_

unique_ptr<DataSampleElementAllocator> OpenDDS::DCPS::ReplayerImpl::sample_list_element_allocator_

private

The cached allocator to allocate DataSampleElement objects.

Definition at line 285 of file ReplayerImpl.h.

Referenced by data_delivered(), data_dropped(), enable(), and write().

◆ sequence_number_

SequenceNumber OpenDDS::DCPS::ReplayerImpl::sequence_number_

private

The sequence number unique in DataWriter scope.

Definition at line 240 of file ReplayerImpl.h.

Referenced by create_sample_data_message(), need_sequence_repair(), and write().

◆ topic_id_

GUID_t OpenDDS::DCPS::ReplayerImpl::topic_id_

private

The associated topic repository id.

Definition at line 220 of file ReplayerImpl.h.

Referenced by init().

◆ topic_name_

CORBA::String_var OpenDDS::DCPS::ReplayerImpl::topic_name_

private

The name of associated topic.

Definition at line 218 of file ReplayerImpl.h.

Referenced by init(), and retrieve_inline_qos_data().

◆ topic_objref_

DDS::Topic_var OpenDDS::DCPS::ReplayerImpl::topic_objref_

private

The object reference of the associated topic.

Definition at line 222 of file ReplayerImpl.h.

Referenced by cleanup(), and init().

◆ topic_servant_

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

private

The topic servant.

Definition at line 224 of file ReplayerImpl.h.

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

◆ type_name_

CORBA::String_var OpenDDS::DCPS::ReplayerImpl::type_name_

private

The type name of associated topic.

Definition at line 190 of file ReplayerImpl.h.

Referenced by init().

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

Classes

Public Member Functions

Public Attributes

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ ReplayerImpl()

◆ ~ReplayerImpl()

Member Function Documentation

◆ add_association()

◆ association_complete_i()

◆ check_transport_qos()

◆ cleanup()

◆ control_delivered()

◆ control_dropped()

◆ create_sample_data_message()

◆ data_delivered()

◆ data_dropped()

◆ domain_id()

◆ enable()

◆ get_guid()

◆ get_ice_endpoint()

◆ get_instance_handle()

◆ get_listener()

◆ get_max_sn()

◆ get_priority_value()

◆ get_qos()

◆ init()

◆ lookup_instance_handles()

◆ need_sequence_repair()

◆ notify_publication_disconnected()

◆ notify_publication_lost() [1/2]

◆ notify_publication_lost() [2/2]

◆ notify_publication_reconnected()

◆ OPENDDS_MAP_CMP() [1/3]

◆ OPENDDS_MAP_CMP() [2/3]

◆ OPENDDS_MAP_CMP() [3/3]

◆ participant()

◆ register_for_reader()

◆ remove_all_associations()

◆ remove_associations()

◆ replay_durable_data_for()

◆ retrieve_inline_qos_data()

◆ set_listener()

◆ set_qos()

◆ unregister_for_reader()

◆ update_incompatible_qos()

◆ update_subscription_params()

◆ write() [1/2]

◆ write() [2/2]

◆ write_to_reader() [1/2]

◆ write_to_reader() [2/2]

Friends And Related Function Documentation

◆ ::DDS_TEST

Member Data Documentation

◆ association_chunk_multiplier_

◆ data_delivered_count_

◆ data_dropped_count_

◆ db_allocator_

◆ domain_id_

◆ empty_condition_

◆ header_allocator_

◆ id_to_handle_map_

◆ idToSequence_

◆ is_bit_

◆ listener_

◆ listener_mask_

◆ lock_

◆ mb_allocator_

◆ n_chunks_

◆ offered_incompatible_qos_status_

◆ participant_servant_

◆ passed_qos_

◆ pending_write_count_

◆ publication_id_

◆ publication_match_status_

◆ publisher_qos_