OpenDDS::DCPS::RecorderImpl Class Reference

Implementation of Recorder functionality. More...

#include <RecorderImpl.h>

Inheritance diagram for OpenDDS::DCPS::RecorderImpl:

[legend]Collaboration diagram for OpenDDS::DCPS::RecorderImpl:

[legend]List of all members.

Public Member Functions
	RecorderImpl ()
virtual	~RecorderImpl ()
DDS::ReturnCode_t	cleanup ()
void	init (TopicDescriptionImpl *a_topic_desc, const DDS::DataReaderQos &qos, RecorderListener_rch a_listener, const DDS::StatusMask &mask, DomainParticipantImpl *participant, DDS::SubscriberQos subqos)
DDS::ReturnCode_t	enable ()
virtual bool	check_transport_qos (const TransportInst &inst)
virtual const RepoId &	get_repo_id () const
DDS::DomainId_t	domain_id () const
virtual CORBA::Long	get_priority_value (const AssociationData &data) const
virtual void	data_received (const ReceivedDataSample &sample)
virtual void	notify_subscription_disconnected (const WriterIdSeq &pubids)
virtual void	notify_subscription_reconnected (const WriterIdSeq &pubids)
virtual void	notify_subscription_lost (const WriterIdSeq &pubids)
virtual void	notify_connection_deleted (const RepoId &)
virtual void	add_association (const RepoId &yourId, const WriterAssociation &writer, bool active)
virtual void	association_complete (const RepoId &remote_id)
virtual void	remove_associations (const WriterIdSeq &writers, CORBA::Boolean callback)
virtual void	update_incompatible_qos (const IncompatibleQosStatus &status)
virtual void	inconsistent_topic ()
virtual void	signal_liveliness (const RepoId &remote_participant)
void	remove_all_associations ()
virtual DDS::ReturnCode_t	repoid_to_bit_key (const DCPS::RepoId &id, DDS::BuiltinTopicKey_t &key)
DDS::ReturnCode_t	set_qos (const ::DDS::SubscriberQos &subscriber_qos, const DDS::DataReaderQos &datareader_qos)
DDS::ReturnCode_t	get_qos (DDS::SubscriberQos &subscriber_qos, DDS::DataReaderQos &datareader_qos)
DDS::ReturnCode_t	set_listener (const RecorderListener_rch &a_listener, DDS::StatusMask mask)
RecorderListener_rch	get_listener ()
DomainParticipantImpl *	participant ()
virtual DDS::InstanceHandle_t	get_instance_handle ()
virtual void	register_for_writer (const RepoId &, const RepoId &, const RepoId &, const TransportLocatorSeq &, DiscoveryListener *)
virtual void	unregister_for_writer (const RepoId &, const RepoId &, const RepoId &)
Protected Member Functions
virtual void	remove_associations_i (const WriterIdSeq &writers, bool callback)
void	remove_or_reschedule (const PublicationId &pub_id)
Private Member Functions
void	notify_subscription_lost (const DDS::InstanceHandleSeq &handles)
bool	lookup_instance_handles (const WriterIdSeq &ids, DDS::InstanceHandleSeq &hdls)
	Lookup the instance handles by the publication repo ids.
void	listener_add_ref ()
void	listener_remove_ref ()
void	_add_ref ()
void	_remove_ref ()
typedef	OPENDDS_MAP_CMP (RepoId, DDS::InstanceHandle_t, GUID_tKeyLessThan) RepoIdToHandleMap
typedef	OPENDDS_MAP_CMP (PublicationId, RcHandle< WriterInfo >, GUID_tKeyLessThan) WriterMapType
	publications writing to this reader.
Private Attributes
DDS::DataReaderQos	qos_
ACE_Recursive_Thread_Mutex	sample_lock_
	lock protecting sample container as well as statuses.
DomainParticipantImpl *	participant_servant_
TopicImpl *	topic_servant_
bool	is_exclusive_ownership_
OwnershipManager *	owner_manager_
DDS::SubscriberQos	subqos_
DDS::TopicDescription_var	topic_desc_
DDS::StatusMask	listener_mask_
RecorderListener_rch	listener_
DDS::DomainId_t	domain_id_
RemoveAssociationSweeper< RecorderImpl > *	remove_association_sweeper_
ACE_Recursive_Thread_Mutex	publication_handle_lock_
RepoIdToHandleMap	id_to_handle_map_
DDS::RequestedIncompatibleQosStatus	requested_incompatible_qos_status_
DDS::SubscriptionMatchedStatus	subscription_match_status_
bool	is_bit_
WriterMapType	writers_
ACE_RW_Thread_Mutex	writers_lock_
	RW lock for reading/writing publications.
Friends
class	RemoveAssociationSweeper< RecorderImpl >
class	::DDS_TEST

---

Detailed Description

Implementation of Recorder functionality.

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

Definition at line 39 of file RecorderImpl.h.

---

Constructor & Destructor Documentation

OpenDDS::DCPS::RecorderImpl::RecorderImpl

(

)

Definition at line 47 of file RecorderImpl.cpp.

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

  : qos_(TheServiceParticipant->initial_DataReaderQos()),
  participant_servant_(0),
  topic_servant_(0),
#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
  is_exclusive_ownership_ (false),
#endif
#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
  owner_manager_ (0),
#endif
  subqos_ (TheServiceParticipant->initial_SubscriberQos()),
  topic_desc_(0),
  listener_mask_(DEFAULT_STATUS_MASK),
  domain_id_(0),
  remove_association_sweeper_(
    new RemoveAssociationSweeper<RecorderImpl>(TheServiceParticipant->reactor(),
                                         TheServiceParticipant->reactor_owner(),
                                         this)),
  is_bit_(false)
{

  requested_incompatible_qos_status_.total_count = 0;
  requested_incompatible_qos_status_.total_count_change = 0;
  requested_incompatible_qos_status_.last_policy_id = 0;
  requested_incompatible_qos_status_.policies.length(0);

  subscription_match_status_.total_count = 0;
  subscription_match_status_.total_count_change = 0;
  subscription_match_status_.current_count = 0;
  subscription_match_status_.current_count_change = 0;
  subscription_match_status_.last_publication_handle =
    DDS::HANDLE_NIL;

}

