#include <UpdateManager.h>

Inheritance diagram for Update::Manager:

Collaboration diagram for Update::Manager:

[legend]

Public Member Functions
	Manager ()

virtual	~Manager ()

virtual int	init (int argc, ACE_TCHAR *argv[])
	Shared object initializer. More...

virtual int	fini ()
	Shared object finalizer. More...

void	add (TAO_DDS_DCPSInfo_i *info)

void	add (Updater *updater)

void	remove ()

void	remove (const Updater *updater)

void	requestImage ()
	Force a clean shutdown. More...

void	pushImage (const DImage &image)
	Downstream request to push image. More...

template<class UType >
void	create (const UType &info)

template<class QosType >
void	update (const IdPath &id, const QosType &qos)

void	destroy (const IdPath &id, ItemType type, ActorType actor=DataWriter)

void	add (const DTopic &topic)

void	add (const DParticipant &participant)

void	add (const DActor &actor)

virtual void	updateLastPartId (PartIdType partId)
	Update Last Participant Id for the repo. More...

Public Member Functions inherited from ACE_Service_Object
	ACE_Service_Object (ACE_Reactor *=0)

virtual	~ACE_Service_Object (void)

virtual int	suspend (void)

virtual int	resume (void)

Public Member Functions inherited from ACE_Event_Handler
virtual	~ACE_Event_Handler (void)

virtual ACE_HANDLE	get_handle (void) const

virtual void	set_handle (ACE_HANDLE)

virtual int	priority (void) const

virtual void	priority (int priority)

