Fix PubSubParticipant rtps deprecated

Signed-off-by: Eugenio Collado <[email protected]>

Author: EugenioCollado

src/cpp/rtps/builtin/discovery/participant/simple/PDPStatelessWriter.hpp

@@ -135,7 +135,7 @@ class PDPStatelessWriter : public StatelessWriter, private fastdds::rtps::IReade
 
     //! Configured initial peers
     fastdds::rtps::LocatorList initial_peers_{};
-    
+
     //! The set of readers interested
     mutable ResourceLimitedVector<GUID_t> interested_readers_;
     //! Whether we have set that all destinations are interested

src/cpp/rtps/participant/RTPSParticipantImpl.cpp

@@ -147,7 +147,8 @@ static void set_builtin_transports_from_env_var(
                         "LARGE_DATA", BuiltinTransports::LARGE_DATA,
                         "LARGE_DATAv6", BuiltinTransports::LARGE_DATAv6))
                 {
-                    EPROSIMA_LOG_ERROR(RTPS_PARTICIPANT, "Wrong value '" << env_value << "' for environment variable '" <<
+                    EPROSIMA_LOG_ERROR(RTPS_PARTICIPANT,
+                            "Wrong value '" << env_value << "' for environment variable '" <<
                             env_var_name << "'. Leaving as DEFAULT");
                 }
                 // Max_msg_size parser
@@ -486,7 +487,8 @@ RTPSParticipantImpl::RTPSParticipantImpl(
                     {
                         EPROSIMA_LOG_INFO(RTPS_PARTICIPANT,
                                 "Participant " << m_att.getName() << " with GUID " << m_guid <<
-                                " tries to create a TCP client for discovery server without providing a proper listening port." <<
+                                " tries to create a TCP client for discovery server without providing a proper listening port."
+                                               <<
                                 " No TCP participants will be able to connect to this participant, but it will be able make connections.");
                     }
                     for (fastdds::rtps::RemoteServerAttributes& it : m_att.builtin.discovery_config.m_DiscoveryServers)
@@ -1428,7 +1430,7 @@ bool RTPSParticipantImpl::createWriter(
                     if (entityId == c_EntityId_SPDPWriter)
                     {
                         return new PDPStatelessWriter(this, guid, param, flow_controller,
-                                        hist, listen);
+                                       hist, listen);
                     }
                     else if (persistence != nullptr)
                     {
@@ -3027,7 +3029,8 @@ void RTPSParticipantImpl::environment_file_has_changed()
     }
     else
     {
-        EPROSIMA_LOG_WARNING(RTPS_QOS_CHECK, "Trying to add Discovery Servers to a participant which is not a SERVER, BACKUP " <<
+        EPROSIMA_LOG_WARNING(RTPS_QOS_CHECK,
+                "Trying to add Discovery Servers to a participant which is not a SERVER, BACKUP " <<
                 "or an overriden CLIENT (SIMPLE participant transformed into CLIENT with the environment variable)");
     }
 }

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

@@ -671,6 +671,13 @@ class PubSubParticipant
         return *this;
     }
 
+    PubSubParticipant& wire_protocol_builtin(
+            const eprosima::fastrtps::rtps::BuiltinAttributes& wire_protocol_builtin)
+    {
+        participant_qos_.wire_protocol().builtin = wire_protocol_builtin;
+        return *this;
+    }
+
     bool update_wire_protocol(
             const eprosima::fastdds::dds::WireProtocolConfigQos& wire_protocol)
     {

test/blackbox/api/fastrtps_deprecated/PubSubParticipant.hpp

@@ -47,6 +47,39 @@ namespace fastrtps {
 template<class TypeSupport>
 class PubSubParticipant
 {
+    class PrtListener : public ParticipantListener
+    {
+        friend class PubSubParticipant;
+
+    public:
+
+        PrtListener(
+                PubSubParticipant* participant)
+            : participant_(participant)
+        {
+        }
+
+        ~PrtListener()
+        {
+        }
+
+        void onParticipantDiscovery(
+                Participant* part,
+                rtps::ParticipantDiscoveryInfo&& info) override
+        {
+            (void)part;
+            (void)info;
+            participant_->prt_matched();
+        }
+
+    private:
+
+        PrtListener& operator =(
+                const PrtListener&) = delete;
+        //! A pointer to the participant
+        PubSubParticipant* participant_;
+    };
+
     class PubListener : public PublisherListener
     {
         friend class PubSubParticipant;
@@ -151,6 +184,7 @@ class PubSubParticipant
         , publisher_attr_()
         , pub_listener_(this)
         , sub_listener_(this)
+        , prt_matched_(0)
         , pub_matched_(0)
         , sub_matched_(0)
         , pub_times_liveliness_lost_(0)
@@ -269,6 +303,59 @@ class PubSubParticipant
         publishers_[index]->assert_liveliness();
     }
 
+    bool wait_discovery(
+            std::chrono::seconds timeout = std::chrono::seconds::zero(),
+            uint8_t matched = 0,
+            bool exact = false)
+    {
+        // No need to wait in this case
+        if (exact && matched == prt_matched_)
+        {
+            return true;
+        }
+
+        std::unique_lock<std::mutex> lock(prt_mutex_);
+        bool ret_value = true;
+        std::cout << "Participant is waiting discovery..." << std::endl;
+
+        if (timeout == std::chrono::seconds::zero())
+        {
+            prt_cv_.wait(lock, [&]()
+                    {
+                        if (exact)
+                        {
+                            return prt_matched_ == matched;
+                        }
+                        return prt_matched_ >= matched;
+                    });
+        }
+        else
+        {
+            if (!prt_cv_.wait_for(lock, timeout, [&]()
+                    {
+                        if (exact)
+                        {
+                            return prt_matched_ == matched;
+                        }
+                        return prt_matched_ >= matched;
+                    }))
+            {
+                ret_value = false;
+            }
+        }
+
+        if (ret_value)
+        {
+            std::cout << "Participant discovery finished successfully..." << std::endl;
+        }
+        else
+        {
+            std::cout << "Participant discovery finished unsuccessfully..." << std::endl;
+        }
+
+        return ret_value;
+    }
+
     void pub_wait_discovery(
             std::chrono::seconds timeout = std::chrono::seconds::zero())
     {
@@ -437,6 +524,13 @@ class PubSubParticipant
         return *this;
     }
 
+    PubSubParticipant& wire_protocol_builtin(
+            const eprosima::fastrtps::rtps::BuiltinAttributes& wire_protocol_builtin)
+    {
+        participant_attr_.rtps.builtin = wire_protocol_builtin;
+        return *this;
+    }
+
     PubSubParticipant& initial_peers(
             const eprosima::fastrtps::rtps::LocatorList_t& initial_peers)
     {
@@ -601,6 +695,20 @@ class PubSubParticipant
     PubSubParticipant& operator =(
             const PubSubParticipant&) = delete;
 
+    void prt_matched()
+    {
+        std::unique_lock<std::mutex> lock(prt_mutex_);
+        ++prt_matched_;
+        prt_cv_.notify_one();
+    }
+
+    void prt_unmatched()
+    {
+        std::unique_lock<std::mutex> lock(prt_mutex_);
+        --prt_matched_;
+        prt_cv_.notify_one();
+    }
+
     void pub_matched()
     {
         std::unique_lock<std::mutex> lock(pub_mutex_);
@@ -654,10 +762,13 @@ class PubSubParticipant
     //! A listener for subscribers
     SubListener sub_listener_;
 
+    std::mutex prt_mutex_;
     std::mutex pub_mutex_;
     std::mutex sub_mutex_;
+    std::condition_variable prt_cv_;
     std::condition_variable pub_cv_;
     std::condition_variable sub_cv_;
+    std::atomic<unsigned int> prt_matched_;
     std::atomic<unsigned int> pub_matched_;
     std::atomic<unsigned int> sub_matched_;
 

test/blackbox/common/BlackboxTestsDiscovery.cpp

@@ -1417,7 +1417,7 @@ TEST_P(Discovery, single_unicast_pdp_response)
 
     // The main participant will use the test transport and a specific announcments configuration
     main_participant->disable_builtin_transport().add_user_transport_to_pparams(test_transport)
-            .wire_protocol(main_wire_protocol);
+            .wire_protocol_builtin(main_wire_protocol.builtin);
 
     // Start the main participant
     ASSERT_TRUE(main_participant->init_participant());
@@ -1446,7 +1446,7 @@ TEST_P(Discovery, single_unicast_pdp_response)
         auto participant = std::make_shared<PubSubParticipant<HelloWorldPubSubType>>(0, 0, 0, 0);
         // All participants use the same transport
         participant->disable_builtin_transport().add_user_transport_to_pparams(udp_localhost_transport)
-                .wire_protocol(wire_protocol);
+                .wire_protocol_builtin(wire_protocol.builtin);
         participants.push_back(participant);
     }