Line data    Source code

       1             : /*
       2             :  *
       3             :  *
       4             :  * Distributed under the OpenDDS License.
       5             :  * See: http://www.opendds.org/license.html
       6             :  */
       7             : 
       8             : #ifndef OPENDDS_DCPS_INSTANCESTATE_H
       9             : #define OPENDDS_DCPS_INSTANCESTATE_H
      10             : 
      11             : #include "dcps_export.h"
      12             : #include "ace/Time_Value.h"
      13             : #include "dds/DdsDcpsInfrastructureC.h"
      14             : #include "Definitions.h"
      15             : #include "GuidUtils.h"
      16             : #include "PoolAllocator.h"
      17             : #include "ReactorInterceptor.h"
      18             : #include "TimeTypes.h"
      19             : 
      20             : #if !defined (ACE_LACKS_PRAGMA_ONCE)
      21             : #pragma once
      22             : #endif /* ACE_LACKS_PRAGMA_ONCE */
      23             : 
      24             : OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL
      25             : 
      26             : namespace OpenDDS {
      27             : namespace DCPS {
      28             : 
      29             : class DataReaderImpl;
      30             : typedef RcHandle<DataReaderImpl> DataReaderImpl_rch;
      31             : typedef WeakRcHandle<DataReaderImpl> DataReaderImpl_wrch;
      32             : 
      33             : class InstanceState;
      34             : typedef RcHandle<InstanceState> InstanceState_rch;
      35             : 
      36             : class ReceivedDataElement;
      37             : 
      38             : /**
      39             :  * @class InstanceState
      40             :  *
      41             :  * @brief manage the states of a received data instance.
      42             :  *
      43             :  * Provide a mechanism to manage the view state and instance
      44             :  * state values for an instance contained within a DataReader.
      45             :  * The instance_state and view_state are managed by this class.
      46             :  * Accessors are provided to query the current value of each of
      47             :  * these states.
      48             :  */
      49             : class OpenDDS_Dcps_Export InstanceState : public ReactorInterceptor {
      50             : public:
      51             :   InstanceState(const DataReaderImpl_rch& reader,
      52             :                 ACE_Recursive_Thread_Mutex& lock,
      53             :                 DDS::InstanceHandle_t handle);
      54             : 
      55             :   virtual ~InstanceState();
      56             : 
      57             :   /// Populate the SampleInfo structure
      58             :   void sample_info(DDS::SampleInfo& si,
      59             :                    const ReceivedDataElement* de);
      60             : 
      61             :   /// Access instance state.
      62             :   DDS::InstanceStateKind instance_state() const;
      63             : 
      64             :   /// Access view state.
      65             :   DDS::ViewStateKind view_state() const;
      66             : 
      67             :   bool match(DDS::ViewStateMask view, DDS::InstanceStateMask inst) const;
      68             : 
      69             :   /// Access disposed generation count
      70             :   size_t disposed_generation_count() const;
      71             : 
      72             :   /// Access no writers generation count
      73             :   size_t no_writers_generation_count() const;
      74             : 
      75             :   /// DISPOSE message received for this instance.
      76             :   /// Return flag indicates whether the instance state was changed.
      77             :   /// This flag is used by concrete DataReader to determine whether
      78             :   /// it should notify listener. If state is not changed, the dispose
      79             :   /// message is ignored.
      80             :   bool dispose_was_received(const GUID_t& writer_id);
      81             : 
      82             :   /// UNREGISTER message received for this instance.
      83             :   /// Return flag indicates whether the instance state was changed.
      84             :   /// This flag is used by concrete DataReader to determine whether
      85             :   /// it should notify listener. If state is not changed, the unregister
      86             :   /// message is ignored.
      87             :   bool unregister_was_received(const GUID_t& writer_id);
      88             : 
      89             :   /// Data sample received for this instance.
      90             :   void data_was_received(const GUID_t& writer_id);
      91             : 
      92             :   /// LIVELINESS message received for this DataWriter.
      93             :   void lively(const GUID_t& writer_id);
      94             : 
      95             :   /// A read or take operation has been performed on this instance.
      96             :   void accessed();
      97             : 
      98             :   bool most_recent_generation(ReceivedDataElement* item) const;
      99             : 
     100             :   /// DataReader has become empty.  Returns true if the instance was released.
     101             :   bool empty(bool value);
     102             : 
     103             :   /// Schedule a pending release of resources.
     104             :   void schedule_pending();
     105             : 
     106             :   /// Schedule an immediate release of resources.
     107             :   void schedule_release();
     108             : 
     109             :   /// Cancel a scheduled or pending release of resources.
     110             :   void cancel_release();
     111             : 
     112             :   /// Remove the instance if it's instance has no samples
     113             :   /// and no writers.
     114             :   /// Returns true if the instance was released.
     115             :   bool release_if_empty();
     116             : 
     117             :   /// Remove the instance immediately.
     118             :   void release();
     119             : 
     120             :   /// Returns true if the writer is a writer of this instance.
     121           0 :   bool writes_instance(const GUID_t& writer_id) const
     122             :   {
     123           0 :     ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, guard, lock_, false);
     124           0 :     return writers_.count(writer_id);
     125           0 :   }
     126             : 
     127             :   WeakRcHandle<DataReaderImpl> data_reader() const;
     128             :   void state_updated() const;
     129             : 
     130             :   virtual int handle_timeout(const ACE_Time_Value& current_time,
     131             :                              const void* arg);
     132             : 
     133             :   void set_owner (const GUID_t& owner);
     134             :   GUID_t get_owner ();
     135             :   bool is_exclusive () const;
     136             :   bool registered();
     137             :   void registered (bool flag);
     138             :   bool is_last (const GUID_t& pub);
     139             : 
     140             :   bool no_writer () const;
     141             : 
     142             :   void reset_ownership (DDS::InstanceHandle_t instance);
     143             : 
     144           0 :   DDS::InstanceHandle_t instance_handle() const { return handle_; }
     145             : 
     146             :   /// Return string of the name of the current instance state
     147             :   const char* instance_state_string() const;
     148             : 
     149             :   /// Return string of the name of the instance state kind passed
     150             :   static const char* instance_state_string(DDS::InstanceStateKind value);
     151             : 
     152             :   /// Return string representation of the instance state mask passed
     153             :   static OPENDDS_STRING instance_state_mask_string(DDS::InstanceStateMask mask);
     154             : 
     155             : private:
     156             :   bool reactor_is_shut_down() const;
     157             : 
     158             :   ACE_Recursive_Thread_Mutex& lock_;
     159             :   ACE_Thread_Mutex owner_lock_;
     160             : 
     161             :   /**
     162             :    * Current instance state.
     163             :    *
     164             :    * Can have values defined as:
     165             :    *
     166             :    *   DDS::ALIVE_INSTANCE_STATE
     167             :    *   DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE
     168             :    *   DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE
     169             :    *
     170             :    * and can be checked with the masks:
     171             :    *
     172             :    *   DDS::ANY_INSTANCE_STATE
     173             :    *   DDS::NOT_ALIVE_INSTANCE_STATE
     174             :    */
     175             :   DDS::InstanceStateKind instance_state_;
     176             : 
     177             :   /**
     178             :    * Current instance view state.
     179             :    *
     180             :    * Can have values defined as:
     181             :    *
     182             :    *   DDS::NEW_VIEW_STATE
     183             :    *   DDS::NOT_NEW_VIEW_STATE
     184             :    *
     185             :    * and can be checked with the mask:
     186             :    *
     187             :    *   DDS::ANY_VIEW_STATE
     188             :    */
     189             :   DDS::ViewStateKind view_state_;
     190             : 
     191             :   /// Number of times the instance state changes
     192             :   /// from NOT_ALIVE_DISPOSED to ALIVE.
     193             :   size_t disposed_generation_count_;
     194             : 
     195             :   /// Number of times the instance state changes
     196             :   /// from NOT_ALIVE_NO_WRITERS to ALIVE.
     197             :   size_t no_writers_generation_count_;
     198             : 
     199             :   /**
     200             :    * Keep track of whether the DataReader is empty or not.
     201             :    */
     202             :   bool empty_;
     203             : 
     204             :   /**
     205             :    * Keep track of whether the instance is waiting to be released.
     206             :    */
     207             :   bool release_pending_;
     208             : 
     209             :   /**
     210             :    * Keep track of a scheduled release timer.
     211             :    */
     212             :   long release_timer_id_;
     213             : 
     214             :   /**
     215             :    * Reference to our containing reader.  This is used to call back
     216             :    * and notify the reader that liveliness has been lost on this
     217             :    * instance.  It is also queried to determine if the DataReader is
     218             :    * empty -- that it contains no more sample data.
     219             :    */
     220             :   WeakRcHandle<DataReaderImpl> reader_;
     221             :   DDS::InstanceHandle_t handle_;
     222             : 
     223             :   RepoIdSet writers_;
     224             :   GUID_t owner_;
     225             :   bool exclusive_;
     226             :   /// registered with participant so it can be called back as
     227             :   /// the owner is updated.
     228             :   bool registered_;
     229             : 
     230             :   struct CommandBase : Command {
     231           0 :     explicit CommandBase(InstanceState* instance_state)
     232           0 :       : instance_state_(instance_state)
     233           0 :     {}
     234             : 
     235             :     InstanceState* instance_state_;
     236             :   };
     237             : 
     238             :   struct CancelCommand : CommandBase {
     239           0 :     explicit CancelCommand(InstanceState* instance_state)
     240           0 :       : CommandBase(instance_state)
     241           0 :     {}
     242             : 
     243             :     void execute();
     244             :   };
     245             : 
     246             :   struct ScheduleCommand : CommandBase {
     247           0 :     ScheduleCommand(InstanceState* instance_state, const TimeDuration& delay)
     248           0 :       : CommandBase(instance_state)
     249           0 :       , delay_(delay)
     250           0 :     {}
     251             : 
     252             :     const TimeDuration delay_;
     253             :     void execute();
     254             :   };
     255             : 
     256             : };
     257             : 
     258             : } // namespace DCPS
     259             : } // namespace OpenDDS
     260             : 
     261             : OPENDDS_END_VERSIONED_NAMESPACE_DECL
     262             : 
     263             : #if defined (__ACE_INLINE__)
     264             : # include "InstanceState.inl"
     265             : #endif  /* __ACE_INLINE__ */
     266             : 
     267             : #endif /* OPENDDS_DCPS_INSTANCESTATE_H */