[14770] SampleRejectedStatus feature

include/fastdds/rtps/common/CacheChange.h

@@ -205,6 +205,14 @@ struct RTPS_DllAPI CacheChange_t
         return first_missing_fragment_ >= fragment_count_;
     }
 
+    /*! Checks if the first fragment is present.
+     * @return true when it contains the first fragment. In other case, false.
+     */
+    bool contains_first_fragment()
+    {
+        return 0 < first_missing_fragment_;
+    }
+
     /*!
      * Fills a FragmentNumberSet_t with the list of missing fragments.
      * @param [out] frag_sns FragmentNumberSet_t where result is stored.

include/fastdds/rtps/history/ReaderHistory.h

@@ -22,6 +22,7 @@
 
 #include <fastdds/rtps/history/History.h>
 #include <fastdds/rtps/common/CacheChange.h>
+#include <fastdds/dds/core/status/SampleRejectedStatus.hpp>
 
 namespace eprosima {
 namespace fastrtps {
@@ -85,6 +86,25 @@ class ReaderHistory : public History
             CacheChange_t* change,
             size_t unknown_missing_changes_up_to);
 
+    /**
+     * Virtual method that is called when a new change is received.
+     * In this implementation this method just calls add_change. The user can overload this method in case
+     * he needs to perform additional checks before adding the change.
+     * @param[in] change Pointer to the change
+     * @param[in] unknown_missing_changes_up_to The number of changes from the same writer with a lower sequence number that
+     *                                      could potentially be received in the future.
+     * @param[out] rejection_reason In case of been rejected the sample, it will contain the reason of the rejection.
+     * @return True if added.
+     */
+    RTPS_DllAPI virtual bool received_change(
+            CacheChange_t* change,
+            size_t unknown_missing_changes_up_to,
+            fastdds::dds::SampleRejectedStatusKind& rejection_reason)
+    {
+        rejection_reason = fastdds::dds::NOT_REJECTED;
+        return received_change(change, unknown_missing_changes_up_to);
+    }
+
     /**
      * Called when a fragmented change is received completely by the Subscriber. Will find its instance and store it.
      * @pre Change should be already present in the history.
@@ -98,6 +118,25 @@ class ReaderHistory : public History
         return true;
     }
 
+    /**
+     * Called when a fragmented change is received completely by the Subscriber. Will find its instance and store it.
+     * @pre Change should be already present in the history.
+     * @param[in] change The received change
+     * @param[in] unknown_missing_changes_up_to Number of missing changes before this one
+     * @param[out] rejection_reason In case of been rejected the sample, it will contain the reason of the rejection.
+     * @return
+     */
+    RTPS_DllAPI virtual bool completed_change(
+            CacheChange_t* change,
+            size_t unknown_missing_changes_up_to,
+            fastdds::dds::SampleRejectedStatusKind& rejection_reason)
+    {
+        (void)change;
+        (void)unknown_missing_changes_up_to;
+        (void)rejection_reason;
+        return true;
+    }
+
     /**
      * Add a CacheChange_t to the ReaderHistory.
      * @param a_change Pointer to the CacheChange to add.

include/fastdds/rtps/reader/ReaderListener.h

@@ -22,6 +22,7 @@
 
 #include <fastdds/dds/core/status/LivelinessChangedStatus.hpp>
 #include <fastdds/dds/core/status/SubscriptionMatchedStatus.hpp>
+#include <fastdds/dds/core/status/SampleRejectedStatus.hpp>
 #include <fastdds/rtps/builtin/data/WriterProxyData.h>
 #include <fastdds/rtps/common/Guid.h>
 #include <fastdds/rtps/common/MatchingInfo.h>
@@ -152,6 +153,23 @@ class RTPS_DllAPI ReaderListener
         static_cast<void>(writer_info);
     }
 
+    /**
+     * This method is called when the reader rejects a samples.
+     *
+     * @param reader                         Pointer to the RTPSReader.
+     * @param change Pointer to the CacheChange_t. This is a const pointer to const data
+     * to indicate that the user should not dispose of this data himself.
+     */
+    virtual void on_sample_rejected(
+            RTPSReader* reader,
+            eprosima::fastdds::dds::SampleRejectedStatusKind reason,
+            const CacheChange_t* const change)
+    {
+        static_cast<void>(reader);
+        static_cast<void>(reason);
+        static_cast<void>(change);
+    }
+
 };
 
 } // namespace rtps

src/cpp/fastdds/subscriber/DataReader.cpp