OpenDDS::DCPS::RecorderImpl::~RecorderImpl

(

)

[virtual]

Definition at line 84 of file RecorderImpl.cpp.

References DBG_ENTRY_LVL, remove_association_sweeper_, and writers_.

{
  DBG_ENTRY_LVL("RecorderImpl","~RecorderImpl",6);
  {
    ACE_READ_GUARD(ACE_RW_Thread_Mutex,
                   read_guard,
                   this->writers_lock_);
    // Cancel any uncancelled sweeper timers to decrement reference count.
    WriterMapType::iterator writer;
    for (writer = writers_.begin(); writer != writers_.end(); ++writer) {
      remove_association_sweeper_->cancel_timer(writer->second);
    }
  }

  remove_association_sweeper_->wait();
  remove_association_sweeper_->destroy();
}

---

Member Function Documentation

void OpenDDS::DCPS::RecorderImpl::_add_ref

(

)

[inline, private, virtual]

Implements OpenDDS::DCPS::Recorder.

Definition at line 155 of file RecorderImpl.h.

{ EntityImpl::_add_ref(); }

void OpenDDS::DCPS::RecorderImpl::_remove_ref

(

)

[inline, private, virtual]

Implements OpenDDS::DCPS::Recorder.

Definition at line 156 of file RecorderImpl.h.

{ EntityImpl::_remove_ref(); }

void OpenDDS::DCPS::RecorderImpl::add_association	(	const RepoId &	yourId,
		const WriterAssociation &	writer,
		bool	active
	)			[virtual]

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 279 of file RecorderImpl.cpp.

References DDS::SubscriptionMatchedStatus::current_count, DDS::SubscriptionMatchedStatus::current_count_change, OpenDDS::DCPS::DCPS_debug_level, DDS::DataWriterQos::durability, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::DomainParticipantImpl::id_to_handle(), id_to_handle_map_, OpenDDS::DCPS::RcHandle< T >::in(), is_bit_, DDS::SubscriptionMatchedStatus::last_publication_handle, listener_, OPENDDS_STRING, participant_servant_, OpenDDS::DCPS::AssociationData::publication_transport_priority_, DDS::DataWriterQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, OpenDDS::DCPS::AssociationData::remote_data_, OpenDDS::DCPS::AssociationData::remote_durable_, OpenDDS::DCPS::AssociationData::remote_id_, OpenDDS::DCPS::AssociationData::remote_reliable_, OpenDDS::DCPS::WriterInfoListener::subscription_id_, subscription_match_status_, TheServiceParticipant, DDS::SubscriptionMatchedStatus::total_count, DDS::SubscriptionMatchedStatus::total_count_change, DDS::DataWriterQos::transport_priority, DDS::VOLATILE_DURABILITY_QOS, OpenDDS::DCPS::WriterAssociation::writerId, OpenDDS::DCPS::WriterAssociation::writerQos, writers_, and OpenDDS::DCPS::WriterAssociation::writerTransInfo.

