diff --git a/src/Discovery.hh b/src/Discovery.hh index efdc4c71c..d9ace3800 100644 --- a/src/Discovery.hh +++ b/src/Discovery.hh @@ -60,12 +60,14 @@ #include #include +#include #include #include #include #include #include #include +#include #include #include @@ -456,6 +458,16 @@ namespace gz // Start the thread that receives discovery information. this->threadReception = std::thread(&Discovery::RecvMessages, this); + + // Request the list of current subscribers so that the cache is + // already complete when the initialization phase finishes. New + // subscriptions are tracked through the SUBSCRIBE announcements. + if constexpr (std::is_same_v) + { + Publisher pub("", "", this->pUuid, "", AdvertiseOptions()); + this->SendMsg( + DestinationType::ALL, msgs::Discovery::SUBSCRIBERS_REQ, pub); + } } #ifdef HAVE_ZENOH @@ -517,9 +529,13 @@ namespace gz /// \sa SetConnectionsCb. /// \sa SetDisconnectionsCb. /// \param[in] _topic Topic name requested. + /// \param[in] _nUuid Node UUID of the subscriber requesting the topic, + /// announced so that other processes can track this subscription. + /// Empty when the request is not tied to a subscription. /// \return True if the method succeeded or false otherwise /// (e.g. if the discovery has not been started). - public: bool Discover(const std::string &_topic) const + public: bool Discover(const std::string &_topic, + const std::string &_nUuid = "") const { DiscoveryCallback cb; bool found; @@ -537,6 +553,7 @@ namespace gz Pub pub; pub.SetTopic(_topic); pub.SetPUuid(this->pUuid); + pub.SetNUuid(_nUuid); // Send a discovery request. this->SendMsg(DestinationType::ALL, msgs::Discovery::SUBSCRIBE, pub); @@ -567,12 +584,35 @@ namespace gz return true; } - /// \brief Send the response to a SUBSCRIBERS_REQ message. - /// \param[in] _pub Information to send. - public: void SendSubscribersRep(const MessagePublisher &_pub) const + /// \brief Send one response of the burst answering a SUBSCRIBERS_REQ + /// message. The burst is a snapshot of all the subscriptions of this + /// process: the receiver knows that the snapshot is complete when + /// _count messages have been received. + /// \param[in] _pub Information to send. An empty topic is used to + /// answer when the process has no subscriptions (_count is zero). + /// \param[in] _count Number of messages in this burst. + /// \param[in] _generation Subscription generation of this process when + /// the snapshot was taken. + public: void SendSubscribersRep(const MessagePublisher &_pub, + const uint32_t _count, + const uint64_t _generation) const { - this->SendMsg( - DestinationType::ALL, msgs::Discovery::SUBSCRIBERS_REP, _pub); + gz::msgs::Discovery discoveryMsg; + discoveryMsg.set_version(this->Version()); + discoveryMsg.set_type(msgs::Discovery::SUBSCRIBERS_REP); + discoveryMsg.set_process_uuid(this->pUuid); + if (!_pub.Topic().empty()) + _pub.FillDiscovery(discoveryMsg); + + auto *snapshot = discoveryMsg.mutable_subscribers_snapshot(); + snapshot->set_count(_count); + snapshot->set_generation(_generation); + + this->SendMulticast(discoveryMsg); + + // Set the RELAY flag in the header and send to the unicast relays. + discoveryMsg.mutable_flags()->set_relay(true); + this->SendUnicast(discoveryMsg); } /// \brief Register a node from this process as a remote subscriber. @@ -813,23 +853,71 @@ namespace gz } /// \brief Get the list of topics currently advertised and subscribed - /// in the network. + /// in the network. The call blocks until every known process has + /// reported a complete snapshot of its subscribers or a short timeout + /// expires. The wait normally finishes in a few milliseconds, when + /// the last snapshot arrives. /// \param[out] _topics List of advertised topics. public: void TopicList(std::vector &_topics) { - if (!this->useZenoh) + [[maybe_unused]] Timestamp requestTime = + std::chrono::steady_clock::now(); + [[maybe_unused]] std::vector knownProcs; + + // Request the list of subscribers. This request is only meaningful + // for message discovery over UDP: the Zenoh backend keeps + // remoteSubscribers updated via liveliness tokens and nothing + // answers this request on the service discovery channel. + if constexpr (std::is_same_v) { - std::lock_guard lock(this->mutex); - this->remoteSubscribers.Clear(); - } + if (!this->useZenoh) + { + { + std::lock_guard lock(this->mutex); + for (const auto &proc : this->activity) + knownProcs.push_back(proc.first); + } - // Request the list of subscribers. - Publisher pub("", "", this->pUuid, "", AdvertiseOptions()); - this->SendMsg( - DestinationType::ALL, msgs::Discovery::SUBSCRIBERS_REQ, pub); + Publisher pub("", "", this->pUuid, "", AdvertiseOptions()); + this->SendMsg( + DestinationType::ALL, msgs::Discovery::SUBSCRIBERS_REQ, pub); + } + } this->WaitForInit(); - std::lock_guard lock(this->mutex); + std::unique_lock lock(this->mutex); + + if constexpr (std::is_same_v) + { + if (!this->useZenoh) + { + // Wait until every process known at request time has reported a + // snapshot after the request or the timeout expires. A lost + // reply or a process running an older version is covered by the + // timeout: the cached information is used instead. A process + // that expired while waiting is not expected to reply. + this->subscribersRepCv.wait_until(lock, + requestTime + + std::chrono::milliseconds(kDefSubscribersRepTimeout), + [&] + { + for (const auto &proc : knownProcs) + { + if (this->activity.find(proc) == this->activity.end()) + continue; + + auto it = this->subscribersSnapshots.find(proc); + if (it == this->subscribersSnapshots.end() || + it->second.completed < requestTime) + { + return false; + } + } + return true; + }); + } + } + this->info.TopicList(_topics); std::vector remoteSubs; @@ -889,6 +977,8 @@ namespace gz { // Remove all the info entries for this process UUID. this->info.DelPublishersByProc(it->first); + this->remoteSubscribers.DelPublishersByProc(it->first); + this->subscribersSnapshots.erase(it->first); uuids.push_back(it->first); @@ -1239,6 +1329,21 @@ namespace gz break; } + // Register the remote subscriber. Subscribers running an older + // version do not announce their node UUID and are only tracked + // through the SUBSCRIBERS_REQ mechanism. + if constexpr (std::is_same_v) + { + if (!msg.sub().n_uuid().empty()) + { + Pub subscriber(recvTopic, "", "", recvPUuid, + msg.sub().n_uuid(), kGenericMessageType, + AdvertiseMessageOptions()); + std::lock_guard lock(this->mutex); + this->remoteSubscribers.AddPublisher(subscriber); + } + } + // Check if at least one of my nodes advertises the topic requested. Addresses_M addresses; { @@ -1284,7 +1389,43 @@ namespace gz { std::lock_guard lock(this->mutex); - this->remoteSubscribers.AddPublisher(publisher); + + if (msg.has_subscribers_snapshot()) + { + const auto &snapshot = msg.subscribers_snapshot(); + auto &progress = this->subscribersSnapshots[recvPUuid]; + + // A new generation starts an authoritative snapshot: + // replace all the entries known for this process. + if (!progress.started || + progress.generation != snapshot.generation()) + { + progress.started = true; + progress.generation = snapshot.generation(); + progress.expected = snapshot.count(); + progress.received = 0; + this->remoteSubscribers.DelPublishersByProc(recvPUuid); + } + + if (!publisher.Topic().empty()) + { + this->remoteSubscribers.AddPublisher(publisher); + ++progress.received; + } + + if (progress.received >= progress.expected) + { + progress.completed = std::chrono::steady_clock::now(); + this->subscribersRepCv.notify_all(); + } + } + else + { + // A process running an older version does not attach the + // snapshot metadata. Its information is merged and its + // completion is covered by the TopicList() timeout. + this->remoteSubscribers.AddPublisher(publisher); + } } break; } @@ -1305,6 +1446,12 @@ namespace gz Pub publisher; publisher.SetFromDiscovery(msg); + { + std::lock_guard lock(this->mutex); + this->remoteSubscribers.DelPublisherByNode( + publisher.Topic(), publisher.PUuid(), publisher.NUuid()); + } + if (unregisterCb) unregisterCb(publisher); @@ -1335,6 +1482,8 @@ namespace gz { std::lock_guard lock(this->mutex); this->info.DelPublishersByProc(recvPUuid); + this->remoteSubscribers.DelPublishersByProc(recvPUuid); + this->subscribersSnapshots.erase(recvPUuid); } break; @@ -1406,6 +1555,7 @@ namespace gz case msgs::Discovery::SUBSCRIBE: { discoveryMsg.mutable_sub()->set_topic(_pub.Topic()); + discoveryMsg.mutable_sub()->set_n_uuid(_pub.NUuid()); break; } case msgs::Discovery::HEARTBEAT: @@ -1637,6 +1787,11 @@ namespace gz /// \sa SetHeartbeatInterval. private: static const unsigned int kDefHeartbeatInterval = 1000; + /// \brief Default maximum time waiting for the subscriber snapshots + /// in TopicList() (ms.). The wait normally finishes much earlier, + /// when every known process has answered. + private: static constexpr unsigned int kDefSubscribersRepTimeout = 100; + /// \brief Default silence interval value (ms.). /// \sa MaxSilenceInterval. /// \sa SetMaxSilenceInterval. @@ -1704,6 +1859,35 @@ namespace gz /// \brief Remote subscribers. private: TopicStorage remoteSubscribers; + /// \brief Progress of the snapshot that a remote process reports in + /// a SUBSCRIBERS_REP burst. + private: struct SubscribersSnapshotProgress + { + /// \brief True when at least one burst has been received. + bool started = false; + + /// \brief Subscription generation of the last burst. + uint64_t generation = 0; + + /// \brief Number of messages expected in the burst. + uint32_t expected = 0; + + /// \brief Number of messages received from the burst. + uint32_t received = 0; + + /// \brief Last time a complete snapshot was received. + Timestamp completed = Timestamp::min(); + }; + + /// \brief Snapshot progress of each remote process, keyed by its + /// process UUID. + private: std::map + subscribersSnapshots; + + /// \brief Condition variable notified every time a remote process + /// completes a subscribers snapshot. + private: mutable std::condition_variable subscribersRepCv; + /// \brief Activity information. Every time there is a message from a /// remote node, its activity information is updated. If we do not hear /// from a node in a while, its entries in 'info' will be invalided. The diff --git a/src/Node.cc b/src/Node.cc index ff150cfb4..17d4fec7e 100644 --- a/src/Node.cc +++ b/src/Node.cc @@ -824,6 +824,23 @@ void Node::TopicList(std::vector &_topics) const this->dataPtr->shared->dataPtr->msgDiscovery->TopicList(allTopics); + // Add the topics subscribed within this process. They are not part of the + // discovery information because a process discards its own discovery + // messages. + { + std::lock_guard lock(this->dataPtr->shared->mutex); + for (const auto &pub : this->dataPtr->shared->localSubscribers.Convert( + this->dataPtr->shared->dataPtr->myAddress, + this->dataPtr->shared->pUuid)) + { + if (std::find(allTopics.begin(), allTopics.end(), pub.Topic()) == + allTopics.end()) + { + allTopics.push_back(pub.Topic()); + } + } + } + for (const auto &fullyQualifiedTopic : allTopics) { std::string partition; @@ -1062,6 +1079,21 @@ bool Node::TopicInfo(const std::string &_topic, convert(subs, _subscribers); } + // Add the subscribers within this process. They are not part of the + // discovery information because a process discards its own discovery + // messages. + for (const auto &pub : this->dataPtr->shared->localSubscribers.Convert( + this->dataPtr->shared->dataPtr->myAddress, + this->dataPtr->shared->pUuid)) + { + if (pub.Topic() == fullyQualifiedTopic && + std::find(_subscribers.begin(), _subscribers.end(), pub) == + _subscribers.end()) + { + _subscribers.push_back(pub); + } + } + return true; } diff --git a/src/NodeShared.cc b/src/NodeShared.cc index 64b192ee6..26acc6462 100644 --- a/src/NodeShared.cc +++ b/src/NodeShared.cc @@ -1215,16 +1215,31 @@ void NodeShared::OnSubscribers() { // Get the list of local subscribers while holding the lock. std::vector pubs; + uint64_t generation; { std::lock_guard lock(this->mutex); pubs = this->localSubscribers.Convert( this->dataPtr->myAddress, this->pUuid); + generation = this->dataPtr->subscriptionsGeneration; } - // Reply to the SUBSCRIBERS_REQ with multiple SUBSCRIBERS_REP. + // Reply to the SUBSCRIBERS_REQ with a snapshot of multiple + // SUBSCRIBERS_REP. An empty snapshot is sent when there are no + // subscriptions, so that the requester can account this process. // Called outside the lock to avoid deadlocks with Discovery::mutex. - for (auto const &publisher : pubs) - this->dataPtr->msgDiscovery->SendSubscribersRep(publisher); + if (pubs.empty()) + { + this->dataPtr->msgDiscovery->SendSubscribersRep( + MessagePublisher(), 0, generation); + } + else + { + for (auto const &publisher : pubs) + { + this->dataPtr->msgDiscovery->SendSubscribersRep( + publisher, static_cast(pubs.size()), generation); + } + } } ////////////////////////////////////////////////// @@ -1960,6 +1975,7 @@ bool NodeShared::Unsubscribe(const std::string &_topic, shouldNotifyPublishers = true; localAddress = this->dataPtr->myAddress; localPUuid = this->pUuid; + ++this->dataPtr->subscriptionsGeneration; } // Notify to the publishers that I am no longer interested in the topic. @@ -2020,13 +2036,15 @@ bool NodeShared::SubscribeHelper(const std::string &_fullyQualifiedTopic, std::lock_guard lk(this->mutex); // Add the topic to the list of subscribed topics (if it was not before). this->dataPtr->topicsSubscribed[_nUuid].insert(_fullyQualifiedTopic); + ++this->dataPtr->subscriptionsGeneration; } - // Discover the list of nodes that publish on the topic. + // Discover the list of nodes that publish on the topic. The node UUID is + // announced so that other processes can track this subscription. std::string impl = this->GzImplementation(); if (impl == "zeromq") { - return this->dataPtr->msgDiscovery->Discover(_fullyQualifiedTopic); + return this->dataPtr->msgDiscovery->Discover(_fullyQualifiedTopic, _nUuid); } return true; } diff --git a/src/NodeSharedPrivate.hh b/src/NodeSharedPrivate.hh index 689103dc1..1910a16f2 100644 --- a/src/NodeSharedPrivate.hh +++ b/src/NodeSharedPrivate.hh @@ -302,6 +302,11 @@ namespace gz::transport /// \brief When true, the reception thread will finish. public: std::atomic exit = false; + /// \brief Subscription generation of this process. Increased on every + /// subscription change, attached to the subscriber snapshots so that + /// remote processes can identify them. + public: std::atomic subscriptionsGeneration = 0; + /// \brief Timeout used for receiving messages (ms.). public: inline static const int Timeout = 250; diff --git a/src/Node_TEST.cc b/src/Node_TEST.cc index 0c5337d55..551cfae59 100644 --- a/src/Node_TEST.cc +++ b/src/Node_TEST.cc @@ -2414,10 +2414,11 @@ TEST(NodeTest, TopicList) // The first TopicList() call might block if the discovery is still // initializing (it may happen if we run this test alone). - // However, the second call should never block. + // However, the second call should finish as soon as all the known + // processes report their subscribers, well below the internal timeout. auto elapsed = end - start; EXPECT_LT(std::chrono::duration_cast - (elapsed).count(), 2); + (elapsed).count(), 50); } ////////////////////////////////////////////////// @@ -2440,6 +2441,46 @@ TEST(NodeTest, TopicListRemap) EXPECT_EQ(g_topic_remap, topics.at(0)); } +////////////////////////////////////////////////// +/// \brief This test creates a node that subscribes to a topic without any +/// publisher. The test verifies that TopicList() includes topics that are +/// only subscribed within this process. +TEST(NodeTest, TopicListSubscriberOnly) +{ + std::vector topics; + transport::Node node; + + EXPECT_TRUE(node.Subscribe(g_topic, cb)); + + node.TopicList(topics); + ASSERT_EQ(1u, topics.size()); + EXPECT_EQ(g_topic, topics.at(0)); + + // After unsubscribing, the topic should not be listed anymore. + EXPECT_TRUE(node.Unsubscribe(g_topic)); + topics.clear(); + node.TopicList(topics); + EXPECT_TRUE(topics.empty()); +} + +////////////////////////////////////////////////// +/// \brief This test creates a node that subscribes to a topic without any +/// publisher. The test verifies that TopicInfo() includes the subscribers +/// from this process. +TEST(NodeTest, TopicInfoSubscriberOnly) +{ + transport::Node node; + + EXPECT_TRUE(node.Subscribe(g_topic, cb)); + + std::vector publishers; + std::vector subscribers; + EXPECT_TRUE(node.TopicInfo(g_topic, publishers, subscribers)); + EXPECT_EQ(publishers.size(), 0u); + ASSERT_EQ(subscribers.size(), 1u); + EXPECT_EQ(subscribers.front().MsgTypeName(), "gz.msgs.Int32"); +} + ////////////////////////////////////////////////// /// \brief This test creates two nodes and advertises some services. The test /// verifies that ServiceList() returns the list of all the services advertised. diff --git a/test/BUILD.bazel b/test/BUILD.bazel index bf4122f36..d4734f501 100644 --- a/test/BUILD.bazel +++ b/test/BUILD.bazel @@ -36,6 +36,7 @@ test_executables = [ "pub_aux", "pub_aux_throttled", "scopedTopicSubscriber_aux", + "subscriberOnly_aux", "twoProcsPublisher_aux", "twoProcsPubSubMixedSubscribers_aux", "twoProcsPubSubSingleSubscriber_aux", @@ -65,6 +66,8 @@ integration_test_sources = [ "scopedTopic.cc", "callback_scope_TEST.cc", "statistics.cc", + "topicListFirstCall.cc", + "topicListStartupSync.cc", "twoProcsPubSub.cc", "twoProcsPubSubStats.cc", "twoProcsSrvCall.cc", @@ -80,6 +83,8 @@ integration_test_sources = [ # Found with # bazel test test:all --runs_per_test 10 flaky_test_srcs = [ + "topicListFirstCall.cc", + "topicListStartupSync.cc", "twoProcsPubSub.cc", "twoProcsSrvCall.cc", "twoProcsSrvCallWithoutInput.cc", @@ -96,6 +101,7 @@ flaky_test_srcs = [ 'PUB_EXE=\\"./test/pub_aux\\"', 'PUB_THROTTLED_EXE=\\"./test/pub_aux_throttled\\"', 'SCOPED_TOPIC_SUBSCRIBER_EXE=\\"./test/scopedTopicSubscriber_aux\\"', + 'SUBSCRIBER_ONLY_EXE=\\"./test/subscriberOnly_aux\\"', 'TWO_PROCS_PUBLISHER_EXE=\\"./test/twoProcsPublisher_aux\\"', 'TWO_PROCS_PUB_SUB_MIXED_SUBSCRIBERS_EXE=\\"./test/twoProcsPubSubMixedSubscribers_aux\\"', 'TWO_PROCS_PUB_SUB_SINGLE_SUBSCRIBER_EXE=\\"./test/twoProcsPubSubSingleSubscriber_aux\\"', diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 0c1bb56e0..cddcdbfb0 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -22,6 +22,7 @@ target_compile_definitions(test_config INTERFACE "PUB_EXE=\"$\"" "PUB_THROTTLED_EXE=\"$\"" "SCOPED_TOPIC_SUBSCRIBER_EXE=\"$\"" + "SUBSCRIBER_ONLY_EXE=\"$\"" "TWO_PROCS_PUBLISHER_EXE=\"$\"" "TWO_PROCS_PUB_SUB_MIXED_SUBSCRIBERS_EXE=\"$\"" "TWO_PROCS_PUB_SUB_SINGLE_SUBSCRIBER_EXE=\"$\"" diff --git a/test/integration/CMakeLists.txt b/test/integration/CMakeLists.txt index 07ad49b43..035d83265 100644 --- a/test/integration/CMakeLists.txt +++ b/test/integration/CMakeLists.txt @@ -5,6 +5,8 @@ set(tests scopedTopic.cc callback_scope_TEST.cc statistics.cc + topicListFirstCall.cc + topicListStartupSync.cc twoProcsPubSub.cc twoProcsPubSubStats.cc twoProcsSrvCall.cc @@ -44,6 +46,7 @@ set(auxiliary_files pub_aux pub_aux_throttled scopedTopicSubscriber_aux + subscriberOnly_aux twoProcsPublisher_aux twoProcsPubSubMixedSubscribers_aux twoProcsPubSubSingleSubscriber_aux diff --git a/test/integration/test_executables/subscriberOnly_aux.cc b/test/integration/test_executables/subscriberOnly_aux.cc new file mode 100644 index 000000000..22a98629a --- /dev/null +++ b/test/integration/test_executables/subscriberOnly_aux.cc @@ -0,0 +1,83 @@ +/* + * Copyright (C) 2026 Open Source Robotics Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * +*/ + +#include + +#include +#include +#include +#include +#include + +#include "gz/transport/Node.hh" + +#include + +#include "test_config.hh" + +using namespace gz; + +static const std::string g_topic = "/subscriber_only"; // NOLINT(*) + +////////////////////////////////////////////////// +/// \brief A callback that is never expected to be executed because nobody +/// publishes on this topic. +void cb(const msgs::Vector3d &) +{ +} + +////////////////////////////////////////////////// +/// \brief Usage: subscriberOnly_aux [topic] [lifetimeSec] +/// [unsubscribeAfterSec] [readyFile]. +/// Subscribe to a topic without any publisher and stay alive for +/// lifetimeSec, giving the test process time to discover this subscription. +/// If unsubscribeAfterSec is positive, unsubscribe after that time while +/// keeping the process alive until lifetimeSec. If readyFile is provided, +/// create that file right after subscribing, so that the test can +/// synchronize without waiting a fixed time. +int main(int argc, char **argv) +{ + if (argc < 2 || argc > 6) + { + std::cerr << "Partition name has not be passed as argument" << std::endl; + return -1; + } + + // Set the partition name for this test. + gz::utils::setenv("GZ_PARTITION", argv[1]); + + const std::string topic = argc > 2 ? argv[2] : g_topic; + const int lifetimeSec = argc > 3 ? std::stoi(argv[3]) : 10; + const int unsubscribeAfterSec = argc > 4 ? std::stoi(argv[4]) : 0; + const std::string readyFile = argc > 5 ? argv[5] : ""; + + transport::Node node; + node.Subscribe(topic, cb); + + if (!readyFile.empty()) + std::ofstream(readyFile) << "ready"; + + int elapsedSec = 0; + if (unsubscribeAfterSec > 0) + { + std::this_thread::sleep_for(std::chrono::seconds(unsubscribeAfterSec)); + node.Unsubscribe(topic); + elapsedSec = unsubscribeAfterSec; + } + + std::this_thread::sleep_for(std::chrono::seconds(lifetimeSec - elapsedSec)); +} diff --git a/test/integration/topicListFirstCall.cc b/test/integration/topicListFirstCall.cc new file mode 100644 index 000000000..e936fd3cf --- /dev/null +++ b/test/integration/topicListFirstCall.cc @@ -0,0 +1,99 @@ +/* + * Copyright (C) 2026 Open Source Robotics Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * +*/ + +#include +#include +#include +#include +#include +#include + +#include "gz/transport/Node.hh" +#include "gz/transport/WaitHelpers.hh" + +#include +#include + +#include "gtest/gtest.h" +#include "test_config.hh" +#include "test_utils.hh" + +using namespace gz; + +static std::string partition; // NOLINT(*) + +////////////////////////////////////////////////// +/// \brief This test spawns a process that only subscribes to a topic. The +/// test verifies that the first TopicList() call of an initialized node +/// already includes the topic, thanks to the subscription announcements and +/// the subscriber snapshots collected before returning. +TEST(topicListFirstCall, SubscriberInFirstCall) +{ + transport::Node node; + + // Let discovery initialize without calling TopicList(). This wait cannot + // poll: any query would exercise the API under test. The value covers + // the two heartbeat initialization phase with margin. + std::this_thread::sleep_for(std::chrono::seconds(4)); + + const std::string readyFile = "subscriberOnly_" + partition + ".ready"; + std::filesystem::remove(readyFile); + auto pi = testing::SubprocessJoinWrapper( + {test_executables::kSubscriberOnly, partition, "/subscriber_only", "12", + "0", readyFile}); + + // Wait until the remote process is subscribed. + ASSERT_TRUE(transport::waitUntil([&readyFile] + { + return std::filesystem::exists(readyFile); + })); + + // The first call should collect the remote subscriber replies. + auto start = std::chrono::steady_clock::now(); + std::vector topics; + node.TopicList(topics); + auto elapsed = std::chrono::steady_clock::now() - start; + + EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/subscriber_only") != + topics.end()); + + // The collection window is bounded. + EXPECT_LT(std::chrono::duration_cast( + elapsed).count(), 2000); + + // The second call should finish as soon as all the known processes + // report their subscribers, well below the internal timeout. + topics.clear(); + auto start2 = std::chrono::steady_clock::now(); + node.TopicList(topics); + auto elapsed2 = std::chrono::steady_clock::now() - start2; + EXPECT_LT(std::chrono::duration_cast( + elapsed2).count(), 50); +} + +////////////////////////////////////////////////// +int main(int argc, char **argv) +{ + // Get a random partition name. + partition = testing::getRandomNumber(); + + // Set the partition name for this process. + gz::utils::setenv("GZ_PARTITION", partition); + + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/integration/topicListStartupSync.cc b/test/integration/topicListStartupSync.cc new file mode 100644 index 000000000..1112e6394 --- /dev/null +++ b/test/integration/topicListStartupSync.cc @@ -0,0 +1,91 @@ +/* + * Copyright (C) 2026 Open Source Robotics Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * +*/ + +#include +#include +#include +#include +#include + +#include "gz/transport/Node.hh" +#include "gz/transport/WaitHelpers.hh" + +#include +#include + +#include "gtest/gtest.h" +#include "test_config.hh" +#include "test_utils.hh" + +using namespace gz; + +static std::string partition; // NOLINT(*) + +////////////////////////////////////////////////// +/// \brief This test spawns a process that only subscribes to a topic before +/// this process starts its discovery. The test verifies that the discovery +/// startup requests the existing subscribers, so the first TopicList() call +/// already includes the topic. This test needs its own process because the +/// discovery of a process only starts once. +TEST(topicListStartupSync, PreexistingSubscriberInFirstCall) +{ + // The remote subscriber exists before this process starts its discovery. + const std::string readyFile = "subscriberOnly_" + partition + ".ready"; + std::filesystem::remove(readyFile); + auto pi = testing::SubprocessJoinWrapper( + {test_executables::kSubscriberOnly, partition, "/subscriber_only", "12", + "0", readyFile}); + + // Wait until the remote process is subscribed. + ASSERT_TRUE(transport::waitUntil([&readyFile] + { + return std::filesystem::exists(readyFile); + })); + + // The first transport node starts the discovery of this process, which + // requests the current subscribers. + transport::Node node; + + std::vector topics; + node.TopicList(topics); + EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/subscriber_only") != + topics.end()); + + // The second call should finish as soon as all the known processes + // report their subscribers, well below the internal timeout. + topics.clear(); + auto start = std::chrono::steady_clock::now(); + node.TopicList(topics); + auto elapsed = std::chrono::steady_clock::now() - start; + EXPECT_TRUE(std::find(topics.begin(), topics.end(), "/subscriber_only") != + topics.end()); + EXPECT_LT(std::chrono::duration_cast( + elapsed).count(), 50); +} + +////////////////////////////////////////////////// +int main(int argc, char **argv) +{ + // Get a random partition name. + partition = testing::getRandomNumber(); + + // Set the partition name for this process. + gz::utils::setenv("GZ_PARTITION", partition); + + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/integration/twoProcsPubSub.cc b/test/integration/twoProcsPubSub.cc index df871f3fe..cbff80a04 100644 --- a/test/integration/twoProcsPubSub.cc +++ b/test/integration/twoProcsPubSub.cc @@ -17,6 +17,7 @@ #include #include +#include #include #include #include @@ -351,7 +352,8 @@ TEST(twoProcPubSub, TopicList) EXPECT_EQ(topics.at(0), g_topic); topics.clear(); - // The second call should never block since discovery already completed. + // The second call should finish as soon as all the known processes + // report their subscribers, well below the internal timeout. auto start2 = std::chrono::steady_clock::now(); node.TopicList(topics); auto end2 = std::chrono::steady_clock::now(); @@ -360,7 +362,78 @@ TEST(twoProcPubSub, TopicList) auto elapsed2 = std::chrono::duration_cast (end2 - start2).count(); - EXPECT_LT(elapsed2, 2); + EXPECT_LT(elapsed2, 50); + + reset(); +} + +////////////////////////////////////////////////// +/// \brief This test spawns a process that only subscribes to a topic, without +/// any publisher involved. The test verifies that a node whose discovery is +/// already initialized eventually reports the topic in TopicList(). +TEST(twoProcPubSub, TopicListSubscriberOnly) +{ + reset(); + + transport::Node node; + std::vector topics; + + // Make sure that discovery is initialized before spawning the subscriber, + // so this test does not benefit from the initialization wait inside the + // first TopicList() call. + node.TopicList(topics); + + auto pi = testing::SubprocessJoinWrapper( + {test_executables::kSubscriberOnly, partition}); + + EXPECT_TRUE(transport::waitForTopic(node, "/subscriber_only", + std::chrono::milliseconds(5000))); + + // The remote subscription should also be visible in TopicInfo(). + std::vector publishers; + std::vector subscribers; + EXPECT_TRUE(node.TopicInfo("/subscriber_only", publishers, subscribers)); + EXPECT_EQ(publishers.size(), 0u); + EXPECT_EQ(subscribers.size(), 1u); + + reset(); +} + +////////////////////////////////////////////////// +/// \brief This test spawns a process that subscribes to a topic and +/// unsubscribes after a while, keeping the process alive. The test verifies +/// that TopicList() stops reporting the topic after the unsubscription. +TEST(twoProcPubSub, TopicListRemovesUnsubscribed) +{ + reset(); + + transport::Node node; + std::vector topics; + + // Make sure that discovery is initialized before spawning the subscriber. + node.TopicList(topics); + + // The remote process subscribes to /unsub_test, unsubscribes after 4 + // seconds and stays alive for 15 seconds. + auto pi = testing::SubprocessJoinWrapper( + {test_executables::kSubscriberOnly, partition, "/unsub_test", "15", "4"}); + + ASSERT_TRUE(transport::waitForTopic(node, "/unsub_test", + std::chrono::milliseconds(4000))); + + // After the remote node unsubscribes, the topic should disappear from + // TopicList() while its process is still alive. Note that the timeout is + // shorter than the remaining lifetime of the remote process, so a topic + // removal triggered by the process termination cannot make this pass. + auto topicAbsent = [&node]() + { + std::vector topicsNow; + node.TopicList(topicsNow); + return std::find(topicsNow.begin(), topicsNow.end(), "/unsub_test") == + topicsNow.end(); + }; + EXPECT_TRUE(transport::waitUntil(topicAbsent, + std::chrono::milliseconds(8000), std::chrono::milliseconds(100))); reset(); } diff --git a/test/test_config.hh.in b/test/test_config.hh.in index 406c9679d..b98683f52 100644 --- a/test/test_config.hh.in +++ b/test/test_config.hh.in @@ -59,6 +59,10 @@ constexpr const char * kPubThrottled = PUB_THROTTLED_EXE; constexpr const char * kScopedTopicSubscriber = SCOPED_TOPIC_SUBSCRIBER_EXE; #endif // SCOPED_TOPIC_SUBSCRIBER_EXE +#ifdef SUBSCRIBER_ONLY_EXE +constexpr const char * kSubscriberOnly = SUBSCRIBER_ONLY_EXE; +#endif // SUBSCRIBER_ONLY_EXE + #ifdef TWO_PROCS_PUBLISHER_EXE constexpr const char * kTwoProcsPublisher = TWO_PROCS_PUBLISHER_EXE; #endif // TWO_PROCS_PUBLISHER_EXE