DomainParticipantImpl.h

Go to the documentation of this file.

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

#ifndef OPENDDS_DCPS_DOMAIN_PARTICIPANT_IMPL_H
#define OPENDDS_DCPS_DOMAIN_PARTICIPANT_IMPL_H

#include "EntityImpl.h"
#include "Definitions.h"
#include "DisjointSequence.h"
#include "TopicImpl.h"
#include "InstanceHandle.h"
#include "OwnershipManager.h"
#include "GuidBuilder.h"
#include "PoolAllocator.h"
#include "Recorder.h"
#include "Replayer.h"
#include "ConditionVariable.h"
#include "TimeTypes.h"
#include "GuidUtils.h"
#include "SporadicTask.h"
#include "XTypes/TypeLookupService.h"
#include "transport/framework/TransportImpl_rch.h"
#include "security/framework/SecurityConfig_rch.h"

#include <dds/DdsDcpsPublicationC.h>
#include <dds/DdsDcpsSubscriptionExtC.h>
#include <dds/DdsDcpsTopicC.h>
#include <dds/DdsDcpsDomainC.h>
#include <dds/DdsDcpsInfoUtilsC.h>
#include <dds/DdsDcpsInfrastructureC.h>
#include <dds/DdsDynamicDataC.h>
#ifndef DDS_HAS_MINIMUM_BIT
#  include <dds/DdsDcpsCoreTypeSupportC.h>
#endif

#include <ace/Null_Mutex.h>
#include <ace/Thread_Mutex.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 PublisherImpl;
class SubscriberImpl;
class DataWriterImpl;
class DomainParticipantFactoryImpl;
class Monitor;
class BitSubscriber;

class RecorderImpl;
class ReplayerImpl;

#ifndef OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE
class FilterEvaluator;
#endif

/**
 * @class DomainParticipantImpl
 *
 * @brief Implements the OpenDDS::DCPS::DomainParticipant interfaces.
 *
 * This class acts as an entrypoint of the service and a factory
 * for publisher, subscriber and topic. It also acts as a container
 * for the publisher, subscriber and topic objects.
 *
 * See the DDS specification, OMG formal/2015-04-10, for a description of
 * the interface this class is implementing.
 */
