Implements the OpenDDS::DCPS::Publisher interfaces. More...

#include <PublisherImpl.h>

Inheritance diagram for OpenDDS::DCPS::PublisherImpl:

[legend]

Collaboration diagram for OpenDDS::DCPS::PublisherImpl:

[legend]

Public Member Functions
	PublisherImpl (DDS::InstanceHandle_t handle, GUID_t id, const DDS::PublisherQos &qos, DDS::PublisherListener_ptr a_listener, const DDS::StatusMask &mask, DomainParticipantImpl *participant)

virtual	~PublisherImpl ()

virtual DDS::InstanceHandle_t	get_instance_handle ()

bool	contains_writer (DDS::InstanceHandle_t a_handle)

virtual DDS::DataWriter_ptr	create_datawriter (DDS::Topic_ptr a_topic, const DDS::DataWriterQos &qos, DDS::DataWriterListener_ptr a_listener, DDS::StatusMask mask)

virtual DDS::ReturnCode_t	delete_datawriter (DDS::DataWriter_ptr a_datawriter)

virtual DDS::DataWriter_ptr	lookup_datawriter (const char *topic_name)

virtual DDS::ReturnCode_t	delete_contained_entities ()

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

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

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

virtual DDS::PublisherListener_ptr	get_listener ()

virtual DDS::ReturnCode_t	suspend_publications ()

virtual DDS::ReturnCode_t	resume_publications ()

virtual DDS::ReturnCode_t	begin_coherent_changes ()

virtual DDS::ReturnCode_t	end_coherent_changes ()

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

virtual DDS::DomainParticipant_ptr	get_participant ()

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

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

virtual DDS::ReturnCode_t	copy_from_topic_qos (DDS::DataWriterQos &a_datawriter_qos, const DDS::TopicQos &a_topic_qos)

virtual DDS::ReturnCode_t	enable ()

ACE_Recursive_Thread_Mutex &	get_pi_lock ()

bool	is_clean (String *leftover_entities=0) const

DDS::ReturnCode_t	writer_enabled (const char topic_name, DataWriterImpl impl)

DDS::PublisherListener_ptr	listener_for (::DDS::StatusKind kind)

DDS::ReturnCode_t	assert_liveliness_by_participant ()

TimeDuration	liveliness_check_interval (DDS::LivelinessQosPolicyKind kind)

bool	participant_liveliness_activity_after (const MonotonicTimePoint &tv)

typedef	OPENDDS_VECTOR (GUID_t) PublicationIdVec

void	get_publication_ids (PublicationIdVec &pubs)

bool	is_suspended () const

virtual RcHandle< EntityImpl >	parent () const

bool	prepare_to_delete_datawriters ()

bool	set_wait_pending_deadline (const MonotonicTimePoint &deadline)

Public Member Functions inherited from DDS::Publisher
DataWriter	create_datawriter (in Topic a_topic, in DataWriterQos qos, in DataWriterListener a_listener, in StatusMask mask)

ReturnCode_t	delete_datawriter (in DataWriter a_datawriter)

DataWriter	lookup_datawriter (in string topic_name)

ReturnCode_t	set_qos (in PublisherQos qos)

ReturnCode_t	get_qos (inout PublisherQos qos)

ReturnCode_t	set_listener (in PublisherListener a_listener, in StatusMask mask)

ReturnCode_t	wait_for_acknowledgments (in Duration_t max_wait)

ReturnCode_t	set_default_datawriter_qos (in DataWriterQos qos)

ReturnCode_t	get_default_datawriter_qos (inout DataWriterQos qos)

ReturnCode_t	copy_from_topic_qos (inout DataWriterQos a_datawriter_qos, in TopicQos a_topic_qos)

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

virtual	~EntityImpl ()

bool	is_enabled () const

virtual DDS::StatusCondition_ptr	get_statuscondition ()

virtual DDS::StatusMask	get_status_changes ()

virtual DDS::DomainId_t	get_domain_id ()

virtual GUID_t	get_id () const

void	set_status_changed_flag (DDS::StatusKind status, bool status_changed_flag)

void	notify_status_condition ()

virtual void	transport_config (const TransportConfig_rch &cfg)

TransportConfig_rch	transport_config () const

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

Observer_rch	get_observer (Observer::Event e)

Static Public Member Functions
static bool	validate_datawriter_qos (const DDS::DataWriterQos &qos, const DDS::DataWriterQos &default_qos, DDS::Topic_ptr a_topic, DDS::DataWriterQos &dw_qos)

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

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

static LocalObject_ptr	_nil (void)

static LocalObject_ptr	_narrow (CORBA::Object_ptr obj)

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

static void	_tao_any_destructor (void *)

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

static void	tao_object_initialize (Object *)

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

static CORBA::Object_ptr	_nil (void)

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

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

Private Types
typedef ACE_Recursive_Thread_Mutex	lock_type

typedef ACE_Reverse_Lock< lock_type >	reverse_lock_type

Private Member Functions
typedef	OPENDDS_MULTIMAP (OPENDDS_STRING, DataWriterImpl_rch) DataWriterMap

typedef	OPENDDS_MAP_CMP (GUID_t, DataWriterImpl_rch, GUID_tKeyLessThan) PublicationMap

typedef	OPENDDS_MAP_CMP (GUID_t, DDS::DataWriterQos, GUID_tKeyLessThan) DwIdToQosMap

typedef	OPENDDS_SET (DataWriterImpl_rch) DataWriterSet

Private Attributes
DDS::InstanceHandle_t	handle_

DDS::PublisherQos	qos_
	Publisher QoS policy list. More...

DDS::DataWriterQos	default_datawriter_qos_
	Default datawriter Qos policy list. More...

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

DDS::StatusMask	listener_mask_

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

DataWriterSet	writers_not_enabled_

DataWriterMap	datawriter_map_
	This map is used to support datawriter lookup by topic name. More...

PublicationMap	publication_map_

std::size_t	change_depth_
	The number of times begin_coherent_changes as been called. More...

DDS::DomainId_t	domain_id_
	Domain in which we are contained. More...

WeakRcHandle< DomainParticipantImpl >	participant_
	The DomainParticipant servant that owns this Publisher. More...

CORBA::Short	suspend_depth_count_
	The suspend depth count. More...

SequenceNumber	sequence_number_

MonotonicTimePoint	aggregation_period_start_
	Start of current aggregation period. - NOT USED IN FIRST IMPL. More...

lock_type	pi_lock_
	The recursive lock to protect datawriter map and suspend count. More...

reverse_lock_type	reverse_pi_lock_

lock_type	pi_suspended_lock_

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

GUID_t	publisher_id_

Friends
class	DataWriterImpl

Additional Inherited Members
Public Types inherited from OpenDDS::DCPS::LocalObject< DDS::Publisher >
typedef DDS::Publisher ::_ptr_type	_ptr_type

typedef DDS::Publisher ::_var_type	_var_type

Public Types inherited from CORBA::LocalObject
typedef LocalObject_ptr	_ptr_type

typedef LocalObject_var	_var_type

typedef LocalObject_out	_out_type

Public Types inherited from CORBA::Object
typedef Object_ptr	_ptr_type

typedef Object_var	_var_type

typedef Object_out	_out_type

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

typedef DDS::Entity ::_var_type	_var_type

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

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

