Implementation of Recorder functionality. More...

#include <RecorderImpl.h>

Inheritance diagram for OpenDDS::DCPS::RecorderImpl:

[legend]

Collaboration diagram for OpenDDS::DCPS::RecorderImpl:

[legend]

Public Member Functions
	RecorderImpl ()

virtual	~RecorderImpl ()

DDS::ReturnCode_t	cleanup ()

void	init (TopicDescriptionImpl a_topic_desc, const DDS::DataReaderQos &qos, RecorderListener_rch a_listener, const DDS::StatusMask &mask, DomainParticipantImpl participant, DDS::SubscriberQos subqos)

DDS::ReturnCode_t	enable ()

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

virtual void	data_received (const ReceivedDataSample &sample)

virtual void	notify_subscription_disconnected (const WriterIdSeq &pubids)

virtual void	notify_subscription_reconnected (const WriterIdSeq &pubids)

virtual void	notify_subscription_lost (const WriterIdSeq &pubids)

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

virtual void	remove_associations (const WriterIdSeq &writers, CORBA::Boolean callback)

virtual void	update_incompatible_qos (const IncompatibleQosStatus &status)

virtual void	signal_liveliness (const GUID_t &remote_participant)

void	remove_all_associations ()

void	add_to_dynamic_type_map (const GUID_t &pub_id, const XTypes::TypeIdentifier &ti)

virtual DDS::ReturnCode_t	repoid_to_bit_key (const DCPS::GUID_t &id, DDS::BuiltinTopicKey_t &key)

DDS::ReturnCode_t	set_qos (const DDS::SubscriberQos &subscriber_qos, const DDS::DataReaderQos &datareader_qos)

DDS::ReturnCode_t	get_qos (DDS::SubscriberQos &subscriber_qos, DDS::DataReaderQos &datareader_qos)

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

RecorderListener_rch	get_listener ()

DomainParticipantImpl *	participant ()

virtual DDS::InstanceHandle_t	get_instance_handle ()

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

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

virtual WeakRcHandle< ICE::Endpoint >	get_ice_endpoint ()

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

long	ref_count () const

WeakObject *	_get_weak_object () const

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

virtual void	transport_discovery_change ()

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

virtual	~DataReaderCallbacks ()

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

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

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

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

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

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

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

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

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

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

DDS::DynamicData_ptr	get_dynamic_data (const RawDataSample &sample)

void	check_encap (bool b)

bool	check_encap () const

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

typedef	OPENDDS_MAP_CMP (GUID_t, RcHandle< WriterInfo >, GUID_tKeyLessThan) WriterMapType
	publications writing to this reader. More...

typedef	OPENDDS_MAP (GUID_t, DDS::DynamicType_var) DynamicTypeByPubId

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

virtual	~WriterInfoListener ()

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

virtual void	writer_became_dead (WriterInfo &info)

virtual void	writer_removed (WriterInfo &info)

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

long	ref_count () const

WeakObject *	_get_weak_object () const

	RcObject ()

Private Attributes
DDS::DataReaderQos	qos_

DDS::DataReaderQos	passed_qos_

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

DomainParticipantImpl *	participant_servant_

TopicDescriptionPtr< TopicImpl >	topic_servant_

bool	is_exclusive_ownership_

OwnershipManager *	owner_manager_

DDS::SubscriberQos	subqos_

DDS::TopicDescription_var	topic_desc_

DDS::StatusMask	listener_mask_

RecorderListener_rch	listener_

DDS::DomainId_t	domain_id_

ACE_Recursive_Thread_Mutex	publication_handle_lock_

RepoIdToHandleMap	id_to_handle_map_

DDS::RequestedIncompatibleQosStatus	requested_incompatible_qos_status_

DDS::SubscriptionMatchedStatus	subscription_match_status_

bool	is_bit_

WriterMapType	writers_

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

DynamicTypeByPubId	dt_map_

bool	check_encap_

TransportMessageBlockAllocator	mb_alloc_

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

TimeDuration	liveliness_lease_duration_

Friends
class	::DDS_TEST

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

Public Types inherited from OpenDDS::DCPS::Recorder
typedef Recorder_ptr	_ptr_type

