Send heartbeat when wait_for_all_acked is called (#6098) (#6103)

* Send heartbeat when wait_for_all_acked is called (#6098)

* Refs #23798. Add test.

Signed-off-by: Miguel Company <[email protected]>

* Refs #23798. Send heartbeat when wait_for_all_acked is called.

Signed-off-by: Miguel Company <[email protected]>

---------

Signed-off-by: Miguel Company <[email protected]>
(cherry picked from commit db8f527)

* Uncrustify

Signed-off-by: Juan Lopez Fernandez <[email protected]>

---------

Signed-off-by: Miguel Company <[email protected]>
Signed-off-by: Juan Lopez Fernandez <[email protected]>
Co-authored-by: Miguel Company <[email protected]>
Co-authored-by: Juan Lopez Fernandez <[email protected]>

Author: 3 people

src/cpp/rtps/writer/StatefulWriter.cpp

@@ -1387,6 +1387,8 @@ bool StatefulWriter::all_readers_updated()
 bool StatefulWriter::wait_for_all_acked(
         const dds::Duration_t& max_wait)
 {
+    send_periodic_heartbeat();
+
     std::unique_lock<RecursiveTimedMutex> lock(mp_mutex);
     std::unique_lock<std::mutex> all_acked_lock(all_acked_mutex_);
 
@@ -1508,7 +1510,8 @@ void StatefulWriter::check_acked_status()
                     listener_->on_writer_change_received_by_all(this, change);
 
                     // Stop if we got to either next_all_acked_notify_sequence_ or the first change
-                } while (seq > end_seq);
+                }
+                while (seq > end_seq);
             }
 
             next_all_acked_notify_sequence_ = min_low_mark + 1;
@@ -2079,7 +2082,8 @@ bool StatefulWriter::ack_timer_expired()
         do
         {
             last_sequence_number_++;
-        } while (!history_->get_change(
+        }
+        while (!history_->get_change(
             last_sequence_number_,
             getGuid(),
             &change) && last_sequence_number_ < next_sequence_number());

test/blackbox/common/BlackboxTestsReliability.cpp

@@ -100,3 +100,45 @@ TEST(Reliability, DisableHeartbeatPiggybackTrue)
 {
     reliability_disable_heartbeat_piggyback(true);
 }
+
+/**
+ * This test checks that waitForAllAcked() triggers a heartbeat if needed.
+ */
+TEST(Reliability, wait_for_all_acked_triggers_heartbeat)
+{
+    PubSubReader<HelloWorldPubSubType> reader(TEST_TOPIC_NAME);
+    PubSubWriter<HelloWorldPubSubType> writer(TEST_TOPIC_NAME);
+
+    // Either ensure that the history will not become full, or disable piggyback heartbeats
+    static constexpr int32_t HISTORY_DEPTH = 10;
+    static constexpr int32_t NUM_SAMPLES = 1;
+    static_assert(NUM_SAMPLES < HISTORY_DEPTH, "NUM_SAMPLES must be less than HISTORY_DEPTH");
+
+    reader.reliability(eprosima::fastdds::dds::RELIABLE_RELIABILITY_QOS)
+            .history_kind(eprosima::fastdds::dds::KEEP_LAST_HISTORY_QOS)
+            .history_depth(HISTORY_DEPTH)
+            .init();
+
+    writer.reliability(eprosima::fastdds::dds::RELIABLE_RELIABILITY_QOS)
+            .history_kind(eprosima::fastdds::dds::KEEP_LAST_HISTORY_QOS)
+            .history_depth(HISTORY_DEPTH)
+            .heartbeat_period_seconds(180000) // Large heartbeat period to avoid interference
+            .init();
+
+    ASSERT_TRUE(reader.isInitialized());
+    ASSERT_TRUE(writer.isInitialized());
+
+    // Wait for discovery.
+    writer.wait_discovery();
+    reader.wait_discovery();
+
+    // Wait for the initial heartbeat/acknack exchange to finish
+    std::this_thread::sleep_for(std::chrono::seconds(1));
+
+    auto data = default_helloworld_data_generator(NUM_SAMPLES);
+    writer.send(data);
+    ASSERT_TRUE(data.empty());
+
+    // Wait until all data is acknowledged. This should trigger a heartbeat.
+    ASSERT_TRUE(writer.waitForAllAcked(std::chrono::seconds(1)));
+}