TAO::Object_Proxy_Broker *	proxy_broker () const

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

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

void	set_deleted (bool state)

bool	get_deleted () const

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

Protected 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

Implements the OpenDDS::DCPS::Publisher interfaces.

This class acts as a factory and container of the datawriter.

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

Definition at line 38 of file PublisherImpl.h.

Member Typedef Documentation

◆ lock_type

typedef ACE_Recursive_Thread_Mutex OpenDDS::DCPS::PublisherImpl::lock_type

private

Definition at line 205 of file PublisherImpl.h.

◆ reverse_lock_type

typedef ACE_Reverse_Lock<lock_type> OpenDDS::DCPS::PublisherImpl::reverse_lock_type

private

Definition at line 206 of file PublisherImpl.h.

Constructor & Destructor Documentation

◆ PublisherImpl()

OpenDDS::DCPS::PublisherImpl::PublisherImpl	(	DDS::InstanceHandle_t	handle,
		GUID_t	id,
		const DDS::PublisherQos &	qos,
		DDS::PublisherListener_ptr	a_listener,
		const DDS::StatusMask &	mask,
		DomainParticipantImpl *	participant
	)

Definition at line 29 of file PublisherImpl.cpp.

References _duplicate(), monitor_, and TheServiceParticipant.

 : handle_(handle),
   qos_(qos),
   default_datawriter_qos_(TheServiceParticipant->initial_DataWriterQos()),
   listener_mask_(mask),
   listener_(DDS::PublisherListener::_duplicate(a_listener)),
 #ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
   change_depth_(0),
 #endif
   domain_id_(participant->get_domain_id()),
   participant_(*participant),
   suspend_depth_count_(0),
   sequence_number_(),
   reverse_pi_lock_(pi_lock_),
   publisher_id_(id)
 {
   monitor_.reset(TheServiceParticipant->monitor_factory_->create_publisher_monitor(this));
 }

◆ ~PublisherImpl()

OpenDDS::DCPS::PublisherImpl::~PublisherImpl ( )

virtual

Definition at line 53 of file PublisherImpl.cpp.

References ACE_ERROR, handle_, is_clean(), LM_WARNING, OpenDDS::DCPS::log_level, participant_, and OpenDDS::DCPS::LogLevel::Warning.

 {
   const RcHandle<DomainParticipantImpl> participant = participant_.lock();
   if (participant) {
     participant->return_handle(handle_);
   }
 
   // The datawriters should be deleted already before calling delete
   // publisher.
   String leftover_entities;
   if (!is_clean(&leftover_entities)) {
     if (log_level >= LogLevel::Warning) {
       ACE_ERROR((LM_WARNING, "(%P|%t) WARNING: PublisherImpl::~PublisherImpl: "
                  "%C still exist\n", leftover_entities.c_str()));
     }
   }
 }

Member Function Documentation

◆ assert_liveliness_by_participant()

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

Definition at line 960 of file PublisherImpl.cpp.

References datawriter_map_, and DDS::RETCODE_OK.

 {
   DDS::ReturnCode_t ret = DDS::RETCODE_OK;
 
   for (DataWriterMap::iterator it(datawriter_map_.begin());
       it != datawriter_map_.end(); ++it) {
     const DDS::ReturnCode_t dw_ret = it->second->assert_liveliness_by_participant();
 
     if (dw_ret != DDS::RETCODE_OK) {
       ret = dw_ret;
     }
   }
 
   return ret;
 }

◆ begin_coherent_changes()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::begin_coherent_changes ( )

virtual

Implements DDS::Publisher.

Definition at line 609 of file PublisherImpl.cpp.

References DDS::PresentationQosPolicy::access_scope, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), change_depth_, DDS::PresentationQosPolicy::coherent_access, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::EntityImpl::enabled_, DDS::INSTANCE_PRESENTATION_QOS, LM_ERROR, pi_lock_, DDS::PublisherQos::presentation, publication_map_, qos_, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, and DDS::RETCODE_OK.

 {
   if (!enabled_) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: PublisherImpl::begin_coherent_changes:")
           ACE_TEXT(" Publisher is not enabled!\n")));
     }
     return DDS::RETCODE_NOT_ENABLED;
   }
 
   if (!qos_.presentation.coherent_access) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: PublisherImpl::begin_coherent_changes:")
           ACE_TEXT(" QoS policy does not support coherent access!\n")));
     }
     return DDS::RETCODE_ERROR;
   }
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       guard,
       this->pi_lock_,
       DDS::RETCODE_ERROR);
 
   ++this->change_depth_;
 
   if (qos_.presentation.access_scope == DDS::INSTANCE_PRESENTATION_QOS) {
     // INSTANCE access scope essentially behaves
     // as a no-op. (see: 7.1.3.6)
     return DDS::RETCODE_OK;
   }
 
   // We should only notify publications on the first
   // and last change to the current change set:
   if (this->change_depth_ == 1) {
     for (PublicationMap::iterator it = this->publication_map_.begin();
         it != this->publication_map_.end(); ++it) {
       it->second->begin_coherent_changes();
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ contains_writer()

bool OpenDDS::DCPS::PublisherImpl::contains_writer ( DDS::InstanceHandle_t a_handle )

Definition at line 78 of file PublisherImpl.cpp.

References ACE_GUARD_RETURN, datawriter_map_, pi_lock_, and DDS::RETCODE_ERROR.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       guard,
       this->pi_lock_,
       DDS::RETCODE_ERROR);
 
   for (DataWriterMap::iterator it(datawriter_map_.begin());
       it != datawriter_map_.end(); ++it) {
     if (a_handle == it->second->get_instance_handle()) {
       return true;
     }
   }
 
   return false;
 }

◆ copy_from_topic_qos()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::copy_from_topic_qos	(	DDS::DataWriterQos &	a_datawriter_qos,
		const DDS::TopicQos &	a_topic_qos
	)

virtual

Definition at line 831 of file PublisherImpl.cpp.