typedef Recorder_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::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::Recorder
static Recorder_ptr	_duplicate (Recorder_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 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 Recorder functionality.

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

Definition at line 45 of file RecorderImpl.h.

Constructor & Destructor Documentation

◆ RecorderImpl()

OpenDDS::DCPS::RecorderImpl::RecorderImpl ( )

Definition at line 52 of file RecorderImpl.cpp.

References DDS::SubscriptionMatchedStatus::current_count, DDS::SubscriptionMatchedStatus::current_count_change, DDS::HANDLE_NIL, DDS::RequestedIncompatibleQosStatus::last_policy_id, DDS::SubscriptionMatchedStatus::last_publication_handle, DDS::RequestedIncompatibleQosStatus::policies, requested_incompatible_qos_status_, subscription_match_status_, DDS::RequestedIncompatibleQosStatus::total_count, DDS::SubscriptionMatchedStatus::total_count, DDS::RequestedIncompatibleQosStatus::total_count_change, and DDS::SubscriptionMatchedStatus::total_count_change.

   : qos_(TheServiceParticipant->initial_DataReaderQos())
   , participant_servant_(0)
   , topic_servant_(0)
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   , is_exclusive_ownership_(false)
 #endif
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   , owner_manager_(0)
 #endif
   , subqos_(TheServiceParticipant->initial_SubscriberQos())
   , topic_desc_(0)
   , listener_mask_(DEFAULT_STATUS_MASK)
   , domain_id_(0)
   , is_bit_(false)
   , check_encap_(true)
   , mb_alloc_(DEFAULT_TRANSPORT_RECEIVE_BUFFERS)
 {
   requested_incompatible_qos_status_.total_count = 0;
   requested_incompatible_qos_status_.total_count_change = 0;
   requested_incompatible_qos_status_.last_policy_id = 0;
   requested_incompatible_qos_status_.policies.length(0);
 
   subscription_match_status_.total_count = 0;
   subscription_match_status_.total_count_change = 0;
   subscription_match_status_.current_count = 0;
   subscription_match_status_.current_count_change = 0;
   subscription_match_status_.last_publication_handle = DDS::HANDLE_NIL;
 }

◆ ~RecorderImpl()

OpenDDS::DCPS::RecorderImpl::~RecorderImpl ( )

virtual

Definition at line 84 of file RecorderImpl.cpp.

References DBG_ENTRY_LVL.

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

Member Function Documentation

◆ add_association()

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

virtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 248 of file RecorderImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), ACE_WRITE_GUARD, OpenDDS::DCPS::DomainParticipantImpl::assign_handle(), OpenDDS::DCPS::TransportClient::associate(), OpenDDS::DCPS::LogGuid::c_str(), DDS::SubscriptionMatchedStatus::current_count, DDS::SubscriptionMatchedStatus::current_count_change, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::AssociationData::discovery_locator_, DDS::DataWriterQos::durability, OpenDDS::DCPS::GUID_UNKNOWN, id_to_handle_map_, OpenDDS::DCPS::RcHandle< T >::in(), is_bit_, DDS::DurabilityQosPolicy::kind, DDS::ReliabilityQosPolicy::kind, DDS::SubscriptionMatchedStatus::last_publication_handle, listener_, LM_DEBUG, LM_WARNING, OpenDDS::DCPS::log_level, OpenDDS::DCPS::RecorderListener::on_recorder_matched(), participant_servant_, publication_handle_lock_, OpenDDS::DCPS::AssociationData::publication_transport_priority_, DDS::DataWriterQos::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_, sample_lock_, OpenDDS::DCPS::WriterInfoListener::subscription_id_, subscription_match_status_, DDS::SubscriptionMatchedStatus::total_count, DDS::SubscriptionMatchedStatus::total_count_change, DDS::DataWriterQos::transport_priority, OpenDDS::DCPS::WriterAssociation::transportContext, DDS::TransportPriorityQosPolicy::value, DDS::VOLATILE_DURABILITY_QOS, OpenDDS::DCPS::LogLevel::Warning, OpenDDS::DCPS::WriterAssociation::writerDiscInfo, OpenDDS::DCPS::WriterAssociation::writerId, OpenDDS::DCPS::WriterAssociation::writerQos, writers_, writers_lock_, and OpenDDS::DCPS::WriterAssociation::writerTransInfo.

 {
   if (DCPS_debug_level >= 4) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) RecorderImpl::add_association: "
                "bit %d local %C remote %C\n",
                is_bit_,
                LogGuid(yourId).c_str(),
                LogGuid(writer.writerId).c_str()));
   }
 
   //
   // This block prevents adding associations to deleted readers.
   // Presumably this is a "good thing(tm)".
   //
   // if (entity_deleted_) {
   //   if (DCPS_debug_level >= 1)
   //     ACE_DEBUG((LM_DEBUG,
   //                ACE_TEXT("(%P|%t) RecorderImpl::add_association")
   //                ACE_TEXT(" This is a deleted datareader, ignoring add.\n")));
   //
   //   return;
   // }
 
   //
   // We are being called back from the repository before we are done
   // processing after our call to the repository that caused this call
   // (from the repository) to be made.
   //
   if (GUID_UNKNOWN == subscription_id_) {
     // add_associations was invoked before DCSPInfoRepo::add_subscription() returned.
     subscription_id_ = yourId;
   }
 
   //
   // We do the following while holding the publication_handle_lock_.
   //
   {
     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, publication_handle_lock_);
 
     //
     // For each writer in the list of writers to associate with, we
     // create a WriterInfo and a WriterStats object and store them in
     // our internal maps.
     //
     {
       ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, writers_lock_);
 
       const GUID_t& writer_id = writer.writerId;
       RcHandle<WriterInfo> info ( make_rch<WriterInfo>(rchandle_from<WriterInfoListener>(this), writer_id, writer.writerQos));
       /*std::pair<WriterMapType::iterator, bool> bpair =*/
       writers_.insert(
         // This insertion is idempotent.
         WriterMapType::value_type(
           writer_id,
           info));
       // statistics_.insert(
       //   StatsMapType::value_type(
       //     writer_id,
       //     WriterStats(
       //       raw_latency_buffer_size_,
       //       raw_latency_buffer_type_)));
 
       // if (DCPS_debug_level > 4) {
       //   GuidConverter converter(writer_id);
       //   ACE_DEBUG((LM_DEBUG,
       //              "(%P|%t) RecorderImpl::add_association: "
       //              "inserted writer %C.return %d\n",
       //              OPENDDS_STRING(converter).c_str(), bpair.second));
       //
       //   WriterMapType::iterator iter = writers_.find(writer_id);
       //   if (iter != writers_.end()) {
       //     // This may not be an error since it could happen that the sample
       //     // is delivered to the datareader after the write is dis-associated
       //     // with this datareader.
       //     GuidConverter reader_converter(subscription_id_);
       //     GuidConverter writer_converter(writer_id);
       //     ACE_DEBUG((LM_DEBUG,
       //               ACE_TEXT("(%P|%t) RecorderImpl::add_association: ")
       //               ACE_TEXT("reader %C is associated with writer %C.\n"),
       //               OPENDDS_STRING(reader_converter).c_str(),
       //               OPENDDS_STRING(writer_converter).c_str()));
       //   }
       // }
     }
 
     //
     // Propagate the add_associations processing down into the Transport
     // layer here.  This will establish the transport support and reserve
     // usage of an existing connection or initiate creation of a new
     // connection if no suitable connection is available.
     //
     AssociationData data;
     data.remote_id_ = writer.writerId;
     data.remote_data_ = writer.writerTransInfo;
     data.discovery_locator_ = writer.writerDiscInfo;
     data.remote_transport_context_ = writer.transportContext;
     data.publication_transport_priority_ =
       writer.writerQos.transport_priority.value;
     data.remote_reliable_ =
       (writer.writerQos.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS);
     data.remote_durable_ =
       (writer.writerQos.durability.kind > DDS::VOLATILE_DURABILITY_QOS);
 
     if (!associate(data, active)) {
       if (log_level >= LogLevel::Warning) {
         ACE_ERROR((LM_WARNING, "(%P|%t) WARNING: RecorderImpl::add_association: "
                    "transport layer failed to associate\n"));
       }
       return;
     }
 
     // Check if any publications have already sent a REQUEST_ACK message.
     // {
     //   ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, writers_lock_);
     //
     //   WriterMapType::iterator where = writers_.find(writer.writerId);
     //
     //   if (where != writers_.end()) {
     //     const MonotonicTimePoint now = MonotonicTimePoint::now();
     //
     //     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
     //
     //     if (where->second->should_ack(now)) {
     //       const SequenceNumber sequence = where->second->ack_sequence();
     //       if (send_sample_ack(writer.writerId, sequence, now.to_dds_time())) {
     //         where->second->clear_acks(sequence);
     //       }
     //     }
     //   }
     // }
 
     //
     // LIVELINESS policy timers are managed here.
     //
     // if (liveliness_lease_duration_ != TimeDuration::zero) {
     //   // this call will start the timer if it is not already set
     //   const MonotonicTimePoint now = MonotonicTimePoint::now();
     //
     //   if (DCPS_debug_level >= 5) {
     //     GuidConverter converter(subscription_id_);
     //     ACE_DEBUG((LM_DEBUG,
     //                ACE_TEXT("(%P|%t) RecorderImpl::add_association: ")
     //                ACE_TEXT("starting/resetting liveliness timer for reader %C\n"),
     //                OPENDDS_STRING(converter).c_str()));
     //   }
     //
     //   handle_timeout(now, this);
     // }
 
     // else - no timer needed when LIVELINESS.lease_duration is INFINITE
 
   }
   //
   // We no longer hold the publication_handle_lock_.
   //
 
   //
   // We only do the following processing for readers that are *not*
   // readers of Builtin Topics.
   //
   if (!is_bit_) {
 
     const DDS::InstanceHandle_t handle = participant_servant_->assign_handle(writer.writerId);
 
     //
     // We acquire the publication_handle_lock_ for the remainder of our
     // processing.
     //
     {
       ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, publication_handle_lock_);
 
       // This insertion is idempotent.
       id_to_handle_map_.insert(
         RepoIdToHandleMap::value_type(writer.writerId, handle));
 
       if (DCPS_debug_level > 4) {
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) RecorderImpl::add_association: ")
                    ACE_TEXT("id_to_handle_map_[ %C] = 0x%x.\n"),
                    LogGuid(writer.writerId).c_str(),
                    handle));
       }
 
       // We need to adjust these after the insertions have all completed
       // since insertions are not guaranteed to increase the number of
       // currently matched publications.
       int matchedPublications = static_cast<int>(id_to_handle_map_.size());
       subscription_match_status_.current_count_change
         = matchedPublications - subscription_match_status_.current_count;
       subscription_match_status_.current_count = matchedPublications;
 
       ++subscription_match_status_.total_count;
       ++subscription_match_status_.total_count_change;
 
       subscription_match_status_.last_publication_handle = handle;
 
       // set_status_changed_flag(DDS::SUBSCRIPTION_MATCHED_STATUS, true);
 
 
       if (listener_.in()) {
         listener_->on_recorder_matched(
           this,
           subscription_match_status_);
 
         // TBD - why does the spec say to change this but not change
         //       the ChangeFlagStatus after a listener call?
 
         // Client will look at it so next time it looks the change should be 0
         subscription_match_status_.total_count_change = 0;
         subscription_match_status_.current_count_change = 0;
       }
 
       // notify_status_condition();
     }
 
     {
       ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
       ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, writers_lock_);
 
       if (!writers_.count(writer.writerId)) {
         return;
       }
 
       writers_[writer.writerId]->handle(handle);
     }
   }
 
   // if (monitor_) {
   //   monitor_->report();
   // }
 }

