[20311] Add missing TypeLookup listeners

include/fastdds/dds/builtin/typelookup/TypeLookupManager.hpp

@@ -286,6 +286,12 @@ class TypeLookupManager
         bool create_secure_endpoints();
      #endif
      */
+
+    void request_cache_change_acked(
+            fastrtps::rtps::CacheChange_t* change);
+
+    void reply_cache_change_acked(
+            fastrtps::rtps::CacheChange_t* change);
 };
 
 } /* namespace builtin */

include/fastdds/dds/builtin/typelookup/TypeLookupReplyListener.hpp

@@ -21,6 +21,8 @@
 #define TYPELOOKUP_REPLY_LISTENER_HPP_
 #ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
 #include <fastrtps/rtps/reader/ReaderListener.h>
+#include <fastrtps/rtps/writer/WriterListener.h>
+
 
 namespace eprosima {
 namespace fastrtps {
@@ -47,7 +49,7 @@ class TypeLookupManager;
  * Class TypeLookupReplyListener that receives the typelookup request messages of remote endpoints.
  * @ingroup TYPES_MODULE
  */
-class TypeLookupReplyListener : public fastrtps::rtps::ReaderListener
+class TypeLookupReplyListener : public fastrtps::rtps::ReaderListener, public fastrtps::rtps::WriterListener
 {
 public:
 
@@ -70,7 +72,16 @@ class TypeLookupReplyListener : public fastrtps::rtps::ReaderListener
      */
     void onNewCacheChangeAdded(
             fastrtps::rtps::RTPSReader* reader,
-            const fastrtps::rtps::CacheChange_t* const  change) override;
+            const fastrtps::rtps::CacheChange_t* const change) override;
+
+    /**
+     * @brief This method is called when all the readers matched with this Writer acknowledge that a cache
+     * change has been received.
+     * @param change The cache change
+     */
+    void onWriterChangeReceivedByAll(
+            fastrtps::rtps::RTPSWriter*,
+            fastrtps::rtps::CacheChange_t* change) override;
 
 private:
 
@@ -85,5 +96,5 @@ class TypeLookupReplyListener : public fastrtps::rtps::ReaderListener
 } /* namespace dds */
 } /* namespace fastdds */
 } /* namespace eprosima */
-#endif
+#endif // ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
 #endif /* TYPELOOKUP_REPLY_LISTENER_HPP_*/

include/fastdds/dds/builtin/typelookup/TypeLookupRequestListener.hpp

@@ -21,6 +21,8 @@
 #define TYPELOOKUP_REQUEST_LISTENER_HPP_
 #ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
 #include <fastrtps/rtps/reader/ReaderListener.h>
+#include <fastrtps/rtps/writer/WriterListener.h>
+
 
 namespace eprosima {
 namespace fastrtps {
@@ -47,7 +49,7 @@ class TypeLookupManager;
  * Class TypeLookupRequestListener that receives the typelookup request messages of remote endpoints.
  * @ingroup TYPES_MODULE
  */
-class TypeLookupRequestListener : public fastrtps::rtps::ReaderListener
+class TypeLookupRequestListener : public fastrtps::rtps::ReaderListener, public fastrtps::rtps::WriterListener
 {
 public:
 
@@ -70,7 +72,16 @@ class TypeLookupRequestListener : public fastrtps::rtps::ReaderListener
      */
     void onNewCacheChangeAdded(
             fastrtps::rtps::RTPSReader* reader,
-            const fastrtps::rtps::CacheChange_t* const  change) override;
+            const fastrtps::rtps::CacheChange_t* const change) override;
+
+    /**
+     * @brief This method is called when all the readers matched with this Writer acknowledge that a cache
+     * change has been received.
+     * @param change The cache change
+     */
+    void onWriterChangeReceivedByAll(
+            fastrtps::rtps::RTPSWriter*,
+            fastrtps::rtps::CacheChange_t* change) override;
 
 private:
 
@@ -86,5 +97,5 @@ class TypeLookupRequestListener : public fastrtps::rtps::ReaderListener
 } /* namespace dds */
 } /* namespace fastdds */
 } /* namespace eprosima */
-#endif
+#endif // ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
 #endif /* TYPELOOKUP_REQUEST_LISTENER_HPP_*/

src/cpp/fastdds/builtin/typelookup/TypeLookupManager.cpp