@@ -348,11 +348,7 @@ ReturnCode_t DataReader::get_sample_lost_status(
 ReturnCode_t DataReader::get_sample_rejected_status(
         SampleRejectedStatus& status) const
 {
-    static_cast<void> (status);
-    return ReturnCode_t::RETCODE_UNSUPPORTED;
-    /*
-       return impl_->get_sample_rejected_status(status);
-     */
+    return impl_->get_sample_rejected_status(status);
 }
 
 ReturnCode_t DataReader::get_subscription_matched_status(

src/cpp/fastdds/subscriber/DataReaderImpl.cpp

@@ -903,6 +903,25 @@ void DataReaderImpl::InnerDataReaderListener::on_sample_lost(
     data_reader_->user_datareader_->get_statuscondition().get_impl()->set_status(notify_status, true);
 }
 
+void DataReaderImpl::InnerDataReaderListener::on_sample_rejected(
+        RTPSReader* /*reader*/,
+        SampleRejectedStatusKind reason,
+        const CacheChange_t* const change_in)
+{
+    data_reader_->update_sample_rejected_status(reason, change_in);
+    StatusMask notify_status = StatusMask::sample_rejected();
+    DataReaderListener* listener = data_reader_->get_listener_for(notify_status);
+    if (listener != nullptr)
+    {
+        SampleRejectedStatus callback_status;
+        if (data_reader_->get_sample_rejected_status(callback_status) == ReturnCode_t::RETCODE_OK)
+        {
+            listener->on_sample_rejected(data_reader_->user_datareader_, callback_status);
+        }
+    }
+    data_reader_->user_datareader_->get_statuscondition().get_impl()->set_status(notify_status, true);
+}
+
 bool DataReaderImpl::on_new_cache_change_added(
         const CacheChange_t* const change)
 {
@@ -1208,16 +1227,24 @@ ReturnCode_t DataReaderImpl::get_sample_lost_status(
     return ReturnCode_t::RETCODE_OK;
 }
 
-/* TODO
-   bool DataReaderImpl::get_sample_rejected_status(
-        SampleRejectedStatus& status) const
-   {
-    (void)status;
-    // TODO Implement
-    // TODO add callback call subscriber_->subscriber_listener_->on_sample_rejected
-    return false;
-   }
- */
+ReturnCode_t DataReaderImpl::get_sample_rejected_status(
+        SampleRejectedStatus& status)
+{
+    if (reader_ == nullptr)
+    {
+        return ReturnCode_t::RETCODE_NOT_ENABLED;
+    }
+
+    {
+        std::lock_guard<RecursiveTimedMutex> lock(reader_->getMutex());
+
+        status = sample_rejected_status_;
+        sample_rejected_status_.total_count_change = 0u;
+    }
+
+    user_datareader_->get_statuscondition().get_impl()->set_status(StatusMask::sample_rejected(), false);
+    return ReturnCode_t::RETCODE_OK;
+}
 
 const Subscriber* DataReaderImpl::get_subscriber() const
 {
@@ -1277,7 +1304,7 @@ LivelinessChangedStatus& DataReaderImpl::update_liveliness_status(
     return liveliness_changed_status_;
 }
 
-SampleLostStatus& DataReaderImpl::update_sample_lost_status(
+const SampleLostStatus& DataReaderImpl::update_sample_lost_status(
         int32_t sample_lost_since_last_update)
 {
     sample_lost_status_.total_count += sample_lost_since_last_update;
@@ -1690,6 +1717,17 @@ InstanceHandle_t DataReaderImpl::lookup_instance(
     return handle;
 }
 
+const SampleRejectedStatus& DataReaderImpl::update_sample_rejected_status(
+        SampleRejectedStatusKind reason,
+        const CacheChange_t* const change_in)
+{
+    ++sample_rejected_status_.total_count;
+    ++sample_rejected_status_.total_count_change;
+    sample_rejected_status_.last_reason = reason;
+    sample_rejected_status_.last_instance_handle = change_in->instanceHandle;
+    return sample_rejected_status_;
+}
+
 } /* namespace dds */
 } /* namespace fastdds */
 } /* namespace eprosima */

src/cpp/fastdds/subscriber/DataReaderImpl.hpp

@@ -258,10 +258,15 @@ class DataReaderImpl
     ReturnCode_t get_sample_lost_status(
             fastdds::dds::SampleLostStatus& status);
 
-    /* TODO
-       bool get_sample_rejected_status(
-       fastrtps::SampleRejectedStatus& status) const;
+    /*!
+     * @brief Get the SAMPLE_REJECTED communication status
+     *
+     * @param[out] status SampleRejectedStatus object where the status is returned.
+     *
+     * @return RETCODE_OK
      */
+    ReturnCode_t get_sample_rejected_status(
+            fastrtps::SampleRejectedStatus& status);
 
     const Subscriber* get_subscriber() const;
 
@@ -383,6 +388,11 @@ class DataReaderImpl
                 fastrtps::rtps::RTPSReader* reader,
                 int32_t sample_lost_since_last_update) override;
 
+        void on_sample_rejected(
+                fastrtps::rtps::RTPSReader* reader,
+                SampleRejectedStatusKind reason,
+                const fastrtps::rtps::CacheChange_t* const change) override;
+
         DataReaderImpl* data_reader_;
     }
     reader_listener_;
@@ -410,6 +420,8 @@ class DataReaderImpl
 
     //! Sample lost status
     SampleLostStatus sample_lost_status_;
+    //! Sample rejected status
+    SampleRejectedStatus sample_rejected_status_;
 
     //! A timed callback to remove expired samples
     fastrtps::rtps::TimedEvent* lifespan_timer_ = nullptr;
@@ -491,9 +503,19 @@ class DataReaderImpl
     LivelinessChangedStatus& update_liveliness_status(
             const fastrtps::LivelinessChangedStatus& status);
 
-    SampleLostStatus& update_sample_lost_status(
+    const SampleLostStatus& update_sample_lost_status(
             int32_t sample_lost_since_last_update);
 
+    /*!
+     * @brief Update SampleRejectedStatus with information about a new rejected sample.
+     *
+     * @param[in] Reason why the new sample was rejected.
+     * @param[in] New sample which was rejected.
+     */
+    const SampleRejectedStatus& update_sample_rejected_status(
+            SampleRejectedStatusKind reason,
+            const fastrtps::rtps::CacheChange_t* const change_in);
+
     /**
      * Returns the most appropriate listener to handle the callback for the given status,
      * or nullptr if there is no appropriate listener.