◆ add_to_dynamic_type_map()

void OpenDDS::DCPS::RecorderImpl::add_to_dynamic_type_map	(	const GUID_t &	pub_id,
		const XTypes::TypeIdentifier &	ti
	)

Definition at line 234 of file RecorderImpl.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DCPS_debug_level, dt_map_, OpenDDS::DCPS::DomainParticipantImpl::get_type_lookup_service(), LM_DEBUG, and participant_servant_.

Referenced by OpenDDS::RTPS::Sedp::match_continue().

 {
   XTypes::TypeLookupService_rch tls = participant_servant_->get_type_lookup_service();
   DDS::DynamicType_var dt = tls->type_identifier_to_dynamic(ti, pub_id);
   if (DCPS_debug_level >= 4) {
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) RecorderImpl::add_association: "
                "DynamicType added to map with guid: %C\n", LogGuid(pub_id).c_str()));
   }
   dt_map_.insert(std::make_pair(pub_id, dt));
 }

◆ check_encap() [1/2]

void OpenDDS::DCPS::RecorderImpl::check_encap ( bool b )

inlineprivatevirtual

Implements OpenDDS::DCPS::Recorder.

Definition at line 161 of file RecorderImpl.h.

161 { check_encap_ = b; }

OpenDDS::DCPS::RecorderImpl::check_encap_

bool check_encap_

Definition: RecorderImpl.h:212

◆ check_encap() [2/2]

bool OpenDDS::DCPS::RecorderImpl::check_encap ( ) const

inlineprivatevirtual

Implements OpenDDS::DCPS::Recorder.

Definition at line 162 of file RecorderImpl.h.

162 { return check_encap_; }

OpenDDS::DCPS::RecorderImpl::check_encap_

bool check_encap_

Definition: RecorderImpl.h:212

◆ check_transport_qos()

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

virtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 156 of file RecorderImpl.cpp.

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

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

◆ cleanup()

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

cleanup the DataWriter.

Definition at line 91 of file RecorderImpl.cpp.