{
    ACE_DEBUG((LM_DEBUG, "RecorderImpl::add_association\n"));
  //
  // The following block is for diagnostic purposes only.
  //
  if (DCPS_debug_level >= 1) {
    GuidConverter reader_converter(yourId);
    GuidConverter writer_converter(writer.writerId);
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) RecorderImpl::add_association - ")
               ACE_TEXT("bit %d local %C remote %C\n"),
               is_bit_,
               OPENDDS_STRING(reader_converter).c_str(),
               OPENDDS_STRING(writer_converter).c_str()));
  }

  //
  // This block prevents adding associations to deleted readers.
  // Presumably this is a "good thing(tm)".
  //
  // if (entity_deleted_ == true) {
  //   if (DCPS_debug_level >= 1)
  //     ACE_DEBUG((LM_DEBUG,
  //                ACE_TEXT("(%P|%t) RecorderImpl::add_association")
  //                ACE_TEXT(" This is a deleted datareader, ignoring add.\n")));
  //
  //   return;
  // }

  //
  // We are being called back from the repository before we are done
  // processing after our call to the repository that caused this call
  // (from the repository) to be made.
  //
  if (GUID_UNKNOWN == subscription_id_) {
    // add_associations was invoked before DCSPInfoRepo::add_subscription() returned.
    subscription_id_ = yourId;
  }

  //
  // We do the following while holding the publication_handle_lock_.
  //
  {
    ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->publication_handle_lock_);

    //
    // For each writer in the list of writers to associate with, we
    // create a WriterInfo and a WriterStats object and store them in
    // our internal maps.
    //
    {
      ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, this->writers_lock_);

      const PublicationId& writer_id = writer.writerId;
      RcHandle<WriterInfo> info = new WriterInfo(this, writer_id, writer.writerQos);
      /*std::pair<WriterMapType::iterator, bool> bpair =*/
      this->writers_.insert(
        // This insertion is idempotent.
        WriterMapType::value_type(
          writer_id,
          info));
      // this->statistics_.insert(
      //   StatsMapType::value_type(
      //     writer_id,
      //     WriterStats(
      //       this->raw_latency_buffer_size_,
      //       this->raw_latency_buffer_type_)));

      // if (DCPS_debug_level > 4) {
      //   GuidConverter converter(writer_id);
      //   ACE_DEBUG((LM_DEBUG,
      //              "(%P|%t) RecorderImpl::add_association: "
      //              "inserted writer %C.return %d \n",
      //              OPENDDS_STRING(converter).c_str(), bpair.second));
      //
      //   WriterMapType::iterator iter = writers_.find(writer_id);
      //   if (iter != writers_.end()) {
      //     // This may not be an error since it could happen that the sample
      //     // is delivered to the datareader after the write is dis-associated
      //     // with this datareader.
      //     GuidConverter reader_converter(subscription_id_);
      //     GuidConverter writer_converter(writer_id);
      //     ACE_DEBUG((LM_DEBUG,
      //               ACE_TEXT("(%P|%t) RecorderImpl::add_association: ")
      //               ACE_TEXT("reader %C is associated with writer %C.\n"),
      //               OPENDDS_STRING(reader_converter).c_str(),
      //               OPENDDS_STRING(writer_converter).c_str()));
      //   }
      // }
    }

    //
    // Propagate the add_associations processing down into the Transport
    // layer here.  This will establish the transport support and reserve
    // usage of an existing connection or initiate creation of a new
    // connection if no suitable connection is available.
    //
    AssociationData data;
    data.remote_id_ = writer.writerId;
    data.remote_data_ = writer.writerTransInfo;
    data.publication_transport_priority_ =
      writer.writerQos.transport_priority.value;
    data.remote_reliable_ =
      (writer.writerQos.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS);
    data.remote_durable_ =
      (writer.writerQos.durability.kind > DDS::VOLATILE_DURABILITY_QOS);

    if (!this->associate(data, active)) {
      if (DCPS_debug_level) {
        ACE_DEBUG((LM_ERROR,
                   ACE_TEXT("(%P|%t) RecorderImpl::add_association: ")
                   ACE_TEXT("ERROR: transport layer failed to associate.\n")));
      }
      return;
    }

    // Check if any publications have already sent a REQUEST_ACK message.
    // {
    //   ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);
    //
    //   WriterMapType::iterator where = this->writers_.find(writer.writerId);
    //
    //   if (where != this->writers_.end()) {
    //     const ACE_Time_Value now = ACE_OS::gettimeofday();
    //
    //     ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_);
    //
    //     if (where->second->should_ack(now)) {
    //       const SequenceNumber sequence = where->second->ack_sequence();
    //       const DDS::Time_t timenow = time_value_to_time(now);
    //       if (this->send_sample_ack(writer.writerId, sequence, timenow)) {
    //         where->second->clear_acks(sequence);
    //       }
    //     }
    //   }
    // }

    //
    // LIVELINESS policy timers are managed here.
    //
    // if (liveliness_lease_duration_ != ACE_Time_Value::zero) {
    //   // this call will start the timer if it is not already set
    //   const ACE_Time_Value now = ACE_OS::gettimeofday();
    //
    //   if (DCPS_debug_level >= 5) {
    //     GuidConverter converter(subscription_id_);
    //     ACE_DEBUG((LM_DEBUG,
    //                ACE_TEXT("(%P|%t) RecorderImpl::add_association: ")
    //                ACE_TEXT("starting/resetting liveliness timer for reader %C\n"),
    //                OPENDDS_STRING(converter).c_str()));
    //   }
    //
    //   this->handle_timeout(now, this);
    // }

    // else - no timer needed when LIVELINESS.lease_duration is INFINITE

  }
  //
  // We no longer hold the publication_handle_lock_.
  //

  //
  // We only do the following processing for readers that are *not*
  // readers of Builtin Topics.
  //
  if (!is_bit_) {

    DDS::InstanceHandle_t handle =
      this->participant_servant_->id_to_handle(writer.writerId);

    //
    // We acquire the publication_handle_lock_ for the remainder of our
    // processing.
    //
    {
      ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->publication_handle_lock_);

      // This insertion is idempotent.
      this->id_to_handle_map_.insert(
        RepoIdToHandleMap::value_type(writer.writerId, handle));

      if (DCPS_debug_level > 4) {
        GuidConverter converter(writer.writerId);
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) RecorderImpl::add_association: ")
                   ACE_TEXT("id_to_handle_map_[ %C] = 0x%x.\n"),
                   OPENDDS_STRING(converter).c_str(),
                   handle));
      }

      // We need to adjust these after the insertions have all completed
      // since insertions are not guaranteed to increase the number of
      // currently matched publications.
      int matchedPublications = static_cast<int>(this->id_to_handle_map_.size());
      this->subscription_match_status_.current_count_change
        = matchedPublications - this->subscription_match_status_.current_count;
      this->subscription_match_status_.current_count = matchedPublications;

      ++this->subscription_match_status_.total_count;
      ++this->subscription_match_status_.total_count_change;

      this->subscription_match_status_.last_publication_handle = handle;

      // set_status_changed_flag(DDS::SUBSCRIPTION_MATCHED_STATUS, true);


      if (listener_.in()) {
        listener_->on_recorder_matched(
          this,
          this->subscription_match_status_);

        // TBD - why does the spec say to change this but not change
        //       the ChangeFlagStatus after a listener call?

        // Client will look at it so next time it looks the change should be 0
        this->subscription_match_status_.total_count_change = 0;
        this->subscription_match_status_.current_count_change = 0;
      }

      // notify_status_condition();
    }

    {
      ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_);

      this->writers_[writer.writerId]->handle_ = handle;
    }
  }

  if (!active) {
    Discovery_rch disco = TheServiceParticipant->get_discovery(this->domain_id_);
    disco->association_complete(this->domain_id_,
                                this->participant_servant_->get_id(),
                                this->subscription_id_, writer.writerId);
  }

  // if (this->monitor_) {
  //   this->monitor_->report();
  // }
}

void OpenDDS::DCPS::RecorderImpl::association_complete

(

const RepoId &

remote_id

)

[virtual]

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 525 of file RecorderImpl.cpp.

{
  // For the current DCPSInfoRepo implementation, the DataReader side will
  // always be passive, so association_complete() will not be called.
}

bool OpenDDS::DCPS::RecorderImpl::check_transport_qos

(

const TransportInst &

inst

)

[virtual]

Implements OpenDDS::DCPS::TransportClient.

Definition at line 199 of file RecorderImpl.cpp.

References OpenDDS::DCPS::TransportInst::is_reliable(), and DDS::RELIABLE_RELIABILITY_QOS.

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

DDS::ReturnCode_t OpenDDS::DCPS::RecorderImpl::cleanup

(

)

cleanup the DataWriter.

Definition at line 104 of file RecorderImpl.cpp.

References OpenDDS::DCPS::DomainParticipantImpl::get_id(), participant_servant_, remove_all_associations(), remove_association_sweeper_, OpenDDS::DCPS::TopicImpl::remove_entity_ref(), DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::WriterInfoListener::subscription_id_, TheServiceParticipant, topic_servant_, and writers_.

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

