Create InitialConnection for TCP initial peers 2.13 & 2.10 & 2.6 (#4947…

…) (#5003) * Create InitialConnection for TCP initial peers 2.13 & 2.10 & 2.6 (#4947) * Refs #20650: Add test Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> * Refs #20650: Create initial connect for initial peers Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> --------- Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> (cherry picked from commit a0a4fee) # Conflicts: # test/blackbox/common/BlackboxTestsTransportTCP.cpp * Remove TCP lock_guard in disconnection Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> --------- Signed-off-by: cferreiragonz <carlosferreira@eprosima.com> Co-authored-by: Carlos Ferreira González <carlosferreira@eprosima.com>
eProsima · Jul 3, 2024 · 1a278d4 · 1a278d4
1 parent df849e4
commit 1a278d4
Show file tree

Hide file tree

Showing 3 changed files with 73 additions and 4 deletions.
diff --git a/src/cpp/rtps/builtin/discovery/participant/PDPSimple.cpp b/src/cpp/rtps/builtin/discovery/participant/PDPSimple.cpp
@@ -378,7 +378,17 @@ bool PDPSimple::create_dcps_participant_endpoints()
 
     WriterAttributes watt = create_builtin_writer_attributes();
     watt.endpoint.reliabilityKind = BEST_EFFORT;
-    watt.endpoint.remoteLocatorList = m_discovery.initialPeersList;
+    if (!m_discovery.initialPeersList.empty())
+    {
+        if (mp_RTPSParticipant->has_tcp_transports())
+        {
+            mp_RTPSParticipant->create_tcp_connections(m_discovery.initialPeersList);
+        }
+        else
+        {
+            watt.endpoint.remoteLocatorList = m_discovery.initialPeersList;
+        }
+    }
 
     if (mp_RTPSParticipant->getRTPSParticipantAttributes().throughputController.bytesPerPeriod != UINT32_MAX &&
             mp_RTPSParticipant->getRTPSParticipantAttributes().throughputController.periodMillisecs != 0)

diff --git a/src/cpp/rtps/transport/TCPChannelResourceBasic.cpp b/src/cpp/rtps/transport/TCPChannelResourceBasic.cpp
@@ -105,7 +105,6 @@ void TCPChannelResourceBasic::disconnect()
 {
     if (eConnecting < change_status(eConnectionStatus::eDisconnected) && alive())
     {
-        std::lock_guard<std::mutex> read_lock(read_mutex_);
         auto socket = socket_;
 
         std::error_code ec;

diff --git a/test/blackbox/common/BlackboxTestsTransportTCP.cpp b/test/blackbox/common/BlackboxTestsTransportTCP.cpp
@@ -607,7 +607,7 @@ TEST(TransportTCP, Client_reconnection)
     delete requester;
 }
 
-// Test copy constructor and copy assignment for TCPv4
+// Test zero listening port for TCPv4
 TEST_P(TransportTCP, TCPv4_autofill_port)
 {
     PubSubReader<HelloWorldPubSubType> p1(TEST_TOPIC_NAME);
@@ -637,7 +637,7 @@ TEST_P(TransportTCP, TCPv4_autofill_port)
     EXPECT_TRUE(IPLocator::getPhysicalPort(p2_locators.begin()[0]) == port);
 }
 
-// Test copy constructor and copy assignment for TCPv6
+// Test zero listening port for TCPv6
 TEST_P(TransportTCP, TCPv6_autofill_port)
 {
     PubSubReader<HelloWorldPubSubType> p1(TEST_TOPIC_NAME);
@@ -1188,6 +1188,66 @@ TEST_P(TransportTCP, send_resource_cleanup_initial_peer)
     client->wait_discovery(2, std::chrono::seconds(0));
 }
 
+// Test CreateInitialConnection for TCP
+TEST_P(TransportTCP, TCP_initial_peers_connection)
+{
+    PubSubWriter<HelloWorldPubSubType> p1(TEST_TOPIC_NAME);
+    PubSubReader<HelloWorldPubSubType> p2(TEST_TOPIC_NAME);
+    PubSubReader<HelloWorldPubSubType> p3(TEST_TOPIC_NAME);
+
+    // Add TCP Transport with listening port
+    auto p1_transport = std::make_shared<eprosima::fastdds::rtps::TCPv4TransportDescriptor>();
+    p1_transport->add_listener_port(global_port);
+    auto p2_transport = std::make_shared<eprosima::fastdds::rtps::TCPv4TransportDescriptor>();
+    p2_transport->add_listener_port(global_port + 1);
+    auto p3_transport = std::make_shared<eprosima::fastdds::rtps::TCPv4TransportDescriptor>();
+    p3_transport->add_listener_port(global_port - 1);
+
+    // Add initial peer to client
+    Locator_t initialPeerLocator;
+    initialPeerLocator.kind = LOCATOR_KIND_TCPv4;
+    IPLocator::setIPv4(initialPeerLocator, 127, 0, 0, 1);
+    initialPeerLocator.port = global_port;
+    LocatorList_t initial_peer_list;
+    initial_peer_list.push_back(initialPeerLocator);
+
+    // Setup participants
+    p1.disable_builtin_transport()
+            .add_user_transport_to_pparams(p1_transport);
+
+    p2.disable_builtin_transport()
+            .initial_peers(initial_peer_list)
+            .add_user_transport_to_pparams(p2_transport);
+
+    p3.disable_builtin_transport()
+            .initial_peers(initial_peer_list)
+            .add_user_transport_to_pparams(p3_transport);
+
+    // Init participants
+    p1.init();
+    p2.init();
+    p3.init();
+    ASSERT_TRUE(p1.isInitialized());
+    ASSERT_TRUE(p2.isInitialized());
+    ASSERT_TRUE(p3.isInitialized());
+
+    // Wait for discovery
+    p1.wait_discovery(2, std::chrono::seconds(0));
+    p2.wait_discovery(std::chrono::seconds(0), 1);
+    p3.wait_discovery(std::chrono::seconds(0), 1);
+
+    // Send and receive data
+    auto data = default_helloworld_data_generator();
+    p2.startReception(data);
+    p3.startReception(data);
+
+    p1.send(data);
+    EXPECT_TRUE(data.empty());
+
+    p2.block_for_all();
+    p3.block_for_all();
+}
+
 #ifdef INSTANTIATE_TEST_SUITE_P
 #define GTEST_INSTANTIATE_TEST_MACRO(x, y, z, w) INSTANTIATE_TEST_SUITE_P(x, y, z, w)
 #else