References ACE_ERROR, domain_id_, OpenDDS::DCPS::DomainParticipantImpl::get_id(), LM_NOTICE, OpenDDS::DCPS::log_level, OpenDDS::DCPS::LogLevel::Notice, participant_servant_, remove_all_associations(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::WriterInfoListener::subscription_id_, and TheServiceParticipant.

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

 {
 
   Discovery_rch disco = TheServiceParticipant->get_discovery(domain_id_);
   if (!disco || !disco->remove_subscription(domain_id_,
                                   participant_servant_->get_id(),
                                   subscription_id_)) {
     if (log_level >= LogLevel::Notice) {
       ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: RecorderImpl::cleanup: "
         "could not remove subscription from discovery\n"));
     }
     return DDS::RETCODE_ERROR;
   }
 
   // Call remove association before unregistering the datareader from the transport,
   // otherwise some callbacks resulted from remove_association may lost.
 
   remove_all_associations();
 
   return DDS::RETCODE_OK;
 }

◆ data_received()

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

virtual

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 175 of file RecorderImpl.cpp.

References ACE_DEBUG, ACE_GUARD, OpenDDS::DCPS::DataSampleHeader::byte_order_, OpenDDS::DCPS::DataSampleHeader::cdr_encapsulation_, check_encap_, OpenDDS::DCPS::ReceivedDataSample::data(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::Encoding::kind(), OpenDDS::DCPS::Encoding::KIND_UNALIGNED_CDR, listener_, LM_DEBUG, mb_alloc_, OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::RecorderListener::on_sample_data_received(), OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::SAMPLE_DATA, sample_lock_, OpenDDS::DCPS::DataSampleHeader::source_timestamp_nanosec_, OpenDDS::DCPS::DataSampleHeader::source_timestamp_sec_, OpenDDS::DCPS::WriterInfoListener::subscription_id_, and OpenDDS::DCPS::to_string().

 {
   DBG_ENTRY_LVL("RecorderImpl","data_received",6);
 
   // Ensure some other thread is not changing the sample container
   // or statuses related to samples.
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
 
   if (DCPS_debug_level >= 8) {
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) RecorderImpl::data_received: "
                "%C received sample: %C\n",
                LogGuid(subscription_id_).c_str(),
                to_string(sample.header_).c_str()));
   }
 
   // we only support SAMPLE_DATA messages
   if (sample.header_.message_id_ == SAMPLE_DATA && listener_.in()) {
     Message_Block_Ptr payload(sample.data(&mb_alloc_));
     Encoding::Kind kind = Encoding::KIND_UNALIGNED_CDR;
     if (sample.header_.cdr_encapsulation_ && check_encap_) {
       Encoding enc;
       Serializer ser(payload.get(), enc);
       EncapsulationHeader encap;
       if (ser >> encap && encap.to_any_encoding(enc)) {
         kind = enc.kind();
       }
     }
     RawDataSample rawSample(sample.header_,
                             static_cast<MessageId> (sample.header_.message_id_),
                             sample.header_.source_timestamp_sec_,
                             sample.header_.source_timestamp_nanosec_,
                             sample.header_.publication_id_,
                             sample.header_.byte_order_,
                             payload.get(),
                             kind);
     listener_->on_sample_data_received(this, rawSample);
   }
 }

◆ domain_id()

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

inlinevirtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 76 of file RecorderImpl.h.

76 { return this->domain_id_; }

OpenDDS::DCPS::RecorderImpl::domain_id_

DDS::DomainId_t domain_id_

Definition: RecorderImpl.h:186

◆ enable()

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

Implements DDS::Entity.

Definition at line 870 of file RecorderImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::TransportClient::connection_info(), OpenDDS::DCPS::DCPS_debug_level, domain_id_, DDS::DataReaderQos::durability, OpenDDS::DCPS::TransportClient::enable_transport(), OpenDDS::DCPS::DomainParticipantImpl::get_id(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::EntityImpl::is_enabled(), DDS::DurabilityQosPolicy::kind, DDS::ReliabilityQosPolicy::kind, LM_DEBUG, LM_WARNING, OpenDDS::DCPS::log_level, participant_servant_, qos_, OpenDDS::DCPS::rchandle_from(), DDS::DataReaderQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, DDS::DataReaderQos::representation, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_enabled(), OpenDDS::DCPS::set_reader_effective_data_rep_qos(), subqos_, OpenDDS::DCPS::WriterInfoListener::subscription_id_, TheServiceParticipant, topic_servant_, DDS::DataRepresentationQosPolicy::value, DDS::VOLATILE_DURABILITY_QOS, and OpenDDS::DCPS::LogLevel::Warning.

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

 {
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) RecorderImpl::enable\n")));
   }
   //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 (is_enabled()) {
     return DDS::RETCODE_OK;
   }
 
   set_enabled();
 
   // if (topic_servant_ && !transport_disabled_) {
   if (topic_servant_) {
     if (DCPS_debug_level >= 2) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) RecorderImpl::enable: enable_transport\n"));
     }
 
     try {
       enable_transport(qos_.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS,
                              qos_.durability.kind > DDS::VOLATILE_DURABILITY_QOS);
     } catch (const Transport::Exception&) {
       if (log_level >= LogLevel::Warning) {
         ACE_ERROR((LM_WARNING, "(%P|%t) WARNING: RecorderImpl::enable: Transport Exception\n"));
       }
       return DDS::RETCODE_ERROR;
     }
 
     const TransportLocatorSeq& trans_conf_info = connection_info();
 
     CORBA::String_var filterClassName = "";
     CORBA::String_var filterExpression = "";
     DDS::StringSeq exprParams;
 
     Discovery_rch disco =
       TheServiceParticipant->get_discovery(domain_id_);
 
     DCPS::set_reader_effective_data_rep_qos(qos_.representation.value);
     if (!topic_servant_->check_data_representation(qos_.representation.value, false)) {
       return DDS::RETCODE_ERROR;
     }
 
     if (DCPS_debug_level >= 2) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) RecorderImpl::enable: add_subscription\n"));
     }
 
     XTypes::TypeInformation type_info;
 
     subscription_id_ =
       disco->add_subscription(domain_id_,
                               participant_servant_->get_id(),
                               topic_servant_->get_id(),
                               rchandle_from(this),
                               qos_,
                               trans_conf_info,
                               subqos_,
                               filterClassName,
                               filterExpression,
                               exprParams,
                               type_info);
 
     if (subscription_id_ == OpenDDS::DCPS::GUID_UNKNOWN) {
       if (log_level >= LogLevel::Warning) {
         ACE_ERROR((LM_WARNING, "(%P|%t) WARNING: RecorderImpl::enable: "
           "add_subscription returned invalid id\n"));
       }
       return DDS::RETCODE_ERROR;
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ get_dynamic_data()

DDS::DynamicData_ptr OpenDDS::DCPS::RecorderImpl::get_dynamic_data ( const RawDataSample & sample )

privatevirtual

Implements OpenDDS::DCPS::Recorder.

Definition at line 1002 of file RecorderImpl.cpp.

References ACE_ERROR, OpenDDS::DCPS::DataSampleHeader::byte_order_, OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::XTypes::DynamicDataXcdrReadImpl::check_xcdr1_mutable(), dt_map_, OpenDDS::DCPS::RawDataSample::encoding_kind_, OpenDDS::DCPS::ENDIAN_BIG, OpenDDS::DCPS::ENDIAN_LITTLE, OpenDDS::DCPS::LogLevel::Error, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), OpenDDS::DCPS::RawDataSample::header_, LM_ERROR, LM_NOTICE, OpenDDS::DCPS::log_level, OpenDDS::DCPS::LogLevel::Notice, OPENDDS_END_VERSIONED_NAMESPACE_DECL, OpenDDS::DCPS::RawDataSample::publication_id_, and OpenDDS::DCPS::RawDataSample::sample_.

 {
   const Encoding enc(sample.encoding_kind_, sample.header_.byte_order_ ? ENDIAN_LITTLE : ENDIAN_BIG);
   const DynamicTypeByPubId::const_iterator dt_found = dt_map_.find(sample.publication_id_);
   if (dt_found == dt_map_.end()) {
     if (log_level >= LogLevel::Error) {
       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: RecorderImpl::get_dynamic_data: "
         "failed to find GUID: %C in DynamicTypeByPubId.\n", LogGuid(sample.publication_id_).c_str()));
     }
     return 0;
   }
 
   DDS::DynamicType_var dt = dt_found->second;
   XTypes::DynamicDataXcdrReadImpl* dd = new XTypes::DynamicDataXcdrReadImpl(sample.sample_.get(), enc, dt);
   DDS::DynamicData_var dd_var = dd;
   if (!dd->check_xcdr1_mutable(dt)) {
     if (log_level >= LogLevel::Notice) {
       ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: RecorderImpl::get_dynamic_data: "
         "Encountered unsupported combination of XCDR1 encoding and mutable extensibility.\n"));
     }
     return 0;
   }
   return dd_var._retn();
 }