{

  Discovery_rch disco = TheServiceParticipant->get_discovery(this->domain_id_);
  if (!disco->remove_subscription(this->domain_id_,
                                  participant_servant_->get_id(),
                                  this->subscription_id_)) {
    ACE_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: ")
                      ACE_TEXT("RecorderImpl::cleanup: ")
                      ACE_TEXT(" could not remove subscription from discovery.\n")),
                     DDS::RETCODE_ERROR);
  }

  // Call remove association before unregistering the datareader from the transport,
  // otherwise some callbacks resulted from remove_association may lost.

  this->remove_all_associations();

  if (topic_servant_) {
    topic_servant_->remove_entity_ref();
    topic_servant_->_remove_ref();
  }
  {
    ACE_READ_GUARD_RETURN(ACE_RW_Thread_Mutex,
                   read_guard,
                   this->writers_lock_,
                   0);
    // Cancel any uncancelled sweeper timers
    WriterMapType::iterator writer;
    for (writer = writers_.begin(); writer != writers_.end(); ++writer) {
      remove_association_sweeper_->cancel_timer(writer->second);
    }
  }

  remove_association_sweeper_->wait();
  return DDS::RETCODE_OK;
}

void OpenDDS::DCPS::RecorderImpl::data_received

(

const ReceivedDataSample &

sample

)

[virtual]

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 218 of file RecorderImpl.cpp.

References OpenDDS::DCPS::DataSampleHeader::byte_order_, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::RcHandle< T >::in(), listener_, OpenDDS::DCPS::DataSampleHeader::message_id_, OPENDDS_STRING, OpenDDS::DCPS::DataSampleHeader::publication_id_, OpenDDS::DCPS::ReceivedDataSample::sample_, OpenDDS::DCPS::SAMPLE_DATA, OpenDDS::DCPS::DataSampleHeader::source_timestamp_nanosec_, OpenDDS::DCPS::DataSampleHeader::source_timestamp_sec_, OpenDDS::DCPS::WriterInfoListener::subscription_id_, and OpenDDS::DCPS::to_string().

{
  DBG_ENTRY_LVL("RecorderImpl","data_received",6);

  // ensure some other thread is not changing the sample container
  // or statuses related to samples.
  ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_);

  if (DCPS_debug_level > 9) {
    GuidConverter converter(subscription_id_);
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) RecorderImpl::data_received: ")
               ACE_TEXT("%C received sample: %C.\n"),
               OPENDDS_STRING(converter).c_str(),
               to_string(sample.header_).c_str()));
  }

  // we only support SAMPLE_DATA messages
  if (sample.header_.message_id_ != SAMPLE_DATA)
    return;

  RawDataSample rawSample(static_cast<MessageId> (sample.header_.message_id_),
                          sample.header_.source_timestamp_sec_,
                          sample.header_.source_timestamp_nanosec_,
                          sample.header_.publication_id_,
                          sample.header_.byte_order_,
                          sample.sample_);

  if (listener_.in()) {
    listener_->on_sample_data_received(this, rawSample);
  }

}

DDS::DomainId_t OpenDDS::DCPS::RecorderImpl::domain_id

(

)

const [inline, virtual]

Implements OpenDDS::DCPS::TransportClient.

Definition at line 73 of file RecorderImpl.h.

{ return this->domain_id_; }

DDS::ReturnCode_t OpenDDS::DCPS::RecorderImpl::enable

(

)

Implements DDS::Entity.

Definition at line 968 of file RecorderImpl.cpp.

References OpenDDS::DCPS::TransportClient::connection_info(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TransportClient::enable_transport(), OpenDDS::DCPS::TopicImpl::get_id(), OpenDDS::DCPS::TopicDescriptionImpl::get_name(), OpenDDS::DCPS::GUID_UNKNOWN, DDS::RELIABLE_RELIABILITY_QOS, DDS::RETCODE_ERROR, DDS::RETCODE_OK, OpenDDS::DCPS::EntityImpl::set_enabled(), subqos_, OpenDDS::DCPS::WriterInfoListener::subscription_id_, TheServiceParticipant, topic_servant_, and DDS::VOLATILE_DURABILITY_QOS.

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

{
  if (DCPS_debug_level >= 1) {

    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) RecorderImpl::enable\n")));
  }
  //According spec:
  // - Calling enable on an already enabled Entity returns OK and has no
  // effect.
  // - Calling enable on an Entity whose factory is not enabled will fail
  // and return PRECONDITION_NOT_MET.

  if (this->is_enabled()) {
    return DDS::RETCODE_OK;
  }

  this->set_enabled();

  // if (topic_servant_ && !transport_disabled_) {
  if (topic_servant_) {

    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) RecorderImpl::enable_transport\n")));

    try {
      this->enable_transport(this->qos_.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS,
                             this->qos_.durability.kind > DDS::VOLATILE_DURABILITY_QOS);
    } catch (const Transport::Exception&) {
      ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: RecorderImpl::enable, ")
                 ACE_TEXT("Transport Exception.\n")));
      return DDS::RETCODE_ERROR;

    }

    const TransportLocatorSeq& trans_conf_info = this->connection_info();

    CORBA::String_var filterClassName = "";
    CORBA::String_var filterExpression = "";
    DDS::StringSeq exprParams;


    Discovery_rch disco =
      TheServiceParticipant->get_discovery(this->domain_id_);

    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) RecorderImpl::add_subscription\n")));

    this->subscription_id_ =
      disco->add_subscription(this->domain_id_,
                              this->participant_servant_->get_id(),
                              this->topic_servant_->get_id(),
                              this,
                              this->qos_,
                              trans_conf_info,
                              this->subqos_,
                              filterClassName,
                              filterExpression,
                              exprParams);

    if (this->subscription_id_ == OpenDDS::DCPS::GUID_UNKNOWN) {
      ACE_ERROR((LM_ERROR,
                 ACE_TEXT("(%P|%t) ERROR: RecorderImpl::enable, ")
                 ACE_TEXT("add_subscription returned invalid id.\n")));
      return DDS::RETCODE_ERROR;
    }
  }

  if (topic_servant_) {
    const CORBA::String_var name = topic_servant_->get_name();
    DDS::ReturnCode_t return_value = DDS::RETCODE_OK;
    //   this->participant_servant_->recorder_enabled(name.in(), this);

    return return_value;
  } else {
    return DDS::RETCODE_OK;
  }
}

DDS::InstanceHandle_t OpenDDS::DCPS::RecorderImpl::get_instance_handle

(

)

[virtual]

