Update::PersistenceUpdater Class Reference

#include <PersistenceUpdater.h>

Inheritance diagram for Update::PersistenceUpdater:

Collaboration diagram for Update::PersistenceUpdater:

Classes
class	IdType_ExtId

Public Types
typedef ACE_Allocator_Adapter< ACE_Malloc< ACE_MMAP_MEMORY_POOL, TAO_SYNCH_MUTEX > >	ALLOCATOR
typedef struct TopicStrt< QosSeq, ACE_CString >	Topic
	Persisted topic data structure. More...
typedef struct ParticipantStrt< QosSeq >	Participant
	Persisted participant data structure. More...
typedef struct ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >	RWActor
	Persisted actor data structure. More...
typedef ACE_Hash_Map_With_Allocator< IdType_ExtId, Topic * >	TopicIndex
typedef ACE_Hash_Map_With_Allocator< IdType_ExtId, Participant * >	ParticipantIndex
typedef ACE_Hash_Map_With_Allocator< IdType_ExtId, RWActor * >	ActorIndex
Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

Public Member Functions
	PersistenceUpdater ()
virtual	~PersistenceUpdater ()
virtual int	init (int argc, ACE_TCHAR *argv[])
	Service object initialization. More...
virtual int	fini ()
	ACE_Task_Base finish method. More...
virtual int	svc ()
	ACE_Task_Base start method. More...
virtual void	requestImage ()
virtual void	create (const UTopic &topic)
	Add topic to be persisted. More...
virtual void	create (const UParticipant &participant)
	Add participant to be persisted. More...
virtual void	create (const URActor &actor)
	Add DataReader to be persisted. More...
virtual void	create (const UWActor &actor)
	Add DataWriter to be persisted. More...
virtual void	create (const OwnershipData &data)
	Add ownership data to be persisted. More...
virtual void	update (const IdPath &id, const DDS::DomainParticipantQos &qos)
	Persist updated Qos parameters for a Participant. More...
virtual void	update (const IdPath &id, const DDS::TopicQos &qos)
	Persist updated Qos parameters for a Topic. More...
virtual void	update (const IdPath &id, const DDS::DataWriterQos &qos)
	Persist updated Qos parameters for a DataWriter. More...
virtual void	update (const IdPath &id, const DDS::PublisherQos &qos)
	Persist updated Qos parameters for a Publisher. More...
virtual void	update (const IdPath &id, const DDS::DataReaderQos &qos)
	Persist updated Qos parameters for a DataReader. More...
virtual void	update (const IdPath &id, const DDS::SubscriberQos &qos)
	Persist updated Qos parameters for a Subscriber. More...
virtual void	update (const IdPath &id, const DDS::StringSeq &exprParams)
	Persist updated subscription exprParams. More...
virtual void	destroy (const IdPath &id, ItemType type, ActorType actor)
	Remove an entity (but not children) from persistence. More...
virtual void	updateLastPartId (PartIdType partId)
	Update Last Participant Id for repo. More...
Public Member Functions inherited from Update::Updater
virtual	~Updater ()
Public Member Functions inherited from ACE_Task_Base
	ACE_Task_Base (ACE_Thread_Manager *=0)
virtual	~ACE_Task_Base (void)
virtual int	open (void *args=0)
virtual int	close (u_long flags=0)
virtual int	module_closed (void)
virtual int	put (ACE_Message_Block *, ACE_Time_Value *=0)
virtual int	activate (long flags=THR_NEW_LWP|THR_JOINABLE|THR_INHERIT_SCHED, int n_threads=1, int force_active=0, long priority=ACE_DEFAULT_THREAD_PRIORITY, int grp_id=-1, ACE_Task_Base *task=0, ACE_hthread_t thread_handles[]=0, void *stack[]=0, size_t stack_size[]=0, ACE_thread_t thread_ids[]=0, const char *thr_name[]=0)
virtual int	wait (void)
virtual int	suspend (void)
virtual int	resume (void)
int	grp_id (void) const
void	grp_id (int)
ACE_Thread_Manager *	thr_mgr (void) const
void	thr_mgr (ACE_Thread_Manager *)
int	is_reader (void) const
int	is_writer (void) const
size_t	thr_count (void) const
ACE_thread_t	last_thread (void) const
Public Member Functions inherited from ACE_Service_Object
	ACE_Service_Object (ACE_Reactor *=0)
