Line data    Source code

       1             : /*
       2             :  * Distributed under the OpenDDS License.
       3             :  * See: http://www.opendds.org/license.html
       4             :  */
       5             : 
       6             : #include "DCPS/DdsDcps_pch.h" //Only the _pch include should start with DCPS/
       7             : 
       8             : #include "SubscriptionInstance.h"
       9             : 
      10             : #include "DataReaderImpl.h"
      11             : 
      12             : OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL
      13             : 
      14             : namespace OpenDDS {
      15             : namespace DCPS {
      16             : 
      17           0 : SubscriptionInstance::SubscriptionInstance(const DataReaderImpl_rch& reader,
      18             :                                            const DDS::DataReaderQos& qos,
      19             :                                            ACE_Recursive_Thread_Mutex& lock,
      20             :                                            DDS::InstanceHandle_t handle,
      21           0 :                                            bool owns_handle)
      22           0 :   : instance_state_(make_rch<InstanceState>(ref(reader), ref(lock), handle))
      23           0 :   , rcvd_samples_(reader, instance_state_)
      24           0 :   , read_sample_count_(0)
      25           0 :   , not_read_sample_count_(0)
      26           0 :   , sample_states_(0)
      27           0 :   , instance_handle_(handle)
      28           0 :   , owns_handle_(owns_handle)
      29             : {
      30           0 :   switch (qos.destination_order.kind) {
      31           0 :   case DDS::BY_RECEPTION_TIMESTAMP_DESTINATIONORDER_QOS:
      32           0 :     rcvd_strategy_.reset(new ReceptionDataStrategy(rcvd_samples_));
      33           0 :     break;
      34             : 
      35           0 :   case DDS::BY_SOURCE_TIMESTAMP_DESTINATIONORDER_QOS:
      36           0 :     rcvd_strategy_.reset(new SourceDataStrategy(rcvd_samples_));
      37           0 :     break;
      38             :   }
      39             : 
      40           0 :   if (!rcvd_strategy_) {
      41           0 :     ACE_ERROR((LM_ERROR,
      42             :                 ACE_TEXT("(%P|%t) ERROR: SubscriptionInstance: ")
      43             :                 ACE_TEXT("unable to allocate ReceiveDataStrategy!\n")));
      44             :   }
      45           0 : }
      46             : 
      47           0 : SubscriptionInstance::~SubscriptionInstance()
      48             : {
      49           0 :   if (owns_handle_) {
      50           0 :     const RcHandle<DataReaderImpl> reader = instance_state_->data_reader().lock();
      51           0 :     if (reader) {
      52           0 :       reader->return_handle(instance_handle_);
      53             :     }
      54           0 :   }
      55           0 : }
      56             : 
      57           0 : bool SubscriptionInstance::matches(CORBA::ULong sample_states, CORBA::ULong view_states, CORBA::ULong instance_states) const
      58             : {
      59           0 :   return instance_state_->match(view_states, instance_states) && rcvd_samples_.matches(sample_states);
      60             : }
      61             : 
      62             : }
      63             : }
      64             : OPENDDS_END_VERSIONED_NAMESPACE_DECL