Implements OpenDDS::DCPS::EntityImpl.

Definition at line 1049 of file RecorderImpl.cpp.

References OpenDDS::DCPS::DomainParticipantImpl::id_to_handle(), participant_servant_, and OpenDDS::DCPS::WriterInfoListener::subscription_id_.

{
  return this->participant_servant_->id_to_handle(subscription_id_);
}

RecorderListener_rch OpenDDS::DCPS::RecorderImpl::get_listener

(

)

[virtual]

Get the listener for this Recorder.

Implements OpenDDS::DCPS::Recorder.

Definition at line 932 of file RecorderImpl.cpp.

References listener_.

{
  return listener_;
}

CORBA::Long OpenDDS::DCPS::RecorderImpl::get_priority_value

(

const AssociationData &

data

)

const [virtual]

Implements OpenDDS::DCPS::TransportClient.

Definition at line 212 of file RecorderImpl.cpp.

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

{
  return data.publication_transport_priority_;
}

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

Get the Quality of Service settings for the Recorder.

Implements OpenDDS::DCPS::Recorder.

Definition at line 912 of file RecorderImpl.cpp.

References DDS::RETCODE_OK, and subqos_.

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

const RepoId & OpenDDS::DCPS::RecorderImpl::get_repo_id

(

)

const [virtual]

Implements OpenDDS::DCPS::TransportClient.

Definition at line 207 of file RecorderImpl.cpp.

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

{
  return this->subscription_id_;
}

void OpenDDS::DCPS::RecorderImpl::inconsistent_topic

(

)

[virtual]

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 817 of file RecorderImpl.cpp.

References OpenDDS::DCPS::TopicImpl::inconsistent_topic(), and topic_servant_.

{
  topic_servant_->inconsistent_topic();
}

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

Definition at line 143 of file RecorderImpl.cpp.

References OpenDDS::DCPS::TopicImpl::add_entity_ref(), OpenDDS::DCPS::BUILT_IN_PARTICIPANT_TOPIC, OpenDDS::DCPS::BUILT_IN_PUBLICATION_TOPIC, OpenDDS::DCPS::BUILT_IN_SUBSCRIPTION_TOPIC, OpenDDS::DCPS::BUILT_IN_TOPIC_TOPIC, OpenDDS::DCPS::DCPS_debug_level, domain_id_, DDS::EXCLUSIVE_OWNERSHIP_QOS, OpenDDS::DCPS::DomainParticipantImpl::get_domain_id(), is_bit_, is_exclusive_ownership_, listener_, listener_mask_, owner_manager_, OpenDDS::DCPS::DomainParticipantImpl::ownership_manager(), participant(), participant_servant_, qos_, subqos_, topic_desc_, and topic_servant_.

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

{
  //
  if (DCPS_debug_level >= 1) {

    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) RecorderImpl::init \n")));
  }


  topic_desc_ = DDS::TopicDescription::_duplicate(a_topic_desc);
  if (TopicImpl* a_topic = dynamic_cast<TopicImpl*>(a_topic_desc)) {
    topic_servant_ = a_topic;
    topic_servant_->_add_ref();

    topic_servant_->add_entity_ref();
  }

  CORBA::String_var topic_name = a_topic_desc->get_name();

#if !defined (DDS_HAS_MINIMUM_BIT)
  is_bit_ = ACE_OS::strcmp(topic_name.in(), BUILT_IN_PARTICIPANT_TOPIC) == 0
            || ACE_OS::strcmp(topic_name.in(), BUILT_IN_TOPIC_TOPIC) == 0
            || ACE_OS::strcmp(topic_name.in(), BUILT_IN_SUBSCRIPTION_TOPIC) == 0
            || ACE_OS::strcmp(topic_name.in(), BUILT_IN_PUBLICATION_TOPIC) == 0;
#endif   // !defined (DDS_HAS_MINIMUM_BIT)

  qos_ = qos;

#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
  is_exclusive_ownership_ = this->qos_.ownership.kind == ::DDS::EXCLUSIVE_OWNERSHIP_QOS;
#endif

  listener_ = a_listener;
  listener_mask_ = mask;

  // Only store the participant pointer, since it is our "grand"
  // parent, we will exist as long as it does
  participant_servant_ = participant;

#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
  if (is_exclusive_ownership_) {
    owner_manager_ = participant_servant_->ownership_manager ();
  }
#endif

  domain_id_ = participant_servant_->get_domain_id();
  subqos_ = subqos;
}

void OpenDDS::DCPS::RecorderImpl::listener_add_ref

(

)

[inline, private, virtual]

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 153 of file RecorderImpl.h.

{ EntityImpl::_add_ref(); }

void OpenDDS::DCPS::RecorderImpl::listener_remove_ref

(

)

[inline, private, virtual]

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 154 of file RecorderImpl.h.

{ EntityImpl::_remove_ref(); }

bool OpenDDS::DCPS::RecorderImpl::lookup_instance_handles	(	const WriterIdSeq &	ids,
		DDS::InstanceHandleSeq &	hdls
	)			[private]

Lookup the instance handles by the publication repo ids.

Definition at line 938 of file RecorderImpl.cpp.

References OpenDDS::DCPS::DCPS_debug_level, and OPENDDS_STRING.

{
  if (DCPS_debug_level > 9) {
    CORBA::ULong const size = ids.length();
    OPENDDS_STRING separator = "";
    OPENDDS_STRING buffer;

    for (unsigned long i = 0; i < size; ++i) {
      buffer += separator + OPENDDS_STRING(GuidConverter(ids[i]));
      separator = ", ";
    }

    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) RecorderImpl::lookup_instance_handles: ")
               ACE_TEXT("searching for handles for writer Ids: %C.\n"),
               buffer.c_str()));
  }

  CORBA::ULong const num_wrts = ids.length();
  hdls.length(num_wrts);

  for (CORBA::ULong i = 0; i < num_wrts; ++i) {
    hdls[i] = this->participant_servant_->id_to_handle(ids[i]);
  }

  return true;
}

void OpenDDS::DCPS::RecorderImpl::notify_connection_deleted

(

const RepoId &

)

[virtual]

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 271 of file RecorderImpl.cpp.

{

}