virtual	~ACE_Service_Object (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 Member Functions
int	parse (int argc, ACE_TCHAR *argv[])
void	storeUpdate (const ACE_Message_Block &data, BinSeq &storage)

Private Attributes
ACE_TString	persistence_file_
bool	reset_
Manager *	um_
OpenDDS::DCPS::unique_ptr< ALLOCATOR >	allocator_
TopicIndex *	topic_index_
	Persisted Topics. More...
ParticipantIndex *	participant_index_
	Persisted Participants. More...
ActorIndex *	actor_index_
	Persisted Readers and Writers. More...
PartIdType *	last_part_id_
	What the last participant id is/was. More...

Additional Inherited Members
Static Public Member Functions inherited from ACE_Task_Base
static ACE_THR_FUNC_RETURN	svc_run (void *)
static void	cleanup (void *object, void *params)
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_Task_Base
size_t	thr_count_
ACE_Thread_Manager *	thr_mgr_
u_long	flags_
int	grp_id_
ACE_thread_t	last_thread_id_
Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Definition at line 33 of file PersistenceUpdater.h.

Member Typedef Documentation

ActorIndex

typedef ACE_Hash_Map_With_Allocator<IdType_ExtId, RWActor*> Update::PersistenceUpdater::ActorIndex

Definition at line 70 of file PersistenceUpdater.h.

ALLOCATOR

typedef ACE_Allocator_Adapter<ACE_Malloc <ACE_MMAP_MEMORY_POOL , TAO_SYNCH_MUTEX> > Update::PersistenceUpdater::ALLOCATOR

Definition at line 55 of file PersistenceUpdater.h.

Participant

typedef struct ParticipantStrt< QosSeq > Update::PersistenceUpdater::Participant

Persisted participant data structure.

Definition at line 61 of file PersistenceUpdater.h.

ParticipantIndex

typedef ACE_Hash_Map_With_Allocator<IdType_ExtId, Participant*> Update::PersistenceUpdater::ParticipantIndex

Definition at line 69 of file PersistenceUpdater.h.

RWActor

typedef struct ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq > Update::PersistenceUpdater::RWActor

Persisted actor data structure.

Definition at line 64 of file PersistenceUpdater.h.

Topic

typedef struct TopicStrt< QosSeq, ACE_CString > Update::PersistenceUpdater::Topic

Persisted topic data structure.

Definition at line 58 of file PersistenceUpdater.h.

TopicIndex

typedef ACE_Hash_Map_With_Allocator<IdType_ExtId, Topic*> Update::PersistenceUpdater::TopicIndex

Definition at line 68 of file PersistenceUpdater.h.

Constructor & Destructor Documentation

PersistenceUpdater()

Update::PersistenceUpdater::PersistenceUpdater

(

)

Definition at line 217 of file PersistenceUpdater.cpp.

References ACE_TEXT().

  : persistence_file_(ACE_TEXT("InforepoPersist"))
  , reset_(false)
  , um_(0)
  , topic_index_(0)
  , participant_index_(0)
  , actor_index_(0)
  , last_part_id_(0)
{}

~PersistenceUpdater()

Update::PersistenceUpdater::~PersistenceUpdater ( )

virtual

Definition at line 227 of file PersistenceUpdater.cpp.

{}

Member Function Documentation

create() [1/5]

void Update::PersistenceUpdater::create ( const UTopic &  topic )

virtual

Add topic to be persisted.

Implements Update::Updater.

Definition at line 478 of file PersistenceUpdater.cpp.

References ACE_ALLOCATOR, allocator_, ACE_Message_Block::base(), ACE_OutputCDR::begin(), ACE_Hash_Map_With_Allocator< class, class >::bind(), ACE_CDR::consolidate(), Update::TopicStrt< Q, S >::dataType, Update::TopicStrt< Q, S >::domainId, ACE_Message_Block::length(), Update::TopicStrt< Q, S >::name, Update::TopicStrt< Q, S >::participantId, topic_index_, Update::TopicStrt< Q, S >::topicId, Update::TopicQos, and Update::TopicStrt< Q, S >::topicQos.

Referenced by create().

{
  // serialize the Topic QOS
  TAO_OutputCDR outCdr;
  outCdr << topic.topicQos;
  ACE_Message_Block dst;
  ACE_CDR::consolidate(&dst, outCdr.begin());

  const BinSeq qos_bin(dst.length(), dst.base());

  const QosSeq p(TopicQos, qos_bin);
  const DTopic topic_data(topic.domainId, topic.topicId, topic.participantId
                          , topic.name.c_str(), topic.dataType.c_str(), p);

  // allocate memory for TopicData
  void* buffer;
  ACE_ALLOCATOR(buffer, allocator_->malloc
                (sizeof(PersistenceUpdater::Topic)));

  // Initialize TopicData
  PersistenceUpdater::Topic* persistent_data
  = new(buffer) PersistenceUpdater::Topic(topic_data, allocator_.get());

  IdType_ExtId ext(topic_data.topicId);

  // bind TopicData with the topicId
  if (topic_index_->bind(ext, persistent_data, allocator_.get()) != 0) {
    allocator_->free((void *) buffer);
    return;
  }
}

create() [2/5]

void Update::PersistenceUpdater::create ( const UParticipant &  participant )

virtual

Add participant to be persisted.

Implements Update::Updater.

Definition at line 511 of file PersistenceUpdater.cpp.

References ACE_ALLOCATOR, allocator_, ACE_Message_Block::base(), ACE_OutputCDR::begin(), ACE_Hash_Map_With_Allocator< class, class >::bind(), ACE_CDR::consolidate(), Update::ParticipantStrt< Q >::domainId, ACE_Message_Block::length(), Update::ParticipantStrt< Q >::owner, participant_index_, Update::ParticipantStrt< Q >::participantId, Update::ParticipantStrt< QosSeq >::participantId, Update::ParticipantQos, and Update::ParticipantStrt< Q >::participantQos.

{
  // serialize the Topic QOS
  TAO_OutputCDR outCdr;
  outCdr << participant.participantQos;
  ACE_Message_Block dst;
  ACE_CDR::consolidate(&dst, outCdr.begin());

  const BinSeq qos_bin(dst.length(), dst.base());

  QosSeq p(ParticipantQos, qos_bin);
  DParticipant participant_data
  (participant.domainId, participant.owner, participant.participantId, p);

  // allocate memory for ParticipantData
  void* buffer;
  ACE_ALLOCATOR(buffer, allocator_->malloc
                (sizeof(PersistenceUpdater::Participant)));

  // Initialize ParticipantData
  PersistenceUpdater::Participant* persistent_data
  = new(buffer) PersistenceUpdater::Participant(participant_data
                                                , allocator_.get());

  IdType_ExtId ext(participant_data.participantId);

  // bind ParticipantData with the participantId
  if (participant_index_->bind(ext, persistent_data, allocator_.get()) != 0) {
    allocator_->free((void *) buffer);
    return;
  }
}

create() [3/5]

void Update::PersistenceUpdater::create ( const URActor &  actor )

virtual

Add DataReader to be persisted.

Implements Update::Updater.

Definition at line 545 of file PersistenceUpdater.cpp.

References ACE_ALLOCATOR, actor_index_, Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::actorId, allocator_, ACE_Message_Block::base(), ACE_OutputCDR::begin(), ACE_Hash_Map_With_Allocator< class, class >::bind(), Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::callback, ACE_CDR::consolidate(), Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::contentSubscriptionProfile, create(), Update::DataReader, Update::DataReaderQos, Update::DataWriter, Update::DataWriterQos, Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::domainId, Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::drdwQos, Update::ContentSubscriptionBin::filterClassName, ACE_Message_Block::length(), Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::participantId, Update::PublisherQos, Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::pubsubQos, ACE_OutputCDR::reset(), Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::serializedTypeInfo, Update::SubscriberQos, Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::topicId, Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::transportContext, and Update::ActorStrt< PSQ, RWQ, C, T, CSP, STI >::transportInterfaceInfo.

{
  TAO_OutputCDR outCdr;
  outCdr << actor.pubsubQos;
  ACE_Message_Block dst;
  ACE_CDR::consolidate(&dst, outCdr.begin());

  const BinSeq pubsub_qos_bin(dst.length(), dst.base());
  const QosSeq pubsub_qos(SubscriberQos, pubsub_qos_bin);

  outCdr.reset();
  outCdr << actor.drdwQos;
  ACE_Message_Block dst2;
  ACE_CDR::consolidate(&dst2, outCdr.begin());

  const BinSeq dwdr_qos_bin(dst2.length(), dst2.base());
  const QosSeq dwdr_qos(DataReaderQos, dwdr_qos_bin);

  outCdr.reset();
  outCdr << actor.transportInterfaceInfo;
  ACE_Message_Block dst3;
  ACE_CDR::consolidate(&dst3, outCdr.begin());

  const BinSeq tr_bin(dst3.length(), dst3.base());

  outCdr.reset();
  outCdr << actor.contentSubscriptionProfile.exprParams;
  ACE_Message_Block dst4;
  ACE_CDR::consolidate(&dst4, outCdr.begin());

  ContentSubscriptionBin csp_bin;
  csp_bin.filterClassName = actor.contentSubscriptionProfile.filterClassName;
  csp_bin.filterExpr = actor.contentSubscriptionProfile.filterExpr;
  csp_bin.exprParams = std::make_pair(dst4.length(), dst4.base());

  outCdr.reset();
  outCdr << actor.serializedTypeInfo;
  ACE_Message_Block dst5;
  ACE_CDR::consolidate(&dst5, outCdr.begin());

  const BinSeq ti_seq(dst5.length(), dst5.base());


  const DActor actor_data(actor.domainId, actor.actorId, actor.topicId
                          , actor.participantId
                          , DataReader, actor.callback.c_str(), pubsub_qos
                          , dwdr_qos, tr_bin, actor.transportContext, csp_bin
                          , ti_seq);

  // allocate memory for ActorData
  void* buffer;
  ACE_ALLOCATOR(buffer, allocator_->malloc
                (sizeof(PersistenceUpdater::RWActor)));

  // Initialize ActorData
  PersistenceUpdater::RWActor* persistent_data =
    new(buffer) PersistenceUpdater::RWActor(actor_data
                                            , allocator_.get());

  IdType_ExtId ext(actor.actorId);

  // bind ActorData with the actorId
  if (actor_index_->bind(ext, persistent_data, allocator_.get()) != 0) {
    allocator_->free((void *) buffer);
    return;
  }
}

create() [4/5]

virtual void Update::PersistenceUpdater::create ( const UWActor &  actor )

virtual

Add DataWriter to be persisted.

Implements Update::Updater.

create() [5/5]

void Update::PersistenceUpdater::create ( const OwnershipData &  data )

virtual

Add ownership data to be persisted.

Implements Update::Updater.

Definition at line 672 of file PersistenceUpdater.cpp.

{
  /* This method intentionally left unimplemented. */
}

destroy()

void Update::PersistenceUpdater::destroy ( const IdPath &  id, ItemType  type, ActorType  actor  )

virtual

Remove an entity (but not children) from persistence.

Implements Update::Updater.

Definition at line 839 of file PersistenceUpdater.cpp.

References ACE_ERROR, ACE_TEXT(), Update::Actor, actor_index_, allocator_, Update::TopicStrt< QosSeq, ACE_CString >::cleanup(), Update::ParticipantStrt< QosSeq >::cleanup(), Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::cleanup(), LM_ERROR, Update::Participant, participant_index_, Update::Topic, topic_index_, and ACE_Hash_Map_With_Allocator< class, class >::unbind().

{
  IdType_ExtId ext(id.id);
  PersistenceUpdater::Topic* topic = 0;
  PersistenceUpdater::Participant* participant = 0;
  PersistenceUpdater::RWActor* actor = 0;

  switch (type) {
  case Update::Topic:

    if (topic_index_->unbind(ext, topic, allocator_.get()) == 0) {
      topic->cleanup(allocator_.get());
      allocator_->free((void *) topic);
    }

    break;
  case Update::Participant:

    if (participant_index_->unbind(ext, participant, allocator_.get()) == 0) {
      participant->cleanup(allocator_.get());
      allocator_->free((void *) participant);
    }

    break;
  case Update::Actor:

    if (actor_index_->unbind(ext, actor, allocator_.get()) == 0) {
      actor->cleanup(allocator_.get());
      allocator_->free((void *) actor);
    }

    break;
  default: {
    OpenDDS::DCPS::RepoIdConverter converter(id.id);
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) PersistenceUpdater::destroy: ")
               ACE_TEXT("unknown entity - %C.\n"),
               std::string(converter).c_str()));
  }
  }
}

