[21670] Fix Data Races on DDS-Pipe (backport #145)

ddspipe_participants/include/ddspipe_participants/participant/dds/CommonParticipant.hpp

@@ -54,7 +54,7 @@ namespace dds {
  * @warning This Participant class does not support RPC so far.
  * @todo TODO
  */
-class CommonParticipant : public core::IParticipant, public fastdds::dds::DomainParticipantListener
+class CommonParticipant : public core::IParticipant
 {
 public:
 
@@ -106,17 +106,56 @@ class CommonParticipant : public core::IParticipant, public fastdds::dds::Domain
     // LISTENER METHODS
     /////////////////////////
 
-    virtual void on_participant_discovery(
-            fastdds::dds::DomainParticipant* participant,
-            fastrtps::rtps::ParticipantDiscoveryInfo&& info) override;
-
-    virtual void on_subscriber_discovery(
-            fastdds::dds::DomainParticipant* participant,
-            fastrtps::rtps::ReaderDiscoveryInfo&& info) override;
-
-    virtual void on_publisher_discovery(
-            fastdds::dds::DomainParticipant* participant,
-            fastrtps::rtps::WriterDiscoveryInfo&& info) override;
+    class DdsListener : public fastdds::dds::DomainParticipantListener
+    {
+    public:
+
+        DDSPIPE_PARTICIPANTS_DllAPI
+        explicit DdsListener(
+                std::shared_ptr<SimpleParticipantConfiguration> conf,
+                std::shared_ptr<core::DiscoveryDatabase> ddb);
+
+        /**
+         * @brief Override method from \c DomainParticipantListener
+         *
+         * This method is only used for debugging purposes.
+         */
+        DDSPIPE_PARTICIPANTS_DllAPI
+        void on_participant_discovery(
+                fastdds::dds::DomainParticipant* participant,
+                fastrtps::rtps::ParticipantDiscoveryInfo&& info) override;
+
+        /**
+         * @brief Override method from \c DomainParticipantListener .
+         *
+         * This method adds to the database the discovered or modified endpoint.
+         */
+        DDSPIPE_PARTICIPANTS_DllAPI
+        void on_subscriber_discovery(
+                fastdds::dds::DomainParticipant* participant,
+                fastrtps::rtps::ReaderDiscoveryInfo&& info) override;
+
+        /**
+         * @brief Override method from \c DomainParticipantListener .
+         *
+         * This method adds to the database the discovered or modified endpoint.
+         */
+        DDSPIPE_PARTICIPANTS_DllAPI
+        void on_publisher_discovery(
+                fastdds::dds::DomainParticipant* participant,
+                fastrtps::rtps::WriterDiscoveryInfo&& info) override;
+
+    protected:
+
+        //! Shared pointer to the configuration of the participant
+        const std::shared_ptr<SimpleParticipantConfiguration> configuration_;
+        //! Shared pointer to the discovery database
+        const std::shared_ptr<core::DiscoveryDatabase> discovery_database_;
+
+    };
+
+    //! Unique pointer to the internal DDS Participant Listener
+    std::unique_ptr<fastdds::dds::DomainParticipantListener> dds_participant_listener_;
 
 protected:
 
@@ -130,14 +169,27 @@ class CommonParticipant : public core::IParticipant, public fastdds::dds::Domain
             const std::shared_ptr<core::PayloadPool>& payload_pool,
             const std::shared_ptr<core::DiscoveryDatabase>& discovery_database);
 
+    /////////////////////////
+    // VIRTUAL METHODS
+    /////////////////////////
+
+    /**
+     * @brief Virtual method that creates a listener for the internal DDS Participant.
+     *        It should be overridden if a different listener is needed.
+     */
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual std::unique_ptr<fastdds::dds::DomainParticipantListener> create_listener_();
+
     /////////////////////////
     // INTERNAL VIRTUAL METHODS
     /////////////////////////
 
+    DDSPIPE_PARTICIPANTS_DllAPI
     virtual
     fastdds::dds::DomainParticipantQos
     reckon_participant_qos_() const;
 
+    DDSPIPE_PARTICIPANTS_DllAPI
     virtual
     fastdds::dds::DomainParticipant*
     create_dds_participant_();

ddspipe_participants/include/ddspipe_participants/participant/dynamic_types/DynTypesParticipant.hpp

@@ -36,7 +36,7 @@ namespace participants {
  *
  * TODO: separate these 2 participants
  */
-class DynTypesParticipant : public rtps::SimpleParticipant, public eprosima::fastdds::dds::DomainParticipantListener
+class DynTypesParticipant : public rtps::SimpleParticipant
 {
 public:
 
@@ -71,33 +71,61 @@ class DynTypesParticipant : public rtps::SimpleParticipant, public eprosima::fas
     std::shared_ptr<core::IReader> create_reader(
             const core::ITopic& topic) override;
 
-    DDSPIPE_PARTICIPANTS_DllAPI
-    void on_type_discovery(
-            eprosima::fastdds::dds::DomainParticipant* participant,
-            const eprosima::fastrtps::rtps::SampleIdentity& request_sample_id,
-            const eprosima::fastrtps::string_255& topic,
-            const eprosima::fastrtps::types::TypeIdentifier* identifier,
-            const eprosima::fastrtps::types::TypeObject* object,
-            eprosima::fastrtps::types::DynamicType_ptr dyn_type) override;
-
-    DDSPIPE_PARTICIPANTS_DllAPI
-    virtual void on_type_information_received(
-            eprosima::fastdds::dds::DomainParticipant* participant,
-            const eprosima::fastrtps::string_255 topic_name,
-            const eprosima::fastrtps::string_255 type_name,
-            const eprosima::fastrtps::types::TypeInformation& type_information) override;
+    class DynTypesRtpsListener : public rtps::CommonParticipant::RtpsListener,
+        public eprosima::fastdds::dds::DomainParticipantListener
+    {
+    public:
+
+        DDSPIPE_PARTICIPANTS_DllAPI
+        explicit DynTypesRtpsListener(
+                std::shared_ptr<ParticipantConfiguration> conf,
+                std::shared_ptr<core::DiscoveryDatabase> ddb,
+                std::shared_ptr<InternalReader> internal_reader);
+
+        DDSPIPE_PARTICIPANTS_DllAPI
+        void on_type_discovery(
+                eprosima::fastdds::dds::DomainParticipant* participant,
+                const eprosima::fastrtps::rtps::SampleIdentity& request_sample_id,
+                const eprosima::fastrtps::string_255& topic,
+                const eprosima::fastrtps::types::TypeIdentifier* identifier,
+                const eprosima::fastrtps::types::TypeObject* object,
+                eprosima::fastrtps::types::DynamicType_ptr dyn_type) override;
+
+        DDSPIPE_PARTICIPANTS_DllAPI
+        virtual void on_type_information_received(
+                eprosima::fastdds::dds::DomainParticipant* participant,
+                const eprosima::fastrtps::string_255 topic_name,
+                const eprosima::fastrtps::string_255 type_name,
+                const eprosima::fastrtps::types::TypeInformation& type_information) override;
+
+        //! Type Object Reader getter
+        inline std::shared_ptr<InternalReader> type_object_reader() const
+        {
+            return type_object_reader_;
+        }
+
+    protected:
+
+        void internal_notify_type_object_(
+                eprosima::fastrtps::types::DynamicType_ptr dynamic_type);
+
+        //! Copy of Type Object Internal Reader
+        std::shared_ptr<InternalReader> type_object_reader_;
+
+    };
 
 protected:
 
-    void internal_notify_type_object_(
-            eprosima::fastrtps::types::DynamicType_ptr dynamic_type);
-
     void initialize_internal_dds_participant_();
 
     eprosima::fastdds::dds::DomainParticipant* dds_participant_;
 
     //! Type Object Internal Reader
     std::shared_ptr<InternalReader> type_object_reader_;
+
+    //! Override method from \c CommonParticipant to create the internal RTPS participant listener
+    std::unique_ptr<fastrtps::rtps::RTPSParticipantListener> create_listener_() override;
+
 };
 
 } /* namespace participants */

ddspipe_participants/include/ddspipe_participants/participant/rtps/CommonParticipant.hpp

@@ -50,7 +50,6 @@ namespace rtps {
  */
 class CommonParticipant
     : public core::IParticipant
-    , public fastrtps::rtps::RTPSParticipantListener
 {
 public:
 
@@ -121,35 +120,56 @@ class CommonParticipant
     // RTPS LISTENER METHODS
     /////////////////////////
 
-    /**
-     * @brief Override method from \c RTPSParticipantListener .
-     *
-     * This method only is for debugging purposes.
-     */
-    DDSPIPE_PARTICIPANTS_DllAPI
-    virtual void onParticipantDiscovery(
-            fastrtps::rtps::RTPSParticipant* participant,
-            fastrtps::rtps::ParticipantDiscoveryInfo&& info) override;
-
-    /**
-     * @brief Override method from \c RTPSParticipantListener .
-     *
-     * This method adds to database the endpoint discovered or modified.
-     */
-    DDSPIPE_PARTICIPANTS_DllAPI
-    virtual void onReaderDiscovery(
-            fastrtps::rtps::RTPSParticipant* participant,
-            fastrtps::rtps::ReaderDiscoveryInfo&& info) override;
-
-    /**
-     * @brief Override method from \c RTPSParticipantListener .
-     *
-     * This method adds to database the endpoint discovered or modified.
-     */
-    DDSPIPE_PARTICIPANTS_DllAPI
-    virtual void onWriterDiscovery(
-            fastrtps::rtps::RTPSParticipant* participant,
-            fastrtps::rtps::WriterDiscoveryInfo&& info) override;
+    class RtpsListener : public fastrtps::rtps::RTPSParticipantListener
+    {
+    public:
+
+        DDSPIPE_PARTICIPANTS_DllAPI
+        explicit RtpsListener(
+                std::shared_ptr<ParticipantConfiguration> conf,
+                std::shared_ptr<core::DiscoveryDatabase> ddb);
+
+        /**
+         * @brief Override method from \c RTPSParticipantListener .
+         *
+         * This method is only used for debugging purposes.
+         */
+        DDSPIPE_PARTICIPANTS_DllAPI
+        virtual void onParticipantDiscovery(
+                fastrtps::rtps::RTPSParticipant* participant,
+                fastrtps::rtps::ParticipantDiscoveryInfo&& info) override;
+
+        /**
+         * @brief Override method from \c RTPSParticipantListener .
+         *
+         * This method adds to database the endpoint discovered or modified.
+         */
+        DDSPIPE_PARTICIPANTS_DllAPI
+        virtual void onReaderDiscovery(
+                fastrtps::rtps::RTPSParticipant* participant,
+                fastrtps::rtps::ReaderDiscoveryInfo&& info) override;
+
+        /**
+         * @brief Override method from \c RTPSParticipantListener .
+         *
+         * This method adds to database the endpoint discovered or modified.
+         */
+        DDSPIPE_PARTICIPANTS_DllAPI
+        virtual void onWriterDiscovery(
+                fastrtps::rtps::RTPSParticipant* participant,
+                fastrtps::rtps::WriterDiscoveryInfo&& info) override;
+
+    protected:
+
+        //! Shared pointer to the configuration of the participant
+        const std::shared_ptr<ParticipantConfiguration> configuration_;
+        //! Shared pointer to the discovery database
+        const std::shared_ptr<core::DiscoveryDatabase> discovery_database_;
+
+    };
+
+    //! Unique pointer to the internal RTPS Participant Listener
+    std::unique_ptr<fastrtps::rtps::RTPSParticipantListener> rtps_participant_listener_;
 
     //////////////////
     // STATIC METHODS
@@ -207,6 +227,15 @@ class CommonParticipant
     static fastrtps::rtps::RTPSParticipantAttributes reckon_participant_attributes_(
             const ParticipantConfiguration* participant_configuration);
 
+    /**
+     * @brief Virtual method that creates a listener for the internal RTPS Participant.
+     *        It should be overridden if a different listener is needed.
+     *        This method must be called after the RTPS Participant is created, otherwise no listener will be set.
+     * @return A unique pointer to an RTPS Participant Listener.
+     */
+    DDSPIPE_PARTICIPANTS_DllAPI
+    virtual std::unique_ptr<fastrtps::rtps::RTPSParticipantListener> create_listener_();
+
     /////
     // VARIABLES