[15118] Adding DataReader::get_unread_count (bool mark_as_read) (backport #2825)

include/fastdds/dds/subscriber/DataReader.hpp

@@ -809,6 +809,16 @@ class DataReader : public DomainEntity
      */
     RTPS_DllAPI uint64_t get_unread_count() const;
 
+    /**
+     * Get the number of samples pending to be read.
+     *
+     * @param mark_as_read  Whether the unread samples should be marked as read or not.
+     *
+     * @return the number of samples on the reader history that have never been read.
+     */
+    RTPS_DllAPI uint64_t get_unread_count(
+            bool mark_as_read) const;
+
     /**
      * Get associated GUID.
      *

include/fastdds/rtps/reader/RTPSReader.h

@@ -238,6 +238,9 @@ class RTPSReader
 
     RTPS_DllAPI uint64_t get_unread_count() const;
 
+    RTPS_DllAPI uint64_t get_unread_count(
+            bool mark_as_read);
+
     /**
      * @return True if the reader expects Inline QOS.
      */

src/cpp/fastdds/subscriber/DataReader.cpp

@@ -252,7 +252,13 @@ ReturnCode_t DataReader::get_first_untaken_info(
 
 uint64_t DataReader::get_unread_count() const
 {
-    return impl_->get_unread_count();
+    return impl_->get_unread_count(false);
+}
+
+uint64_t DataReader::get_unread_count(
+        bool mark_as_read) const
+{
+    return impl_->get_unread_count(mark_as_read);
 }
 
 const GUID_t& DataReader::guid()

src/cpp/fastdds/subscriber/DataReaderImpl.cpp

@@ -710,9 +710,10 @@ ReturnCode_t DataReaderImpl::get_first_untaken_info(
     return ReturnCode_t::RETCODE_NO_DATA;
 }
 
-uint64_t DataReaderImpl::get_unread_count() const
+uint64_t DataReaderImpl::get_unread_count(
+        bool mark_as_read)
 {
-    return reader_ ? reader_->get_unread_count() : 0;
+    return reader_ ? history_.get_unread_count(mark_as_read) : 0;
 }
 
 const GUID_t& DataReaderImpl::guid() const

src/cpp/fastdds/subscriber/DataReaderImpl.hpp

@@ -196,9 +196,14 @@ class DataReaderImpl
             SampleInfo* info);
 
     /**
-     * @return the number of samples pending to be read.
+     * Get the number of samples pending to be read.
+     *
+     * @param mark_as_read  Whether the unread samples should be marked as read or not.
+     *
+     * @return the number of samples on the reader history that have never been read.
      */
-    uint64_t get_unread_count() const;
+    uint64_t get_unread_count(
+            bool mark_as_read);
 
     /**
      * Get associated GUID

src/cpp/fastdds/subscriber/history/DataReaderHistory.cpp

@@ -518,6 +518,12 @@ bool DataReaderHistory::get_next_deadline(
     return true;
 }
 
+uint64_t DataReaderHistory::get_unread_count(
+        bool mark_as_read)
+{
+    return mp_reader->get_unread_count(mark_as_read);
+}
+
 std::pair<bool, DataReaderHistory::instance_info> DataReaderHistory::lookup_instance(
         const InstanceHandle_t& handle,
         bool exact) const

src/cpp/fastdds/subscriber/history/DataReaderHistory.hpp

@@ -187,6 +187,16 @@ class DataReaderHistory : public eprosima::fastrtps::rtps::ReaderHistory
             InstanceHandle_t& handle,
             std::chrono::steady_clock::time_point& next_deadline_us);
 
+    /**
+     * Get the number of samples pending to be read.
+     *
+     * @param mark_as_read  Whether the unread samples should be marked as read or not.
+     *
+     * @return the number of samples on the reader history that have never been read.
+     */
+    uint64_t get_unread_count(
+            bool mark_as_read);
+
     /**
      * @brief Get the list of changes corresponding to an instance handle.
      * @param handle The handle to the instance.

src/cpp/rtps/reader/RTPSReader.cpp

@@ -367,6 +367,28 @@ uint64_t RTPSReader::get_unread_count() const
     return total_unread_;
 }
 
+uint64_t RTPSReader::get_unread_count(
+        bool mark_as_read)
+{
+    std::unique_lock<RecursiveTimedMutex> lock(mp_mutex);
+    uint64_t ret_val = total_unread_;
+
+    if (mark_as_read)
+    {
+        for (auto it = mp_history->changesBegin(); 0 < total_unread_ && it != mp_history->changesEnd(); ++it)
+        {
+            CacheChange_t* change = *it;
+            if (!change->isRead)
+            {
+                change->isRead = true;
+                --total_unread_;
+            }
+        }
+        assert(0 == total_unread_);
+    }
+    return ret_val;
+}
+
 bool RTPSReader::is_datasharing_compatible_with(
         const WriterProxyData& wdata)
 {

test/mock/rtps/RTPSReader/fastdds/rtps/reader/RTPSReader.h

@@ -115,6 +115,8 @@ class RTPSReader : public Endpoint
 
     MOCK_METHOD0(get_unread_count, uint64_t());
 
+    MOCK_METHOD1(get_unread_count, uint64_t(bool));
+
     MOCK_METHOD1(set_content_filter, void (eprosima::fastdds::rtps::IReaderDataFilter* filter));
 
     // *INDENT-ON*

test/unittest/dds/subscriber/DataReaderTests.cpp

@@ -1550,6 +1550,80 @@ TEST_F(DataReaderTests, read_unread)
     }
 }
 
+/*
+ * This test checks the behaviour of the two overloads of get_unread_count.
+ */
+TEST_F(DataReaderTests, get_unread_count)
+{
+    static const Duration_t time_to_wait(0, 100 * 1000 * 1000);
+    static constexpr int32_t num_samples = 10;
+    static constexpr uint64_t num_samples_check = static_cast<uint64_t>(num_samples);
+
+    const ReturnCode_t& ok_code = ReturnCode_t::RETCODE_OK;
+
+    DataWriterQos writer_qos = DATAWRITER_QOS_DEFAULT;
+    writer_qos.history().kind = KEEP_LAST_HISTORY_QOS;
+    writer_qos.history().depth = num_samples;
+    writer_qos.publish_mode().kind = SYNCHRONOUS_PUBLISH_MODE;
+    writer_qos.reliability().kind = RELIABLE_RELIABILITY_QOS;
+
+    DataReaderQos reader_qos = DATAREADER_QOS_DEFAULT;
+    reader_qos.reliability().kind = RELIABLE_RELIABILITY_QOS;
+    reader_qos.history().kind = KEEP_ALL_HISTORY_QOS;
+    reader_qos.resource_limits().max_instances = 1;
+    reader_qos.resource_limits().max_samples_per_instance = num_samples;
+    reader_qos.resource_limits().max_samples = 3 * num_samples;
+
+    create_instance_handles();
+    create_entities(nullptr, reader_qos, SUBSCRIBER_QOS_DEFAULT, writer_qos);
+
+    FooType data;
+    data.index(0);
+    data.message()[1] = '\0';
+
+    // Send a bunch of samples
+    for (char i = 0; i < num_samples; ++i)
+    {
+        data.message()[0] = i + '0';
+        EXPECT_EQ(ok_code, data_writer_->write(&data, handle_ok_));
+    }
+
+    // Reader should have 10 unread samples
+
+    // There are unread samples, so wait_for_unread should be ok
+    EXPECT_TRUE(data_reader_->wait_for_unread_message(time_to_wait));
+
+    // Calling get_unread_count() several times should always return the same value
+    for (char i = 0; i < num_samples; ++i)
+    {
+        EXPECT_EQ(num_samples_check, data_reader_->get_unread_count());
+    }
+
+    SampleInfo sample_info;
+    ASSERT_EQ(ReturnCode_t::RETCODE_OK, data_reader_->get_first_untaken_info(&sample_info));
+    ASSERT_EQ(SampleStateKind::NOT_READ_SAMPLE_STATE, sample_info.sample_state);
+
+    // Calling get_unread_count(false) several times should always return the same value
+    for (char i = 0; i < num_samples; ++i)
+    {
+        EXPECT_EQ(num_samples_check, data_reader_->get_unread_count(false));
+    }
+
+    ASSERT_EQ(ReturnCode_t::RETCODE_OK, data_reader_->get_first_untaken_info(&sample_info));
+    ASSERT_EQ(SampleStateKind::NOT_READ_SAMPLE_STATE, sample_info.sample_state);
+
+    // Calling get_unread_count(true) once will return the correct value
+    EXPECT_EQ(num_samples_check, data_reader_->get_unread_count(true));
+
+    ASSERT_EQ(ReturnCode_t::RETCODE_OK, data_reader_->get_first_untaken_info(&sample_info));
+    ASSERT_EQ(SampleStateKind::READ_SAMPLE_STATE, sample_info.sample_state);
+
+    // All variants should then return 0
+    EXPECT_EQ(0, data_reader_->get_unread_count(true));
+    EXPECT_EQ(0, data_reader_->get_unread_count(false));
+    EXPECT_EQ(0, data_reader_->get_unread_count());
+}
+
 template<typename DataType>
 void lookup_instance_test(
         DataType& data,