fini()

int Update::PersistenceUpdater::fini ( void  )

virtual

ACE_Task_Base finish method.

Reimplemented from ACE_Shared_Object.

Definition at line 393 of file PersistenceUpdater.cpp.

{
  return 0;
}

init()

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

virtual

Service object initialization.

Reimplemented from ACE_Shared_Object.

Definition at line 280 of file PersistenceUpdater.cpp.

References ACE_DEFAULT_BASE_ADDR, ACE_ERROR, ACE_NEW_RETURN, ACE_TEXT(), actor_index_, Update::Manager::add(), allocator_, ACE_String_Base< char >::c_str(), Update::createIndex(), Update::index_cleanup(), ACE_Dynamic_Service< class >::instance(), last_part_id_, LM_ERROR, parse(), participant_index_, persistence_file_, reset_, topic_index_, and um_.

{
  // discover the UpdateManager
  um_ = ACE_Dynamic_Service<Update::Manager>::instance("UpdateManagerSvc");

  if (um_ == 0) {
    ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: PersistenceUpdater::init ")
               ACE_TEXT("No UpdateManager discovered.\n")));
    return -1;
  }

  this->parse(argc, argv);

#if defined ACE_HAS_MAC_OSX && defined __x86_64__ && __x86_64__
  ACE_MMAP_Memory_Pool::OPTIONS options((void*)0x200000000);