virtual int	handle_input (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_output (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_exception (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_timeout (const ACE_Time_Value &current_time, const void *act=0)

virtual int	handle_exit (ACE_Process *)

virtual int	handle_close (ACE_HANDLE handle, ACE_Reactor_Mask close_mask)

virtual int	handle_signal (int signum, siginfo_t =0, ucontext_t =0)

virtual int	resume_handler (void)

virtual int	handle_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual int	handle_group_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual void	reactor (ACE_Reactor *reactor)

virtual ACE_Reactor *	reactor (void) const

virtual ACE_Reactor_Timer_Interface *	reactor_timer_interface (void) const

virtual Reference_Count	add_reference (void)

virtual Reference_Count	remove_reference (void)

Reference_Counting_Policy &	reference_counting_policy (void)

Public Member Functions inherited from ACE_Shared_Object
	ACE_Shared_Object (void)

virtual	~ACE_Shared_Object (void)

virtual int	info (ACE_TCHAR **info_string, size_t length=0) const

Private Types
typedef std::set< Updater * >	Updaters

Private Attributes
TAO_DDS_DCPSInfo_i *	info_

Updaters	updaters_

Additional Inherited Members
Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

Static Public Member Functions inherited from ACE_Event_Handler
static ACE_THR_FUNC_RETURN	read_adapter (void *event_handler)

static int	register_stdin_handler (ACE_Event_Handler eh, ACE_Reactor reactor, ACE_Thread_Manager *thr_mgr, int flags=THR_DETACHED)

static int	remove_stdin_handler (ACE_Reactor reactor, ACE_Thread_Manager thr_mgr)

Public Attributes inherited from ACE_Service_Object
	ACE_ALLOC_HOOK_DECLARE

Public Attributes inherited from ACE_Event_Handler
	LO_PRIORITY

	HI_PRIORITY

	NULL_MASK

	READ_MASK

	WRITE_MASK

	EXCEPT_MASK

	ACCEPT_MASK

	CONNECT_MASK

	TIMER_MASK

	QOS_MASK

	GROUP_QOS_MASK

	SIGNAL_MASK

	ALL_EVENTS_MASK

	RWE_MASK

	DONT_CALL

	ACE_EVENT_HANDLER_NOT_RESUMED

	ACE_REACTOR_RESUMES_HANDLER

	ACE_APPLICATION_RESUMES_HANDLER

Protected Types inherited from ACE_Event_Handler
typedef ACE_Atomic_Op< ACE_SYNCH_MUTEX, Reference_Count >	Atomic_Reference_Count

Protected Member Functions inherited from ACE_Event_Handler
	ACE_Event_Handler (ACE_Reactor *=0, int priority=ACE_Event_Handler::LO_PRIORITY)

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Definition at line 32 of file UpdateManager.h.

Member Typedef Documentation

◆ Updaters

typedef std::set<Updater*> Update::Manager::Updaters

private

Definition at line 83 of file UpdateManager.h.

Constructor & Destructor Documentation

◆ Manager()

Update::Manager::Manager ( )

Definition at line 22 of file UpdateManager.cpp.

   : info_(0)
 {
 }

◆ ~Manager()

Update::Manager::~Manager ( )

virtual

Definition at line 27 of file UpdateManager.cpp.

28 {

29 }

Member Function Documentation

◆ add() [1/5]

void Update::Manager::add ( TAO_DDS_DCPSInfo_i * info )

Definition at line 32 of file UpdateManager.cpp.

References ACE_Shared_Object::info(), and info_.

Referenced by TAO_DDS_DCPSInfo_i::add(), Update::PersistenceUpdater::init(), and TAO_DDS_DCPSInfo_i::init_persistence().

 {
   info_ = info;
 }

◆ add() [2/5]

void Update::Manager::add ( Updater * updater )

Definition at line 38 of file UpdateManager.cpp.

References updaters_.

 {
   // push new element to the back.
   updaters_.insert(updater);
 }

◆ add() [3/5]

void Update::Manager::add ( const DTopic & topic )

Definition at line 315 of file UpdateManager.cpp.

References TAO_DDS_DCPSInfo_i::add_topic(), Update::TopicStrt< Q, S >::dataType, Update::TopicStrt< Q, S >::domainId, info_, Update::TopicStrt< Q, S >::name, Update::TopicStrt< Q, S >::participantId, Update::TopicStrt< Q, S >::topicId, and Update::TopicStrt< Q, S >::topicQos.

 {
   if (info_ == NULL) {
     return;
   }
 
   // Demarshal QOS data
   TAO_InputCDR in_cdr(topic.topicQos.second.second
                       , topic.topicQos.second.first);
 
   DDS::TopicQos qos;
   in_cdr >> qos;
 
   // Pass topic info to infoRepo.
   info_->add_topic(topic.topicId, topic.domainId
                    , topic.participantId, topic.name.c_str()
                    , topic.dataType.c_str(), qos);
 }

◆ add() [4/5]

void Update::Manager::add ( const DParticipant & participant )

Definition at line 335 of file UpdateManager.cpp.

References TAO_DDS_DCPSInfo_i::add_domain_participant(), Update::ParticipantStrt< QosSeq >::domainId, info_, Update::ParticipantStrt< QosSeq >::participantId, and Update::ParticipantStrt< QosSeq >::participantQos.

 {
   if (info_ == NULL) {
     return;
   }
 
   // Demarshal QOS data
   TAO_InputCDR in_cdr(participant.participantQos.second.second
                       , participant.participantQos.second.first);
 
   DDS::DomainParticipantQos qos;
   in_cdr >> qos;
 
   // Pass participant info to infoRepo.
   info_->add_domain_participant(participant.domainId
                                 , participant.participantId
                                 , qos);
 }

◆ add() [5/5]

void Update::Manager::add ( const DActor & actor )

Definition at line 355 of file UpdateManager.cpp.

 {
   if (info_ == NULL) {
     return;
   }
 
   // Demarshal QOS data
   TAO_InputCDR pubSubCdr(actor.pubsubQos.second.second
                          , actor.pubsubQos.second.first);
 
   TAO_InputCDR drdwCdr(actor.drdwQos.second.second
                        , actor.drdwQos.second.first);
 
   std::string callback(actor.callback.c_str());
 
   TAO_InputCDR transportCdr(actor.transportInterfaceInfo.second
                             , actor.transportInterfaceInfo.first);
 
   OpenDDS::DCPS::TransportLocatorSeq transport_info;
   transportCdr >> transport_info;
 
   TAO_InputCDR typeInfoCdr(actor.serializedTypeInfo.second, actor.serializedTypeInfo.first);
 
   DDS::OctetSeq serializedTypeInfo;
   typeInfoCdr >> serializedTypeInfo;
 
   if (actor.type == DataReader) {
     DDS::SubscriberQos sub_qos;
     DDS::DataReaderQos reader_qos;
 
     pubSubCdr >> sub_qos;
     drdwCdr >> reader_qos;
 
     Update::ContentSubscriptionInfo csi;
     csi.filterClassName = actor.contentSubscriptionProfile.filterClassName.c_str();
     csi.filterExpr = actor.contentSubscriptionProfile.filterExpr.c_str();
     TAO_InputCDR cspCdr(actor.contentSubscriptionProfile.exprParams.second,
                         actor.contentSubscriptionProfile.exprParams.first);
     cspCdr >> csi.exprParams;
 
     // Pass actor to InfoRepo.
     info_->add_subscription(actor.domainId, actor.participantId
                             , actor.topicId, actor.actorId
                             , callback.c_str(), reader_qos
                             , transport_info, actor.transportContext, sub_qos
                             , csi.filterClassName, csi.filterExpr, csi.exprParams
                             , serializedTypeInfo);
 
   } else if (actor.type == DataWriter) {
     DDS::PublisherQos pub_qos;
     DDS::DataWriterQos writer_qos;
 
     pubSubCdr >> pub_qos;
     drdwCdr >> writer_qos;
 
     // Pass actor info to infoRepo.
     info_->add_publication(actor.domainId, actor.participantId
                            , actor.topicId, actor.actorId
                            , callback.c_str(), writer_qos
                            , transport_info, actor.transportContext, pub_qos
                            , serializedTypeInfo);
   }
 }

◆ create()

template<class UType >

void Update::Manager::create ( const UType & info )

Definition at line 12 of file UpdateManager_T.cpp.

References updaters_.

Referenced by TAO_DDS_DCPSInfo_i::add_publication(), TAO_DDS_DCPSInfo_i::add_subscription(), TAO_DDS_DCPSInfo_i::assert_topic(), and DCPS_IR_Participant::takeOwnership().

 {
   // Invoke add on each of the iterators.
   for (Updaters::iterator iter = updaters_.begin();
        iter != updaters_.end();
        iter++) {
     (*iter)->create(info);
   }
 }

◆ destroy()

void Update::Manager::destroy	(	const IdPath &	id,
		ItemType	type,
		ActorType	actor = `DataWriter`
	)

Definition at line 304 of file UpdateManager.cpp.

References updaters_.

Referenced by DCPS_IR_Participant::remove_all_dependents(), TAO_DDS_DCPSInfo_i::remove_domain_participant(), TAO_DDS_DCPSInfo_i::remove_publication(), TAO_DDS_DCPSInfo_i::remove_subscription(), and TAO_DDS_DCPSInfo_i::remove_topic().

 {
   // Invoke remove on each of the iterators.
   for (Updaters::iterator iter = updaters_.begin();
        iter != updaters_.end();
        iter++) {
     (*iter)->destroy(id, type, actor);
   }
 }

◆ fini()

int Update::Manager::fini ( void )

virtual

Shared object finalizer.

Reimplemented from ACE_Shared_Object.

Definition at line 72 of file UpdateManager.cpp.

 {
   return 0;
 }

◆ init()

int Update::Manager::init	(	int	argc,
		ACE_TCHAR *	argv[]
	)

virtual

Shared object initializer.

Reimplemented from ACE_Shared_Object.

Definition at line 66 of file UpdateManager.cpp.

 {
   return 0;
 }

◆ pushImage()

void Update::Manager::pushImage ( const DImage & image )

Downstream request to push image.

Definition at line 112 of file UpdateManager.cpp.

Referenced by Update::PersistenceUpdater::requestImage().

 {
   if (info_ == NULL) {
     return;
   }
 
   // image to be propagated.
   UImage u_image;
 
   /***************************
   // The downstream image needs to be converted to a
   // format compatible with the upstream layers (Dimage -> UImage)
   // The Uimage cotains a lot of references to the complex data
   // types. These are collecetd in several buckets (see below) and
   // passed by reference. Usage of a custom guard class 'SeqGuard'
   // automates memory cleanup.
   ***************************/
 
   // Participant buckets
   SeqGuard<DDS::DomainParticipantQos> part_qos_guard;
   SeqGuard<DDS::DomainParticipantQos>::Seq& part_qos = part_qos_guard.seq();
 
   SeqGuard<UParticipant> part_guard;
   SeqGuard<UParticipant>::Seq& parts = part_guard.seq();
 
   for (DImage::ParticipantSeq::const_iterator iter = image.participants.begin();
        iter != image.participants.end(); iter++) {
     const DParticipant& part = *iter;
 
     TAO_InputCDR in_cdr(part.participantQos.second.second
                         , part.participantQos.second.first);
 
     DDS::DomainParticipantQos* qos;
     ACE_NEW_NORETURN(qos, DDS::DomainParticipantQos);
     in_cdr >> *qos;
     part_qos.push_back(qos);
 
     UParticipant* u_part;
     ACE_NEW_NORETURN(u_part, UParticipant(part.domainId
                                           , part.owner
                                           , part.participantId
                                           , *qos));
     parts.push_back(u_part);
 
     // push newly created UParticipant into UImage Participant bucket
     u_image.participants.push_back(u_part);
   }
 
   // Topic buckets
   SeqGuard<DDS::TopicQos> topics_qos_guard;
   SeqGuard<DDS::TopicQos>::Seq& topics_qos = topics_qos_guard.seq();
 
   SeqGuard<UTopic> topics_guard;
   SeqGuard<UTopic>::Seq& topics = topics_guard.seq();
 
   for (DImage::TopicSeq::const_iterator iter = image.topics.begin();
        iter != image.topics.end(); iter++) {
     const DTopic& topic = *iter;
 
     TAO_InputCDR in_cdr(topic.topicQos.second.second
                         , topic.topicQos.second.first);
 
     DDS::TopicQos* qos;
     ACE_NEW_NORETURN(qos, DDS::TopicQos);
     in_cdr >> *qos;
     topics_qos.push_back(qos);
 
     UTopic* u_topic;
     ACE_NEW_NORETURN(u_topic
                      , UTopic(topic.domainId, topic.topicId
                               , topic.participantId
                               , topic.name.c_str()
                               , topic.dataType.c_str(), *qos));
     topics.push_back(u_topic);
 
     // Push newly created UTopic into UImage Topic bucket
     u_image.topics.push_back(u_topic);
   }
 
   // Actor buckets
   SeqGuard<DDS::PublisherQos> pub_qos_guard;
   SeqGuard<DDS::PublisherQos>::Seq& pub_qos_seq = pub_qos_guard.seq();
   SeqGuard<DDS::DataWriterQos> dw_qos_guard;
   SeqGuard<DDS::DataWriterQos>::Seq& dw_qos_seq = dw_qos_guard.seq();
 
   SeqGuard<DDS::SubscriberQos> sub_qos_guard;
   SeqGuard<DDS::SubscriberQos>::Seq& sub_qos_seq = sub_qos_guard.seq();
   SeqGuard<DDS::DataReaderQos> dr_qos_guard;
   SeqGuard<DDS::DataReaderQos>::Seq& dr_qos_seq = dr_qos_guard.seq();
 
   SeqGuard<OpenDDS::DCPS::TransportLocatorSeq> trans_guard;
   SeqGuard<OpenDDS::DCPS::TransportLocatorSeq>::Seq& transports = trans_guard.seq();
 
   SeqGuard<DDS::OctetSeq> type_info_guard;
   SeqGuard<DDS::OctetSeq>::Seq& serializedTypeInfoSeq = type_info_guard.seq();
 
   SeqGuard<URActor> reader_guard;
   SeqGuard<URActor>::Seq& readers = reader_guard.seq();
   SeqGuard<UWActor> writer_guard;
   SeqGuard<UWActor>::Seq& writers = writer_guard.seq();
 
   ContentSubscriptionInfo csi;
 
   for (DImage::ReaderSeq::const_iterator iter = image.actors.begin();
        iter != image.actors.end(); iter++) {
     const DActor& actor = *iter;
 
     TAO_InputCDR in_cdr(actor.transportInterfaceInfo.second
                         , actor.transportInterfaceInfo.first);
     OpenDDS::DCPS::TransportLocatorSeq* trans;
     ACE_NEW_NORETURN(trans, OpenDDS::DCPS::TransportLocatorSeq);
     transports.push_back(trans);
     in_cdr >> *trans;
 
     TAO_InputCDR ti_cdr(actor.serializedTypeInfo.second
                         , actor.serializedTypeInfo.first);
     DDS::OctetSeq* type_info_ptr;
     ACE_NEW_NORETURN(type_info_ptr, DDS::OctetSeq);
     serializedTypeInfoSeq.push_back(type_info_ptr);
     ti_cdr >> *type_info_ptr;
 
     DDS::PublisherQos* pub_qos = 0;
     DDS::DataWriterQos* writer_qos = 0;
     DDS::SubscriberQos* sub_qos = 0;
     DDS::DataReaderQos* reader_qos = 0;
 
     if (actor.type == DataReader) {
       TAO_InputCDR sub_cdr(actor.pubsubQos.second.second
                            , actor.pubsubQos.second.first);
       ACE_NEW_NORETURN(sub_qos, DDS::SubscriberQos);
       sub_qos_seq.push_back(sub_qos);
       sub_cdr >> *sub_qos;
 
       TAO_InputCDR read_cdr(actor.drdwQos.second.second
                             , actor.drdwQos.second.first);
       ACE_NEW_NORETURN(reader_qos, DDS::DataReaderQos);
       dr_qos_seq.push_back(reader_qos);
       read_cdr >> *reader_qos;
 
       csi.filterClassName = actor.contentSubscriptionProfile.filterClassName.c_str();
       csi.filterExpr = actor.contentSubscriptionProfile.filterExpr.c_str();
       TAO_InputCDR csp_cdr(actor.contentSubscriptionProfile.exprParams.second,
                            actor.contentSubscriptionProfile.exprParams.first);
       csp_cdr >> csi.exprParams;
 
       URActor* reader;
       ACE_NEW_NORETURN(reader
                        , URActor(actor.domainId, actor.actorId
                                  , actor.topicId, actor.participantId
                                  , actor.type, actor.callback.c_str()
                                  , *sub_qos, *reader_qos
                                  , *trans, actor.transportContext, csi
                                  , *type_info_ptr));
       readers.push_back(reader);
       u_image.actors.push_back(reader);
 
     } else if (actor.type == DataWriter) {
       TAO_InputCDR pub_cdr(actor.pubsubQos.second.second
                            , actor.pubsubQos.second.first);
       ACE_NEW_NORETURN(pub_qos, DDS::PublisherQos);
       pub_qos_seq.push_back(pub_qos);
       pub_cdr >> *pub_qos;
 
       TAO_InputCDR write_cdr(actor.drdwQos.second.second
                              , actor.drdwQos.second.first);
       ACE_NEW_NORETURN(writer_qos, DDS::DataWriterQos);
       dw_qos_seq.push_back(writer_qos);
       write_cdr >> *writer_qos;
 
       UWActor* writer;
       ACE_NEW_NORETURN(writer
                        , UWActor(actor.domainId, actor.actorId
                                  , actor.topicId, actor.participantId
                                  , actor.type, actor.callback.c_str()
                                  , *pub_qos, *writer_qos
                                  , *trans, actor.transportContext, csi
                                  , *type_info_ptr));
       writers.push_back(writer);
       u_image.wActors.push_back(writer);
 
     } else {
       ACE_ERROR((LM_ERROR, "Update::Manager::pushImage> unknown "
                  "actor type.\n"));
     }
   }
 
   u_image.lastPartId = image.lastPartId;
 
   info_->receive_image(u_image);
 }

◆ remove() [1/2]

void Update::Manager::remove ( void )

Definition at line 45 of file UpdateManager.cpp.

References info_.

 {
   // Clean the reference to the InfoRepo.
   info_ = 0;
 }

◆ remove() [2/2]

void Update::Manager::remove ( const Updater * updater )

Definition at line 52 of file UpdateManager.cpp.

References updaters_.

 {
   // check if the Updaters is part of the list.
   Updaters::iterator iter = updaters_.find(const_cast<Updater*>(updater));
 
   if (iter == updaters_.end()) {
     return;
   }
 
   // remove the element
   updaters_.erase(iter);
 }

◆ requestImage()

void Update::Manager::requestImage ( )

Force a clean shutdown.

Upstream request for a fresh image Currently handled synchronously via 'pushImage' TBD: Replace with an asynchronous model.

Definition at line 78 of file UpdateManager.cpp.

References updaters_.

Referenced by TAO_DDS_DCPSInfo_i::init_persistence().

 {
   for (Updaters::iterator iter = updaters_.begin();
        iter != updaters_.end();
        iter++) {
     (*iter)->requestImage();
   }
 }

◆ update()

template<class QosType >

void Update::Manager::update	(	const IdPath &	id,
		const QosType &	qos
	)

Definition at line 24 of file UpdateManager_T.cpp.

References updaters_.

Referenced by TAO_DDS_DCPSInfo_i::update_domain_participant_qos(), TAO_DDS_DCPSInfo_i::update_publication_qos(), TAO_DDS_DCPSInfo_i::update_subscription_params(), TAO_DDS_DCPSInfo_i::update_subscription_qos(), and TAO_DDS_DCPSInfo_i::update_topic_qos().

 {
   // Invoke update on each of the iterators.
   for (Updaters::iterator iter = updaters_.begin();
        iter != updaters_.end();
        iter++) {
     (*iter)->update(id, qos);
   }
 }

◆ updateLastPartId()

void Update::Manager::updateLastPartId ( PartIdType partId )

virtual

Update Last Participant Id for the repo.

Definition at line 419 of file UpdateManager.cpp.

References updaters_.

 {
   for (Updaters::iterator iter = updaters_.begin();
        iter != updaters_.end();
        iter++) {
     (*iter)->updateLastPartId(partId);
   }
 }

Member Data Documentation

◆ info_

TAO_DDS_DCPSInfo_i* Update::Manager::info_

private

Definition at line 88 of file UpdateManager.h.

Referenced by add(), pushImage(), and remove().

◆ updaters_

Updaters Update::Manager::updaters_

private

Definition at line 89 of file UpdateManager.h.

Referenced by add(), create(), destroy(), remove(), requestImage(), update(), and updateLastPartId().

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

Public Member Functions

Private Types

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ Updaters

Constructor & Destructor Documentation

◆ Manager()

◆ ~Manager()

Member Function Documentation

◆ add() [1/5]

◆ add() [2/5]

◆ add() [3/5]

◆ add() [4/5]

◆ add() [5/5]

◆ create()

◆ destroy()

◆ fini()

◆ init()

◆ pushImage()

◆ remove() [1/2]

◆ remove() [2/2]

◆ requestImage()

◆ update()

◆ updateLastPartId()

Member Data Documentation

◆ info_

◆ updaters_