◆ get_guid()

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

virtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 164 of file RecorderImpl.cpp.

References OpenDDS::DCPS::WriterInfoListener::subscription_id_.

 {
   return subscription_id_;
 }

◆ get_ice_endpoint()

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

inlinevirtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 145 of file RecorderImpl.h.

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

◆ get_instance_handle()

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

virtual

Implements OpenDDS::DCPS::EntityImpl.

Definition at line 950 of file RecorderImpl.cpp.

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

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

◆ get_listener()

RecorderListener_rch OpenDDS::DCPS::RecorderImpl::get_listener ( )

virtual

Get the listener for this Recorder.

Implements OpenDDS::DCPS::Recorder.

Definition at line 836 of file RecorderImpl.cpp.

References listener_.

 {
   return listener_;
 }

◆ get_priority_value()

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

virtual

Implements OpenDDS::DCPS::TransportClient.

Definition at line 169 of file RecorderImpl.cpp.

References OpenDDS::DCPS::AssociationData::publication_transport_priority_.

 {
   return data.publication_transport_priority_;
 }

◆ get_qos()

DDS::ReturnCode_t OpenDDS::DCPS::RecorderImpl::get_qos	(	DDS::SubscriberQos &	subscriber_qos,
		DDS::DataReaderQos &	datareader_qos
	)

virtual

Get the Quality of Service settings for the Recorder.

Implements OpenDDS::DCPS::Recorder.

Definition at line 816 of file RecorderImpl.cpp.

References passed_qos_, DDS::RETCODE_OK, and subqos_.

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

◆ init()

void OpenDDS::DCPS::RecorderImpl::init	(	TopicDescriptionImpl *	a_topic_desc,
		const DDS::DataReaderQos &	qos,
		RecorderListener_rch	a_listener,
		const DDS::StatusMask &	mask,
		DomainParticipantImpl *	participant,
		DDS::SubscriberQos	subqos
	)

Definition at line 113 of file RecorderImpl.cpp.

References ACE_DEBUG, OpenDDS::DCPS::DCPS_debug_level, domain_id_, DDS::EXCLUSIVE_OWNERSHIP_QOS, OpenDDS::DCPS::DomainParticipantImpl::get_domain_id(), OpenDDS::DCPS::TopicDescriptionImpl::get_name(), is_exclusive_ownership_, DDS::OwnershipQosPolicy::kind, listener_, listener_mask_, LM_DEBUG, owner_manager_, DDS::DataReaderQos::ownership, OpenDDS::DCPS::DomainParticipantImpl::ownership_manager(), participant(), participant_servant_, passed_qos_, qos_, subqos_, topic_desc_, and topic_servant_.

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

 {
   if (DCPS_debug_level >= 2) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) RecorderImpl::init\n"));
   }
 
 
   topic_desc_ = DDS::TopicDescription::_duplicate(a_topic_desc);
   if (TopicImpl* a_topic = dynamic_cast<TopicImpl*>(a_topic_desc)) {
     topic_servant_ = a_topic;
   }
 
   CORBA::String_var topic_name = a_topic_desc->get_name();
   qos_ = qos;
   passed_qos_ = qos;
 
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   is_exclusive_ownership_ = qos_.ownership.kind == ::DDS::EXCLUSIVE_OWNERSHIP_QOS;
 #endif
 
   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;
 
 #ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
   if (is_exclusive_ownership_) {
     owner_manager_ = participant_servant_->ownership_manager ();
   }
 #endif
 
   domain_id_ = participant_servant_->get_domain_id();
   subqos_ = subqos;
 }

◆ lookup_instance_handles()

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

private

Lookup the instance handles by the publication repo ids.