References OpenDDS::DCPS::Qos_Helper::copy_from_topic_qos(), DDS::RETCODE_INCONSISTENT_POLICY, and DDS::RETCODE_OK.

 {
   if (Qos_Helper::copy_from_topic_qos(a_datawriter_qos, a_topic_qos)) {
     return DDS::RETCODE_OK;
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ create_datawriter()

DDS::DataWriter_ptr OpenDDS::DCPS::PublisherImpl::create_datawriter	(	DDS::Topic_ptr	a_topic,
		const DDS::DataWriterQos &	qos,
		DDS::DataWriterListener_ptr	a_listener,
		DDS::StatusMask	mask
	)

virtual

Definition at line 96 of file PublisherImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), DDS::EntityFactoryQosPolicy::autoenable_created_entities, OpenDDS::DCPS::DCPS_debug_level, default_datawriter_qos_, OpenDDS::DCPS::DataWriterImpl::enable(), OpenDDS::DCPS::EntityImpl::enabled_, DDS::PublisherQos::entity_factory, OpenDDS::DCPS::TopicDescriptionImpl::get_name(), OpenDDS::DCPS::TopicDescriptionImpl::get_type_support(), TAO::String_var< charT >::in(), OpenDDS::DCPS::DataWriterImpl::init(), LM_ERROR, LM_WARNING, name, participant_, pi_lock_, qos_, OpenDDS::DCPS::rchandle_from(), DDS::RETCODE_OK, validate_datawriter_qos(), and writers_not_enabled_.

 {
   DDS::DataWriterQos dw_qos;
 
   if (!validate_datawriter_qos(qos, default_datawriter_qos_, a_topic, dw_qos)) {
     return DDS::DataWriter::_nil();
   }
 
   TopicImpl* topic_servant = dynamic_cast<TopicImpl*>(a_topic);
 
   if (!topic_servant) {
     if (DCPS_debug_level > 0) {
       CORBA::String_var name = a_topic->get_name();
       ACE_ERROR((LM_ERROR,
         ACE_TEXT("(%P|%t) ERROR: ")
         ACE_TEXT("PublisherImpl::create_datawriter, ")
         ACE_TEXT("topic_servant(topic_name=%C) is nil.\n"),
         name.in()));
     }
     return 0;
   }
 
   OpenDDS::DCPS::TypeSupport_ptr typesupport =
       topic_servant->get_type_support();
 
   if (typesupport == 0) {
     if (DCPS_debug_level > 0) {
       CORBA::String_var name = topic_servant->get_name();
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("PublisherImpl::create_datawriter, ")
           ACE_TEXT("typesupport(topic_name=%C) is nil.\n"),
           name.in()));
     }
     return DDS::DataWriter::_nil();
   }
 
   DDS::DataWriter_var dw_obj = typesupport->create_datawriter();
 
   DataWriterImpl* dw_servant =
       dynamic_cast <DataWriterImpl*>(dw_obj.in());
 
   if (dw_servant == 0) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("PublisherImpl::create_datawriter, ")
           ACE_TEXT("servant is nil.\n")));
     }
     return DDS::DataWriter::_nil();
   }
 
   dw_servant->init(
       topic_servant,
       dw_qos,
       a_listener,
       mask,
       participant_,
       this);
 
   if (enabled_ && qos_.entity_factory.autoenable_created_entities) {
     const DDS::ReturnCode_t ret = dw_servant->enable();
 
     if (ret != DDS::RETCODE_OK) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_WARNING,
             ACE_TEXT("(%P|%t) WARNING: ")
             ACE_TEXT("PublisherImpl::create_datawriter, ")
             ACE_TEXT("enable failed.\n")));
       }
       return DDS::DataWriter::_nil();
     }
   } else {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, pi_lock_, 0);
     writers_not_enabled_.insert(rchandle_from(dw_servant));
   }
 
   return DDS::DataWriter::_duplicate(dw_obj.in());
 }

◆ delete_contained_entities()

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

virtual

Implements DDS::Publisher.

Definition at line 369 of file PublisherImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), datawriter_map_, OpenDDS::DCPS::DCPS_debug_level, delete_datawriter(), OpenDDS::DCPS::EntityImpl::get_deleted(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::RcHandle< T >::in(), LM_ERROR, pi_lock_, prepare_to_delete_datawriters(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_deleted(), set_wait_pending_deadline(), and TheServiceParticipant.

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

 {
   // If the call isn't part of another delete, prepare the datawriters to be
   // deleted and set the pending deadline on all the writers.
   if (!get_deleted()) {
     // mark that the entity is being deleted
     set_deleted(true);
 
     if (!prepare_to_delete_datawriters()) {
       return DDS::RETCODE_ERROR;
     }
     if (!set_wait_pending_deadline(TheServiceParticipant->new_pending_timeout_deadline())) {
       return DDS::RETCODE_ERROR;
     }
   }
 
   while (true) {
     GUID_t pub_id = GUID_UNKNOWN;
     DataWriterImpl_rch a_datawriter;
 
     {
       ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
           guard,
           this->pi_lock_,
           DDS::RETCODE_ERROR);
 
       if (datawriter_map_.empty()) {
         break;
       } else {
         a_datawriter = datawriter_map_.begin()->second;
         pub_id = a_datawriter->get_guid();
       }
     }
 
     const DDS::ReturnCode_t ret = delete_datawriter(a_datawriter.in());
 
     if (ret != DDS::RETCODE_OK) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
             ACE_TEXT("(%P|%t) ERROR: ")
             ACE_TEXT("PublisherImpl::")
             ACE_TEXT("delete_contained_entities: ")
             ACE_TEXT("failed to delete ")
             ACE_TEXT("datawriter %C.\n"),
             LogGuid(pub_id).c_str()));
       }
       return ret;
     }
   }
 
   // the publisher can now start creating new publications
   set_deleted(false);
 
   return DDS::RETCODE_OK;
 }

◆ delete_datawriter()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::delete_datawriter ( DDS::DataWriter_ptr a_datawriter )

virtual

Definition at line 181 of file PublisherImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DataWriterImpl::cleanup(), datawriter_map_, OpenDDS::DCPS::DCPS_debug_level, domain_id_, OpenDDS::DCPS::EntityImpl::get_deleted(), OpenDDS::DCPS::DataWriterImpl::get_guid(), OpenDDS::DCPS::DataWriterImpl::get_publisher(), OpenDDS::DCPS::GUID_UNKNOWN, LM_ERROR, monitor_, participant_, pi_lock_, OpenDDS::DCPS::DataWriterImpl::prepare_to_delete(), publication_map_, OpenDDS::DCPS::DataWriterImpl::remove_all_associations(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, DDS::RETCODE_PRECONDITION_NOT_MET, reverse_pi_lock_, OpenDDS::DCPS::DataWriterImpl::set_wait_pending_deadline(), TheServiceParticipant, OpenDDS::DCPS::DataWriterImpl::unregister_all(), and OpenDDS::DCPS::DataWriterImpl::wait_pending().

Referenced by delete_contained_entities().

 {
   DataWriterImpl* dw_servant = dynamic_cast<DataWriterImpl*>(a_datawriter);
   if (!dw_servant) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
                 "(%P|%t) PublisherImpl::delete_datawriter - dynamic cast to DataWriterImpl failed\n"));
     }
     return DDS::RETCODE_ERROR;
   }
 
   {
     DDS::Publisher_var dw_publisher(dw_servant->get_publisher());
 
     if (dw_publisher.in() != this) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
             ACE_TEXT("(%P|%t) PublisherImpl::delete_datawriter: ")
             ACE_TEXT("the data writer %C doesn't ")
             ACE_TEXT("belong to this subscriber\n"),
             LogGuid(dw_servant->get_guid()).c_str()));
       }
       return DDS::RETCODE_PRECONDITION_NOT_MET;
     }
   }
 
   if (!dw_servant->get_deleted()) {
     dw_servant->prepare_to_delete();
     dw_servant->set_wait_pending_deadline(TheServiceParticipant->new_pending_timeout_deadline());
   }
 
   // Wait for any data and control messages to be transported during
   // unregistering of instances.
   dw_servant->wait_pending();
 
   GUID_t publication_id  = GUID_UNKNOWN;
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
         guard,
         this->pi_lock_,
         DDS::RETCODE_ERROR);
 
     publication_id = dw_servant->get_guid();
 
     PublicationMap::iterator it = publication_map_.find(publication_id);
 
     if (it == publication_map_.end()) {
       if (DCPS_debug_level > 0) {
         ACE_ERROR((LM_ERROR,
             ACE_TEXT("(%P|%t) ERROR: ")
             ACE_TEXT("PublisherImpl::delete_datawriter, ")
             ACE_TEXT("datawriter %C not found.\n"),
             LogGuid(publication_id).c_str()));
       }
       return DDS::RETCODE_ERROR;
     }
 
     // We can not erase the datawriter from datawriter map by the topic name
     // because the map might have multiple datawriters with the same topic
     // name.
     // Find the iterator to the datawriter in the datawriter map and erase
     // by the iterator.
     DataWriterMap::iterator writ;
     DataWriterMap::iterator the_writ = datawriter_map_.end();
 
     for (writ = datawriter_map_.begin();
         writ != datawriter_map_.end();
         ++writ) {
       if (writ->second == it->second) {
         the_writ = writ;
         break;
       }
     }
 
     if (the_writ != datawriter_map_.end()) {
       datawriter_map_.erase(the_writ);
     }
 
     publication_map_.erase(it);
 
     // not just unregister but remove any pending writes/sends.
     dw_servant->unregister_all();
 
     // Release pi_lock_ before making call to transport layer to avoid
     // some deadlock situations that threads acquire locks(PublisherImpl
     // pi_lock_, TransportClient reservation_lock and TransportImpl
     // lock_) in reverse order.
     ACE_GUARD_RETURN(reverse_lock_type, reverse_monitor, this->reverse_pi_lock_,
         DDS::RETCODE_ERROR);
     // Wait for pending samples to drain prior to removing associations
     // and unregistering the publication.
     dw_servant->wait_pending();
 
     // Call remove association before unregistering the datawriter
     // with the transport, otherwise some callbacks resulted from
     // remove_association may lost.
     dw_servant->remove_all_associations();
     dw_servant->cleanup();
   }
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 
   RcHandle<DomainParticipantImpl> participant = this->participant_.lock();
 
   Discovery_rch disco = TheServiceParticipant->get_discovery(this->domain_id_);
   if (!disco->remove_publication(
       this->domain_id_,
       participant->get_id(),
       publication_id)) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("PublisherImpl::delete_datawriter, ")
           ACE_TEXT("publication not removed from discovery.\n")));
     }
     return DDS::RETCODE_ERROR;
   }
 
   participant->remove_adjust_liveliness_timers();
 
   return DDS::RETCODE_OK;
 }

