SubscriberImpl.h

Go to the documentation of this file.

/*
 * Distributed under the OpenDDS License.
 * See: http://www.opendds.org/license.html
 */

#ifndef OPENDDS_DCPS_SUBSCRIBERIMPL_H
#define OPENDDS_DCPS_SUBSCRIBERIMPL_H

#include "DataReaderCallbacks.h"
#include "EntityImpl.h"
#include "Definitions.h"
#include "DataCollector_T.h"
#include "DataReaderImpl.h"
#include "PoolAllocator.h"

#include <dds/DdsDcpsInfoUtilsC.h>

#include <ace/Recursive_Thread_Mutex.h>

#ifndef ACE_LACKS_PRAGMA_ONCE
#  pragma once
#endif

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL

namespace OpenDDS {
namespace DCPS {

class DomainParticipantImpl;
class Monitor;

#ifndef OPENDDS_NO_MULTI_TOPIC
class MultiTopicImpl;
#endif

class OpenDDS_Dcps_Export SubscriberImpl
  : public virtual OpenDDS::DCPS::LocalObject<DDS::Subscriber>
  , public virtual EntityImpl {
public:

  SubscriberImpl(DDS::InstanceHandle_t handle,
                 const DDS::SubscriberQos& qos,
                 DDS::SubscriberListener_ptr a_listener,
                 const DDS::StatusMask& mask,
                 DomainParticipantImpl* participant);

  virtual ~SubscriberImpl();

  virtual DDS::InstanceHandle_t get_instance_handle();

  bool contains_reader(DDS::InstanceHandle_t a_handle);

  virtual DDS::DataReader_ptr create_datareader(
    DDS::TopicDescription_ptr a_topic_desc,
    const DDS::DataReaderQos& qos,
    DDS::DataReaderListener_ptr a_listener,
    DDS::StatusMask mask);

  virtual DDS::ReturnCode_t delete_datareader(
    DDS::DataReader_ptr a_datareader);

  virtual DDS::ReturnCode_t delete_contained_entities();

  virtual DDS::DataReader_ptr lookup_datareader(
    const char* topic_name);

  virtual DDS::ReturnCode_t get_datareaders(
    DDS::DataReaderSeq& readers,
    DDS::SampleStateMask sample_states,
    DDS::ViewStateMask view_states,
    DDS::InstanceStateMask instance_states);

  virtual DDS::ReturnCode_t notify_datareaders();

  virtual DDS::ReturnCode_t set_qos(
    const DDS::SubscriberQos& qos);

  virtual DDS::ReturnCode_t get_qos(
    DDS::SubscriberQos& qos);

  virtual DDS::ReturnCode_t set_listener(
    DDS::SubscriberListener_ptr a_listener,
    DDS::StatusMask mask);

  virtual DDS::SubscriberListener_ptr get_listener();

#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE

  virtual DDS::ReturnCode_t begin_access();

  virtual DDS::ReturnCode_t end_access();

#endif

  virtual DDS::DomainParticipant_ptr get_participant();

  virtual DDS::ReturnCode_t set_default_datareader_qos(
    const DDS::DataReaderQos& qos);

  virtual DDS::ReturnCode_t get_default_datareader_qos(
    DDS::DataReaderQos& qos);

  virtual DDS::ReturnCode_t copy_from_topic_qos(
    DDS::DataReaderQos& a_datareader_qos,
    const DDS::TopicQos& a_topic_qos);

  virtual DDS::ReturnCode_t enable();

  /**
   * This method is not defined in the IDL and is defined for
   * internal use.
   * Check if there is any datareader associated with it.
   */
  bool is_clean(String* leftover_entities = 0) const;

  // called by DataReaderImpl::data_received
  void data_received(DataReaderImpl* reader);

  DDS::ReturnCode_t reader_enabled(const char* topic_name,
                                   DataReaderImpl* reader);

#ifndef OPENDDS_NO_MULTI_TOPIC
  DDS::ReturnCode_t multitopic_reader_enabled(DDS::DataReader_ptr reader);
  void remove_from_datareader_set(DataReaderImpl* reader);
#endif

  DDS::SubscriberListener_ptr listener_for(DDS::StatusKind kind);

  /// @name Raw Latency Statistics Configuration Interfaces
  /// @{

  /// Configure the size of the raw data collection buffer.
  unsigned int& raw_latency_buffer_size();

  /// Configure the type of the raw data collection buffer.
  DataCollector<double>::OnFull& raw_latency_buffer_type();

  /// @}

  typedef OPENDDS_VECTOR(GUID_t) SubscriptionIdVec;
  /// Populates a std::vector with the SubscriptionIds (GUIDs)
  /// of this Subscriber's Data Readers
  void get_subscription_ids(SubscriptionIdVec& subs);

#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
  void update_ownership_strength (const GUID_t& pub_id,
                                  const CORBA::Long& ownership_strength);
#endif

#ifndef OPENDDS_NO_OBJECT_MODEL_PROFILE
  void coherent_change_received(const GUID_t& publisher_id,
                                DataReaderImpl* reader,
                                Coherent_State& group_state);
#endif

  virtual RcHandle<EntityImpl> parent() const;

  static bool validate_datareader_qos(const DDS::DataReaderQos & qos,
                                      const DDS::DataReaderQos & default_qos,
                                      DDS::Topic_ptr a_topic,
                                      DDS::DataReaderQos & result_qos, bool mt);

private:

  /// Keep track of all the DataReaders attached to this
  /// Subscriber: key is the topic_name
  typedef OPENDDS_MULTIMAP(OPENDDS_STRING, DataReaderImpl_rch) DataReaderMap;

  /// Keep track of DataReaders with data
  /// std::set for now, want to encapsulate
  /// this so we can switch between a set or
  /// list depending on Presentation QoS.
  typedef OPENDDS_SET(DataReaderImpl_rch) DataReaderSet;

  /// DataReader id to qos map.
  typedef OPENDDS_MAP_CMP(GUID_t, DDS::DataReaderQos, GUID_tKeyLessThan) DrIdToQosMap;

  DDS::InstanceHandle_t        handle_;

  DDS::SubscriberQos           qos_;
  DDS::DataReaderQos           default_datareader_qos_;

  ACE_Thread_Mutex             listener_mutex_;
  DDS::StatusMask              listener_mask_;
  DDS::SubscriberListener_var  listener_;

  DataReaderSet                readers_not_enabled_;
  DataReaderMap                datareader_map_;
  DataReaderSet                datareader_set_;

#ifndef OPENDDS_NO_MULTI_TOPIC
  typedef OPENDDS_MAP(String, DDS::DataReader_var) MultitopicReaderMap;
  MultitopicReaderMap multitopic_reader_map_;
#endif

  WeakRcHandle<DomainParticipantImpl> participant_;

  DDS::DomainId_t              domain_id_;
  GUID_t                       dp_id_;

  /// Bound (or initial reservation) of raw latency buffers.
  unsigned int raw_latency_buffer_size_;

  /// Type of raw latency data buffers.
  DataCollector<double>::OnFull raw_latency_buffer_type_;

  /// This lock protects datareader_set_. Only this lock needs to
  /// be acquired if only datareader_set_ is accessed.
  ACE_Recursive_Thread_Mutex dr_set_lock_;

  /// General lock protects the data structures in this class.
  /// If datareader_set_ is accessed together with other data members,
  /// acquire dr_set_lock_ in the scope of this lock.
  ACE_Recursive_Thread_Mutex   si_lock_;

  /// Monitor object for this entity
  unique_ptr<Monitor> monitor_;

  int access_depth_;
};

} // namespace DCPS
} // namespace OpenDDS

OPENDDS_END_VERSIONED_NAMESPACE_DECL

#endif /* OPENDDS_DCPS_SUBSCRIBER_H  */