[20276] Discard already processed samples on PDPListener

include/fastdds/rtps/builtin/data/ParticipantProxyData.h

@@ -120,6 +120,8 @@ class ParticipantProxyData
     //!
     ProxyHashTable<WriterProxyData>* m_writers = nullptr;
 
+    SampleIdentity m_sample_identity;
+
     /**
      * Update the data.
      * @param pdata Object to copy the data from

src/cpp/rtps/builtin/data/ParticipantProxyData.cpp

@@ -65,6 +65,7 @@ ParticipantProxyData::ParticipantProxyData(
     , should_check_lease_duration(false)
     , m_readers(new ProxyHashTable<ReaderProxyData>(allocation.readers))
     , m_writers(new ProxyHashTable<WriterProxyData>(allocation.writers))
+    , m_sample_identity()
 {
     m_userData.set_max_size(static_cast<uint32_t>(allocation.data_limits.max_user_data));
 }
@@ -99,6 +100,7 @@ ParticipantProxyData::ParticipantProxyData(
     // so there is no need to copy m_readers and m_writers
     , m_readers(nullptr)
     , m_writers(nullptr)
+    , m_sample_identity(pdata.m_sample_identity)
     , lease_duration_(pdata.lease_duration_)
 {
 }
@@ -758,6 +760,7 @@ void ParticipantProxyData::copy(
     isAlive = pdata.isAlive;
     m_userData = pdata.m_userData;
     m_properties = pdata.m_properties;
+    m_sample_identity = pdata.m_sample_identity;
 
     // This method is only called when a new participant is discovered.The destination of the copy
     // will always be a new ParticipantProxyData or one from the pool, so there is no need for

src/cpp/rtps/builtin/discovery/participant/PDPListener.cpp

@@ -126,16 +126,33 @@ void PDPListener::onNewCacheChangeAdded(
 
             // Check if participant already exists (updated info)
             ParticipantProxyData* pdata = nullptr;
+            bool already_processed = false;
             for (ParticipantProxyData* it : parent_pdp_->participant_proxies_)
             {
                 if (guid == it->m_guid)
                 {
                     pdata = it;
+
+                    // This means this is the same DATA(p) that we have already processed.
+                    // We do not compare sample_identity directly because it is not properly filled
+                    // in the change during desearialization.
+                    if (it->m_sample_identity.writer_guid() == change->writerGUID &&
+                            it->m_sample_identity.sequence_number() == change->sequenceNumber)
+                    {
+                        already_processed = true;
+                    }
+
                     break;
                 }
             }
 
-            process_alive_data(pdata, temp_participant_data_, writer_guid, reader, lock);
+            // Only notified the DATA(p) if it is not a repeated one
+            if (!already_processed)
+            {
+                temp_participant_data_.m_sample_identity.writer_guid(change->writerGUID);
+                temp_participant_data_.m_sample_identity.sequence_number(change->sequenceNumber);
+                process_alive_data(pdata, temp_participant_data_, writer_guid, reader, lock);
+            }
         }
     }
     else if (reader->matched_writer_is_matched(writer_guid))