Fix topic info in generic subscription callback using zenoh

include/gz/transport/SubscriptionHandler.hh

@@ -168,7 +168,7 @@ namespace gz::transport
     /// \param[in] _topic The topic.
     public: void CreateGenericZenohSubscriber(
       std::shared_ptr<zenoh::Session> _session,
-      const std::string &_topic);
+      const FullyQualifiedTopic &_fullyQualifiedTopic);
 #endif
   };
 
@@ -225,10 +225,10 @@ namespace gz::transport
     /// \param[in] _topic The topic associated to this callback.
     public: void SetCallback(const MsgCallback<T> &_cb,
                              std::shared_ptr<zenoh::Session> _session,
-                             const std::string &_topic)
+                             const FullyQualifiedTopic &_fullyQualifiedTopic)
     {
       this->SetCallback(std::move(_cb));
-      this->CreateGenericZenohSubscriber(_session, _topic);
+      this->CreateGenericZenohSubscriber(_session, _fullyQualifiedTopic);
     }
 #endif
 
@@ -356,10 +356,10 @@ namespace gz::transport
     /// \param[in] _topic The topic associated to this callback.
     public: void SetCallback(const MsgCallback<ProtoMsg> &_cb,
                              std::shared_ptr<zenoh::Session> _session,
-                             const std::string &_topic)
+                             const FullyQualifiedTopic &_fullyQualifiedTopic)
     {
       this->SetCallback(std::move(_cb));
-      this->CreateGenericZenohSubscriber(_session, _topic);
+      this->CreateGenericZenohSubscriber(_session, _fullyQualifiedTopic);
     }
 #endif
 

include/gz/transport/detail/Node.hh

@@ -148,9 +148,10 @@ namespace gz::transport
     std::string topic = _topic;
     this->Options().TopicRemap(_topic, topic);
 
-    std::string fullyQualifiedTopic;
-    if (!TopicUtils::FullyQualifiedName(this->Options().Partition(),
-      this->Options().NameSpace(), topic, fullyQualifiedTopic))
+    FullyQualifiedTopic fullyQualifiedTopic(
+      this->Options().Partition(), this->Options().NameSpace(), topic);
+
+    if (!fullyQualifiedTopic.FullTopic())
     {
       std::cerr << "Topic [" << topic << "] is not valid." << std::endl;
       return nullptr;
@@ -184,9 +185,9 @@ namespace gz::transport
     // it will recover the subscription handler associated to the topic and
     // will invoke the callback.
     this->Shared()->localSubscribers.normal.AddHandler(
-      fullyQualifiedTopic, this->NodeUuid(), subscrHandlerPtr);
+      *fullyQualifiedTopic.FullTopic(), this->NodeUuid(), subscrHandlerPtr);
 
-    if (!this->SubscribeHelper(fullyQualifiedTopic))
+    if (!this->SubscribeHelper(*fullyQualifiedTopic.FullTopic()))
       return nullptr;
 
     return subscrHandlerPtr;

src/Node_TEST.cc

@@ -126,12 +126,15 @@ void rawCbInfo(const char *_msgData, const size_t _size,
 
 //////////////////////////////////////////////////
 /// \brief A generic callback.
-void genericCb(const transport::ProtoMsg &_msg)
+void genericCb(const transport::ProtoMsg &_msg,
+               const transport::MessageInfo &_info)
 {
   std::string content;
   ASSERT_TRUE(google::protobuf::TextFormat::PrintToString(_msg, &content));
   EXPECT_TRUE(content.find(std::to_string(data)) != std::string::npos);
   genericCbExecuted = true;
+
+  EXPECT_EQ(_info.Topic().find("@"), std::string::npos);
 }
 
 //////////////////////////////////////////////////

src/SubscriptionHandler.cc

@@ -155,10 +155,10 @@ namespace gz::transport
   /////////////////////////////////////////////////
   void ISubscriptionHandler::CreateGenericZenohSubscriber(
     std::shared_ptr<zenoh::Session> _session,
-    const std::string &_topic)
+    const FullyQualifiedTopic &_fullyQualifiedTopic)
   {
     MessageInfo msgInfo;
-    msgInfo.SetTopic(_topic);
+    msgInfo.SetTopic(_fullyQualifiedTopic.Topic());
     msgInfo.SetType(this->TypeName());
     auto dataHandler = [this, msgInfo](const zenoh::Sample &_sample)
     {
@@ -186,10 +186,11 @@ namespace gz::transport
 
     this->dataPtr->zSub = std::make_unique<zenoh::Subscriber<void>>(
       _session->declare_subscriber(
-        _topic, dataHandler, zenoh::closures::none));
+        *_fullyQualifiedTopic.FullTopic(), dataHandler, zenoh::closures::none));
 
     std::string token = TopicUtils::CreateLivelinessToken(
-      _topic, this->ProcUuid(), this->NodeUuid(), "MS", this->TypeName());
+      *_fullyQualifiedTopic.FullTopic(), this->ProcUuid(), this->NodeUuid(),
+      "MS", this->TypeName());
 
     if (token.empty())
       return;

test/integration/twoProcsPubSub.cc

@@ -79,10 +79,13 @@ void cbInfo(const msgs::Int32 &_msg,
 
 //////////////////////////////////////////////////
 /// \brief A generic callback.
-void genericCb(const transport::ProtoMsg &/*_msg*/)
+void genericCb(const transport::ProtoMsg &/*_msg*/,
+               const transport::MessageInfo &_info)
 {
   genericCbExecuted = true;
   ++counter;
+
+  EXPECT_EQ(_info.Topic().find("@"), std::string::npos);
 }
 
 //////////////////////////////////////////////////
@@ -210,9 +213,10 @@ TEST(twoProcPubSub, PubSubWrongTypesTwoRawSubscribers)
   };
 
   auto genericCb = [&](const char *, const size_t /*_size*/,
-                       const transport::MessageInfo &)
+                       const transport::MessageInfo &_info)
   {
     genericRawCbExecuted = true;
+    EXPECT_EQ(_info.Topic().find("@"), std::string::npos);
   };
 
   transport::Node node1;