◆ enable()

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

virtual

Implements DDS::Entity.

Definition at line 842 of file PublisherImpl.cpp.

References ACE_GUARD_RETURN, DDS::EntityFactoryQosPolicy::autoenable_created_entities, DDS::PublisherQos::entity_factory, OpenDDS::DCPS::EntityImpl::is_enabled(), monitor_, participant_, pi_lock_, qos_, DDS::RETCODE_ERROR, DDS::RETCODE_OK, DDS::RETCODE_PRECONDITION_NOT_MET, OpenDDS::DCPS::EntityImpl::set_enabled(), and writers_not_enabled_.

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

 {
   //According spec:
   // - Calling enable on an already enabled Entity returns OK and has no
   // effect.
   // - Calling enable on an Entity whose factory is not enabled will fail
   // and return PRECONDITION_NOT_MET.
 
   if (this->is_enabled()) {
     return DDS::RETCODE_OK;
   }
 
   RcHandle<DomainParticipantImpl> participant = this->participant_.lock();
   if (!participant || !participant->is_enabled()) {
     return DDS::RETCODE_PRECONDITION_NOT_MET;
   }
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 
   this->set_enabled();
 
   if (qos_.entity_factory.autoenable_created_entities) {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, pi_lock_, DDS::RETCODE_ERROR);
     DataWriterSet writers;
     writers_not_enabled_.swap(writers);
     for (DataWriterSet::iterator it = writers.begin(); it != writers.end(); ++it) {
       (*it)->enable();
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ end_coherent_changes()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::end_coherent_changes ( )

virtual

Implements DDS::Publisher.

Definition at line 655 of file PublisherImpl.cpp.

References DDS::PresentationQosPolicy::access_scope, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), change_depth_, DDS::PresentationQosPolicy::coherent_access, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::EntityImpl::enabled_, DDS::INSTANCE_PRESENTATION_QOS, LM_ERROR, pi_lock_, DDS::PublisherQos::presentation, publication_map_, qos_, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, and DDS::RETCODE_PRECONDITION_NOT_MET.

 {
   if (!enabled_) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: PublisherImpl::end_coherent_changes:")
           ACE_TEXT(" Publisher is not enabled!\n")));
     }
     return DDS::RETCODE_NOT_ENABLED;
   }
 
   if (!qos_.presentation.coherent_access) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: PublisherImpl::end_coherent_changes:")
           ACE_TEXT(" QoS policy does not support coherent access!\n")));
     }
     return DDS::RETCODE_ERROR;
   }
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       guard,
       this->pi_lock_,
       DDS::RETCODE_ERROR);
 
   if (this->change_depth_ == 0) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: PublisherImpl::end_coherent_changes:")
           ACE_TEXT(" No matching call to begin_coherent_changes!\n")));
     }
     return DDS::RETCODE_PRECONDITION_NOT_MET;
   }
 
   --this->change_depth_;
 
   if (qos_.presentation.access_scope == DDS::INSTANCE_PRESENTATION_QOS) {
     // INSTANCE access scope essentially behaves
     // as a no-op. (see: 7.1.3.6)
     return DDS::RETCODE_OK;
   }
 
   // We should only notify publications on the first
   // and last change to the current change set:
   if (this->change_depth_ == 0) {
     GroupCoherentSamples group_samples;
     for (PublicationMap::iterator it = this->publication_map_.begin();
         it != this->publication_map_.end(); ++it) {
 
       if (it->second->coherent_samples_ == 0) {
         continue;
       }
 
       std::pair<GroupCoherentSamples::iterator, bool> pair =
           group_samples.insert(GroupCoherentSamples::value_type(
               it->second->get_guid(),
               WriterCoherentSample(it->second->coherent_samples_,
                   it->second->sequence_number_)));
 
       if (!pair.second) {
         if (DCPS_debug_level > 0) {
           ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) ERROR: PublisherImpl::end_coherent_changes: ")
               ACE_TEXT("failed to insert to GroupCoherentSamples.\n")));
         }
         return DDS::RETCODE_ERROR;
       }
     }
 
     for (PublicationMap::iterator it = this->publication_map_.begin();
         it != this->publication_map_.end(); ++it) {
       if (it->second->coherent_samples_ == 0) {
         continue;
       }
 
       it->second->end_coherent_changes(group_samples);
     }
   }
 
   return DDS::RETCODE_OK;
 }

◆ get_default_datawriter_qos()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::get_default_datawriter_qos ( DDS::DataWriterQos & qos )

virtual

Definition at line 824 of file PublisherImpl.cpp.

References default_datawriter_qos_, and DDS::RETCODE_OK.

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

◆ get_instance_handle()

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

virtual

Implements OpenDDS::DCPS::EntityImpl.

Definition at line 72 of file PublisherImpl.cpp.

References handle_.

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

 {
   return handle_;
 }

◆ get_listener()

DDS::PublisherListener_ptr OpenDDS::DCPS::PublisherImpl::get_listener ( )

virtual

Implements DDS::Publisher.

Definition at line 525 of file PublisherImpl.cpp.

References listener_, and listener_mutex_.

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

◆ get_participant()

DDS::DomainParticipant_ptr OpenDDS::DCPS::PublisherImpl::get_participant ( )

