Fix heartbeat and ACKs issues <2.0.x> [11006]

include/fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h

@@ -30,17 +30,17 @@ class TransportInterface;
  */
 typedef struct SharedMemTransportDescriptor : public TransportDescriptorInterface
 {
-    virtual ~SharedMemTransportDescriptor() 
+    virtual ~SharedMemTransportDescriptor()
     {
-        
+
     }
 
     virtual TransportInterface* create_transport() const override;
     uint32_t min_send_buffer_size() const override
     {
-        return 0;
+        return segment_size_;
     }
-    
+
     RTPS_DllAPI SharedMemTransportDescriptor();
 
     RTPS_DllAPI SharedMemTransportDescriptor(
@@ -50,16 +50,16 @@ typedef struct SharedMemTransportDescriptor : public TransportDescriptorInterfac
     {
         return segment_size_;
     }
-    
+
     RTPS_DllAPI void segment_size(
             uint32_t segment_size)
     {
         segment_size_ = segment_size;
     }
 
     virtual uint32_t max_message_size() const override
-    { 
-        return maxMessageSize; 
+    {
+        return maxMessageSize;
     }
 
     RTPS_DllAPI void max_message_size(

src/cpp/rtps/network/ReceiverResource.cpp

@@ -17,25 +17,25 @@
 #include <cassert>
 #include <fastdds/dds/log/Log.hpp>
 
-#define IDSTRING "(ID:" << std::this_thread::get_id() <<") "<<
+#define IDSTRING "(ID:" << std::this_thread::get_id() << ") " <<
 
 using namespace std;
 using namespace eprosima::fastdds::rtps;
 
-namespace eprosima{
-namespace fastrtps{
-namespace rtps{
+namespace eprosima {
+namespace fastrtps {
+namespace rtps {
 
 ReceiverResource::ReceiverResource(
-			TransportInterface& transport,
-			const Locator_t& locator,
-			uint32_t max_recv_buffer_size)
-        : Cleanup(nullptr)
-        , LocatorMapsToManagedChannel(nullptr)
-        , mValid(false)
-        , mtx()
-        , receiver(nullptr)
-        , max_message_size_(max_recv_buffer_size)
+        TransportInterface& transport,
+        const Locator_t& locator,
+        uint32_t max_recv_buffer_size)
+    : Cleanup(nullptr)
+    , LocatorMapsToManagedChannel(nullptr)
+    , mValid(false)
+    , mtx()
+    , receiver(nullptr)
+    , max_message_size_(max_recv_buffer_size)
 {
     // Internal channel is opened and assigned to this resource.
     mValid = transport.OpenInputChannel(locator, this, max_message_size_);
@@ -45,12 +45,18 @@ ReceiverResource::ReceiverResource(
     }
 
     // Implementation functions are bound to the right transport parameters
-    Cleanup = [&transport, locator]() { transport.CloseInputChannel(locator); };
+    Cleanup = [&transport, locator]()
+            {
+                transport.CloseInputChannel(locator);
+            };
     LocatorMapsToManagedChannel = [&transport, locator](const Locator_t& locatorToCheck) -> bool
-    { return transport.DoInputLocatorsMatch(locator, locatorToCheck); };
+            {
+                return locator.kind == locatorToCheck.kind && transport.DoInputLocatorsMatch(locator, locatorToCheck);
+            };
 }
 
-ReceiverResource::ReceiverResource(ReceiverResource&& rValueResource)
+ReceiverResource::ReceiverResource(
+        ReceiverResource&& rValueResource)
 {
     Cleanup.swap(rValueResource.Cleanup);
     LocatorMapsToManagedChannel.swap(rValueResource.LocatorMapsToManagedChannel);
@@ -61,7 +67,8 @@ ReceiverResource::ReceiverResource(ReceiverResource&& rValueResource)
     max_message_size_ = rValueResource.max_message_size_;
 }
 
-bool ReceiverResource::SupportsLocator(const Locator_t& localLocator)
+bool ReceiverResource::SupportsLocator(
+        const Locator_t& localLocator)
 {
     if (LocatorMapsToManagedChannel)
     {
@@ -70,22 +77,31 @@ bool ReceiverResource::SupportsLocator(const Locator_t& localLocator)
     return false;
 }
 
-void ReceiverResource::RegisterReceiver(MessageReceiver* rcv)
+void ReceiverResource::RegisterReceiver(
+        MessageReceiver* rcv)
 {
     std::unique_lock<std::mutex> lock(mtx);
     if (receiver == nullptr)
+    {
         receiver = rcv;
+    }
 }
 
-void ReceiverResource::UnregisterReceiver(MessageReceiver* rcv)
+void ReceiverResource::UnregisterReceiver(
+        MessageReceiver* rcv)
 {
     std::unique_lock<std::mutex> lock(mtx);
     if (receiver == rcv)
+    {
         receiver = nullptr;
+    }
 }
 
-void ReceiverResource::OnDataReceived(const octet * data, const uint32_t size,
-    const Locator_t & localLocator, const Locator_t & remoteLocator)
+void ReceiverResource::OnDataReceived(
+        const octet* data,
+        const uint32_t size,
+        const Locator_t& localLocator,
+        const Locator_t& remoteLocator)
 {
     (void)localLocator;
 

src/cpp/rtps/participant/RTPSParticipantImpl.cpp

@@ -952,7 +952,6 @@ bool RTPSParticipantImpl::assignEndpoint2LocatorList(
        one of the supported Locators is needed to make the match, and the case of new ListenResources being created has been removed
        since its the NetworkFactory the one that takes care of Resource creation.
      */
-    LocatorList_t finalList;
     for (auto lit = list.begin(); lit != list.end(); ++lit)
     {
         //Iteration of all Locators within the Locator list passed down as argument

src/cpp/rtps/reader/StatefulReader.cpp

@@ -141,6 +141,12 @@ bool StatefulReader::matched_writer_add(
         matched_writers_pool_.pop_back();
     }
 
+    SequenceNumber_t initial_sequence;
+    add_persistence_guid(wdata.guid(), wdata.persistence_guid());
+    initial_sequence = get_last_notified(wdata.guid());
+
+    wp->start(wdata, initial_sequence);
+
     if (!is_same_process)
     {
         for (const Locator_t& locator : wp->remote_locators_shrinked())
@@ -149,12 +155,6 @@ bool StatefulReader::matched_writer_add(
         }
     }
 
-    SequenceNumber_t initial_sequence;
-    add_persistence_guid(wdata.guid(), wdata.persistence_guid());
-    initial_sequence = get_last_notified(wdata.guid());
-
-    wp->start(wdata, initial_sequence);
-
     matched_writers_.push_back(wp);
 
     if (liveliness_lease_duration_ < c_TimeInfinite)

test/blackbox/BlackboxTestsPubSubBasic.cpp

@@ -253,6 +253,7 @@ TEST_P(PubSubBasic, PubSubMoreThan256Unacknowledged)
 
     reader.startReception(expected_data);
     reader.block_for_all();
+    EXPECT_TRUE(writer.waitForAllAcked(std::chrono::seconds(10)));
 }
 
 TEST_P(PubSubBasic, PubSubAsReliableHelloworldMulticastDisabled)
@@ -286,6 +287,7 @@ TEST_P(PubSubBasic, PubSubAsReliableHelloworldMulticastDisabled)
     ASSERT_TRUE(data.empty());
     // Block reader until reception finished or timeout.
     reader.block_for_all();
+    EXPECT_TRUE(writer.waitForAllAcked(std::chrono::seconds(10)));
 }
 
 TEST_P(PubSubBasic, ReceivedDynamicDataWithNoSizeLimit)