#else
  ACE_MMAP_Memory_Pool::OPTIONS options(ACE_DEFAULT_BASE_ADDR);
#endif

  // Create the allocator with the appropriate options.  The name used
  // for  the lock is the same as one used for the file.
  ALLOCATOR* allocator;
  ACE_NEW_RETURN(allocator,
                 ALLOCATOR(persistence_file_.c_str(),
                           persistence_file_.c_str(),
                           &options),
                 -1);
  allocator_.reset(allocator);

  bool exists = false;

  char* topic_index = (char*)createIndex("TopicIndex", *allocator_
                                         , sizeof(TopicIndex), exists);
  if (!topic_index) {
    return -1;
  }

  char* participant_index = (char*)createIndex("ParticipantIndex", *allocator_
                                               , sizeof(ParticipantIndex), exists);
  if (!participant_index) {
    return -1;
  }

  char* actor_index = (char*)createIndex("ActorIndex", *allocator_
                                         , sizeof(ActorIndex), exists);
  if (!actor_index) {
    return -1;
  }

  void* last_part_id = createIndex(
    "LastParticipantId", *allocator_, sizeof(PartIdType), exists);
  if (!last_part_id) {
    return -1;
  }
  last_part_id_ = reinterpret_cast<PartIdType*>(last_part_id);

  if (exists) {
    topic_index_ = reinterpret_cast<TopicIndex*>(topic_index);
    participant_index_ = reinterpret_cast<ParticipantIndex*>(participant_index);
    actor_index_ = reinterpret_cast<ActorIndex*>(actor_index);
  } else {
    topic_index_ = new(topic_index) TopicIndex(allocator_.get());
    participant_index_ = new(participant_index) ParticipantIndex(allocator_.get());
    actor_index_ = new(actor_index) ActorIndex(allocator_.get());
    *last_part_id_ = 0;
  }

  if (reset_) {
    index_cleanup(topic_index_, allocator_.get());
    index_cleanup(participant_index_, allocator_.get());
    index_cleanup(actor_index_, allocator_.get());
    *last_part_id_ = 0;
  }

  // lastly register the callback
  um_->add(this);

  return 0;
}