virtual

Implements DDS::Publisher.

Definition at line 806 of file PublisherImpl.cpp.

References participant_.

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

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

◆ get_pi_lock()

ACE_Recursive_Thread_Mutex& OpenDDS::DCPS::PublisherImpl::get_pi_lock ( )

inline

Definition at line 113 of file PublisherImpl.h.

References OpenDDS::DCPS::OPENDDS_MAP_CMP(), OPENDDS_MULTIMAP, OPENDDS_STRING, and OpenDDS::DCPS::OPENDDS_VECTOR().

113 { return pi_lock_; }

OpenDDS::DCPS::PublisherImpl::pi_lock_

lock_type pi_lock_

The recursive lock to protect datawriter map and suspend count.

Definition: PublisherImpl.h:208

◆ get_publication_ids()

void OpenDDS::DCPS::PublisherImpl::get_publication_ids ( PublicationIdVec & pubs )

Populates a std::vector with the GUID_ts of this Publisher's Data Writers

Definition at line 1000 of file PublisherImpl.cpp.

References ACE_GUARD_RETURN, pi_lock_, and publication_map_.

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

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       guard,
       this->pi_lock_,
   );
 
   pubs.reserve(publication_map_.size());
   for (PublicationMap::iterator iter = publication_map_.begin();
       iter != publication_map_.end();
       ++iter) {
     pubs.push_back(iter->first);
   }
 }

◆ get_qos()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::get_qos ( DDS::PublisherQos & qos )

virtual

Definition at line 507 of file PublisherImpl.cpp.

References qos_, and DDS::RETCODE_OK.

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

◆ is_clean()

bool OpenDDS::DCPS::PublisherImpl::is_clean ( String * leftover_entities = 0 ) const

This method is not defined in the IDL and is defined for internal use. Check if there is any datawriter associated with this publisher.

Definition at line 877 of file PublisherImpl.cpp.

References ACE_GUARD_RETURN, datawriter_map_, pi_lock_, publication_map_, and OpenDDS::DCPS::to_dds_string().

Referenced by OpenDDS::DCPS::DomainParticipantImpl::delete_publisher(), and ~PublisherImpl().

 {
   if (leftover_entities) {
     leftover_entities->clear();
   }
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, pi_lock_, false);
 
   const size_t writer_count = datawriter_map_.size();
   if (leftover_entities && writer_count) {
     *leftover_entities += to_dds_string(writer_count) + " writer(s)";
   }
 
   const size_t publication_count = publication_map_.size();
   if (leftover_entities && publication_count) {
     if (leftover_entities->size()) {
       *leftover_entities += ", ";
     }
     *leftover_entities += to_dds_string(publication_count) + " publication(s)";
   }
 
   return writer_count == 0 && publication_count == 0;
 }

◆ is_suspended()

bool OpenDDS::DCPS::PublisherImpl::is_suspended ( void ) const

Definition at line 553 of file PublisherImpl.cpp.

References ACE_GUARD_RETURN, pi_suspended_lock_, and suspend_depth_count_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       suspend_guard,
       this->pi_suspended_lock_,
       false);
   return suspend_depth_count_;
 }

◆ listener_for()

DDS::PublisherListener_ptr OpenDDS::DCPS::PublisherImpl::listener_for ( ::DDS::StatusKind kind )

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

Definition at line 939 of file PublisherImpl.cpp.

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

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

◆ liveliness_check_interval()

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

Definition at line 977 of file PublisherImpl.cpp.

References datawriter_map_, and OpenDDS::DCPS::TimeDuration::max_value.

 {
   TimeDuration tv = TimeDuration::max_value;
   for (DataWriterMap::iterator it(datawriter_map_.begin());
       it != datawriter_map_.end(); ++it) {
     tv = std::min(tv, it->second->liveliness_check_interval(kind));
   }
   return tv;
 }

◆ lookup_datawriter()

DDS::DataWriter_ptr OpenDDS::DCPS::PublisherImpl::lookup_datawriter ( const char * topic_name )

virtual

Definition at line 307 of file PublisherImpl.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), datawriter_map_, OpenDDS::DCPS::DCPS_debug_level, LM_DEBUG, and pi_lock_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       guard,
       this->pi_lock_,
       DDS::DataWriter::_nil());
 
   // If multiple entries whose key is "topic_name" then which one is
   // returned ? Spec does not limit which one should give.
   DataWriterMap::iterator it = datawriter_map_.find(topic_name);
 
   if (it == datawriter_map_.end()) {
     if (DCPS_debug_level >= 2) {
       ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) ")
           ACE_TEXT("PublisherImpl::lookup_datawriter, ")
           ACE_TEXT("The datawriter(topic_name=%C) is not found\n"),
           topic_name));
     }
 
     return DDS::DataWriter::_nil();
 
   } else {
     return DDS::DataWriter::_duplicate(it->second.in());
   }
 }

◆ OPENDDS_MAP_CMP() [1/2]

typedef OpenDDS::DCPS::PublisherImpl::OPENDDS_MAP_CMP	(	GUID_t	,
		DataWriterImpl_rch	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MAP_CMP() [2/2]

typedef OpenDDS::DCPS::PublisherImpl::OPENDDS_MAP_CMP	(	GUID_t	,
		DDS::DataWriterQos	,
		GUID_tKeyLessThan
	)

private

◆ OPENDDS_MULTIMAP()

typedef OpenDDS::DCPS::PublisherImpl::OPENDDS_MULTIMAP	(	OPENDDS_STRING	,
		DataWriterImpl_rch
	)

private

◆ OPENDDS_SET()

typedef OpenDDS::DCPS::PublisherImpl::OPENDDS_SET ( DataWriterImpl_rch )

private

◆ OPENDDS_VECTOR()

typedef OpenDDS::DCPS::PublisherImpl::OPENDDS_VECTOR ( GUID_t )

◆ parent()

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

virtual

Reimplemented from OpenDDS::DCPS::EntityImpl.

Definition at line 1016 of file PublisherImpl.cpp.

References participant_.

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

◆ participant_liveliness_activity_after()

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

Definition at line 988 of file PublisherImpl.cpp.

References datawriter_map_.

 {
   for (DataWriterMap::iterator it(datawriter_map_.begin());
       it != datawriter_map_.end(); ++it) {
     if (it->second->participant_liveliness_activity_after(tv)) {
       return true;
     }
   }
   return false;
 }

◆ prepare_to_delete_datawriters()

bool OpenDDS::DCPS::PublisherImpl::prepare_to_delete_datawriters ( )

Definition at line 334 of file PublisherImpl.cpp.

References ACE_GUARD_RETURN, datawriter_map_, OpenDDS::DCPS::EntityImpl::get_deleted(), pi_lock_, and OpenDDS::DCPS::DataWriterImpl::prepare_to_delete().

Referenced by delete_contained_entities().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, pi_lock_, false);
   bool result = true;
   const DataWriterMap::iterator end = datawriter_map_.end();
   for (DataWriterMap::iterator i = datawriter_map_.begin(); i != end; ++i) {
     DataWriterImpl* const writer = dynamic_cast<DataWriterImpl*>(i->second.in());
     if (writer) {
       if (!writer->get_deleted()) {
         writer->prepare_to_delete();
       }
     } else {
       result = false;
     }
   }
 
   return result;
 }

