Merge pull request #4085 from jrw972/internal-dds-improvements

Internal DDS improvements

Author: jrw972

dds/DCPS/InternalDataReader.h

@@ -22,14 +22,34 @@ OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL
 namespace OpenDDS {
 namespace DCPS {
 
+/*
+  The InternalDataWriter supports the following QoS:
+
+  DurabilityQosPolicy durability; => VOLATILE, TRANSIENT_LOCAL
+  DurabilityServiceQosPolicy durability_service; => None
+  DeadlineQosPolicy deadline; => None
+  LatencyBudgetQosPolicy latency_budget => None
+  LivelinessQosPolicy liveliness => None
+  ReliabilityQosPolicy reliability => RELIABLE
+  DestinationOrderQosPolicy destination_order; => None
+  HistoryQosPolicy history; => KEEP_LAST_HISTORY, KEEP_ALL_HISTORY
+  ResourceLimitsQosPolicy resource_limits; => None
+  TransportPriorityQosPolicy transport_priority; => None
+  LifespanQosPolicy lifespan; => None
+  UserDataQosPolicy user_data; => None
+  OwnershipQosPolicy ownership; => None
+  OwnershipStrengthQosPolicy ownership_strength; => None
+  WriterDataLifecycleQosPolicy writer_data_lifecycle; => Yes
+*/
+
 template <typename T>
 class InternalDataWriter : public InternalEntity {
 public:
   typedef RcHandle<InternalDataReader<T> > InternalDataReader_rch;
   typedef WeakRcHandle<InternalDataReader<T> > InternalDataReader_wrch;
 
-  explicit InternalDataWriter(bool durable)
-    : durable_(durable)
+  explicit InternalDataWriter(const DDS::DataWriterQos& qos)
+    : qos_(qos)
   {}
 
   /// @name InternalTopic Interface
@@ -39,14 +59,10 @@ class InternalDataWriter : public InternalEntity {
     ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
     readers_.insert(reader);
 
-    if (durable_ && reader->durable()) {
-      for (typename InstanceMap::const_iterator pos = instance_map_.begin(), limit = instance_map_.end();
+    if (qos_.durability.kind == DDS::TRANSIENT_LOCAL_DURABILITY_QOS && reader->durable()) {
+      for (typename InstanceMap::iterator pos = instance_map_.begin(), limit = instance_map_.end();
            pos != limit; ++pos) {
-        if (pos->second.valid_data) {
-          reader->write(static_rchandle_cast<InternalEntity>(rchandle_from(this)), pos->second.sample);
-        } else {
-          reader->register_instance(static_rchandle_cast<InternalEntity>(rchandle_from(this)), pos->first);
-        }
+        pos->second.add_reader(reader, static_rchandle_cast<InternalEntity>(rchandle_from(this)));
       }
     }
   }
@@ -55,7 +71,7 @@ class InternalDataWriter : public InternalEntity {
   {
     ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
     if (readers_.erase(reader)) {
-      reader->remove_publication(static_rchandle_cast<InternalEntity>(rchandle_from(this)));
+      reader->remove_publication(static_rchandle_cast<InternalEntity>(rchandle_from(this)), qos_.writer_data_lifecycle.autodispose_unregistered_instances);
     }
   }
 
@@ -73,30 +89,13 @@ class InternalDataWriter : public InternalEntity {
 
   /// @name User Interface
   /// @{
-  void register_instance(const T& sample)
-  {
-    ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
-
-    if (durable_) {
-      instance_map_.insert(std::make_pair(sample, SampleHolder()));
-    }
-
-    for (typename ReaderSet::const_iterator pos = readers_.begin(), limit = readers_.end(); pos != limit; ++pos) {
-      InternalDataReader_rch reader = pos->lock();
-      if (reader) {
-        reader->register_instance(static_rchandle_cast<InternalEntity>(rchandle_from(this)), sample);
-      }
-    }
-  }
-
   void write(const T& sample)
   {
     ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
-    if (durable_) {
+    if (qos_.durability.kind == DDS::TRANSIENT_LOCAL_DURABILITY_QOS) {
       const std::pair<typename InstanceMap::iterator, bool> p = instance_map_.insert(std::make_pair(sample, SampleHolder()));
-      p.first->second.sample = sample;
-      p.first->second.valid_data = true;
+      p.first->second.write(sample, qos_);
     }
 
     for (typename ReaderSet::const_iterator pos = readers_.begin(), limit = readers_.end(); pos != limit; ++pos) {
@@ -107,52 +106,81 @@ class InternalDataWriter : public InternalEntity {
     }
   }
 
-  void unregister_instance(const T& sample)
+  void dispose(const T& sample)
   {
     ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
-    if (durable_) {
-      instance_map_.erase(sample);
+    if (qos_.durability.kind == DDS::TRANSIENT_LOCAL_DURABILITY_QOS) {
+      typename InstanceMap::iterator pos = instance_map_.find(sample);
+      if (pos != instance_map_.end()) {
+        pos->second.dispose();
+      }
     }
 
     for (typename ReaderSet::const_iterator pos = readers_.begin(), limit = readers_.end(); pos != limit; ++pos) {
       InternalDataReader_rch reader = pos->lock();
       if (reader) {
-        reader->unregister_instance(static_rchandle_cast<InternalEntity>(rchandle_from(this)), sample);
+        reader->dispose(static_rchandle_cast<InternalEntity>(rchandle_from(this)), sample);
       }
     }
   }
 
-  void dispose(const T& sample)
+  void unregister_instance(const T& sample)
   {
     ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
-    if (durable_) {
+    if (qos_.durability.kind == DDS::TRANSIENT_LOCAL_DURABILITY_QOS) {
       instance_map_.erase(sample);
     }
 
     for (typename ReaderSet::const_iterator pos = readers_.begin(), limit = readers_.end(); pos != limit; ++pos) {
       InternalDataReader_rch reader = pos->lock();
       if (reader) {
-        reader->dispose(static_rchandle_cast<InternalEntity>(rchandle_from(this)), sample);
+        if (qos_.writer_data_lifecycle.autodispose_unregistered_instances) {
+          reader->dispose(static_rchandle_cast<InternalEntity>(rchandle_from(this)), sample);
+        }
+        reader->unregister_instance(static_rchandle_cast<InternalEntity>(rchandle_from(this)), sample);
       }
     }
   }
   /// @}
 
 private:
-  const bool durable_;
+  const DDS::DataWriterQos qos_;
 
   typedef OPENDDS_SET(InternalDataReader_wrch) ReaderSet;
   ReaderSet readers_;
 
-  struct SampleHolder {
-    T sample;
-    bool valid_data;
+  class SampleHolder {
+  public:
+    bool empty() const { return samples_.empty(); }
+
+    void add_reader(InternalDataReader_rch reader, RcHandle<InternalEntity> writer)
+    {
+      for (typename SampleList::const_iterator pos = samples_.begin(), limit = samples_.end(); pos != limit; ++pos) {
+        reader->write(writer, *pos);
+      }
+    }
+
+    void write(const T& sample,
+               const DDS::DataWriterQos& qos)
+    {
+      samples_.push_back(sample);
+      if (qos.history.kind == DDS::KEEP_LAST_HISTORY_QOS) {
+        while (samples_.size() > static_cast<std::size_t>(qos.history.depth)) {
+          samples_.pop_front();
+        }
+      }
+    }
+
+    void dispose()
+    {
+      samples_.clear();
+    }
 
-    SampleHolder()
-      : valid_data(false)
-    {}
+  private:
+    typedef OPENDDS_LIST(T) SampleList;
+    SampleList samples_;
   };
 
   typedef OPENDDS_MAP_T(T, SampleHolder) InstanceMap;