Refs #21185: Send User Data to share the information about the kind of writer

Signed-off-by: elianalf <62831776+elianalf@users.noreply.github.com>

Author: elianalf

examples/cpp/flow_control/PublisherApp.cpp

@@ -90,6 +90,9 @@ PublisherApp::PublisherApp(
     // Disable Data Sharing to force the communication on a Transport,
     // since a Flow Control is applied only on Transports.
     wsqos.data_sharing().off();
+    // Set a user data value to share with the DataReader the information about the kind of Data Writer
+    // (0 for Slow and 1 for Fast)
+    wsqos.user_data().data_vec({0});
     slow_writer_ = publisher_->create_datawriter(topic_, wsqos, this, StatusMask::all());
     if (slow_writer_ == nullptr)
     {
@@ -104,7 +107,9 @@ PublisherApp::PublisherApp(
     wfqos.publish_mode().kind = ASYNCHRONOUS_PUBLISH_MODE;
     // Disable Data Sharing to be consistent with the Slow Writer
     wfqos.data_sharing().off();
-
+    // Set a user data value to share with the DataReader the information about the kind of Data Writer
+    // (0 for Slow and 1 for Fast)
+    wfqos.user_data().data_vec({1});
     fast_writer_ = publisher_->create_datawriter(topic_, wfqos, this, StatusMask::all());
     if (fast_writer_ == nullptr)
     {

examples/cpp/flow_control/SubscriberApp.cpp

@@ -44,9 +44,14 @@ SubscriberApp::SubscriberApp(
     , type_(new FlowControlPubSubType())
     , samples_(config.samples)
     , stop_(false)
+    , slow_writer_guid({0})
+    , fast_writer_guid({0})
 {
+    StatusMask status_mask = StatusMask::none();
+    status_mask << StatusMask::data_available();
+    status_mask << StatusMask::subscription_matched();
     // Create Participant
-    participant_ = DomainParticipantFactory::get_instance()->create_participant(0, PARTICIPANT_QOS_DEFAULT);
+    participant_ = DomainParticipantFactory::get_instance()->create_participant(0, PARTICIPANT_QOS_DEFAULT, this, status_mask);
     if (participant_ == nullptr)
     {
         throw std::runtime_error("Participant initialization failed");
@@ -125,15 +130,15 @@ void SubscriberApp::on_data_available(
             static unsigned int fastMessages = 0;
             static unsigned int slowMessages = 0;
 
-            if (msg.wasFast())
+            if (info.sample_identity.writer_guid().entityId == fast_writer_guid)
             {
                 fastMessages++;
-                std::cout << "Sample RECEIVED from fast writer, count=" << fastMessages << std::endl;
+                std::cout << "Sample RECEIVED from FAST writer, count=" << fastMessages << std::endl;
             }
-            else
+            else if (info.sample_identity.writer_guid().entityId == slow_writer_guid)
             {
                 slowMessages++;
-                std::cout << "Sample RECEIVED from slow writer, count=" << slowMessages << std::endl;
+                std::cout << "Sample RECEIVED from SLOW writer, count=" << slowMessages << std::endl;
             }
 
             if ((samples_ > 0) && (fastMessages >= samples_) && (slowMessages >= samples_))
@@ -144,6 +149,26 @@ void SubscriberApp::on_data_available(
     }
 }
 
+void SubscriberApp::on_data_writer_discovery(
+        DomainParticipant* /*participant*/,
+        eprosima::fastdds::rtps::WriterDiscoveryInfo&& info,
+        bool& /*should_be_ignored*/)
+{
+    std::vector<eprosima::fastdds::rtps::octet> slow_writer_id = {0};
+    std::vector<eprosima::fastdds::rtps::octet> fast_writer_id = {1};
+
+    if (info.info.m_qos.m_userData.data_vec() == fast_writer_id)
+    {
+        fast_writer_guid = info.info.guid().entityId;
+        std::cout << "Fast writer id " << fast_writer_guid << std::endl;
+    }
+    else if(info.info.m_qos.m_userData.data_vec() == slow_writer_id)
+    {
+        slow_writer_guid = info.info.guid().entityId;
+        std::cout << "Slow writer id " << slow_writer_guid << std::endl;
+    }
+}
+
 void SubscriberApp::run()
 {
     std::unique_lock<std::mutex> lck(terminate_cv_mtx_);

examples/cpp/flow_control/SubscriberApp.hpp

@@ -24,6 +24,7 @@
 #include <condition_variable>
 
 #include <fastdds/dds/domain/DomainParticipant.hpp>
+#include <fastdds/dds/domain/DomainParticipantListener.hpp>
 #include <fastdds/dds/subscriber/Subscriber.hpp>
 #include <fastdds/dds/subscriber/DataReader.hpp>
 #include <fastdds/dds/subscriber/DataReaderListener.hpp>
@@ -40,7 +41,7 @@ namespace fastdds {
 namespace examples {
 namespace flow_control {
 
-class SubscriberApp : public Application,  public DataReaderListener
+class SubscriberApp : public Application,  public DomainParticipantListener
 {
 public:
 
@@ -58,6 +59,11 @@ class SubscriberApp : public Application,  public DataReaderListener
             DataReader* reader,
             const SubscriptionMatchedStatus& info) override;
 
+    void on_data_writer_discovery(
+            DomainParticipant* participant,
+            eprosima::fastdds::rtps::WriterDiscoveryInfo&& info,
+            bool& should_be_ignored) override;
+
     //! Run subscriber
     void run() override;
 
@@ -86,6 +92,11 @@ class SubscriberApp : public Application,  public DataReaderListener
     mutable std::mutex terminate_cv_mtx_;
 
     std::condition_variable terminate_cv_;
+
+    eprosima::fastdds::rtps::EntityId_t slow_writer_guid;
+
+    eprosima::fastdds::rtps::EntityId_t fast_writer_guid;
+
 };
 
 } // namespace flow_control