Fix heartbeat and ACKs issues [11003]

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

@@ -47,7 +47,7 @@ struct SharedMemTransportDescriptor : public TransportDescriptorInterface
     //! Minimum size of the send buffer
     uint32_t min_send_buffer_size() const override
     {
-        return 0;
+        return segment_size_;
     }
 
     //! Constructor

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

@@ -1196,7 +1196,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

@@ -219,6 +219,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())
@@ -227,12 +233,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);
-
     if (is_datasharing)
     {
         if (datasharing_listener_->add_datasharing_writer(wdata.guid(),

test/blackbox/common/BlackboxTestsPubSubBasic.cpp

@@ -276,6 +276,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)
@@ -309,6 +310,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)