parse()

int Update::PersistenceUpdater::parse ( int  argc, ACE_TCHAR *  argv[]  )

private

Definition at line 361 of file PersistenceUpdater.cpp.

References ACE_DEBUG, ACE_TEXT(), ACE_OS::atoi(), LM_DEBUG, persistence_file_, reset_, and ACE_OS::strcasecmp().

Referenced by init().

{
  for (ssize_t count = 0; count < argc; count++) {
    if (ACE_OS::strcasecmp(argv[count], ACE_TEXT("-file")) == 0) {
      if ((count + 1) < argc) {
        persistence_file_ = argv[count+1];
        count++;
      }

    } else if (ACE_OS::strcasecmp(argv[count], ACE_TEXT("-reset")) == 0) {
      if ((count + 1) < argc) {
        int val = ACE_OS::atoi(argv[count+1]);
        reset_ = true;

        if (val == 0) {
          reset_ = false;
        }

        count++;
      }

    } else {
      ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) PersistenceUpdater::parse: Unknown option %s\n")
                 , argv[count]));
      return -1;
    }
  }

  return 0;
}

requestImage()

void Update::PersistenceUpdater::requestImage ( )

virtual

Request an image refresh to be sent upstream. This is currently done synchronously. TBD: Move to an asynchronous model

Implements Update::Updater.

Definition at line 405 of file PersistenceUpdater.cpp.

References ACE_DEBUG, actor_index_, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::actorId, Update::ImageData< T, P, A, W >::actors, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, ACE_Hash< EXT_ID >, ACE_Equal_To< EXT_ID >, ACE_Null_Mutex >::begin(), ACE_String_Base< char >::c_str(), Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::callback, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::contentSubscriptionProfile, Update::DataReader, Update::TopicStrt< QosSeq, ACE_CString >::dataType, OpenDDS::DCPS::DCPS_debug_level, Update::TopicStrt< QosSeq, ACE_CString >::domainId, Update::ParticipantStrt< QosSeq >::domainId, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::domainId, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::drdwQos, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, ACE_Hash< EXT_ID >, ACE_Equal_To< EXT_ID >, ACE_Null_Mutex >::end(), Update::ContentSubscriptionBin::exprParams, Update::ContentSubscriptionBin::filterClassName, Update::ContentSubscriptionBin::filterExpr, last_part_id_, Update::ImageData< T, P, A, W >::lastPartId, LM_DEBUG, Update::TopicStrt< QosSeq, ACE_CString >::name, OPENDDS_STRING, Update::ParticipantStrt< QosSeq >::owner, participant_index_, Update::TopicStrt< QosSeq, ACE_CString >::participantId, Update::ParticipantStrt< QosSeq >::participantId, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::participantId, Update::ParticipantQos, Update::ParticipantStrt< QosSeq >::participantQos, Update::ImageData< T, P, A, W >::participants, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::pubsubQos, Update::Manager::pushImage(), Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::serializedTypeInfo, topic_index_, Update::TopicStrt< QosSeq, ACE_CString >::topicId, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::topicId, Update::TopicQos, Update::TopicStrt< QosSeq, ACE_CString >::topicQos, Update::ImageData< T, P, A, W >::topics, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::transportContext, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::transportInterfaceInfo, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::type, and um_.