◆ resume_publications()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::resume_publications ( )

virtual

Implements DDS::Publisher.

Definition at line 563 of file PublisherImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::EntityImpl::enabled_, LM_ERROR, pi_lock_, pi_suspended_lock_, publication_map_, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, DDS::RETCODE_PRECONDITION_NOT_MET, and suspend_depth_count_.

 {
   if (!enabled_) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("PublisherImpl::resume_publications, ")
           ACE_TEXT(" Entity is not enabled.\n")));
     }
     return DDS::RETCODE_NOT_ENABLED;
   }
 
   PublicationMap publication_map_copy;
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
         suspend_guard,
         this->pi_suspended_lock_,
         DDS::RETCODE_ERROR);
     --suspend_depth_count_;
 
     if (suspend_depth_count_ < 0) {
       suspend_depth_count_ = 0;
       return DDS::RETCODE_PRECONDITION_NOT_MET;
     }
     if (suspend_depth_count_ == 0) {
       suspend_guard.release();
       ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
           guard,
           this->pi_lock_,
           DDS::RETCODE_ERROR);
 
       publication_map_copy = publication_map_;
     }
   }
 
   for (PublicationMap::const_iterator it = publication_map_copy.begin();
       it != publication_map_copy.end(); ++it) {
     it->second->send_suspended_data();
   }
 
   return DDS::RETCODE_OK;
 }

◆ set_default_datawriter_qos()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::set_default_datawriter_qos ( const DDS::DataWriterQos & qos )

virtual

Definition at line 812 of file PublisherImpl.cpp.

References OpenDDS::DCPS::Qos_Helper::consistent(), default_datawriter_qos_, DDS::RETCODE_INCONSISTENT_POLICY, DDS::RETCODE_OK, and OpenDDS::DCPS::Qos_Helper::valid().

 {
   if (Qos_Helper::valid(qos) && Qos_Helper::consistent(qos)) {
     default_datawriter_qos_ = qos;
     return DDS::RETCODE_OK;
 
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ set_listener()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::set_listener	(	DDS::PublisherListener_ptr	a_listener,
		DDS::StatusMask	mask
	)

virtual

Definition at line 514 of file PublisherImpl.cpp.

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

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

◆ set_qos()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::set_qos ( const DDS::PublisherQos & qos )

virtual

Definition at line 426 of file PublisherImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::Qos_Helper::changeable(), OpenDDS::DCPS::Qos_Helper::consistent(), OpenDDS::DCPS::DCPS_debug_level, domain_id_, OpenDDS::DCPS::EntityImpl::enabled_, LM_ERROR, OPENDDS_NO_OBJECT_MODEL_PROFILE_COMPATIBILITY_CHECK, participant_, pi_lock_, publication_map_, 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(qos, DDS::RETCODE_UNSUPPORTED);
 
   if (Qos_Helper::valid(qos) && Qos_Helper::consistent(qos)) {
     if (qos_ == qos)
       return DDS::RETCODE_OK;
 
     // for the not changeable qos, it can be changed before enable
     if (!Qos_Helper::changeable(qos_, qos) && enabled_) {
       return DDS::RETCODE_IMMUTABLE_POLICY;
 
     } else {
       qos_ = qos;
 
       DwIdToQosMap idToQosMap;
       {
         ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
             guard,
             this->pi_lock_,
             DDS::RETCODE_ERROR);
 
         for (PublicationMap::iterator iter = publication_map_.begin();
             iter != publication_map_.end();
             ++iter) {
           DDS::DataWriterQos qos = iter->second->qos_;
           GUID_t id = iter->second->get_guid();
           std::pair<DwIdToQosMap::iterator, bool> pair =
               idToQosMap.insert(DwIdToQosMap::value_type(id, qos));
 
           if (!pair.second) {
             if (DCPS_debug_level > 0) {
               ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) ")
                   ACE_TEXT("PublisherImpl::set_qos: ")
                   ACE_TEXT("insert id %C to DwIdToQosMap ")
                   ACE_TEXT("failed.\n"),
                   LogGuid(id).c_str()));
             }
             return DDS::RETCODE_ERROR;
           }
         }
       }
 
       DwIdToQosMap::iterator iter = idToQosMap.begin();
 
       while (iter != idToQosMap.end()) {
         Discovery_rch disco = TheServiceParticipant->get_discovery(this->domain_id_);
         bool status = false;
 
         RcHandle<DomainParticipantImpl> participant = this->participant_.lock();
         if (participant)
           status = disco->update_publication_qos(
               participant->get_domain_id(),
               participant->get_id(),
               iter->first,
               iter->second,
               this->qos_);
 
         if (!status) {
           if (DCPS_debug_level > 0) {
             ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) PublisherImpl::set_qos, ")
                 ACE_TEXT("failed.\n")));
           }
           return DDS::RETCODE_ERROR;
         }
 
         ++iter;
       }
     }
 
     return DDS::RETCODE_OK;
 
   } else {
     return DDS::RETCODE_INCONSISTENT_POLICY;
   }
 }

◆ set_wait_pending_deadline()

bool OpenDDS::DCPS::PublisherImpl::set_wait_pending_deadline ( const MonotonicTimePoint & deadline )

Definition at line 353 of file PublisherImpl.cpp.

References ACE_GUARD_RETURN, datawriter_map_, pi_lock_, and OpenDDS::DCPS::DataWriterImpl::set_wait_pending_deadline().

Referenced by delete_contained_entities().

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, pi_lock_, false);
   bool result = true;
   const DataWriterMap::iterator end = datawriter_map_.end();
   for (DataWriterMap::iterator i = datawriter_map_.begin(); i != end; ++i) {
     DataWriterImpl* const writer = dynamic_cast<DataWriterImpl*>(i->second.in());
     if (writer) {
       writer->set_wait_pending_deadline(deadline);
     } else {
       result = false;
     }
   }
   return result;
 }

◆ suspend_publications()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::suspend_publications ( )

virtual

Implements DDS::Publisher.

Definition at line 532 of file PublisherImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::EntityImpl::enabled_, LM_ERROR, pi_suspended_lock_, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, and suspend_depth_count_.

 {
   if (!enabled_) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("PublisherImpl::suspend_publications, ")
           ACE_TEXT(" Entity is not enabled.\n")));
     }
     return DDS::RETCODE_NOT_ENABLED;
   }
 
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       suspend_guard,
       this->pi_suspended_lock_,
       DDS::RETCODE_ERROR);
   ++suspend_depth_count_;
   return DDS::RETCODE_OK;
 }

◆ validate_datawriter_qos()

bool OpenDDS::DCPS::PublisherImpl::validate_datawriter_qos	(	const DDS::DataWriterQos &	qos,
		const DDS::DataWriterQos &	default_qos,
		DDS::Topic_ptr	a_topic,
		DDS::DataWriterQos &	dw_qos
	)

static

Definition at line 1022 of file PublisherImpl.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::Qos_Helper::consistent(), OpenDDS::DCPS::Qos_Helper::copy_from_topic_qos(), DATAWRITER_QOS_DEFAULT, DATAWRITER_QOS_USE_TOPIC_QOS, OpenDDS::DCPS::DCPS_debug_level, CORBA::is_nil(), LM_ERROR, OPENDDS_END_VERSIONED_NAMESPACE_DECL, OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK, OPENDDS_NO_DURABILITY_SERVICE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK, OPENDDS_NO_OWNERSHIP_STRENGTH_COMPATIBILITY_CHECK, and OpenDDS::DCPS::Qos_Helper::valid().