Definition at line 842 of file RecorderImpl.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_i().

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

◆ notify_subscription_disconnected()

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

virtual

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 215 of file RecorderImpl.cpp.

216 {

217 }

◆ notify_subscription_lost() [1/2]

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

virtual

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 228 of file RecorderImpl.cpp.

Referenced by remove_associations_i().

229 {

230 }

◆ notify_subscription_lost() [2/2]

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

private

Definition at line 224 of file RecorderImpl.cpp.

225 {

226 }

◆ notify_subscription_reconnected()

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

virtual

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 219 of file RecorderImpl.cpp.

220 {

221 }

◆ OPENDDS_MAP()

typedef OpenDDS::DCPS::RecorderImpl::OPENDDS_MAP	(	GUID_t	,
		DDS::DynamicType_var
	)

private

◆ OPENDDS_MAP_CMP() [1/2]

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

private

◆ OPENDDS_MAP_CMP() [2/2]

typedef OpenDDS::DCPS::RecorderImpl::OPENDDS_MAP_CMP	(	GUID_t	,
		RcHandle< WriterInfo >	,
		GUID_tKeyLessThan
	)

private

publications writing to this reader.

◆ participant()

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

inline

Definition at line 129 of file RecorderImpl.h.

Referenced by OpenDDS::DCPS::Service_Participant::delete_recorder(), and init().

                                                 {
     return participant_servant_;
   }

◆ register_for_writer()

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

virtual

Reimplemented from OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 956 of file RecorderImpl.cpp.

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

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

◆ remove_all_associations()

void OpenDDS::DCPS::RecorderImpl::remove_all_associations ( )

Definition at line 650 of file RecorderImpl.cpp.

References ACE_GUARD, ACE_READ_GUARD, DBG_ENTRY_LVL, publication_handle_lock_, remove_associations(), OpenDDS::DCPS::TransportClient::transport_stop(), writers_, and writers_lock_.

Referenced by cleanup().

 {
   DBG_ENTRY_LVL("RecorderImpl","remove_all_associations",6);
 
   OpenDDS::DCPS::WriterIdSeq writers;
   int size;
 
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, publication_handle_lock_);
 
   {
     ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, writers_lock_);
 
     size = static_cast<int>(writers_.size());
     writers.length(size);
 
     WriterMapType::iterator curr_writer = writers_.begin();
     WriterMapType::iterator end_writer = writers_.end();
 
     int i = 0;
 
     while (curr_writer != end_writer) {
       writers[i++] = curr_writer->first;
       ++curr_writer;
     }
   }
 
   try {
     CORBA::Boolean dont_notify_lost = false;
 
     if (0 < size) {
       remove_associations(writers, dont_notify_lost);
     }
 
   } catch (const CORBA::Exception&) {
   }
 
   transport_stop();
 }

◆ remove_associations()

void OpenDDS::DCPS::RecorderImpl::remove_associations	(	const WriterIdSeq &	writers,
		CORBA::Boolean	callback
	)

virtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 483 of file RecorderImpl.cpp.

References ACE_DEBUG, ACE_TEXT(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::EntityImpl::get_deleted(), is_bit_, LM_DEBUG, remove_associations_i(), OpenDDS::DCPS::TransportClient::stop_associating(), and OpenDDS::DCPS::WriterInfoListener::subscription_id_.

Referenced by remove_all_associations().

 {
   DBG_ENTRY_LVL("RecorderImpl", "remove_associations", 6);
   if (writers.length() == 0) {
     return;
   }
 
   if (DCPS_debug_level >= 4) {
     ACE_DEBUG((LM_DEBUG,
                ACE_TEXT("(%P|%t) RecorderImpl::remove_associations: ")
                ACE_TEXT("bit %d local %C remote %C num remotes %d\n"),
                is_bit_,
                LogGuid(subscription_id_).c_str(),
                LogGuid(writers[0]).c_str(),
                writers.length()));
   }
   if (!get_deleted()) {
     // stop pending associations for these writer ids
     stop_associating(writers.get_buffer(), writers.length());
   }
 
   remove_associations_i(writers, notify_lost);
 }

◆ remove_associations_i()

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

protectedvirtual

Section 7.1.4.1: total_count will not decrement.

: Reconcile this with the verbiage in section 7.1.4.1

Definition at line 509 of file RecorderImpl.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), ACE_WRITE_GUARD, DDS::SubscriptionMatchedStatus::current_count, DDS::SubscriptionMatchedStatus::current_count_change, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TransportClient::disassociate(), dt_map_, id_to_handle_map_, OpenDDS::DCPS::RcHandle< T >::in(), is_bit_, DDS::SubscriptionMatchedStatus::last_publication_handle, listener_, LM_DEBUG, lookup_instance_handles(), notify_subscription_lost(), OpenDDS::DCPS::RecorderListener::on_recorder_matched(), participant_servant_, publication_handle_lock_, OpenDDS::DCPS::push_back(), OpenDDS::DCPS::DomainParticipantImpl::return_handle(), OpenDDS::DCPS::WriterInfoListener::subscription_id_, subscription_match_status_, DDS::SubscriptionMatchedStatus::total_count_change, writers_, and writers_lock_.

Referenced by remove_associations().

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

◆ repoid_to_bit_key()

DDS::ReturnCode_t OpenDDS::DCPS::RecorderImpl::repoid_to_bit_key	(	const DCPS::GUID_t &	id,
		DDS::BuiltinTopicKey_t &	key
	)

virtual

Find the bit key for a given repo id.

Implements OpenDDS::DCPS::Recorder.

Definition at line 975 of file RecorderImpl.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::BUILT_IN_PUBLICATION_TOPIC, OpenDDS::DCPS::DomainParticipantImpl::lookup_handle(), participant_servant_, publication_handle_lock_, DDS::RETCODE_ERROR, and DDS::RETCODE_OK.

 {
   const DDS::InstanceHandle_t publication_handle = participant_servant_->lookup_handle(id);
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                    guard,
                    publication_handle_lock_,
                    DDS::RETCODE_ERROR);
 
   DDS::PublicationBuiltinTopicDataSeq data;
 
   DDS::ReturnCode_t const ret = instance_handle_to_bit_data<DDS::PublicationBuiltinTopicDataDataReader_var>(
                             participant_servant_,
                             BUILT_IN_PUBLICATION_TOPIC,
                             publication_handle,
                             data);
 
   if (ret == DDS::RETCODE_OK) {
     key = data[0].key;
   }
 
   return ret;
 }