class OpenDDS_Dcps_Export DomainParticipantImpl
  : public virtual OpenDDS::DCPS::LocalObject<DDS::DomainParticipant>
  , public virtual OpenDDS::DCPS::EntityImpl
  , public virtual ACE_Event_Handler
{
public:
  typedef Objref_Servant_Pair <SubscriberImpl, DDS::Subscriber,
                               DDS::Subscriber_ptr, DDS::Subscriber_var> Subscriber_Pair;

  typedef Objref_Servant_Pair <PublisherImpl, DDS::Publisher,
                               DDS::Publisher_ptr, DDS::Publisher_var> Publisher_Pair;

  typedef Objref_Servant_Pair <TopicImpl, DDS::Topic,
                               DDS::Topic_ptr, DDS::Topic_var> Topic_Pair;

  typedef OPENDDS_SET(Subscriber_Pair) SubscriberSet;
  typedef OPENDDS_SET(Publisher_Pair) PublisherSet;

  class OpenDDS_Dcps_Export RepoIdSequence {
  public:
    explicit RepoIdSequence(const GUID_t& base);
    GUID_t next();
  private:
    GUID_t base_;          // will be combined with serial to produce next
    long serial_;          // will be incremented each time
    GuidBuilder builder_;  // used to modify base
  };

  struct RefCounted_Topic {
    RefCounted_Topic()
      : client_refs_(0)
    {
    }

    explicit RefCounted_Topic(const Topic_Pair& pair)
      : pair_(pair),
      client_refs_(1)
    {
    }

    /// The topic object reference.
    Topic_Pair pair_;
    /// The reference count on the obj_.
    CORBA::ULong client_refs_;
  };

  typedef OPENDDS_MULTIMAP(OPENDDS_STRING, RefCounted_Topic) TopicMap;
  typedef TopicMap::iterator TopicMapIterator;
  typedef std::pair<TopicMapIterator, TopicMapIterator> TopicMapIteratorPair;

  typedef OPENDDS_MAP(OPENDDS_STRING, DDS::TopicDescription_var) TopicDescriptionMap;

  DomainParticipantImpl(InstanceHandleGenerator&           handle_generator,
                        const DDS::DomainId_t&             domain_id,
                        const DDS::DomainParticipantQos &  qos,
                        DDS::DomainParticipantListener_ptr a_listener,
                        const DDS::StatusMask &            mask);

  virtual ~DomainParticipantImpl();

  virtual DDS::InstanceHandle_t get_instance_handle();

  virtual DDS::Publisher_ptr create_publisher(
    const DDS::PublisherQos &  qos,
    DDS::PublisherListener_ptr a_listener,
    DDS::StatusMask            mask);

  virtual DDS::ReturnCode_t delete_publisher(
    DDS::Publisher_ptr p);

  virtual DDS::Subscriber_ptr create_subscriber(
    const DDS::SubscriberQos &  qos,
    DDS::SubscriberListener_ptr a_listener,
    DDS::StatusMask             mask);

  virtual DDS::ReturnCode_t delete_subscriber(
    DDS::Subscriber_ptr s);

  virtual DDS::Subscriber_ptr get_builtin_subscriber();

  RcHandle<DCPS::BitSubscriber> get_builtin_subscriber_proxy();

  virtual DDS::Topic_ptr create_topic(
    const char *           topic_name,
    const char *           type_name,
    const DDS::TopicQos &  qos,
    DDS::TopicListener_ptr a_listener,
    DDS::StatusMask        mask);

  virtual DDS::ReturnCode_t delete_topic(
    DDS::Topic_ptr a_topic);

  virtual DDS::Topic_ptr find_topic(
    const char *            topic_name,
    const DDS::Duration_t & timeout);

  virtual DDS::TopicDescription_ptr lookup_topicdescription(
    const char * name);

#ifndef OPENDDS_NO_CONTENT_FILTERED_TOPIC

  virtual DDS::ContentFilteredTopic_ptr create_contentfilteredtopic(
    const char *           name,
    DDS::Topic_ptr         related_topic,
    const char *           filter_expression,
    const DDS::StringSeq & expression_parameters);

  virtual DDS::ReturnCode_t delete_contentfilteredtopic(
    DDS::ContentFilteredTopic_ptr a_contentfilteredtopic);

#endif

#ifndef OPENDDS_NO_MULTI_TOPIC

  virtual DDS::MultiTopic_ptr create_multitopic(
    const char *           name,
    const char *           type_name,
    const char *           subscription_expression,
    const DDS::StringSeq & expression_parameters);

  virtual DDS::ReturnCode_t delete_multitopic(DDS::MultiTopic_ptr a_multitopic);

#endif

#ifndef OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE

  RcHandle<FilterEvaluator> get_filter_eval(const char* filter);
  void deref_filter_eval(const char* filter);

#endif

  virtual DDS::ReturnCode_t delete_contained_entities();

  virtual CORBA::Boolean contains_entity(DDS::InstanceHandle_t a_handle);

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

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

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

  virtual DDS::DomainParticipantListener_ptr get_listener();

  virtual DDS::ReturnCode_t ignore_participant(
    DDS::InstanceHandle_t handle);

  virtual DDS::ReturnCode_t ignore_topic(
    DDS::InstanceHandle_t handle);

  virtual DDS::ReturnCode_t ignore_publication(
    DDS::InstanceHandle_t handle);

  virtual DDS::ReturnCode_t ignore_subscription(
    DDS::InstanceHandle_t handle);

  virtual DDS::DomainId_t get_domain_id();

  virtual DDS::ReturnCode_t assert_liveliness();

  virtual DDS::ReturnCode_t set_default_publisher_qos(
    const DDS::PublisherQos & qos);

  virtual DDS::ReturnCode_t get_default_publisher_qos(
    DDS::PublisherQos & qos);

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

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

  virtual DDS::ReturnCode_t set_default_topic_qos(
    const DDS::TopicQos & qos);

  virtual DDS::ReturnCode_t get_default_topic_qos(
    DDS::TopicQos & qos);

  /**
   * Set Argument to Current System Time
   */
  virtual DDS::ReturnCode_t get_current_time(DDS::Time_t& current_time);

#if !defined (DDS_HAS_MINIMUM_BIT)

  virtual DDS::ReturnCode_t get_discovered_participants(
    DDS::InstanceHandleSeq & participant_handles);

  virtual DDS::ReturnCode_t get_discovered_participant_data(
    DDS::ParticipantBuiltinTopicData & participant_data,
    DDS::InstanceHandle_t              participant_handle);

  virtual DDS::ReturnCode_t get_discovered_topics(
    DDS::InstanceHandleSeq & topic_handles);

  virtual DDS::ReturnCode_t get_discovered_topic_data(
    DDS::TopicBuiltinTopicData & topic_data,
    DDS::InstanceHandle_t        topic_handle);

#endif

  virtual DDS::ReturnCode_t enable();

  /// Following methods are not the idl interfaces and are
  /// local operations.

  /**
   *  Return the id given by discovery.
   */
  GUID_t get_id() const;

  /**
   * Return a unique string based on repo ID.
   */
  OPENDDS_STRING get_unique_id();

  /**
   * Assign an instance handle, optionally representing a GUID.
   * If a GUID is provided (not GUID_UNKNOWN), other calls to assign_handle
   * for this GUID will return the same handle, as will subsequent calls to
   * lookup_handle.
   *
   * If this method returns a valid (non-HANDLE_NIL) handle, it must be
   * returned by calling return_handle.
   */
  DDS::InstanceHandle_t assign_handle(const GUID_t& id = GUID_UNKNOWN);

  /**
   * Get a handle that was previously mapped to a GUID or HANDLE_NIL if none exists.
   *
   * Handles returned from this method should not be passed to return_handle.
   */
  DDS::InstanceHandle_t lookup_handle(const GUID_t& id) const;

  /**
   * Similar to lookup_handle in that it will return a previously mapped handle,
   * but will coorindate with assign_handle when a desired handle has not yet
   * been mapped, but is expected to be. The optional max_wait argument can be
   * supplied to limit the time spent waiting for a handle. If the wait times out,
   * a value of HANDLE_NIL is returned.
   */
  DDS::InstanceHandle_t await_handle(const GUID_t& id, TimeDuration max_wait = TimeDuration::zero_value) const;

  /**
   * Return a previously-assigned handle.
   */
  void return_handle(DDS::InstanceHandle_t handle);

  /**
   * Obtain a GUID representing a local hande.
   * @return GUID_UNKNOWN if not found.
   */
  GUID_t get_repoid(DDS::InstanceHandle_t id) const;

  /**
   * Check to see if the Participant has any entities left in it.
   * leftover_entities will be set with a description of what is left.
   */
  bool is_clean(String* leftover_entities = 0) const;

  /**
   * This is used to retrieve the listener for a certain status change.
   * If this DomainParticipant has a registered listener and the status
   * kind is in the listener mask then the listener is returned.
   * Otherwise, return nil.
   */
  DDS::DomainParticipantListener_ptr listener_for(DDS::StatusKind kind);

  typedef OPENDDS_VECTOR(GUID_t) TopicIdVec;
  /**
   * Populates an std::vector with the GUID_t of the topics this
   * participant has created/found.
   */
  void get_topic_ids(TopicIdVec& topics);

#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE

  /** Accessor for ownership manager.
   */
  OwnershipManager* ownership_manager();


  /**
   * Called upon receiving new BIT publication data to
   * update the ownership strength of a publication.
   */
  void update_ownership_strength(const GUID_t& pub_id,
                                 const CORBA::Long&   ownership_strength);

#endif

  bool federated() const {
    return this->federated_;
  }


  Recorder_ptr create_recorder(DDS::Topic_ptr               a_topic,
                               const DDS::SubscriberQos &   subscriber_qos,
                               const DDS::DataReaderQos &   datareader_qos,
                               const RecorderListener_rch & a_listener,
                               DDS::StatusMask              mask);

  Replayer_ptr create_replayer(DDS::Topic_ptr               a_topic,
                               const DDS::PublisherQos &    publisher_qos,
                               const DDS::DataWriterQos &   datawriter_qos,
                               const ReplayerListener_rch & a_listener,
                               DDS::StatusMask              mask);

  DDS::Topic_ptr create_typeless_topic(
    const char *           topic_name,
    const char *           type_name,
    bool                   type_has_keys,
    const DDS::TopicQos &  qos,
    DDS::TopicListener_ptr a_listener,
    DDS::StatusMask        mask);

  void delete_recorder(Recorder_ptr recorder);
  void delete_replayer(Replayer_ptr replayer);

  void add_adjust_liveliness_timers(DataWriterImpl* writer);
  void remove_adjust_liveliness_timers();

  XTypes::TypeLookupService_rch get_type_lookup_service() { return type_lookup_service_; }

#if defined(OPENDDS_SECURITY)
  Security::SecurityConfig_rch get_security_config() const
  {
    return security_config_;
  }
  DDS::Security::PermissionsHandle permissions_handle() const
  {
    return perm_handle_;
  }
  DDS::Security::ParticipantCryptoHandle crypto_handle() const
  {
    return part_crypto_handle_;
  }
#endif

  bool prepare_to_delete_datawriters();
  bool set_wait_pending_deadline(const MonotonicTimePoint& deadline);

#ifndef OPENDDS_SAFETY_PROFILE
  DDS::ReturnCode_t get_dynamic_type(
    DDS::DynamicType_var& type, const DDS::BuiltinTopicKey_t& key);
#endif

private:
  bool validate_publisher_qos(DDS::PublisherQos & publisher_qos);
  bool validate_subscriber_qos(DDS::SubscriberQos & subscriber_qos);

  /** The implementation of create_topic.
   */

  ///{@
  /// constants for the topic_mask argument to create_topic_i
  static const int TOPIC_TYPE_HAS_KEYS = 1;
  static const int TOPIC_TYPELESS = 2;
  ///@}

  DDS::Topic_ptr create_topic_i(
    const char *           topic_name,
    const char *           type_name,
    const DDS::TopicQos &  qos,
    DDS::TopicListener_ptr a_listener,
    DDS::StatusMask        mask,
    int                    topic_mask);

  DDS::Topic_ptr create_new_topic(
    const char *                   topic_name,
    const char *                   type_name,
    const DDS::TopicQos &          qos,
    DDS::TopicListener_ptr         a_listener,
    const DDS::StatusMask &        mask,
    OpenDDS::DCPS::TypeSupport_ptr type_support);

  /** Delete the topic with option of whether the
   *  topic object reference should be removed.
   */
  DDS::ReturnCode_t delete_topic_i(
    DDS::Topic_ptr a_topic,
    bool           remove_objref);

  /// The default topic qos.
  DDS::TopicQos default_topic_qos_;
  /// The default publisher qos.
  DDS::PublisherQos default_publisher_qos_;
  /// The default subscriber qos.
  DDS::SubscriberQos default_subscriber_qos_;

  /// The qos of this DomainParticipant.
  DDS::DomainParticipantQos qos_;
  /// Mutex to protect listener info
  ACE_Thread_Mutex listener_mutex_;
  /// Used to notify the entity for relevant events.
  DDS::DomainParticipantListener_var listener_;
  /// The StatusKind bit mask indicates which status condition change
  /// can be notified by the listener of this entity.
  DDS::StatusMask listener_mask_;

  #if defined(OPENDDS_SECURITY)
  /// This participant id handle given by authentication.
  DDS::Security::IdentityHandle id_handle_;
  /// This participant permissions handle given by access constrol.
  DDS::Security::PermissionsHandle perm_handle_;
  /// This participant crypto handle given by crypto
  DDS::Security::ParticipantCryptoHandle part_crypto_handle_;
  #endif

  /// The id of the domain that creates this participant.
  const DDS::DomainId_t domain_id_;
  /// This participant id given by discovery.
  GUID_t dp_id_;

  /// Whether this DomainParticipant is attached to a federated
  /// repository.
  bool federated_;

  /// Collection of publishers.
  PublisherSet publishers_;
  /// Collection of subscribers.
  SubscriberSet subscribers_;
  /// Collection of topics.
  TopicMap topics_;
#if !defined(OPENDDS_NO_CONTENT_FILTERED_TOPIC) || !defined(OPENDDS_NO_MULTI_TOPIC)
  /// Collection of TopicDescriptions which are not also Topics
  TopicDescriptionMap topic_descrs_;
#endif

  typedef std::pair<DDS::InstanceHandle_t, unsigned int> HandleWithCounter;
  typedef OPENDDS_MAP_CMP(GUID_t, HandleWithCounter, GUID_tKeyLessThan) CountedHandleMap;
  typedef OPENDDS_MAP(DDS::InstanceHandle_t, GUID_t) RepoIdMap;

  /// Instance handles assigned which are mapped to GUIDs (use handle_protector_)
  CountedHandleMap handles_;
  /// By-handle lookup of instance handles assigned to GUIDs (use handle_protector_)
  RepoIdMap repoIds_;

  typedef OPENDDS_MAP_CMP(GUID_t, DDS::InstanceHandle_t, GUID_tKeyLessThan) HandleMap;

  /// Collection of ignored participants.
  HandleMap ignored_participants_;
  /// Collection of ignored topics.
  HandleMap ignored_topics_;
  /// Protect the publisher collection.
  ACE_Recursive_Thread_Mutex publishers_protector_;
  /// Protect the subscriber collection.
  ACE_Recursive_Thread_Mutex subscribers_protector_;
  /// Protect the topic collection.
  ACE_Recursive_Thread_Mutex topics_protector_;
  /// Protect the handle collection.
  mutable ACE_Thread_Mutex handle_protector_;

  mutable ConditionVariable<ACE_Thread_Mutex> handle_waiters_;

  /// Protect the shutdown.
  ACE_Thread_Mutex shutdown_mutex_;
  ConditionVariable<ACE_Thread_Mutex> shutdown_condition_;
  DDS::ReturnCode_t shutdown_result_;
  bool shutdown_complete_;

  /// The built in topic subscriber.
  RcHandle<BitSubscriber> bit_subscriber_;

  /// Get instances handles from DomainParticipantFactory (use handle_protector_)
  InstanceHandleGenerator& participant_handles_;

  /// Keep track of handles that can be reused (use handle_protector_)
  DisjointSequence::OrderedRanges<DDS::InstanceHandle_t> reusable_handles_;

  unique_ptr<Monitor> monitor_;

#ifndef OPENDDS_NO_OWNERSHIP_KIND_EXCLUSIVE
  OwnershipManager owner_man_;
#endif

  /// Publisher ID generator.
  RepoIdSequence pub_id_gen_;

#ifndef OPENDDS_NO_CONTENT_SUBSCRIPTION_PROFILE
  ACE_Thread_Mutex filter_cache_lock_;
  OPENDDS_MAP(OPENDDS_STRING, RcHandle<FilterEvaluator> ) filter_cache_;
#endif

  typedef OPENDDS_SET_CMP(Recorder_var, VarLess<Recorder> ) RecorderSet;
  typedef OPENDDS_SET_CMP(Replayer_var, VarLess<Replayer> ) ReplayerSet;

  RecorderSet recorders_;
  ReplayerSet replayers_;

#if defined(OPENDDS_SECURITY)
  Security::SecurityConfig_rch security_config_;
#endif

  /// Protect the recorders collection.
  ACE_Recursive_Thread_Mutex recorders_protector_;
  /// Protect the replayers collection.
  ACE_Recursive_Thread_Mutex replayers_protector_;

  class LivelinessTimer : public virtual RcObject {
  public:
    LivelinessTimer(DomainParticipantImpl& impl, DDS::LivelinessQosPolicyKind kind);
    virtual ~LivelinessTimer();
    void add_adjust(OpenDDS::DCPS::DataWriterImpl* writer);
    void remove_adjust();
    void execute(const MonotonicTimePoint& now);
    virtual void dispatch(const MonotonicTimePoint& tv) = 0;
    virtual void cancel() = 0;

  protected:
    DomainParticipantImpl& impl_;
    const DDS::LivelinessQosPolicyKind kind_;

    TimeDuration interval () const { return interval_; }

    virtual void schedule(const TimeDuration& interval) = 0;

  private:
    TimeDuration interval_;
    bool recalculate_interval_;
    MonotonicTimePoint last_liveliness_check_;
    bool scheduled_;
    ACE_Thread_Mutex lock_;
  };

  class AutomaticLivelinessTimer : public LivelinessTimer {
  public:
    AutomaticLivelinessTimer(DomainParticipantImpl& impl);
    virtual void dispatch(const MonotonicTimePoint& tv);

    void cancel()
    {
      impl_.automatic_liveliness_task_->cancel();
    }

  private:
    void schedule(const TimeDuration& interval)
    {
      impl_.automatic_liveliness_task_->schedule(interval);
    }
  };
  RcHandle<AutomaticLivelinessTimer> automatic_liveliness_timer_;
  typedef PmfSporadicTask<AutomaticLivelinessTimer> AutomaticLivelinessTask;
  RcHandle<AutomaticLivelinessTask> automatic_liveliness_task_;

  class ParticipantLivelinessTimer : public LivelinessTimer {
  public:
    ParticipantLivelinessTimer(DomainParticipantImpl& impl);
    virtual void dispatch(const MonotonicTimePoint& tv);

    void cancel()
    {
      impl_.participant_liveliness_task_->cancel();
    }

  private:
    void schedule(const TimeDuration& interval)
    {
      impl_.participant_liveliness_task_->schedule(interval);
    }
  };
  RcHandle<ParticipantLivelinessTimer> participant_liveliness_timer_;
  typedef PmfSporadicTask<ParticipantLivelinessTimer> ParticipantLivelinessTask;
  RcHandle<ParticipantLivelinessTask> participant_liveliness_task_;

  TimeDuration liveliness_check_interval(DDS::LivelinessQosPolicyKind kind);
  bool participant_liveliness_activity_after(const MonotonicTimePoint& tv);
  void signal_liveliness(DDS::LivelinessQosPolicyKind kind);

  MonotonicTimePoint last_liveliness_activity_;

  virtual int handle_exception(ACE_HANDLE fd);

  XTypes::TypeLookupService_rch type_lookup_service_;
};

} // namespace DCPS
} // namespace OpenDDS

OPENDDS_END_VERSIONED_NAMESPACE_DECL

#endif /* OPENDDS_DCPS_DOMAIN_PARTICIPANT_IMPL_H  */