Referenced by create_datawriter(), and OpenDDS::DCPS::DomainParticipantImpl::create_replayer().

 {
   if (CORBA::is_nil(a_topic)) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("PublisherImpl::create_datawriter, ")
           ACE_TEXT("topic is nil.\n")));
     }
     return DDS::DataWriter::_nil();
   }
 
   if (qos == DATAWRITER_QOS_DEFAULT) {
     dw_qos = default_qos;
 
   } else if (qos == DATAWRITER_QOS_USE_TOPIC_QOS) {
     DDS::TopicQos topic_qos;
     a_topic->get_qos(topic_qos);
     dw_qos = default_qos;
 
     Qos_Helper::copy_from_topic_qos(dw_qos, topic_qos);
 
   } else {
     dw_qos = qos;
   }
 
   OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE_COMPATIBILITY_CHECK(dw_qos, DDS::DataWriter::_nil());
   OPENDDS_NO_OWNERSHIP_STRENGTH_COMPATIBILITY_CHECK(dw_qos, DDS::DataWriter::_nil());
   OPENDDS_NO_OWNERSHIP_PROFILE_COMPATIBILITY_CHECK(dw_qos, DDS::DataWriter::_nil());
   OPENDDS_NO_DURABILITY_SERVICE_COMPATIBILITY_CHECK(dw_qos, DDS::DataWriter::_nil());
   OPENDDS_NO_DURABILITY_KIND_TRANSIENT_PERSISTENT_COMPATIBILITY_CHECK(dw_qos, DDS::DataWriter::_nil());
 
   if (!Qos_Helper::valid(dw_qos)) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("PublisherImpl::create_datawriter, ")
           ACE_TEXT("invalid qos.\n")));
     }
     return DDS::DataWriter::_nil();
   }
 
   if (!Qos_Helper::consistent(dw_qos)) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("PublisherImpl::create_datawriter, ")
           ACE_TEXT("inconsistent qos.\n")));
     }
     return DDS::DataWriter::_nil();
   }
   return true;
 }

◆ wait_for_acknowledgments()

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

virtual

Definition at line 740 of file PublisherImpl.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), datawriter_map_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::EntityImpl::enabled_, OpenDDS::DCPS::RcHandle< T >::in(), LM_DEBUG, LM_ERROR, OpenDDS::DCPS::OPENDDS_MAP(), pi_lock_, DDS::RELIABLE_RELIABILITY_QOS, DDS::RETCODE_ERROR, DDS::RETCODE_NOT_ENABLED, and DDS::RETCODE_OK.

 {
   if (!enabled_) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: PublisherImpl::wait_for_acknowledgments, ")
           ACE_TEXT("Entity is not enabled.\n")));
     }
     return DDS::RETCODE_NOT_ENABLED;
   }
 
   typedef OPENDDS_MAP(DataWriterImpl*, DataWriterImpl::AckToken) DataWriterAckMap;
   DataWriterAckMap ack_writers;
   {
     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
         guard,
         this->pi_lock_,
         DDS::RETCODE_ERROR);
 
     // Collect writers to request acks
     for (DataWriterMap::iterator it(this->datawriter_map_.begin());
         it != this->datawriter_map_.end(); ++it) {
       DataWriterImpl_rch writer = it->second;
       if (writer->qos_.reliability.kind != DDS::RELIABLE_RELIABILITY_QOS)
         continue;
       if (writer->should_ack()) {
         DataWriterImpl::AckToken token = writer->create_ack_token(max_wait);
 
         std::pair<DataWriterAckMap::iterator, bool> pair =
             ack_writers.insert(DataWriterAckMap::value_type(writer.in(), token));
 
         if (!pair.second) {
           if (DCPS_debug_level > 0) {
             ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: PublisherImpl::wait_for_acknowledgments, ")
                 ACE_TEXT("Unable to insert AckToken into DataWriterAckMap!\n")));
           }
           return DDS::RETCODE_ERROR;
         }
       }
     }
   }
 
   if (ack_writers.empty()) {
     if (DCPS_debug_level > 0) {
       ACE_DEBUG((LM_DEBUG,
           ACE_TEXT("(%P|%t) PublisherImpl::wait_for_acknowledgments() - ")
           ACE_TEXT("not blocking due to no writers requiring acks.\n")));
     }
 
     return DDS::RETCODE_OK;
   }
 
   // Wait for ack responses from all associated readers
   for (DataWriterAckMap::iterator it(ack_writers.begin());
       it != ack_writers.end(); ++it) {
     DataWriterImpl::AckToken token = it->second;
 
     it->first->wait_for_specific_ack(token);
   }
 
   return DDS::RETCODE_OK;
 }

◆ writer_enabled()

DDS::ReturnCode_t OpenDDS::DCPS::PublisherImpl::writer_enabled	(	const char *	topic_name,
		DataWriterImpl *	impl
	)

This method is called when the datawriter created by this publisher was enabled.

Definition at line 902 of file PublisherImpl.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT(), datawriter_map_, OpenDDS::DCPS::DCPS_debug_level, LM_ERROR, monitor_, pi_lock_, publication_map_, OpenDDS::DCPS::rchandle_from(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, and writers_not_enabled_.

 {
   ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
       guard,
       this->pi_lock_,
       DDS::RETCODE_ERROR);
   DataWriterImpl_rch writer = rchandle_from(writer_ptr);
   writers_not_enabled_.erase(writer);
 
   datawriter_map_.insert(DataWriterMap::value_type(topic_name, writer));
 
   const GUID_t publication_id = writer->get_guid();
 
   std::pair<PublicationMap::iterator, bool> pair =
       publication_map_.insert(PublicationMap::value_type(publication_id, writer));
 
   if (!pair.second) {
     if (DCPS_debug_level > 0) {
       ACE_ERROR((LM_ERROR,
           ACE_TEXT("(%P|%t) ERROR: ")
           ACE_TEXT("PublisherImpl::writer_enabled: ")
           ACE_TEXT("insert publication %C failed.\n"),
           LogGuid(publication_id).c_str()));
     }
     return DDS::RETCODE_ERROR;
   }
 
   if (this->monitor_) {
     this->monitor_->report();
   }
 
   return DDS::RETCODE_OK;
 }

Friends And Related Function Documentation

◆ DataWriterImpl

friend class DataWriterImpl

friend

Definition at line 43 of file PublisherImpl.h.

Member Data Documentation

◆ aggregation_period_start_

MonotonicTimePoint OpenDDS::DCPS::PublisherImpl::aggregation_period_start_

private

Start of current aggregation period. - NOT USED IN FIRST IMPL.

Definition at line 203 of file PublisherImpl.h.

◆ change_depth_

std::size_t OpenDDS::DCPS::PublisherImpl::change_depth_

private

The number of times begin_coherent_changes as been called.

Definition at line 191 of file PublisherImpl.h.

Referenced by begin_coherent_changes(), and end_coherent_changes().

◆ datawriter_map_

DataWriterMap OpenDDS::DCPS::PublisherImpl::datawriter_map_

private

This map is used to support datawriter lookup by topic name.

Definition at line 185 of file PublisherImpl.h.

