Added discovery regression test [5479] (#542)

* Refs #5453 Adding a test for discovery with thirty participants

* Refs #5453 Updating the test to check for matched readers/writers

* Refs #5453. Added test for multiple participants with multiple topics.

* Refs #5479. Fixing warnings on Mac.

* Refs #5479. Changed test to fail by timeout pattern.

* Refs #5479. Taking removal and unmatching into account.

* Refs #5479. Styling and no copy.

* Refs #5479. Fixing valgrind.

* Refs #5479. Avoiding possible port collisions.

* Refs #5479. Removing debug print methods.

* Refs #5479. Adding debug messages to persistence tests.

* Refs #5479. Closing participants.

* Refs #5479. Adding more info to persistence tests.

* Refs #5479. Fix build errors after rebase.

* Refs #4579. Addressing review comments.

Author: MiguelCompany

test/blackbox/BlackboxTestsDiscovery.cpp

@@ -14,6 +14,7 @@
 
 #include "BlackboxTests.hpp"
 
+#include "PubSubWriterReader.hpp"
 #include "PubSubReader.hpp"
 #include "PubSubWriter.hpp"
 
@@ -481,3 +482,120 @@ TEST(BlackBox, PubSubAsReliableHelloworldUserData)
 
     reader.wait_discovery_result();
 }
+
+//! Tests discovery of 20 participants, having one publisher and one subscriber each
+TEST(Discovery, TwentyParticipants)
+{
+    // Number of participants
+    constexpr size_t n_participants = 20;
+    // Wait time for discovery
+    constexpr unsigned int wait_ms = 20;
+
+    std::vector<std::shared_ptr<PubSubWriterReader<HelloWorldType>>> pubsub;
+    pubsub.reserve(n_participants);
+
+    for (unsigned int i=0; i<n_participants; i++)
+    {
+        pubsub.emplace_back(std::make_shared<PubSubWriterReader<HelloWorldType>>(TEST_TOPIC_NAME));
+    }
+
+    // Initialization of all the participants
+    for (auto& ps : pubsub)
+    {
+        ps->init();
+        ASSERT_EQ(ps->isInitialized(), true);
+    }
+
+    bool all_discovered = false;
+    while (!all_discovered)
+    {
+        all_discovered = true;
+
+        for (auto& ps : pubsub)
+        {
+            if ((ps->get_num_discovered_participants() < n_participants - 1) ||
+                (ps->get_num_discovered_publishers() < n_participants) ||
+                (ps->get_num_discovered_subscribers() < n_participants) ||
+                (ps->get_publication_matched() < n_participants) ||
+                (ps->get_subscription_matched() < n_participants))
+            {
+                all_discovered = false;
+                break;
+            }
+        }
+
+        if (!all_discovered)
+        {
+            // Give some time so that participants can discover each other
+            std::this_thread::sleep_for(std::chrono::milliseconds(wait_ms));
+        }
+    }
+
+    // Test will fail by timeout if a discovery error happens
+    std::cout << "All discovered. Closing participants..." << std::endl;
+    for (auto& ps : pubsub)
+    {
+        ps->destroy();
+    }
+}
+
+//! Regression for ROS2 #280 and #281
+TEST(Discovery, TwentyParticipantsSeveralEndpoints)
+{
+    // Number of participants
+    constexpr size_t n_participants = 20;
+    // Number of endpoints
+    constexpr size_t n_topics = 10;
+    // Total number of discovered endpoints
+    constexpr size_t n_total_endpoints = n_participants * n_topics;
+    // Wait time for discovery
+    constexpr unsigned int wait_ms = 20;
+
+    std::vector<std::shared_ptr<PubSubWriterReader<HelloWorldType>>> pubsub;
+    pubsub.reserve(n_participants);
+
+    for (unsigned int i = 0; i < n_participants; i++)
+    {
+        pubsub.emplace_back(std::make_shared<PubSubWriterReader<HelloWorldType>>(TEST_TOPIC_NAME));
+    }
+
+    // Initialization of all the participants
+    for (auto& ps : pubsub)
+    {
+        ps->init();
+        ASSERT_EQ(ps->isInitialized(), true);
+        ASSERT_TRUE(ps->create_additional_topics(n_topics - 1));
+    }
+
+    bool all_discovered = false;
+    while (!all_discovered)
+    {
+        all_discovered = true;
+
+        for (auto& ps : pubsub)
+        {
+            if ((ps->get_num_discovered_participants() < n_participants - 1) ||
+                (ps->get_num_discovered_publishers() < n_total_endpoints) ||
+                (ps->get_num_discovered_subscribers() < n_total_endpoints) ||
+                (ps->get_publication_matched() < n_total_endpoints) ||
+                (ps->get_subscription_matched() < n_total_endpoints))
+            {
+                all_discovered = false;
+                break;
+            }
+        }
+
+        if (!all_discovered)
+        {
+            // Give some time so that participants can discover each other
+            std::this_thread::sleep_for(std::chrono::milliseconds(wait_ms));
+        }
+    }
+
+    // Test will fail by timeout if a discovery error happens
+    std::cout << "All discovered. Closing participants..." << std::endl;
+    for (auto& ps : pubsub)
+    {
+        ps->destroy();
+    }
+}

test/blackbox/BlackboxTestsPersistence.cpp

@@ -42,7 +42,10 @@ class BlackBoxPersistence : public ::testing::Test
         writer.wait_discovery();
         reader.wait_discovery();
 
+        std::cout << "Discovery finished." << std::endl;
+
         auto data = default_helloworld_data_generator();
+        size_t n_samples = data.size();
         not_received_data.insert(not_received_data.end(), data.begin(), data.end());
 
         reader.expected_data(not_received_data);
@@ -56,21 +59,28 @@ class BlackBoxPersistence : public ::testing::Test
         // Block reader until reception finished or timeout.
         if (seq_check > 0)
         {
+            std::cout << "Reader waiting for sequence " << seq_check << "." << std::endl;
             reader.block_until_seq_number_greater_or_equal({ 0,seq_check });
         }
         else
         {
             if (reliable)
             {
+                std::cout << "Reader waiting for " << n_samples << " samples." << std::endl;
                 reader.block_for_all();
             }
             else
             {
+                std::cout << "Reader waiting for 2 samples." << std::endl;
                 reader.block_for_at_least(2);
             }
         }
 
+        std::cout << "Last received sequence was " << reader.get_last_received_sequence_number() << std::endl;
+
+        std::cout << "Destroying reader..." << std::endl;
         reader.destroy();
+        std::cout << "Destroying writer..." << std::endl;
         writer.destroy();
 
         data = reader.not_received_data();