void OpenDDS::DCPS::RecorderImpl::notify_subscription_disconnected

(

const WriterIdSeq &

pubids

)

[virtual]

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 252 of file RecorderImpl.cpp.

{
}

void OpenDDS::DCPS::RecorderImpl::notify_subscription_lost

(

const DDS::InstanceHandleSeq &

handles

)

[private]

Definition at line 262 of file RecorderImpl.cpp.

{
}

void OpenDDS::DCPS::RecorderImpl::notify_subscription_lost

(

const WriterIdSeq &

pubids

)

[virtual]

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 266 of file RecorderImpl.cpp.

Referenced by remove_associations_i().

{

}

void OpenDDS::DCPS::RecorderImpl::notify_subscription_reconnected

(

const WriterIdSeq &

pubids

)

[virtual]

Implements OpenDDS::DCPS::TransportReceiveListener.

Definition at line 256 of file RecorderImpl.cpp.

{

}

typedef OpenDDS::DCPS::RecorderImpl::OPENDDS_MAP_CMP	(	PublicationId	,
		RcHandle< WriterInfo >	,
		GUID_tKeyLessThan
	)			[private]

publications writing to this reader.

typedef OpenDDS::DCPS::RecorderImpl::OPENDDS_MAP_CMP	(	RepoId	,
		DDS::InstanceHandle_t	,
		GUID_tKeyLessThan
	)			[private]

DomainParticipantImpl* OpenDDS::DCPS::RecorderImpl::participant

(

)

[inline]

Definition at line 125 of file RecorderImpl.h.

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

                                                {
    return participant_servant_;
  }

void OpenDDS::DCPS::RecorderImpl::register_for_writer	(	const RepoId &	,
		const RepoId &	,
		const RepoId &	,
		const TransportLocatorSeq &	,
		DiscoveryListener *
	)			[virtual]

Reimplemented from OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 1055 of file RecorderImpl.cpp.

References participant(), and OpenDDS::DCPS::TransportClient::register_for_writer().

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

void OpenDDS::DCPS::RecorderImpl::remove_all_associations

(

)

Definition at line 739 of file RecorderImpl.cpp.

References DBG_ENTRY_LVL, remove_associations(), and writers_.

Referenced by cleanup().

{
  DBG_ENTRY_LVL("RecorderImpl","remove_all_associations",6);

  OpenDDS::DCPS::WriterIdSeq writers;
  int size;

  ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->publication_handle_lock_);

  {
    ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);

    size = static_cast<int>(writers_.size());
    writers.length(size);

    WriterMapType::iterator curr_writer = writers_.begin();
    WriterMapType::iterator end_writer = writers_.end();

    int i = 0;

    while (curr_writer != end_writer) {
      writers[i++] = curr_writer->first;
      ++curr_writer;
    }
  }

  try {
    CORBA::Boolean dont_notify_lost = 0;

    if (0 < size) {
      remove_associations(writers, dont_notify_lost);
    }

  } catch (const CORBA::Exception&) {
  }
}

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

Implements OpenDDS::DCPS::DataReaderCallbacks.

Referenced by remove_all_associations(), and remove_or_reschedule().

void OpenDDS::DCPS::RecorderImpl::remove_associations_i	(	const WriterIdSeq &	writers,
		bool	callback
	)			[protected, virtual]

Definition at line 601 of file RecorderImpl.cpp.

References DDS::SubscriptionMatchedStatus::current_count, DDS::SubscriptionMatchedStatus::current_count_change, DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, id_to_handle_map_, OpenDDS::DCPS::RcHandle< T >::in(), is_bit_, DDS::SubscriptionMatchedStatus::last_publication_handle, listener_, notify_subscription_lost(), OPENDDS_STRING, OpenDDS::DCPS::push_back(), remove_association_sweeper_, OpenDDS::DCPS::WriterInfoListener::subscription_id_, subscription_match_status_, DDS::SubscriptionMatchedStatus::total_count_change, and writers_.

Referenced by remove_or_reschedule().

{
  DBG_ENTRY_LVL("RecorderImpl", "remove_associations_i", 6);

  if (writers.length() == 0) {
    return;
  }

  if (DCPS_debug_level >= 1) {
    GuidConverter reader_converter(subscription_id_);
    GuidConverter writer_converter(writers[0]);
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) RecorderImpl::remove_associations_i: ")
               ACE_TEXT("bit %d local %C remote %C num remotes %d \n"),
               is_bit_,
               OPENDDS_STRING(reader_converter).c_str(),
               OPENDDS_STRING(writer_converter).c_str(),
               writers.length()));
  }
  DDS::InstanceHandleSeq handles;

  ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->publication_handle_lock_);

  // This is used to hold the list of writers which were actually
  // removed, which is a proper subset of the writers which were
  // requested to be removed.
  WriterIdSeq updated_writers;

  CORBA::ULong wr_len;

  //Remove the writers from writer list. If the supplied writer
  //is not in the cached writers list then it is already removed.
  //We just need remove the writers in the list that have not been
  //removed.
  {
    ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, this->writers_lock_);

    wr_len = writers.length();

    for (CORBA::ULong i = 0; i < wr_len; i++) {
      PublicationId writer_id = writers[i];

      WriterMapType::iterator it = this->writers_.find(writer_id);

      if (it != this->writers_.end()) {
        it->second->removed();
        remove_association_sweeper_->cancel_timer(it->second);
      }

      if (this->writers_.erase(writer_id) == 0) {
        if (DCPS_debug_level >= 1) {
          GuidConverter converter(writer_id);
          ACE_DEBUG((LM_DEBUG,
                     ACE_TEXT("(%P|%t) RecorderImpl::remove_associations_i: ")
                     ACE_TEXT("the writer local %C was already removed.\n"),
                     OPENDDS_STRING(converter).c_str()));
        }

      } else {
        push_back(updated_writers, writer_id);
      }
    }
  }

  wr_len = updated_writers.length();

  // Return now if the supplied writers have been removed already.
  if (wr_len == 0) {
    return;
  }

  if (!is_bit_) {
    // The writer should be in the id_to_handle map at this time.  Note
    // it if it not there.
    if (this->lookup_instance_handles(updated_writers, handles) == false) {
      if (DCPS_debug_level > 4) {
        ACE_DEBUG((LM_DEBUG,
                   ACE_TEXT("(%P|%t) RecorderImpl::remove_associations_i: ")
                   ACE_TEXT("lookup_instance_handles failed.\n")));
      }
    }
    for (CORBA::ULong i = 0; i < wr_len; ++i) {
      id_to_handle_map_.erase(updated_writers[i]);
    }
  }
  for (CORBA::ULong i = 0; i < updated_writers.length(); ++i) {
    this->disassociate(updated_writers[i]);
  }

  // Mirror the add_associations SUBSCRIPTION_MATCHED_STATUS processing.
  if (!this->is_bit_) {
    // Derive the change in the number of publications writing to this reader.
    int matchedPublications = static_cast<int>(this->id_to_handle_map_.size());
    this->subscription_match_status_.current_count_change
      = matchedPublications - this->subscription_match_status_.current_count;

    // Only process status if the number of publications has changed.
    if (this->subscription_match_status_.current_count_change != 0) {
      this->subscription_match_status_.current_count = matchedPublications;
      /// Section 7.1.4.1: total_count will not decrement.

      /// @TODO: Reconcile this with the verbiage in section 7.1.4.1
      this->subscription_match_status_.last_publication_handle
        = handles[ wr_len - 1];

      // set_status_changed_flag(DDS::SUBSCRIPTION_MATCHED_STATUS, true);

      // DDS::DataReaderListener_var listener
      // = listener_for(DDS::SUBSCRIPTION_MATCHED_STATUS);

      if (listener_.in()) {
        listener_->on_recorder_matched(
          this,
          this->subscription_match_status_);

        // Client will look at it so next time it looks the change should be 0
        this->subscription_match_status_.total_count_change = 0;
        this->subscription_match_status_.current_count_change = 0;
      }

      // notify_status_condition();
    }
  }

  // If this remove_association is invoked when the InfoRepo
  // detects a lost writer then make a callback to notify
  // subscription lost.
  if (notify_lost) {
    this->notify_subscription_lost(handles);
  }

  // if (this->monitor_) {
  //   this->monitor_->report();
  // }
}