Referenced by assert_liveliness_by_participant(), contains_writer(), delete_contained_entities(), delete_datawriter(), is_clean(), liveliness_check_interval(), lookup_datawriter(), participant_liveliness_activity_after(), prepare_to_delete_datawriters(), set_wait_pending_deadline(), wait_for_acknowledgments(), and writer_enabled().

◆ default_datawriter_qos_

DDS::DataWriterQos OpenDDS::DCPS::PublisherImpl::default_datawriter_qos_

private

Default datawriter Qos policy list.

Definition at line 171 of file PublisherImpl.h.

Referenced by create_datawriter(), get_default_datawriter_qos(), and set_default_datawriter_qos().

◆ domain_id_

DDS::DomainId_t OpenDDS::DCPS::PublisherImpl::domain_id_

private

Domain in which we are contained.

Definition at line 194 of file PublisherImpl.h.

Referenced by delete_datawriter(), and set_qos().

◆ handle_

DDS::InstanceHandle_t OpenDDS::DCPS::PublisherImpl::handle_

private

Definition at line 166 of file PublisherImpl.h.

Referenced by get_instance_handle(), and ~PublisherImpl().

◆ listener_

DDS::PublisherListener_var OpenDDS::DCPS::PublisherImpl::listener_

private

Used to notify the entity for relevant events.

Definition at line 179 of file PublisherImpl.h.

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

◆ listener_mask_

DDS::StatusMask OpenDDS::DCPS::PublisherImpl::listener_mask_

private

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

Definition at line 177 of file PublisherImpl.h.

Referenced by listener_for(), and set_listener().

◆ listener_mutex_

ACE_Thread_Mutex OpenDDS::DCPS::PublisherImpl::listener_mutex_

private

Mutex to protect listener info.

Definition at line 174 of file PublisherImpl.h.

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

◆ monitor_

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

private

Monitor object for this entity.

Definition at line 213 of file PublisherImpl.h.

Referenced by delete_datawriter(), enable(), PublisherImpl(), and writer_enabled().

◆ participant_

WeakRcHandle<DomainParticipantImpl> OpenDDS::DCPS::PublisherImpl::participant_

private

The DomainParticipant servant that owns this Publisher.

Definition at line 196 of file PublisherImpl.h.

Referenced by create_datawriter(), delete_datawriter(), enable(), get_participant(), listener_for(), parent(), set_qos(), and ~PublisherImpl().

◆ pi_lock_

lock_type OpenDDS::DCPS::PublisherImpl::pi_lock_

mutableprivate

The recursive lock to protect datawriter map and suspend count.

Definition at line 208 of file PublisherImpl.h.

Referenced by begin_coherent_changes(), contains_writer(), create_datawriter(), delete_contained_entities(), delete_datawriter(), enable(), end_coherent_changes(), get_publication_ids(), is_clean(), lookup_datawriter(), prepare_to_delete_datawriters(), resume_publications(), set_qos(), set_wait_pending_deadline(), wait_for_acknowledgments(), and writer_enabled().

◆ pi_suspended_lock_

lock_type OpenDDS::DCPS::PublisherImpl::pi_suspended_lock_

mutableprivate

Definition at line 210 of file PublisherImpl.h.

Referenced by is_suspended(), resume_publications(), and suspend_publications().

◆ publication_map_

PublicationMap OpenDDS::DCPS::PublisherImpl::publication_map_

private

This map is used to support datawriter lookup by datawriter repository id.

Definition at line 188 of file PublisherImpl.h.

Referenced by begin_coherent_changes(), delete_datawriter(), end_coherent_changes(), get_publication_ids(), is_clean(), resume_publications(), set_qos(), and writer_enabled().

◆ publisher_id_

GUID_t OpenDDS::DCPS::PublisherImpl::publisher_id_

private

Note: The publisher_id_ is not generated by repository, it's unique in DomainParticipant scope.

Definition at line 217 of file PublisherImpl.h.

◆ qos_

DDS::PublisherQos OpenDDS::DCPS::PublisherImpl::qos_

private

Publisher QoS policy list.

Definition at line 169 of file PublisherImpl.h.

Referenced by begin_coherent_changes(), create_datawriter(), enable(), end_coherent_changes(), get_qos(), and set_qos().

◆ reverse_pi_lock_

reverse_lock_type OpenDDS::DCPS::PublisherImpl::reverse_pi_lock_

private

Definition at line 209 of file PublisherImpl.h.

Referenced by delete_datawriter().

◆ sequence_number_

SequenceNumber OpenDDS::DCPS::PublisherImpl::sequence_number_

private

Unique sequence number used when the scope_access = GROUP.

NOT USED IN FIRST IMPL - not supporting GROUP scope

Definition at line 201 of file PublisherImpl.h.

◆ suspend_depth_count_

CORBA::Short OpenDDS::DCPS::PublisherImpl::suspend_depth_count_

private

The suspend depth count.

Definition at line 198 of file PublisherImpl.h.

Referenced by is_suspended(), resume_publications(), and suspend_publications().

◆ writers_not_enabled_

DataWriterSet OpenDDS::DCPS::PublisherImpl::writers_not_enabled_

private

Definition at line 182 of file PublisherImpl.h.

Referenced by create_datawriter(), enable(), and writer_enabled().

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

Public Member Functions

Static Public Member Functions

Private Types

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ lock_type

◆ reverse_lock_type

Constructor & Destructor Documentation

◆ PublisherImpl()

◆ ~PublisherImpl()

Member Function Documentation

◆ assert_liveliness_by_participant()

◆ begin_coherent_changes()

◆ contains_writer()

◆ copy_from_topic_qos()

◆ create_datawriter()

◆ delete_contained_entities()

◆ delete_datawriter()

◆ enable()

◆ end_coherent_changes()

◆ get_default_datawriter_qos()

◆ get_instance_handle()

◆ get_listener()

◆ get_participant()

◆ get_pi_lock()

◆ get_publication_ids()

◆ get_qos()

◆ is_clean()

◆ is_suspended()

◆ listener_for()

◆ liveliness_check_interval()

◆ lookup_datawriter()

◆ OPENDDS_MAP_CMP() [1/2]

◆ OPENDDS_MAP_CMP() [2/2]

◆ OPENDDS_MULTIMAP()

◆ OPENDDS_SET()

◆ OPENDDS_VECTOR()

◆ parent()

◆ participant_liveliness_activity_after()

◆ prepare_to_delete_datawriters()

◆ resume_publications()

◆ set_default_datawriter_qos()

◆ set_listener()

◆ set_qos()

◆ set_wait_pending_deadline()

◆ suspend_publications()

◆ validate_datawriter_qos()

◆ wait_for_acknowledgments()

◆ writer_enabled()

Friends And Related Function Documentation

◆ DataWriterImpl

Member Data Documentation

◆ aggregation_period_start_

◆ change_depth_

◆ datawriter_map_

◆ default_datawriter_qos_

◆ domain_id_

◆ handle_

◆ listener_

◆ listener_mask_

◆ listener_mutex_

◆ monitor_

◆ participant_

◆ pi_lock_

◆ pi_suspended_lock_

◆ publication_map_

◆ publisher_id_

◆ qos_

◆ reverse_pi_lock_

◆ sequence_number_

◆ suspend_depth_count_

◆ writers_not_enabled_