{
  if (um_ == NULL) {
    return;
  }

  DImage image;

  for (ParticipantIndex::ITERATOR iter = participant_index_->begin();
       iter != participant_index_->end(); iter++) {
    const PersistenceUpdater::Participant* participant = iter->int_id_;

    const BinSeq& in_seq = participant->participantQos.second;
    const QosSeq qos(ParticipantQos, in_seq);
    const DParticipant dparticipant(participant->domainId
                                    , participant->owner
                                    , participant->participantId
                                    , qos);
    image.participants.push_back(dparticipant);
    if (OpenDDS::DCPS::DCPS_debug_level >= 2)  {
      OpenDDS::DCPS::RepoIdConverter conv(participant->participantId);
      ACE_DEBUG((LM_DEBUG,
        "(%P|%t) PersistenceUpdater::requestImage(): loaded participant %C\n",
        OPENDDS_STRING(conv).c_str()));
    }
  }

  for (TopicIndex::ITERATOR iter = topic_index_->begin();
       iter != topic_index_->end(); iter++) {
    const PersistenceUpdater::Topic* topic = iter->int_id_;

    const BinSeq& in_seq = topic->topicQos.second;
    const QosSeq qos(TopicQos, in_seq);
    const DTopic dTopic(topic->domainId, topic->topicId
                        , topic->participantId, topic->name.c_str()
                        , topic->dataType.c_str(), qos);
    image.topics.push_back(dTopic);
  }

  for (ActorIndex::ITERATOR iter = actor_index_->begin();
       iter != actor_index_->end(); iter++) {
    const PersistenceUpdater::RWActor* actor = iter->int_id_;

    const BinSeq& in_pubsub_seq = actor->pubsubQos.second;
    const QosSeq pubsub_qos(actor->pubsubQos.first, in_pubsub_seq);

    const BinSeq& in_drdw_seq = actor->drdwQos.second;
    const QosSeq drdw_qos(actor->drdwQos.first, in_drdw_seq);

    const BinSeq& in_transport_seq = actor->transportInterfaceInfo;
    const BinSeq& in_type_info = actor->serializedTypeInfo;

    ContentSubscriptionBin in_csp_bin;
    if (actor->type == DataReader) {
      in_csp_bin.filterClassName = actor->contentSubscriptionProfile.filterClassName;
      in_csp_bin.filterExpr = actor->contentSubscriptionProfile.filterExpr;
      in_csp_bin.exprParams = actor->contentSubscriptionProfile.exprParams;
    }

    const DActor dActor(actor->domainId, actor->actorId, actor->topicId
                        , actor->participantId
                        , actor->type, actor->callback.c_str()
                        , pubsub_qos, drdw_qos, in_transport_seq, actor->transportContext, in_csp_bin
                        , in_type_info);
    image.actors.push_back(dActor);
  }

  image.lastPartId = *last_part_id_;

  um_->pushImage(image);
}

storeUpdate()

void Update::PersistenceUpdater::storeUpdate ( const ACE_Message_Block &  data, BinSeq &  storage  )

private

Definition at line 882 of file PersistenceUpdater.cpp.

References ACE_ALLOCATOR, allocator_, ACE_Message_Block::base(), ACE_Message_Block::length(), and ACE_OS::memcpy().

Referenced by update().

{
  size_t len = data.length();

  void* buffer;
  ACE_ALLOCATOR(buffer, this->allocator_->malloc(len));
  ACE_OS::memcpy(buffer, data.base(), len);

  storage.first  = len;
  storage.second = static_cast<char*>(buffer);
}

svc()

int Update::PersistenceUpdater::svc ( void  )

virtual

ACE_Task_Base start method.

Reimplemented from ACE_Task_Base.

Definition at line 399 of file PersistenceUpdater.cpp.

{
  return 0;
}

update() [1/7]

void Update::PersistenceUpdater::update ( const IdPath &  id, const DDS::DomainParticipantQos &  qos  )

virtual

Persist updated Qos parameters for a Participant.

Implements Update::Updater.

Definition at line 678 of file PersistenceUpdater.cpp.

References ACE_ERROR, ACE_TEXT(), ACE_OutputCDR::begin(), ACE_CDR::consolidate(), ACE_Hash_Map_With_Allocator< class, class >::find(), LM_ERROR, participant_index_, Update::ParticipantStrt< QosSeq >::participantQos, and storeUpdate().

{
  IdType_ExtId ext(id.id);
  PersistenceUpdater::Participant* part_data = 0;

  if (this->participant_index_->find(ext, part_data, this->allocator_.get()) ==  0) {
    TAO_OutputCDR outCdr;
    outCdr << qos;
    ACE_Message_Block dst;
    ACE_CDR::consolidate(&dst, outCdr.begin());

    this->storeUpdate(dst, part_data->participantQos.second);

  } else {
    OpenDDS::DCPS::RepoIdConverter converter(id.id);
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) PersistenceUpdater::update: ")
               ACE_TEXT("participant %C not found\n"),
               std::string(converter).c_str()));
  }
}

update() [2/7]