void OpenDDS::DCPS::RecorderImpl::remove_or_reschedule

(

const PublicationId &

pub_id

)

[protected]

Definition at line 581 of file RecorderImpl.cpp.

References OpenDDS::DCPS::push_back(), remove_associations(), remove_associations_i(), and writers_.

{
  ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, this->writers_lock_);
  WriterMapType::iterator where = writers_.find(pub_id);
  if (writers_.end() != where) {
    WriterInfo& info = *where->second;
    WriterIdSeq writers;
    push_back(writers, pub_id);
    bool notify = info.notify_lost_;
    if (info.removal_deadline_ < ACE_OS::gettimeofday()) {
      write_guard.release();
      remove_associations_i(writers, notify);
    } else {
      write_guard.release();
      remove_associations(writers, notify);
    }
  }
}

DDS::ReturnCode_t OpenDDS::DCPS::RecorderImpl::repoid_to_bit_key	(	const DCPS::RepoId &	id,
		DDS::BuiltinTopicKey_t &	key
	)			[virtual]

Find the bit key for a given repo id.

Implements OpenDDS::DCPS::Recorder.

Definition at line 1074 of file RecorderImpl.cpp.

References OpenDDS::DCPS::BUILT_IN_PUBLICATION_TOPIC, OpenDDS::DCPS::DomainParticipantImpl::id_to_handle(), participant_servant_, DDS::RETCODE_ERROR, and DDS::RETCODE_OK.

{
  DDS::InstanceHandle_t publication_handle = this->participant_servant_->id_to_handle(id);

  ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,
                   guard,
                   this->publication_handle_lock_,
                   DDS::RETCODE_ERROR);

  BIT_Helper_1 < DDS::PublicationBuiltinTopicDataDataReader,
                 DDS::PublicationBuiltinTopicDataDataReader_var,
                 DDS::PublicationBuiltinTopicDataSeq > hh;

  DDS::PublicationBuiltinTopicDataSeq data;

  DDS::ReturnCode_t ret
    = hh.instance_handle_to_bit_data(participant_servant_,
                                     BUILT_IN_PUBLICATION_TOPIC,
                                     publication_handle,
                                     data);

  if (ret == DDS::RETCODE_OK) {
    key = data[0].key;
  }

  return ret;
}

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

Change the listener for this Recorder.

Implements OpenDDS::DCPS::Recorder.

Definition at line 922 of file RecorderImpl.cpp.

References listener_, listener_mask_, and DDS::RETCODE_OK.

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

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

Set the Quality of Service settings for the Recorder.

Implements OpenDDS::DCPS::Recorder.

void OpenDDS::DCPS::RecorderImpl::signal_liveliness

(

const RepoId &

remote_participant

)

[virtual]

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 823 of file RecorderImpl.cpp.

References OpenDDS::DCPS::GUID_t::entityId, OpenDDS::DCPS::GUID_t::guidPrefix, OpenDDS::DCPS::TransportClient::OPENDDS_VECTOR(), and writers_.

{
  RepoId prefix = remote_participant;
  prefix.entityId = EntityId_t();

  ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_);

  typedef std::pair<RepoId, RcHandle<WriterInfo> > WriterSetElement;
  typedef OPENDDS_VECTOR(WriterSetElement) WriterSet;
  WriterSet writers;

  {
    ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);
    for (WriterMapType::iterator pos = writers_.lower_bound(prefix),
           limit = writers_.end();
         pos != limit && GuidPrefixEqual() (pos->first.guidPrefix, prefix.guidPrefix);
         ++pos) {
      writers.push_back(std::make_pair(pos->first, pos->second));
    }
  }

  ACE_Time_Value when = ACE_OS::gettimeofday();
  for (WriterSet::iterator pos = writers.begin(), limit = writers.end();
       pos != limit;
       ++pos) {
    pos->second->received_activity(when);
  }
}

void OpenDDS::DCPS::RecorderImpl::unregister_for_writer	(	const RepoId &	,
		const RepoId &	,
		const RepoId &
	)			[virtual]

Reimplemented from OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 1065 of file RecorderImpl.cpp.

References participant(), and OpenDDS::DCPS::TransportClient::unregister_for_writer().

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