@@ -326,6 +326,8 @@ ReaderHistory* TypeLookupManager::get_builtin_reply_reader_history()
  */
 bool TypeLookupManager::create_endpoints()
 {
+    bool ret = true;
+
     const RTPSParticipantAttributes& pattr = participant_->getRTPSParticipantAttributes();
 
     // Built-in history attributes.
@@ -348,14 +350,15 @@ bool TypeLookupManager::create_endpoints()
     // Built-in request writer
     if (builtin_protocols_->m_att.typelookup_config.use_client)
     {
+        request_listener_ = new TypeLookupRequestListener(this);
         builtin_request_writer_history_ = new WriterHistory(hatt);
 
         RTPSWriter* req_writer;
         if (participant_->createWriter(
                     &req_writer,
                     watt,
                     builtin_request_writer_history_,
-                    nullptr,
+                    request_listener_,
                     fastrtps::rtps::c_EntityId_TypeLookup_request_writer,
                     true))
         {
@@ -365,23 +368,22 @@ bool TypeLookupManager::create_endpoints()
         else
         {
             EPROSIMA_LOG_ERROR(TYPELOOKUP_SERVICE, "Typelookup request writer creation failed.");
-            delete builtin_request_writer_history_;
-            builtin_request_writer_history_ = nullptr;
-            return false;
+            ret = false;
         }
     }
 
     // Built-in reply writer
-    if (builtin_protocols_->m_att.typelookup_config.use_server)
+    if (ret && builtin_protocols_->m_att.typelookup_config.use_server)
     {
+        reply_listener_ = new TypeLookupReplyListener(this);
         builtin_reply_writer_history_ = new WriterHistory(hatt);
 
         RTPSWriter* rep_writer;
         if (participant_->createWriter(
                     &rep_writer,
                     watt,
                     builtin_reply_writer_history_,
-                    nullptr,
+                    reply_listener_,
                     fastrtps::rtps::c_EntityId_TypeLookup_reply_writer,
                     true))
         {
@@ -391,9 +393,7 @@ bool TypeLookupManager::create_endpoints()
         else
         {
             EPROSIMA_LOG_ERROR(TYPELOOKUP_SERVICE, "Typelookup reply writer creation failed.");
-            delete builtin_reply_writer_history_;
-            builtin_reply_writer_history_ = nullptr;
-            return false;
+            ret = false;
         }
     }
 
@@ -410,9 +410,12 @@ bool TypeLookupManager::create_endpoints()
     ratt.endpoint.durabilityKind = fastrtps::rtps::VOLATILE;
 
     // Built-in request reader
-    if (builtin_protocols_->m_att.typelookup_config.use_server)
+    if (ret && builtin_protocols_->m_att.typelookup_config.use_server)
     {
-        request_listener_ = new TypeLookupRequestListener(this);
+        if (nullptr == request_listener_)
+        {
+            request_listener_ = new TypeLookupRequestListener(this);
+        }
         builtin_request_reader_history_ = new ReaderHistory(hatt);
 
         RTPSReader* req_reader;
@@ -430,18 +433,17 @@ bool TypeLookupManager::create_endpoints()
         else
         {
             EPROSIMA_LOG_ERROR(TYPELOOKUP_SERVICE, "Typelookup request reader creation failed.");
-            delete builtin_request_reader_history_;
-            builtin_request_reader_history_ = nullptr;
-            delete request_listener_;
-            request_listener_ = nullptr;
-            return false;
+            ret = false;
         }
     }
 
     // Built-in reply reader
-    if (builtin_protocols_->m_att.typelookup_config.use_client)
+    if (ret && builtin_protocols_->m_att.typelookup_config.use_client)
     {
-        reply_listener_ = new TypeLookupReplyListener(this);
+        if (nullptr == reply_listener_)
+        {
+            reply_listener_ = new TypeLookupReplyListener(this);
+        }
         builtin_reply_reader_history_ = new ReaderHistory(hatt);
 
         RTPSReader* rep_reader;
@@ -459,15 +461,50 @@ bool TypeLookupManager::create_endpoints()
         else
         {
             EPROSIMA_LOG_ERROR(TYPELOOKUP_SERVICE, "Typelookup reply reader creation failed.");
+            ret = false;
+        }
+    }
+
+    // Clean up if something failed.
+    if (!ret)
+    {
+        if (nullptr != builtin_request_writer_history_)
+        {
+            delete builtin_request_writer_history_;
+            builtin_request_writer_history_ = nullptr;
+        }
+
+        if (nullptr != builtin_reply_writer_history_)
+        {
+            delete builtin_reply_writer_history_;
+            builtin_reply_writer_history_ = nullptr;
+        }
+
+        if (nullptr != builtin_request_reader_history_)
+        {
+            delete builtin_request_reader_history_;
+            builtin_request_reader_history_ = nullptr;
+        }
+
+        if (nullptr != builtin_reply_reader_history_)
+        {
             delete builtin_reply_reader_history_;
             builtin_reply_reader_history_ = nullptr;
+        }
+
+        if (nullptr != request_listener_)
+        {
+            delete request_listener_;
+            request_listener_ = nullptr;
+        }
+        if (nullptr != reply_listener_)
+        {
             delete reply_listener_;
             reply_listener_ = nullptr;
-            return false;
         }
     }
 
-    return true;
+    return ret;
 }
 
 /* TODO Implement if security is needed.
@@ -567,7 +604,13 @@ bool TypeLookupManager::send_request(
         SerializedPayload_t payload;
         payload.max_size = change->serializedPayload.max_size - 4;
         payload.data = change->serializedPayload.data + 4;
-        if (valid && request_type_.serialize(&req, &payload, DataRepresentationId_t::XCDR2_DATA_REPRESENTATION))
+
+        bool serialize_ret = request_type_.serialize(&req, &payload, DataRepresentationId_t::XCDR2_DATA_REPRESENTATION);
+        if (!serialize_ret)
+        {
+            payload.data = nullptr;
+        }
+        else if (valid)
         {
             change->serializedPayload.length += payload.length;
             change->serializedPayload.pos += payload.pos;
@@ -610,7 +653,13 @@ bool TypeLookupManager::send_reply(
         SerializedPayload_t payload;
         payload.max_size = change->serializedPayload.max_size - 4;
         payload.data = change->serializedPayload.data + 4;
-        if (valid && reply_type_.serialize(&rep, &payload, DataRepresentationId_t::XCDR2_DATA_REPRESENTATION))
+
+        bool serialize_ret = reply_type_.serialize(&rep, &payload, DataRepresentationId_t::XCDR2_DATA_REPRESENTATION);
+        if (!serialize_ret)
+        {
+            payload.data = nullptr;
+        }
+        else if (valid)
         {
             change->serializedPayload.length += payload.length;
             change->serializedPayload.pos += payload.pos;
@@ -695,6 +744,18 @@ const fastrtps::rtps::GUID_t& TypeLookupManager::get_builtin_request_writer_guid
     return c_Guid_Unknown;
 }
 
+void TypeLookupManager::request_cache_change_acked(
+        fastrtps::rtps::CacheChange_t* change)
+{
+    builtin_request_writer_history_->remove_change(change);
+}
+
+void TypeLookupManager::reply_cache_change_acked(
+        fastrtps::rtps::CacheChange_t* change)
+{
+    builtin_reply_writer_history_->remove_change(change);
+}
+
 } // namespace builtin
 } // namespace dds
 } // namespace fastdds

src/cpp/fastdds/builtin/typelookup/TypeLookupReplyListener.cpp

@@ -116,6 +116,13 @@ void TypeLookupReplyListener::onNewCacheChangeAdded(
     reader->getHistory()->remove_change(change);
 }
 
+void TypeLookupReplyListener::onWriterChangeReceivedByAll(
+        fastrtps::rtps::RTPSWriter*,
+        fastrtps::rtps::CacheChange_t* change)
+{
+    tlm_->reply_cache_change_acked(change);
+}
+
 } // namespace builtin
 } // namespace dds
 } // namespace fastdds

src/cpp/fastdds/builtin/typelookup/TypeLookupRequestListener.cpp

@@ -150,6 +150,13 @@ void TypeLookupRequestListener::onNewCacheChangeAdded(
     reader->getHistory()->remove_change(change);
 }
 
+void TypeLookupRequestListener::onWriterChangeReceivedByAll(
+        fastrtps::rtps::RTPSWriter*,
+        fastrtps::rtps::CacheChange_t* change)
+{
+    tlm_->request_cache_change_acked(change);
+}
+
 } // namespace builtin
 } // namespace dds
 } // namespace fastdds