Set different content filter signatures for each requester (#5972) (#5984)

* Refs #23568: Set different content filter signatures for each requester



* Refs #23568: Add test



---------


(cherry picked from commit 5e01f49)

Signed-off-by: Carlosespicur <[email protected]>
Co-authored-by: Carlos Espinoza Curto <[email protected]>

Author: mergify[bot]

src/cpp/fastdds/rpc/RequesterImpl.cpp

@@ -26,6 +26,7 @@
 #include <fastdds/dds/domain/qos/RequesterQos.hpp>
 #include <fastdds/dds/log/Log.hpp>
 #include <fastdds/dds/rpc/RequestInfo.hpp>
+#include <fastdds/dds/topic/ContentFilteredTopic.hpp>
 #include <fastdds/rtps/common/Guid.hpp>
 #include <fastdds/rtps/common/SequenceNumber.hpp>
 #include <fastdds/rtps/common/WriteParams.hpp>
@@ -248,16 +249,24 @@ ReturnCode_t RequesterImpl::create_dds_entities(
         return RETCODE_ERROR;
     }
 
+    ContentFilteredTopic* reply_topic = service_->get_reply_filtered_topic();
+
     requester_reader_ =
             service_->get_subscriber()->create_datareader(
-        service_->get_reply_filtered_topic(), qos.reader_qos, this, StatusMask::subscription_matched());
+        reply_topic, qos.reader_qos, this, StatusMask::subscription_matched());
 
     if (!requester_reader_)
     {
         EPROSIMA_LOG_ERROR(REQUESTER, "Error creating reply reader");
         return RETCODE_ERROR;
     }
 
+    // Set the content filter signature to be different from the one used in other requesters
+    std::stringstream guid;
+    guid << requester_reader_->guid();
+    std::vector<std::string> expression_parameters;
+    reply_topic->set_filter_expression(guid.str(), expression_parameters);
+
     // Set the related entity key on both entities
     requester_reader_->set_related_datawriter(requester_writer_);
     requester_writer_->set_related_datareader(requester_reader_);

test/blackbox/api/dds-pim/ReqRepHelloWorldReplier.cpp

@@ -134,14 +134,21 @@ void ReqRepHelloWorldReplier::init_with_custom_qos(
 }
 
 void ReqRepHelloWorldReplier::wait_discovery()
+{
+    wait_discovery(1, 1);
+}
+
+void ReqRepHelloWorldReplier::wait_discovery(
+        unsigned int min_pub_matched,
+        unsigned int min_sub_matched)
 {
     std::unique_lock<std::mutex> lock(mutexDiscovery_);
 
     std::cout << "Replier is waiting discovery..." << std::endl;
 
     cvDiscovery_.wait(lock, [&]()
             {
-                return pub_matched_ > 0 && sub_matched_ > 0;
+                return pub_matched_ >= min_pub_matched && sub_matched_ >= min_sub_matched;
             });
 
     std::cout << "Replier discovery finished..." << std::endl;

test/blackbox/api/dds-pim/ReqRepHelloWorldReplier.hpp

@@ -70,6 +70,10 @@ class ReqRepHelloWorldReplier
 
     void wait_discovery();
 
+    void wait_discovery(
+            unsigned int min_pub_matched,
+            unsigned int min_sub_matched);
+
     void matched(
             bool is_pub);
 

test/blackbox/api/dds-pim/ReqRepHelloWorldRequester.cpp

@@ -149,14 +149,21 @@ void ReqRepHelloWorldRequester::block(
 }
 
 void ReqRepHelloWorldRequester::wait_discovery()
+{
+    wait_discovery(1, 1);
+}
+
+void ReqRepHelloWorldRequester::wait_discovery(
+        unsigned int min_pub_matched,
+        unsigned int min_sub_matched)
 {
     std::unique_lock<std::mutex> lock(mutexDiscovery_);
 
     std::cout << "Requester is waiting discovery..." << std::endl;
 
     cvDiscovery_.wait(lock, [&]()
             {
-                return pub_matched_ > 0 && sub_matched_ > 0;
+                return pub_matched_ >= min_pub_matched && sub_matched_ >= min_sub_matched;
             });
 
     std::cout << "Requester discovery finished..." << std::endl;

test/blackbox/api/dds-pim/ReqRepHelloWorldRequester.hpp

@@ -76,6 +76,10 @@ class ReqRepHelloWorldRequester
 
     void wait_discovery();
 
+    void wait_discovery(
+            unsigned int min_pub_matched,
+            unsigned int min_sub_matched);
+
     void matched(
             bool is_pub);
 

test/blackbox/common/DDSBlackboxTestsRPC.cpp

@@ -236,3 +236,37 @@ TEST(RPC, requester_unmatched_during_request_processing)
     // Check that the reply took at least the wait_matching timeout (3 secs)
     ASSERT_GT(reply_elapsed, Duration_t{2});
 }
+
+/**
+ * Test RPC communication with multiple requesters and one replier.
+ *
+ * This test checks that multiple requesters can send requests to a single replier
+ * and receive replies correctly.
+ */
+TEST(RPC, multiple_requesters_one_replier)
+{
+    ReqRepHelloWorldRequester requester_1;
+    ReqRepHelloWorldRequester requester_2;
+    ReqRepHelloWorldReplier replier;
+
+    // Initialize the requesters and the replier
+    requester_1.init();
+    ASSERT_TRUE(requester_1.isInitialized());
+    requester_2.init();
+    ASSERT_TRUE(requester_2.isInitialized());
+    replier.init();
+    ASSERT_TRUE(replier.isInitialized());
+
+    // Wait for discovery
+    requester_1.wait_discovery();
+    requester_2.wait_discovery();
+    replier.wait_discovery(2, 2);
+
+    // Send requests from both requesters
+    requester_1.send(1);
+    requester_2.send(2);
+
+    // Block to wait for replies
+    requester_1.block(std::chrono::seconds(5));
+    requester_2.block(std::chrono::seconds(5));
+}