void OpenDDS::DCPS::RecorderImpl::update_incompatible_qos

(

const IncompatibleQosStatus &

status

)

[virtual]

Implements OpenDDS::DCPS::DataReaderCallbacks.

Definition at line 777 of file RecorderImpl.cpp.

References OpenDDS::DCPS::IncompatibleQosStatus::count_since_last_send, OpenDDS::DCPS::IncompatibleQosStatus::last_policy_id, DDS::RequestedIncompatibleQosStatus::last_policy_id, OpenDDS::DCPS::IncompatibleQosStatus::policies, DDS::RequestedIncompatibleQosStatus::policies, requested_incompatible_qos_status_, DDS::RequestedIncompatibleQosStatus::total_count, OpenDDS::DCPS::IncompatibleQosStatus::total_count, and DDS::RequestedIncompatibleQosStatus::total_count_change.

{


  ACE_GUARD(ACE_Recursive_Thread_Mutex,
            guard,
            this->publication_handle_lock_);

  if (this->requested_incompatible_qos_status_.total_count == status.total_count) {
    // This test should make the method idempotent.
    return;
  }

  // set_status_changed_flag(DDS::REQUESTED_INCOMPATIBLE_QOS_STATUS,
  //                         true);

  // copy status and increment change
  requested_incompatible_qos_status_.total_count = status.total_count;
  requested_incompatible_qos_status_.total_count_change +=
    status.count_since_last_send;
  requested_incompatible_qos_status_.last_policy_id =
    status.last_policy_id;
  requested_incompatible_qos_status_.policies = status.policies;

  // if (!CORBA::is_nil(listener.in())) {
  //   listener->on_requested_incompatible_qos(dr_local_objref_.in(),
  //                                           requested_incompatible_qos_status_);
  //
  //   // TBD - why does the spec say to change total_count_change but not
  //   // change the ChangeFlagStatus after a listener call?
  //
  //   // client just looked at it so next time it looks the
  //   // change should be 0
  //   requested_incompatible_qos_status_.total_count_change = 0;
  // }
  //
  // notify_status_condition();
}

---

Friends And Related Function Documentation

friend class ::DDS_TEST [friend]

Reimplemented from OpenDDS::DCPS::TransportClient.

Definition at line 176 of file RecorderImpl.h.

friend class RemoveAssociationSweeper< RecorderImpl > [friend]

Definition at line 174 of file RecorderImpl.h.

---

Member Data Documentation

DDS::DomainId_t OpenDDS::DCPS::RecorderImpl::domain_id_ [private]

Definition at line 181 of file RecorderImpl.h.

Referenced by init().

RepoIdToHandleMap OpenDDS::DCPS::RecorderImpl::id_to_handle_map_ [private]

Definition at line 187 of file RecorderImpl.h.

Referenced by add_association(), and remove_associations_i().

bool OpenDDS::DCPS::RecorderImpl::is_bit_ [private]

Flag indicates that this datareader is a builtin topic datareader.

Definition at line 194 of file RecorderImpl.h.

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

bool OpenDDS::DCPS::RecorderImpl::is_exclusive_ownership_ [private]

Definition at line 167 of file RecorderImpl.h.

Referenced by init().

RecorderListener_rch OpenDDS::DCPS::RecorderImpl::listener_ [private]

Definition at line 180 of file RecorderImpl.h.

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

DDS::StatusMask OpenDDS::DCPS::RecorderImpl::listener_mask_ [private]

Definition at line 179 of file RecorderImpl.h.

Referenced by init(), and set_listener().

OwnershipManager* OpenDDS::DCPS::RecorderImpl::owner_manager_ [private]

Definition at line 169 of file RecorderImpl.h.

Referenced by init().

DomainParticipantImpl* OpenDDS::DCPS::RecorderImpl::participant_servant_ [private]

Definition at line 163 of file RecorderImpl.h.

Referenced by add_association(), cleanup(), get_instance_handle(), init(), and repoid_to_bit_key().

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::RecorderImpl::publication_handle_lock_ [private]

Definition at line 184 of file RecorderImpl.h.

DDS::DataReaderQos OpenDDS::DCPS::RecorderImpl::qos_ [private]

Definition at line 158 of file RecorderImpl.h.

Referenced by init().

RemoveAssociationSweeper<RecorderImpl>* OpenDDS::DCPS::RecorderImpl::remove_association_sweeper_ [private]

Definition at line 182 of file RecorderImpl.h.

Referenced by cleanup(), remove_associations_i(), and ~RecorderImpl().

DDS::RequestedIncompatibleQosStatus OpenDDS::DCPS::RecorderImpl::requested_incompatible_qos_status_ [private]

Definition at line 189 of file RecorderImpl.h.

Referenced by RecorderImpl(), and update_incompatible_qos().

ACE_Recursive_Thread_Mutex OpenDDS::DCPS::RecorderImpl::sample_lock_ [private]

lock protecting sample container as well as statuses.

Definition at line 161 of file RecorderImpl.h.

DDS::SubscriberQos OpenDDS::DCPS::RecorderImpl::subqos_ [private]

Definition at line 172 of file RecorderImpl.h.

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

DDS::SubscriptionMatchedStatus OpenDDS::DCPS::RecorderImpl::subscription_match_status_ [private]

Definition at line 190 of file RecorderImpl.h.

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

DDS::TopicDescription_var OpenDDS::DCPS::RecorderImpl::topic_desc_ [private]

Definition at line 178 of file RecorderImpl.h.

Referenced by init().

TopicImpl* OpenDDS::DCPS::RecorderImpl::topic_servant_ [private]

Definition at line 164 of file RecorderImpl.h.

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

WriterMapType OpenDDS::DCPS::RecorderImpl::writers_ [private]

Definition at line 199 of file RecorderImpl.h.

Referenced by add_association(), cleanup(), remove_all_associations(), remove_associations_i(), remove_or_reschedule(), signal_liveliness(), and ~RecorderImpl().

ACE_RW_Thread_Mutex OpenDDS::DCPS::RecorderImpl::writers_lock_ [private]

RW lock for reading/writing publications.

Definition at line 202 of file RecorderImpl.h.

---

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

• RecorderImpl.h
• RecorderImpl.cpp

---