void Update::PersistenceUpdater::update ( const IdPath &  id, const DDS::TopicQos &  qos  )

virtual

Persist updated Qos parameters for a Topic.

Implements Update::Updater.

Definition at line 701 of file PersistenceUpdater.cpp.

References ACE_ERROR, ACE_TEXT(), ACE_OutputCDR::begin(), ACE_CDR::consolidate(), ACE_Hash_Map_With_Allocator< class, class >::find(), LM_ERROR, storeUpdate(), topic_index_, and Update::TopicStrt< QosSeq, ACE_CString >::topicQos.

{
  IdType_ExtId ext(id.id);
  PersistenceUpdater::Topic* topic_data = 0;

  if (this->topic_index_->find(ext, topic_data, this->allocator_.get()) ==  0) {
    TAO_OutputCDR outCdr;
    outCdr << qos;
    ACE_Message_Block dst;
    ACE_CDR::consolidate(&dst, outCdr.begin());

    this->storeUpdate(dst, topic_data->topicQos.second);

  } else {
    OpenDDS::DCPS::RepoIdConverter converter(id.id);
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) PersistenceUpdater::update: ")
               ACE_TEXT("topic %C not found\n"),
               std::string(converter).c_str()));
  }
}

update() [3/7]

void Update::PersistenceUpdater::update ( const IdPath &  id, const DDS::DataWriterQos &  qos  )

virtual

Persist updated Qos parameters for a DataWriter.

Implements Update::Updater.

Definition at line 724 of file PersistenceUpdater.cpp.

References ACE_ERROR, ACE_TEXT(), actor_index_, ACE_OutputCDR::begin(), ACE_CDR::consolidate(), Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::drdwQos, ACE_Hash_Map_With_Allocator< class, class >::find(), LM_ERROR, and storeUpdate().

{
  IdType_ExtId ext(id.id);
  PersistenceUpdater::RWActor* actor_data = 0;

  if (this->actor_index_->find(ext, actor_data, this->allocator_.get()) ==  0) {
    TAO_OutputCDR outCdr;
    outCdr << qos;
    ACE_Message_Block dst;
    ACE_CDR::consolidate(&dst, outCdr.begin());

    this->storeUpdate(dst, actor_data->drdwQos.second);

  } else {
    OpenDDS::DCPS::RepoIdConverter converter(id.id);
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) PersistenceUpdater::update(writerQos): ")
               ACE_TEXT("publication %C not found\n"),
               std::string(converter).c_str()));
  }
}

update() [4/7]

void Update::PersistenceUpdater::update ( const IdPath &  id, const DDS::PublisherQos &  qos  )

virtual

Persist updated Qos parameters for a Publisher.

Implements Update::Updater.

Definition at line 747 of file PersistenceUpdater.cpp.

References ACE_ERROR, ACE_TEXT(), actor_index_, ACE_OutputCDR::begin(), ACE_CDR::consolidate(), ACE_Hash_Map_With_Allocator< class, class >::find(), LM_ERROR, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::pubsubQos, and storeUpdate().

{
  IdType_ExtId ext(id.id);
  PersistenceUpdater::RWActor* actor_data = 0;

  if (this->actor_index_->find(ext, actor_data, this->allocator_.get()) ==  0) {
    TAO_OutputCDR outCdr;
    outCdr << qos;
    ACE_Message_Block dst;
    ACE_CDR::consolidate(&dst, outCdr.begin());

    this->storeUpdate(dst, actor_data->pubsubQos.second);

  } else {
    OpenDDS::DCPS::RepoIdConverter converter(id.id);
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) PersistenceUpdater::update(publisherQos): ")
               ACE_TEXT("publication %C not found\n"),
               std::string(converter).c_str()));
  }
}

update() [5/7]

void Update::PersistenceUpdater::update ( const IdPath &  id, const DDS::DataReaderQos &  qos  )

virtual

Persist updated Qos parameters for a DataReader.

Implements Update::Updater.

Definition at line 770 of file PersistenceUpdater.cpp.

References ACE_ERROR, ACE_TEXT(), actor_index_, ACE_OutputCDR::begin(), ACE_CDR::consolidate(), Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::drdwQos, ACE_Hash_Map_With_Allocator< class, class >::find(), LM_ERROR, and storeUpdate().

{
  IdType_ExtId ext(id.id);
  PersistenceUpdater::RWActor* actor_data = 0;

  if (this->actor_index_->find(ext, actor_data, this->allocator_.get()) ==  0) {
    TAO_OutputCDR outCdr;
    outCdr << qos;
    ACE_Message_Block dst;
    ACE_CDR::consolidate(&dst, outCdr.begin());

    this->storeUpdate(dst, actor_data->drdwQos.second);

  } else {
    OpenDDS::DCPS::RepoIdConverter converter(id.id);
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) PersistenceUpdater::update(readerQos): ")
               ACE_TEXT("subscription %C not found\n"),
               std::string(converter).c_str()));
  }
}