◆ set_listener()

DDS::ReturnCode_t OpenDDS::DCPS::RecorderImpl::set_listener	(	const RecorderListener_rch &	a_listener,
		DDS::StatusMask	mask
	)

virtual

Change the listener for this Recorder.

Implements OpenDDS::DCPS::Recorder.

Definition at line 826 of file RecorderImpl.cpp.

References listener_, listener_mask_, and DDS::RETCODE_OK.

 {
   listener_mask_ = mask;
   //note: OK to duplicate  a nil object ref
   listener_ = a_listener;
   return DDS::RETCODE_OK;
 }

◆ set_qos()

DDS::ReturnCode_t OpenDDS::DCPS::RecorderImpl::set_qos	(	const DDS::SubscriberQos &	subscriber_qos,
		const DDS::DataReaderQos &	datareader_qos
	)

virtual

Set the Quality of Service settings for the Recorder.

Implements OpenDDS::DCPS::Recorder.

Definition at line 757 of file RecorderImpl.cpp.

References ACE_ERROR, 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(), OpenDDS::DCPS::EntityImpl::is_enabled(), LM_NOTICE, OpenDDS::DCPS::log_level, OpenDDS::DCPS::LogLevel::Notice, OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK, OPENDDS_NO_OBJECT_MODEL_PROFILE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK, participant_servant_, qos_, DDS::RETCODE_ERROR, DDS::RETCODE_IMMUTABLE_POLICY, DDS::RETCODE_INCONSISTENT_POLICY, DDS::RETCODE_OK, DDS::RETCODE_UNSUPPORTED, subqos_, OpenDDS::DCPS::WriterInfoListener::subscription_id_, TheServiceParticipant, and OpenDDS::DCPS::Qos_Helper::valid().

 {
   OPENDDS_NO_OBJECT_MODEL_PROFILE_COMPATIBILITY_CHECK(subscriber_qos, DDS::RETCODE_UNSUPPORTED);
 
   if (Qos_Helper::valid(subscriber_qos) && Qos_Helper::consistent(subscriber_qos)) {
     if (subqos_ != subscriber_qos) {
       // for the not changeable qos, it can be changed before enable
       if (!Qos_Helper::changeable(subqos_, subscriber_qos) && enabled_) {
         return DDS::RETCODE_IMMUTABLE_POLICY;
 
       } else {
         subqos_ = subscriber_qos;
       }
     }
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 
   OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
   OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
   OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK(qos, DDS::RETCODE_UNSUPPORTED);
 
   if (Qos_Helper::valid(qos) && Qos_Helper::consistent(qos)) {
     if (qos_ == qos)
       return DDS::RETCODE_OK;
 
     if (!Qos_Helper::changeable(qos_, qos) && is_enabled()) {
       return DDS::RETCODE_IMMUTABLE_POLICY;
 
     } else {
       Discovery_rch disco = TheServiceParticipant->get_discovery(domain_id_);
       const bool status =
         disco->update_subscription_qos(
           participant_servant_->get_domain_id(),
           participant_servant_->get_id(),
           subscription_id_,
           qos,
           subscriber_qos);
       if (!status) {
         if (log_level >= LogLevel::Notice) {
           ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: RecorderImpl::set_qos: qos not updated\n"));
         }
         return DDS::RETCODE_ERROR;
       }
     }
 
     qos_ = qos;
     subqos_ = subscriber_qos;
 
     return DDS::RETCODE_OK;
 
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ signal_liveliness()

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

virtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 728 of file RecorderImpl.cpp.

References ACE_GUARD, ACE_READ_GUARD, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::equal_guid_prefixes(), OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OpenDDS::DCPS::TransportClient::OPENDDS_VECTOR(), sample_lock_, writers_, and writers_lock_.

 {
   GUID_t prefix = remote_participant;
   prefix.entityId = EntityId_t();
 
   ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, sample_lock_);
 
   typedef std::pair<GUID_t, RcHandle<WriterInfo> > WriterSetElement;
   typedef OPENDDS_VECTOR(WriterSetElement) WriterSet;
   WriterSet writers;
 
   {
     ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, writers_lock_);
     for (WriterMapType::iterator pos = writers_.lower_bound(prefix),
            limit = writers_.end();
          pos != limit && equal_guid_prefixes(pos->first, prefix);
          ++pos) {
       writers.push_back(std::make_pair(pos->first, pos->second));
     }
   }
 
   const MonotonicTimePoint now = MonotonicTimePoint::now();
   for (WriterSet::iterator pos = writers.begin(), limit = writers.end();
        pos != limit;
        ++pos) {
     pos->second->received_activity(now);
   }
 }

◆ unregister_for_writer()

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

virtual

Reimplemented from OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 966 of file RecorderImpl.cpp.

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

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

◆ update_incompatible_qos()

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

virtual

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 690 of file RecorderImpl.cpp.

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

Friends And Related Function Documentation

◆ ::DDS_TEST

friend class ::DDS_TEST

friend

Definition at line 181 of file RecorderImpl.h.

Member Data Documentation

◆ check_encap_

bool OpenDDS::DCPS::RecorderImpl::check_encap_

private

Definition at line 212 of file RecorderImpl.h.

Referenced by data_received().

◆ domain_id_

DDS::DomainId_t OpenDDS::DCPS::RecorderImpl::domain_id_

private

Definition at line 186 of file RecorderImpl.h.

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

◆ dt_map_

DynamicTypeByPubId OpenDDS::DCPS::RecorderImpl::dt_map_

private

Definition at line 210 of file RecorderImpl.h.

Referenced by add_to_dynamic_type_map(), get_dynamic_data(), and remove_associations_i().

◆ id_to_handle_map_

RepoIdToHandleMap OpenDDS::DCPS::RecorderImpl::id_to_handle_map_

private

Definition at line 191 of file RecorderImpl.h.

Referenced by add_association(), and remove_associations_i().

◆ is_bit_

bool OpenDDS::DCPS::RecorderImpl::is_bit_

private

Flag indicates that this datareader is a builtin topic datareader.

Definition at line 198 of file RecorderImpl.h.

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

◆ is_exclusive_ownership_

bool OpenDDS::DCPS::RecorderImpl::is_exclusive_ownership_

private

Definition at line 174 of file RecorderImpl.h.

Referenced by init().

◆ listener_

RecorderListener_rch OpenDDS::DCPS::RecorderImpl::listener_

private

Definition at line 185 of file RecorderImpl.h.

Referenced by add_association(), data_received(), get_listener(), init(), remove_associations_i(), and set_listener().

◆ listener_mask_

DDS::StatusMask OpenDDS::DCPS::RecorderImpl::listener_mask_

private

Definition at line 184 of file RecorderImpl.h.

Referenced by init(), and set_listener().

◆ mb_alloc_

TransportMessageBlockAllocator OpenDDS::DCPS::RecorderImpl::mb_alloc_

private

Definition at line 214 of file RecorderImpl.h.

Referenced by data_received().

◆ owner_manager_

OwnershipManager* OpenDDS::DCPS::RecorderImpl::owner_manager_

private

Definition at line 176 of file RecorderImpl.h.

Referenced by init().

◆ participant_servant_

DomainParticipantImpl* OpenDDS::DCPS::RecorderImpl::participant_servant_

private

Definition at line 170 of file RecorderImpl.h.

Referenced by add_association(), add_to_dynamic_type_map(), cleanup(), enable(), get_instance_handle(), init(), lookup_instance_handles(), remove_associations_i(), repoid_to_bit_key(), and set_qos().

◆ passed_qos_

DDS::DataReaderQos OpenDDS::DCPS::RecorderImpl::passed_qos_

private

Definition at line 165 of file RecorderImpl.h.

Referenced by get_qos(), and init().

◆ publication_handle_lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::RecorderImpl::publication_handle_lock_

private

Definition at line 188 of file RecorderImpl.h.

Referenced by add_association(), remove_all_associations(), remove_associations_i(), repoid_to_bit_key(), and update_incompatible_qos().

◆ qos_

DDS::DataReaderQos OpenDDS::DCPS::RecorderImpl::qos_

private

Definition at line 164 of file RecorderImpl.h.

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

◆ requested_incompatible_qos_status_

DDS::RequestedIncompatibleQosStatus OpenDDS::DCPS::RecorderImpl::requested_incompatible_qos_status_

private

Definition at line 193 of file RecorderImpl.h.

Referenced by RecorderImpl(), and update_incompatible_qos().

◆ sample_lock_

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::RecorderImpl::sample_lock_

private

lock protecting sample container as well as statuses.

Definition at line 168 of file RecorderImpl.h.

Referenced by add_association(), data_received(), and signal_liveliness().

◆ subqos_

DDS::SubscriberQos OpenDDS::DCPS::RecorderImpl::subqos_

private

Definition at line 179 of file RecorderImpl.h.

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

◆ subscription_match_status_

DDS::SubscriptionMatchedStatus OpenDDS::DCPS::RecorderImpl::subscription_match_status_

private

Definition at line 194 of file RecorderImpl.h.

Referenced by add_association(), RecorderImpl(), and remove_associations_i().

◆ topic_desc_

DDS::TopicDescription_var OpenDDS::DCPS::RecorderImpl::topic_desc_

private

Definition at line 183 of file RecorderImpl.h.

Referenced by init().

◆ topic_servant_

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

private

Definition at line 171 of file RecorderImpl.h.

Referenced by enable(), and init().

◆ writers_

WriterMapType OpenDDS::DCPS::RecorderImpl::writers_

private

Definition at line 203 of file RecorderImpl.h.

Referenced by add_association(), remove_all_associations(), remove_associations_i(), and signal_liveliness().

◆ writers_lock_

ACE_RW_Thread_Mutex OpenDDS::DCPS::RecorderImpl::writers_lock_

private

RW lock for reading/writing publications.

Definition at line 206 of file RecorderImpl.h.

Referenced by add_association(), remove_all_associations(), remove_associations_i(), and signal_liveliness().

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

Public Member Functions

Protected Member Functions

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ RecorderImpl()

◆ ~RecorderImpl()

Member Function Documentation

◆ add_association()

◆ add_to_dynamic_type_map()

◆ check_encap() [1/2]

◆ check_encap() [2/2]

◆ check_transport_qos()

◆ cleanup()

◆ data_received()

◆ domain_id()

◆ enable()

◆ get_dynamic_data()

◆ get_guid()

◆ get_ice_endpoint()

◆ get_instance_handle()

◆ get_listener()

◆ get_priority_value()

◆ get_qos()

◆ init()

◆ lookup_instance_handles()

◆ notify_subscription_disconnected()

◆ notify_subscription_lost() [1/2]

◆ notify_subscription_lost() [2/2]

◆ notify_subscription_reconnected()

◆ OPENDDS_MAP()

◆ OPENDDS_MAP_CMP() [1/2]

◆ OPENDDS_MAP_CMP() [2/2]

◆ participant()

◆ register_for_writer()

◆ remove_all_associations()

◆ remove_associations()

◆ remove_associations_i()

◆ repoid_to_bit_key()

◆ set_listener()

◆ set_qos()

◆ signal_liveliness()

◆ unregister_for_writer()

◆ update_incompatible_qos()

Friends And Related Function Documentation

◆ ::DDS_TEST

Member Data Documentation

◆ check_encap_

◆ domain_id_

◆ dt_map_

◆ id_to_handle_map_

◆ is_bit_

◆ is_exclusive_ownership_

◆ listener_

◆ listener_mask_

◆ mb_alloc_

◆ owner_manager_

◆ participant_servant_

◆ passed_qos_

◆ publication_handle_lock_

◆ qos_

◆ requested_incompatible_qos_status_

◆ sample_lock_

◆ subqos_

◆ subscription_match_status_

◆ topic_desc_

◆ topic_servant_

◆ writers_

◆ writers_lock_