update() [6/7]

void Update::PersistenceUpdater::update ( const IdPath &  id, const DDS::SubscriberQos &  qos  )

virtual

Persist updated Qos parameters for a Subscriber.

Implements Update::Updater.

Definition at line 793 of file PersistenceUpdater.cpp.

References ACE_ERROR, ACE_TEXT(), actor_index_, ACE_OutputCDR::begin(), ACE_CDR::consolidate(), ACE_Hash_Map_With_Allocator< class, class >::find(), LM_ERROR, Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::pubsubQos, and storeUpdate().

{
  IdType_ExtId ext(id.id);
  PersistenceUpdater::RWActor* actor_data = 0;

  if (this->actor_index_->find(ext, actor_data, this->allocator_.get()) ==  0) {
    TAO_OutputCDR outCdr;
    outCdr << qos;
    ACE_Message_Block dst;
    ACE_CDR::consolidate(&dst, outCdr.begin());

    this->storeUpdate(dst, actor_data->pubsubQos.second);

  } else {
    OpenDDS::DCPS::RepoIdConverter converter(id.id);
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) PersistenceUpdater::update(subscriberQos): ")
               ACE_TEXT("subscription %C not found\n"),
               std::string(converter).c_str()));
  }
}

update() [7/7]

void Update::PersistenceUpdater::update ( const IdPath &  id, const DDS::StringSeq &  exprParams  )

virtual

Persist updated subscription exprParams.

Implements Update::Updater.

Definition at line 816 of file PersistenceUpdater.cpp.

References ACE_ERROR, ACE_TEXT(), actor_index_, allocator_, ACE_OutputCDR::begin(), ACE_CDR::consolidate(), Update::ActorStrt< QosSeq, QosSeq, ACE_CString, BinSeq, ContentSubscriptionBin, BinSeq >::contentSubscriptionProfile, Update::ContentSubscriptionBin::exprParams, ACE_Hash_Map_With_Allocator< class, class >::find(), LM_ERROR, and storeUpdate().

{
  IdType_ExtId ext(id.id);
  PersistenceUpdater::RWActor* actor_data = 0;

  if (actor_index_->find(ext, actor_data, allocator_.get()) ==  0) {
    TAO_OutputCDR outCdr;
    outCdr << exprParams;
    ACE_Message_Block dst;
    ACE_CDR::consolidate(&dst, outCdr.begin());

    storeUpdate(dst, actor_data->contentSubscriptionProfile.exprParams);

  } else {
    OpenDDS::DCPS::RepoIdConverter converter(id.id);
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) PersistenceUpdater::update(readerQos): ")
               ACE_TEXT("subscription %C not found\n"),
               std::string(converter).c_str()));
  }
}

updateLastPartId()

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

virtual

Update Last Participant Id for repo.

Reimplemented from Update::Updater.

Definition at line 894 of file PersistenceUpdater.cpp.

References last_part_id_.

{
  *last_part_id_ = partId;
}

Member Data Documentation

actor_index_

ActorIndex* Update::PersistenceUpdater::actor_index_

private

Persisted Readers and Writers.

Definition at line 150 of file PersistenceUpdater.h.

Referenced by create(), destroy(), init(), requestImage(), and update().

allocator_

OpenDDS::DCPS::unique_ptr<ALLOCATOR> Update::PersistenceUpdater::allocator_

private

Definition at line 141 of file PersistenceUpdater.h.

Referenced by create(), destroy(), init(), storeUpdate(), and update().

last_part_id_

PartIdType* Update::PersistenceUpdater::last_part_id_

private

What the last participant id is/was.

Definition at line 153 of file PersistenceUpdater.h.

Referenced by init(), requestImage(), and updateLastPartId().

participant_index_

ParticipantIndex* Update::PersistenceUpdater::participant_index_

private

Persisted Participants.

Definition at line 147 of file PersistenceUpdater.h.

Referenced by create(), destroy(), init(), requestImage(), and update().

persistence_file_

ACE_TString Update::PersistenceUpdater::persistence_file_

private

Definition at line 136 of file PersistenceUpdater.h.

Referenced by init(), and parse().

reset_

bool Update::PersistenceUpdater::reset_

private

Definition at line 137 of file PersistenceUpdater.h.

Referenced by init(), and parse().

topic_index_

TopicIndex* Update::PersistenceUpdater::topic_index_

private

Persisted Topics.

Definition at line 144 of file PersistenceUpdater.h.

Referenced by create(), destroy(), init(), requestImage(), and update().

um_

Manager* Update::PersistenceUpdater::um_

private

Definition at line 139 of file PersistenceUpdater.h.

Referenced by init(), and requestImage().

---

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

• PersistenceUpdater.h
• PersistenceUpdater.cpp