From dc033adde3cfdfeb83f3b2837c941491b2431fb9 Mon Sep 17 00:00:00 2001 From: "mergify[bot]" <37929162+mergify[bot]@users.noreply.github.com> Date: Thu, 28 Mar 2024 09:37:44 +0100 Subject: [PATCH] TCP first message loss (#4454) (#4560) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Refs #20508: add remaining add_logical_port calls * Refs #20508: Add unittests * Refs #20508: Add blackbox test + uncomment section (commented for testing purposes) * Refs #20508: Add blackbox tests + functional fixes * Refs #20508: Fix tests * Refs #20508: Add cv + fix windows tests * Refs #20508. Use lock_guard where apropiate. * Refs #20508. Proxy is_local_port_opened. * Refs #20508. Refactor wait_logical_port_under_negotiation. * Refs #20508: Apply suggestions * Refs #20508: Fix typo * Refs #20508: Add wait_for_logical_port_negotiation_ms to transport descriptor * Refs #20508: Apply suggestions - firs message loss related * Refs #20508: Apply suggestions - transport descriptor related * Refs #20508: Reuse unused transport descriptor tcp_negotiation_timeout * Refs #20508: Change timeout behavior * Refs #20508: Uncrustify * Refs #20508: Fix xml parser * Refs #20508: Undo fastcdr commit change --------- Signed-off-by: Jesus Perez Signed-off-by: Miguel Company Co-authored-by: Jesús Pérez <78275223+jepemi@users.noreply.github.com> Co-authored-by: Miguel Company --- .../rtps/transport/TCPTransportDescriptor.h | 8 +- include/fastrtps/xmlparser/XMLParserCommon.h | 1 + resources/xsd/fastRTPS_profiles.xsd | 66 +++++------ .../attributes/RTPSParticipantAttributes.cpp | 2 + src/cpp/rtps/transport/TCPChannelResource.cpp | 70 ++++++++++-- src/cpp/rtps/transport/TCPChannelResource.h | 17 +++ .../rtps/transport/TCPTransportInterface.cpp | 106 +++++++++++++----- .../rtps/transport/TCPTransportInterface.h | 14 +++ .../rtps/transport/tcp/RTCPMessageManager.cpp | 3 + src/cpp/rtps/xmlparser/XMLParser.cpp | 12 ++ src/cpp/rtps/xmlparser/XMLParserCommon.cpp | 1 + test/blackbox/api/dds-pim/PubSubReader.hpp | 13 ++- test/blackbox/api/dds-pim/PubSubWriter.hpp | 13 ++- .../api/fastrtps_deprecated/PubSubReader.hpp | 13 ++- .../api/fastrtps_deprecated/PubSubWriter.hpp | 13 ++- .../common/BlackboxTestsTransportTCP.cpp | 105 +++++++++++++++++ .../transport/TCPTransportDescriptor.h | 2 + .../XMLTesterExample_profile.xml | 1 + .../tools/xmlvalidation/all_profile.xml | 2 + test/unittest/transport/TCPv4Tests.cpp | 101 +++++++++++++++++ test/unittest/transport/TCPv6Tests.cpp | 99 ++++++++++++++++ .../transport/mock/MockTCPv4Transport.h | 5 + .../transport/mock/MockTCPv6Transport.h | 5 + test/unittest/xmlparser/XMLParserTests.cpp | 4 + 24 files changed, 599 insertions(+), 77 deletions(-) diff --git a/include/fastdds/rtps/transport/TCPTransportDescriptor.h b/include/fastdds/rtps/transport/TCPTransportDescriptor.h index 013441ef90a..f002ac4bad0 100644 --- a/include/fastdds/rtps/transport/TCPTransportDescriptor.h +++ b/include/fastdds/rtps/transport/TCPTransportDescriptor.h @@ -53,6 +53,8 @@ namespace rtps { * * - \c tls_config: Configuration for TLS. * + * - \c tcp_negotiation_timeout: time to wait for logical port negotiation (in ms). + * * @ingroup TRANSPORT_MODULE */ struct TCPTransportDescriptor : public SocketTransportDescriptor @@ -251,7 +253,11 @@ struct TCPTransportDescriptor : public SocketTransportDescriptor //! Increment between logical ports to try during RTCP negotiation uint16_t logical_port_increment; - FASTDDS_TODO_BEFORE(3, 0, "Eliminate tcp_negotiation_timeout, variable not in use.") + /** + * Time to wait for logical port negotiation (ms). If a logical port is under negotiation, it waits for the + * negotiation to finish up to this timeout before trying to send a message to that port. + * Zero value means no waiting (default). + */ uint32_t tcp_negotiation_timeout; //! Enables the TCP_NODELAY socket option diff --git a/include/fastrtps/xmlparser/XMLParserCommon.h b/include/fastrtps/xmlparser/XMLParserCommon.h index 266e818c526..2ea21c53fcf 100644 --- a/include/fastrtps/xmlparser/XMLParserCommon.h +++ b/include/fastrtps/xmlparser/XMLParserCommon.h @@ -76,6 +76,7 @@ extern const char* CALCULATE_CRC; extern const char* CHECK_CRC; extern const char* KEEP_ALIVE_THREAD; extern const char* ACCEPT_THREAD; +extern const char* TCP_NEGOTIATION_TIMEOUT; extern const char* SEGMENT_SIZE; extern const char* PORT_QUEUE_CAPACITY; extern const char* PORT_OVERFLOW_POLICY; diff --git a/resources/xsd/fastRTPS_profiles.xsd b/resources/xsd/fastRTPS_profiles.xsd index 5dfe86d0968..9b9e596b788 100644 --- a/resources/xsd/fastRTPS_profiles.xsd +++ b/resources/xsd/fastRTPS_profiles.xsd @@ -844,38 +844,39 @@ + ├ transport_id [string], + ├ type [string] ("UDPv4", "UDPv6", "TCPv4", "TCPv6", "SHM"), + ├ sendBufferSize [uint32], + ├ receiveBufferSize [uint32], + ├ maxMessageSize [uint32], + ├ maxInitialPeersRange [uint32], + ├ interfaceWhiteList [0~*], (NOT available for SHM type) + | └ address [ipv4Address|ipv6Address] + ├ TTL [uint8], (ONLY available for UDP type) + ├ non_blocking_send [boolean], (ONLY available for UDP type) + ├ output_port [uint16], (ONLY available for UDP type) + ├ wan_addr [ipv4AddressFormat], (ONLY available for TCPv4 type) + ├ keep_alive_frequency_ms [uint32], (ONLY available for TCP type) + ├ keep_alive_timeout_ms [uint32], (ONLY available for TCP type) + ├ max_logical_port [uint16], (ONLY available for TCP type) + ├ logical_port_range [uint16], (ONLY available for TCP type) + ├ logical_port_increment [uint16], (ONLY available for TCP type) + ├ listening_ports [0~*], (ONLY available for TCP type) + | └ port [uint16] (ONLY available for TCP type) + ├ tls [0~1], (ONLY available for TCP type) + ├ calculate_crc [bool], (ONLY available for TCP type) + ├ check_crc [bool], (ONLY available for TCP type) + ├ enable_tcp_nodelay [bool], (ONLY available for TCP type) + ├ keep_alive_thread [threadSettingsType], (ONLY available for TCP type) + ├ accept_thread [threadSettingsType], (ONLY available for TCP type) + ├ tcp_negotiation_timeout [uint32], (ONLY available for TCP type) + ├ segment_size [uint32], (ONLY available for SHM type) + ├ port_queue_capacity [uint32], (ONLY available for SHM type) + ├ healthy_check_timeout_ms [uint32], (ONLY available for SHM type) + ├ rtps_dump_file [string] (ONLY available for SHM type) + ├ default_reception_threads [threadSettingsType] + ├ reception_threads [receptionThreadsListType] (ONLY available for SHM type) + └ dump_thread [threadSettingsType] (ONLY available for SHM type) --> @@ -931,6 +932,7 @@ + diff --git a/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp b/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp index d941b64ea47..afde927e27c 100644 --- a/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp +++ b/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp @@ -101,6 +101,7 @@ static std::shared_ptr create_tcpv4_tra descriptor->check_crc = false; descriptor->apply_security = false; descriptor->enable_tcp_nodelay = true; + descriptor->tcp_negotiation_timeout = 0; descriptor->default_reception_threads(att.builtin_transports_reception_threads); descriptor->accept_thread = att.builtin_transports_reception_threads; @@ -120,6 +121,7 @@ static std::shared_ptr create_tcpv6_tra descriptor->check_crc = false; descriptor->apply_security = false; descriptor->enable_tcp_nodelay = true; + descriptor->tcp_negotiation_timeout = 0; descriptor->default_reception_threads(att.builtin_transports_reception_threads); descriptor->accept_thread = att.builtin_transports_reception_threads; diff --git a/src/cpp/rtps/transport/TCPChannelResource.cpp b/src/cpp/rtps/transport/TCPChannelResource.cpp index 315e9f6cbe0..2bb1e4b5db6 100644 --- a/src/cpp/rtps/transport/TCPChannelResource.cpp +++ b/src/cpp/rtps/transport/TCPChannelResource.cpp @@ -97,7 +97,7 @@ ResponseCode TCPChannelResource::process_bind_request( void TCPChannelResource::set_all_ports_pending() { - std::unique_lock scopedLock(pending_logical_mutex_); + std::lock_guard scopedLock(pending_logical_mutex_); pending_logical_output_ports_.insert(pending_logical_output_ports_.end(), logical_output_ports_.begin(), logical_output_ports_.end()); @@ -107,24 +107,75 @@ void TCPChannelResource::set_all_ports_pending() bool TCPChannelResource::is_logical_port_opened( uint16_t port) { - std::unique_lock scopedLock(pending_logical_mutex_); + std::lock_guard scopedLock(pending_logical_mutex_); + return is_logical_port_opened_nts(port); +} + +bool TCPChannelResource::is_logical_port_opened_nts( + uint16_t port) +{ return std::find(logical_output_ports_.begin(), logical_output_ports_.end(), port) != logical_output_ports_.end(); } bool TCPChannelResource::is_logical_port_added( uint16_t port) { - std::unique_lock scopedLock(pending_logical_mutex_); + std::lock_guard scopedLock(pending_logical_mutex_); return std::find(logical_output_ports_.begin(), logical_output_ports_.end(), port) != logical_output_ports_.end() || std::find(pending_logical_output_ports_.begin(), pending_logical_output_ports_.end(), port) != pending_logical_output_ports_.end(); } +bool TCPChannelResource::wait_logical_port_under_negotiation( + uint16_t port, + const std::chrono::milliseconds& timeout) +{ + std::unique_lock scopedLock(pending_logical_mutex_); + + // Early return if the port is already opened. + if (is_logical_port_opened_nts(port)) + { + return true; + } + + // Early return if the timeout is 0. + if (timeout == std::chrono::milliseconds(0)) + { + return false; + } + + // The port is under negotiation if it's in the pending list and in the negotiation list. + bool found_in_negotiating_list = negotiating_logical_ports_.end() != std::find_if( + negotiating_logical_ports_.begin(), + negotiating_logical_ports_.end(), + [port](const decltype(negotiating_logical_ports_)::value_type& item) + { + return item.second == port; + }); + + if (found_in_negotiating_list && + pending_logical_output_ports_.end() != std::find( + pending_logical_output_ports_.begin(), + pending_logical_output_ports_.end(), + port)) + { + // Wait for the negotiation to finish. The condition variable might get notified if other logical port is opened. In such case, + // it should wait again with the respective remaining time. + auto wait_predicate = [this, port]() -> bool + { + return is_logical_port_opened_nts(port); + }; + logical_output_ports_updated_cv.wait_for(scopedLock, timeout, wait_predicate); + } + + return is_logical_port_opened_nts(port); +} + void TCPChannelResource::add_logical_port( uint16_t port, RTCPMessageManager* rtcp_manager) { - std::unique_lock scopedLock(pending_logical_mutex_); + std::lock_guard scopedLock(pending_logical_mutex_); // Already opened? if (std::find(logical_output_ports_.begin(), logical_output_ports_.end(), port) == logical_output_ports_.end()) { @@ -150,7 +201,7 @@ void TCPChannelResource::add_logical_port( void TCPChannelResource::send_pending_open_logical_ports( RTCPMessageManager* rtcp_manager) { - std::unique_lock scopedLock(pending_logical_mutex_); + std::lock_guard scopedLock(pending_logical_mutex_); if (!pending_logical_output_ports_.empty()) { for (uint16_t port : pending_logical_output_ports_) @@ -180,6 +231,7 @@ void TCPChannelResource::add_logical_port_response( { pending_logical_output_ports_.erase(portIt); logical_output_ports_.push_back(port); + logical_output_ports_updated_cv.notify_all(); EPROSIMA_LOG_INFO(RTCP, "OpenedLogicalPort: " << port); } else @@ -217,7 +269,7 @@ void TCPChannelResource::prepare_send_check_logical_ports_req( // Don't add ports just tested and already pendings if (p <= max_port && p != closedPort) { - std::unique_lock scopedLock(pending_logical_mutex_); + std::lock_guard scopedLock(pending_logical_mutex_); auto pendingIt = std::find(pending_logical_output_ports_.begin(), pending_logical_output_ports_.end(), p); if (pendingIt == pending_logical_output_ports_.end()) { @@ -233,7 +285,7 @@ void TCPChannelResource::prepare_send_check_logical_ports_req( else { TCPTransactionId id = rtcp_manager->sendCheckLogicalPortsRequest(this, candidatePorts); - std::unique_lock scopedLock(pending_logical_mutex_); + std::lock_guard scopedLock(pending_logical_mutex_); last_checked_logical_port_[id] = candidatePorts.back(); } } @@ -268,7 +320,7 @@ void TCPChannelResource::process_check_logical_ports_response( void TCPChannelResource::set_logical_port_pending( uint16_t port) { - std::unique_lock scopedLock(pending_logical_mutex_); + std::lock_guard scopedLock(pending_logical_mutex_); auto it = std::find(logical_output_ports_.begin(), logical_output_ports_.end(), port); if (it != logical_output_ports_.end()) { @@ -280,7 +332,7 @@ void TCPChannelResource::set_logical_port_pending( bool TCPChannelResource::remove_logical_port( uint16_t port) { - std::unique_lock scopedLock(pending_logical_mutex_); + std::lock_guard scopedLock(pending_logical_mutex_); if (!is_logical_port_added(port)) { return false; diff --git a/src/cpp/rtps/transport/TCPChannelResource.h b/src/cpp/rtps/transport/TCPChannelResource.h index 1f795c37c6e..c0a9e97f7bb 100644 --- a/src/cpp/rtps/transport/TCPChannelResource.h +++ b/src/cpp/rtps/transport/TCPChannelResource.h @@ -70,6 +70,7 @@ class TCPChannelResource : public ChannelResource std::map last_checked_logical_port_; std::vector pending_logical_output_ports_; // Must be accessed after lock pending_logical_mutex_ std::vector logical_output_ports_; + std::condition_variable_any logical_output_ports_updated_cv; std::mutex read_mutex_; std::recursive_mutex pending_logical_mutex_; std::atomic connection_status_; @@ -94,6 +95,19 @@ class TCPChannelResource : public ChannelResource bool is_logical_port_added( uint16_t port); + /** + * This method checks if a logical port is under negotiation. If it is, it waits for the negotiation to finish up to a timeout. + * Independently if being under negotiation or not, it returns true if the port is opened, false otherwise. + * + * @param port The logical port to check. + * @param timeout The maximum time to wait for the negotiation to finish. Zero value means no wait + * + * @return true if the port is opened, false otherwise. + */ + bool wait_logical_port_under_negotiation( + uint16_t port, + const std::chrono::milliseconds& timeout); + bool connection_established() { return connection_status_ == eConnectionStatus::eEstablished; @@ -227,6 +241,9 @@ class TCPChannelResource : public ChannelResource private: + bool is_logical_port_opened_nts( + uint16_t port); + void prepare_send_check_logical_ports_req( uint16_t closedPort, RTCPMessageManager* rtcp_manager); diff --git a/src/cpp/rtps/transport/TCPTransportInterface.cpp b/src/cpp/rtps/transport/TCPTransportInterface.cpp index f9cd60425e3..9e686fcd44b 100644 --- a/src/cpp/rtps/transport/TCPTransportInterface.cpp +++ b/src/cpp/rtps/transport/TCPTransportInterface.cpp @@ -88,10 +88,6 @@ static const int s_default_keep_alive_frequency = 5000; // 5 SECONDS static const int s_default_keep_alive_timeout = 15000; // 15 SECONDS //static const int s_clean_deleted_sockets_pool_timeout = 100; // 100 MILLISECONDS -FASTDDS_TODO_BEFORE(3, 0, - "Eliminate s_default_tcp_negotitation_timeout, variable used to initialize deprecate attribute.") -static const int s_default_tcp_negotitation_timeout = 5000; // 5 Seconds - TCPTransportDescriptor::TCPTransportDescriptor() : SocketTransportDescriptor(s_maximumMessageSize, s_maximumInitialPeersRange) , keep_alive_frequency_ms(s_default_keep_alive_frequency) @@ -99,7 +95,7 @@ TCPTransportDescriptor::TCPTransportDescriptor() , max_logical_port(100) , logical_port_range(20) , logical_port_increment(2) - , tcp_negotiation_timeout(s_default_tcp_negotitation_timeout) + , tcp_negotiation_timeout(0) , enable_tcp_nodelay(false) , wait_for_tcp_negotiation(false) , calculate_crc(true) @@ -732,6 +728,8 @@ bool TCPTransportInterface::OpenOutputChannel( Locator physical_locator = IPLocator::toPhysicalLocator(locator); + std::lock_guard socketsLock(sockets_map_mutex_); + // We try to find a SenderResource that has this locator. // Note: This is done in this level because if we do in NetworkFactory level, we have to mantain what transport // already reuses a SenderResource. @@ -744,7 +742,26 @@ bool TCPTransportInterface::OpenOutputChannel( IPLocator::WanToLanLocator(physical_locator) == tcp_sender_resource->locator()))) { - // If missing, logical port will be added in first send() + // Add logical port to channel if it's not there yet + auto channel_resource = channel_resources_.find(physical_locator); + + // Maybe as WAN? + if (channel_resource == channel_resources_.end() && IPLocator::hasWan(locator)) + { + Locator wan_locator = IPLocator::WanToLanLocator(locator); + channel_resource = channel_resources_.find(IPLocator::toPhysicalLocator(wan_locator)); + } + + if (channel_resource != channel_resources_.end()) + { + channel_resource->second->add_logical_port(logical_port, rtcp_message_manager_.get()); + } + else + { + std::lock_guard channelPendingLock(channel_pending_logical_ports_mutex_); + channel_pending_logical_ports_[physical_locator].insert(logical_port); + } + statistics_info_.add_entry(locator); return true; } @@ -758,7 +775,6 @@ bool TCPTransportInterface::OpenOutputChannel( << IPLocator::getLogicalPort( locator) << ") @ " << IPLocator::to_string(locator)); - std::lock_guard socketsLock(sockets_map_mutex_); auto channel_resource = channel_resources_.find(physical_locator); // Maybe as WAN? @@ -848,6 +864,8 @@ bool TCPTransportInterface::OpenOutputChannel( EPROSIMA_LOG_INFO(OpenOutputChannel, "OpenOutputChannel: [WAIT_CONNECTION] (physical: " << IPLocator::getPhysicalPort(locator) << "; logical: " << IPLocator::getLogicalPort(locator) << ") @ " << IPLocator::to_string(locator)); + std::lock_guard channelPendingLock(channel_pending_logical_ports_mutex_); + channel_pending_logical_ports_[physical_locator].insert(logical_port); } } @@ -1346,7 +1364,7 @@ bool TCPTransportInterface::send( bool success = false; - std::lock_guard scoped_lock(sockets_map_mutex_); + std::unique_lock scoped_lock(sockets_map_mutex_); auto channel_resource = channel_resources_.find(locator); if (channel_resource == channel_resources_.end()) { @@ -1374,31 +1392,42 @@ bool TCPTransportInterface::send( if (channel->is_logical_port_added(logical_port)) { - if (channel->is_logical_port_opened(logical_port)) + // If tcp_negotiation_timeout is setted, wait until logical port is opened or timeout. Negative timeout means + // waiting indefinitely. + if (!channel->is_logical_port_opened(logical_port)) { - TCPHeader tcp_header; - statistics_info_.set_statistics_message_data(remote_locator, send_buffer, send_buffer_size); - fill_rtcp_header(tcp_header, send_buffer, send_buffer_size, logical_port); - + // Logical port might be under negotiation. Wait a little and check again. This prevents from + // losing first messages. + scoped_lock.unlock(); + bool logical_port_opened = channel->wait_logical_port_under_negotiation(logical_port, std::chrono::milliseconds( + configuration()->tcp_negotiation_timeout)); + if (!logical_port_opened) { - asio::error_code ec; - size_t sent = channel->send( - (octet*)&tcp_header, - static_cast(TCPHeader::size()), - send_buffer, - send_buffer_size, - ec); - - if (sent != static_cast(TCPHeader::size() + send_buffer_size) || ec) - { - EPROSIMA_LOG_WARNING(DEBUG, "Failed to send RTCP message (" << sent << " of " << - TCPHeader::size() + send_buffer_size << " b): " << ec.message()); - success = false; - } - else - { - success = true; - } + return success; + } + scoped_lock.lock(); + } + TCPHeader tcp_header; + statistics_info_.set_statistics_message_data(remote_locator, send_buffer, send_buffer_size); + fill_rtcp_header(tcp_header, send_buffer, send_buffer_size, logical_port); + { + asio::error_code ec; + size_t sent = channel->send( + (octet*)&tcp_header, + static_cast(TCPHeader::size()), + send_buffer, + send_buffer_size, + ec); + + if (sent != static_cast(TCPHeader::size() + send_buffer_size) || ec) + { + EPROSIMA_LOG_WARNING(DEBUG, "Failed to send RTCP message (" << sent << " of " << + TCPHeader::size() + send_buffer_size << " b): " << ec.message()); + success = false; + } + else + { + success = true; } } } @@ -1918,6 +1947,21 @@ void TCPTransportInterface::CloseOutputChannel( } } +void TCPTransportInterface::send_channel_pending_logical_ports( + std::shared_ptr& channel) +{ + std::lock_guard channelPendingLock(channel_pending_logical_ports_mutex_); + auto logical_ports = channel_pending_logical_ports_.find(channel->locator()); + if (logical_ports != channel_pending_logical_ports_.end()) + { + for (auto logical_port : logical_ports->second) + { + channel->add_logical_port(logical_port, rtcp_message_manager_.get()); + } + channel_pending_logical_ports_.erase(channel->locator()); + } +} + } // namespace rtps } // namespace fastrtps } // namespace eprosima diff --git a/src/cpp/rtps/transport/TCPTransportInterface.h b/src/cpp/rtps/transport/TCPTransportInterface.h index 316e9dc38f9..a81266af6e8 100644 --- a/src/cpp/rtps/transport/TCPTransportInterface.h +++ b/src/cpp/rtps/transport/TCPTransportInterface.h @@ -121,6 +121,12 @@ class TCPTransportInterface : public TransportInterface eprosima::fastdds::statistics::rtps::OutputTrafficManager statistics_info_; + // Map containging the logical ports that must be added to a channel that has not been created yet. This could happen + // with acceptor channels that are created after their output channel has been opened (LARGE_DATA case). + // The key is the physical locator associated with the sender resource, and later to the channel. + std::map> channel_pending_logical_ports_; + std::mutex channel_pending_logical_ports_mutex_; + TCPTransportInterface( int32_t transport_kind); @@ -496,6 +502,14 @@ class TCPTransportInterface : public TransportInterface SendResourceList& send_resource_list, const LocatorList& remote_participant_locators, const LocatorList& participant_initial_peers) const; + + /** + * Method to add the logical ports associated to a channel that was not available + * when the logical ports were obtained. + * @param channel Channel that might add the logical ports if available. + */ + void send_channel_pending_logical_ports( + std::shared_ptr& channel); }; } // namespace rtps diff --git a/src/cpp/rtps/transport/tcp/RTCPMessageManager.cpp b/src/cpp/rtps/transport/tcp/RTCPMessageManager.cpp index 3bbc00c5bbf..2bf2e828a73 100644 --- a/src/cpp/rtps/transport/tcp/RTCPMessageManager.cpp +++ b/src/cpp/rtps/transport/tcp/RTCPMessageManager.cpp @@ -472,6 +472,9 @@ ResponseCode RTCPMessageManager::processBindConnectionRequest( sendData(channel, BIND_CONNECTION_RESPONSE, transaction_id, &payload, code); + // Add pending logical ports to the channel + mTransport->send_channel_pending_logical_ports(channel); + return RETCODE_OK; } diff --git a/src/cpp/rtps/xmlparser/XMLParser.cpp b/src/cpp/rtps/xmlparser/XMLParser.cpp index c9b7056eee7..c787a88675e 100644 --- a/src/cpp/rtps/xmlparser/XMLParser.cpp +++ b/src/cpp/rtps/xmlparser/XMLParser.cpp @@ -249,6 +249,7 @@ XMLP_ret XMLParser::parseXMLTransportData( + @@ -430,6 +431,7 @@ XMLP_ret XMLParser::validateXMLTransportElements( strcmp(name, KEEP_ALIVE_THREAD) == 0 || strcmp(name, ACCEPT_THREAD) == 0 || strcmp(name, ENABLE_TCP_NODELAY) == 0 || + strcmp(name, TCP_NEGOTIATION_TIMEOUT) == 0 || strcmp(name, TLS) == 0 || strcmp(name, SEGMENT_SIZE) == 0 || strcmp(name, PORT_QUEUE_CAPACITY) == 0 || @@ -763,6 +765,16 @@ XMLP_ret XMLParser::parseXMLCommonTCPTransportData( return XMLP_ret::XML_ERROR; } } + else if (strcmp(name, TCP_NEGOTIATION_TIMEOUT) == 0) + { + // tcp_negotiation_timeout - uint32Type + int iTimeout(0); + if (XMLP_ret::XML_OK != getXMLInt(p_aux0, &iTimeout, 0)) + { + return XMLP_ret::XML_ERROR; + } + pTCPDesc->tcp_negotiation_timeout = static_cast(iTimeout); + } } } else diff --git a/src/cpp/rtps/xmlparser/XMLParserCommon.cpp b/src/cpp/rtps/xmlparser/XMLParserCommon.cpp index 0faa752d9b7..4743f10abf1 100644 --- a/src/cpp/rtps/xmlparser/XMLParserCommon.cpp +++ b/src/cpp/rtps/xmlparser/XMLParserCommon.cpp @@ -63,6 +63,7 @@ const char* CALCULATE_CRC = "calculate_crc"; const char* CHECK_CRC = "check_crc"; const char* KEEP_ALIVE_THREAD = "keep_alive_thread"; const char* ACCEPT_THREAD = "accept_thread"; +const char* TCP_NEGOTIATION_TIMEOUT = "tcp_negotiation_timeout"; const char* SEGMENT_SIZE = "segment_size"; const char* PORT_QUEUE_CAPACITY = "port_queue_capacity"; const char* PORT_OVERFLOW_POLICY = "port_overflow_policy"; diff --git a/test/blackbox/api/dds-pim/PubSubReader.hpp b/test/blackbox/api/dds-pim/PubSubReader.hpp index a14754e2c4d..6c26120cc06 100644 --- a/test/blackbox/api/dds-pim/PubSubReader.hpp +++ b/test/blackbox/api/dds-pim/PubSubReader.hpp @@ -969,7 +969,8 @@ class PubSubReader PubSubReader& setup_large_data_tcp( bool v6 = false, - const uint16_t& port = 0) + const uint16_t& port = 0, + const uint32_t& tcp_negotiation_timeout = 0) { participant_qos_.transport().use_builtin_transports = false; @@ -985,6 +986,11 @@ class PubSubReader auto data_transport = std::make_shared(); data_transport->add_listener_port(tcp_listening_port); + data_transport->calculate_crc = false; + data_transport->check_crc = false; + data_transport->apply_security = false; + data_transport->enable_tcp_nodelay = true; + data_transport->tcp_negotiation_timeout = tcp_negotiation_timeout; participant_qos_.transport().user_transports.push_back(data_transport); } else @@ -994,6 +1000,11 @@ class PubSubReader auto data_transport = std::make_shared(); data_transport->add_listener_port(tcp_listening_port); + data_transport->calculate_crc = false; + data_transport->check_crc = false; + data_transport->apply_security = false; + data_transport->enable_tcp_nodelay = true; + data_transport->tcp_negotiation_timeout = tcp_negotiation_timeout; participant_qos_.transport().user_transports.push_back(data_transport); } diff --git a/test/blackbox/api/dds-pim/PubSubWriter.hpp b/test/blackbox/api/dds-pim/PubSubWriter.hpp index 6fa2e3f5ab4..27c027233a2 100644 --- a/test/blackbox/api/dds-pim/PubSubWriter.hpp +++ b/test/blackbox/api/dds-pim/PubSubWriter.hpp @@ -957,7 +957,8 @@ class PubSubWriter PubSubWriter& setup_large_data_tcp( bool v6 = false, - const uint16_t& port = 0) + const uint16_t& port = 0, + const uint32_t& tcp_negotiation_timeout = 0) { participant_qos_.transport().use_builtin_transports = false; @@ -973,6 +974,11 @@ class PubSubWriter auto data_transport = std::make_shared(); data_transport->add_listener_port(tcp_listening_port); + data_transport->calculate_crc = false; + data_transport->check_crc = false; + data_transport->apply_security = false; + data_transport->enable_tcp_nodelay = true; + data_transport->tcp_negotiation_timeout = tcp_negotiation_timeout; participant_qos_.transport().user_transports.push_back(data_transport); } else @@ -982,6 +988,11 @@ class PubSubWriter auto data_transport = std::make_shared(); data_transport->add_listener_port(tcp_listening_port); + data_transport->calculate_crc = false; + data_transport->check_crc = false; + data_transport->apply_security = false; + data_transport->enable_tcp_nodelay = true; + data_transport->tcp_negotiation_timeout = tcp_negotiation_timeout; participant_qos_.transport().user_transports.push_back(data_transport); } diff --git a/test/blackbox/api/fastrtps_deprecated/PubSubReader.hpp b/test/blackbox/api/fastrtps_deprecated/PubSubReader.hpp index 0c384253bb2..3031ade192f 100644 --- a/test/blackbox/api/fastrtps_deprecated/PubSubReader.hpp +++ b/test/blackbox/api/fastrtps_deprecated/PubSubReader.hpp @@ -745,7 +745,8 @@ class PubSubReader PubSubReader& setup_large_data_tcp( bool v6 = false, - const uint16_t& port = 0) + const uint16_t& port = 0, + const uint32_t& tcp_negotiation_timeout = 0) { participant_attr_.rtps.useBuiltinTransports = false; @@ -761,6 +762,11 @@ class PubSubReader auto data_transport = std::make_shared(); data_transport->add_listener_port(tcp_listening_port); + data_transport->calculate_crc = false; + data_transport->check_crc = false; + data_transport->apply_security = false; + data_transport->enable_tcp_nodelay = true; + data_transport->tcp_negotiation_timeout = tcp_negotiation_timeout; participant_attr_.rtps.userTransports.push_back(data_transport); } else @@ -770,6 +776,11 @@ class PubSubReader auto data_transport = std::make_shared(); data_transport->add_listener_port(tcp_listening_port); + data_transport->calculate_crc = false; + data_transport->check_crc = false; + data_transport->apply_security = false; + data_transport->enable_tcp_nodelay = true; + data_transport->tcp_negotiation_timeout = tcp_negotiation_timeout; participant_attr_.rtps.userTransports.push_back(data_transport); } diff --git a/test/blackbox/api/fastrtps_deprecated/PubSubWriter.hpp b/test/blackbox/api/fastrtps_deprecated/PubSubWriter.hpp index cf7fbe4f3b6..af0070f9a99 100644 --- a/test/blackbox/api/fastrtps_deprecated/PubSubWriter.hpp +++ b/test/blackbox/api/fastrtps_deprecated/PubSubWriter.hpp @@ -756,7 +756,8 @@ class PubSubWriter PubSubWriter& setup_large_data_tcp( bool v6 = false, - const uint16_t& port = 0) + const uint16_t& port = 0, + const uint32_t& tcp_negotiation_timeout = 0) { participant_attr_.rtps.useBuiltinTransports = false; @@ -772,6 +773,11 @@ class PubSubWriter auto data_transport = std::make_shared(); data_transport->add_listener_port(tcp_listening_port); + data_transport->calculate_crc = false; + data_transport->check_crc = false; + data_transport->apply_security = false; + data_transport->enable_tcp_nodelay = true; + data_transport->tcp_negotiation_timeout = tcp_negotiation_timeout; participant_attr_.rtps.userTransports.push_back(data_transport); } else @@ -781,6 +787,11 @@ class PubSubWriter auto data_transport = std::make_shared(); data_transport->add_listener_port(tcp_listening_port); + data_transport->calculate_crc = false; + data_transport->check_crc = false; + data_transport->apply_security = false; + data_transport->enable_tcp_nodelay = true; + data_transport->tcp_negotiation_timeout = tcp_negotiation_timeout; participant_attr_.rtps.userTransports.push_back(data_transport); } diff --git a/test/blackbox/common/BlackboxTestsTransportTCP.cpp b/test/blackbox/common/BlackboxTestsTransportTCP.cpp index ff174a62729..4a5ad0def83 100644 --- a/test/blackbox/common/BlackboxTestsTransportTCP.cpp +++ b/test/blackbox/common/BlackboxTestsTransportTCP.cpp @@ -1133,6 +1133,111 @@ TEST_P(TransportTCP, send_resource_cleanup_initial_peer) client->wait_discovery(2, std::chrono::seconds(0)); } +// Test TCP transport on large message with best effort reliability +TEST_P(TransportTCP, large_message_send_receive) +{ + // Prepare data to be sent before participants discovery so it is ready to be sent as soon as possible. + std::list data; + data = default_data300kb_data_generator(1); + + uint16_t writer_port = global_port; + + /* Test configuration */ + PubSubReader reader(TEST_TOPIC_NAME); + PubSubWriter writer(TEST_TOPIC_NAME); + + std::shared_ptr writer_transport; + std::shared_ptr reader_transport; + Locator_t initialPeerLocator; + if (use_ipv6) + { + reader_transport = std::make_shared(); + writer_transport = std::make_shared(); + initialPeerLocator.kind = LOCATOR_KIND_TCPv6; + IPLocator::setIPv6(initialPeerLocator, "::1"); + } + else + { + reader_transport = std::make_shared(); + writer_transport = std::make_shared(); + initialPeerLocator.kind = LOCATOR_KIND_TCPv4; + IPLocator::setIPv4(initialPeerLocator, 127, 0, 0, 1); + } + writer_transport->tcp_negotiation_timeout = 100; + reader_transport->tcp_negotiation_timeout = 100; + + // Add listener port to server + writer_transport->add_listener_port(writer_port); + + // Add initial peer to client + initialPeerLocator.port = writer_port; + LocatorList_t initial_peer_list; + initial_peer_list.push_back(initialPeerLocator); + + // Setup participants + writer.disable_builtin_transport() + .add_user_transport_to_pparams(writer_transport); + + reader.disable_builtin_transport() + .initial_peers(initial_peer_list) + .add_user_transport_to_pparams(reader_transport); + + // Init participants + writer.init(); + reader.init(); + ASSERT_TRUE(writer.isInitialized()); + ASSERT_TRUE(reader.isInitialized()); + + // Wait for discovery + writer.wait_discovery(1, std::chrono::seconds(0)); + reader.wait_discovery(std::chrono::seconds(0), 1); + + // Send and receive data + reader.startReception(data); + + writer.send(data); + EXPECT_TRUE(data.empty()); + + reader.block_for_all(); +} + +// Test TCP transport on large message with best effort reliability and LARGE_DATA mode +TEST_P(TransportTCP, large_message_large_data_send_receive) +{ + // Prepare data to be sent. before participants discovery so it is ready to be sent as soon as possible. + // The writer might try to send the data before the reader has negotiated the connection. + // If the negotiation timeout is too short, the writer will fail to send the data and the reader will not receive it. + // LARGE_DATA participant discovery is tipically faster than tcp negotiation. + std::list data; + data = default_data300kb_data_generator(1); + + /* Test configuration */ + PubSubReader reader(TEST_TOPIC_NAME); + PubSubWriter writer(TEST_TOPIC_NAME); + + uint32_t tcp_negotiation_timeout = 100; + writer.setup_large_data_tcp(use_ipv6, 0, tcp_negotiation_timeout); + reader.setup_large_data_tcp(use_ipv6, 0, tcp_negotiation_timeout); + + // Init participants + writer.init(); + reader.init(); + ASSERT_TRUE(writer.isInitialized()); + ASSERT_TRUE(reader.isInitialized()); + + // Wait for discovery + writer.wait_discovery(1, std::chrono::seconds(0)); + reader.wait_discovery(std::chrono::seconds(0), 1); + + // Send and receive data + reader.startReception(data); + + writer.send(data); + EXPECT_TRUE(data.empty()); + + reader.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 diff --git a/test/mock/rtps/TCPTransportDescriptor/fastrtps/transport/TCPTransportDescriptor.h b/test/mock/rtps/TCPTransportDescriptor/fastrtps/transport/TCPTransportDescriptor.h index 404c85b727b..4d147b5a611 100644 --- a/test/mock/rtps/TCPTransportDescriptor/fastrtps/transport/TCPTransportDescriptor.h +++ b/test/mock/rtps/TCPTransportDescriptor/fastrtps/transport/TCPTransportDescriptor.h @@ -178,6 +178,8 @@ typedef struct TCPTransportDescriptor : public SocketTransportDescriptor fastdds::rtps::ThreadSettings keep_alive_thread; fastdds::rtps::ThreadSettings accept_thread; + uint32_t tcp_negotiation_timeout; + void add_listener_port( uint16_t port) { diff --git a/test/system/tools/xmlvalidation/XMLTesterExample_profile.xml b/test/system/tools/xmlvalidation/XMLTesterExample_profile.xml index 9fcda9e5a55..608dc76b7c4 100644 --- a/test/system/tools/xmlvalidation/XMLTesterExample_profile.xml +++ b/test/system/tools/xmlvalidation/XMLTesterExample_profile.xml @@ -51,6 +51,7 @@ false false false + 100 diff --git a/test/system/tools/xmlvalidation/all_profile.xml b/test/system/tools/xmlvalidation/all_profile.xml index f90f3907d81..9943e52ccd2 100644 --- a/test/system/tools/xmlvalidation/all_profile.xml +++ b/test/system/tools/xmlvalidation/all_profile.xml @@ -893,6 +893,7 @@ false false false + 100 @@ -943,6 +944,7 @@ false false false + 100 diff --git a/test/unittest/transport/TCPv4Tests.cpp b/test/unittest/transport/TCPv4Tests.cpp index bdaf86b385a..5a5ac22ce2f 100644 --- a/test/unittest/transport/TCPv4Tests.cpp +++ b/test/unittest/transport/TCPv4Tests.cpp @@ -2167,6 +2167,107 @@ TEST_F(TCPv4Tests, remove_from_send_resource_list) } } +// This test verifies the logical port passed to OpenOutputChannel is correctly added to the channel pending list or the +// trasnport's pending channel logical ports map. +TEST_F(TCPv4Tests, add_logical_port_on_send_resource_creation) +{ + eprosima::fastdds::dds::Log::SetVerbosity(eprosima::fastdds::dds::Log::Warning); + + // TCP Client + { + uint16_t port = 12345; + TCPv4TransportDescriptor clientDescriptor; + std::unique_ptr clientTransportUnderTest(new MockTCPv4Transport(clientDescriptor)); + clientTransportUnderTest->init(); + + // Add initial peer to the client + Locator_t initialPeerLocator; + IPLocator::createLocator(LOCATOR_KIND_TCPv4, "127.0.0.1", port, initialPeerLocator); + IPLocator::setLogicalPort(initialPeerLocator, 7410); + + // OpenOutputChannel + SendResourceList client_resource_list; + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, initialPeerLocator)); + IPLocator::setLogicalPort(initialPeerLocator, 7411); + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, initialPeerLocator)); + ASSERT_FALSE(client_resource_list.empty()); + auto channel = clientTransportUnderTest->get_channel_resources().begin()->second; + ASSERT_TRUE(channel->is_logical_port_added(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7411)); + auto channel_pending_logical_ports = clientTransportUnderTest->get_channel_pending_logical_ports(); + ASSERT_TRUE(channel_pending_logical_ports.empty()); + + client_resource_list.clear(); + } + + // TCP Server - LARGE_DATA + { + uint16_t port = 12345; + // Discovered participant physical port has to have a lower value than the listening port to behave as a server + uint16_t participantPhysicalLocator = 12344; + // Create a TCP Server transport + TCPv4TransportDescriptor serverDescriptor; + serverDescriptor.add_listener_port(port); + std::unique_ptr serverTransportUnderTest(new MockTCPv4Transport(serverDescriptor)); + serverTransportUnderTest->init(); + + // Add participant discovered (from UDP discovery for example) + Locator_t discoveredParticipantLocator; + IPLocator::createLocator(LOCATOR_KIND_TCPv4, "127.0.0.1", participantPhysicalLocator, + discoveredParticipantLocator); + IPLocator::setLogicalPort(discoveredParticipantLocator, 7410); + + // OpenOutputChannel + SendResourceList server_resource_list; + ASSERT_TRUE(serverTransportUnderTest->OpenOutputChannel(server_resource_list, discoveredParticipantLocator)); + IPLocator::setLogicalPort(discoveredParticipantLocator, 7411); + ASSERT_TRUE(serverTransportUnderTest->OpenOutputChannel(server_resource_list, discoveredParticipantLocator)); + ASSERT_FALSE(server_resource_list.empty()); + ASSERT_TRUE(serverTransportUnderTest->get_channel_resources().empty()); + auto channel_pending_logical_ports = serverTransportUnderTest->get_channel_pending_logical_ports(); + ASSERT_EQ(channel_pending_logical_ports.size(), 1); + ASSERT_EQ(channel_pending_logical_ports.begin()->second.size(), 2); + ASSERT_TRUE(channel_pending_logical_ports.begin()->second.find( + 7410) != channel_pending_logical_ports.begin()->second.end()); + ASSERT_TRUE(channel_pending_logical_ports.begin()->second.find( + 7411) != channel_pending_logical_ports.begin()->second.end()); + + server_resource_list.clear(); + } + + // TCP Client - LARGE_DATA + { + uint16_t port = 12345; + // Discovered participant physical port has to have a larger value than the listening port to behave as a client + uint16_t participantPhysicalLocator = 12346; + // Create a TCP Client transport + TCPv4TransportDescriptor clientDescriptor; + clientDescriptor.add_listener_port(port); + std::unique_ptr clientTransportUnderTest(new MockTCPv4Transport(clientDescriptor)); + clientTransportUnderTest->init(); + + // Add participant discovered (from UDP discovery for example) + Locator_t discoveredParticipantLocator; + IPLocator::createLocator(LOCATOR_KIND_TCPv4, "127.0.0.1", participantPhysicalLocator, + discoveredParticipantLocator); + IPLocator::setLogicalPort(discoveredParticipantLocator, 7410); + + // OpenOutputChannel + SendResourceList client_resource_list; + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, discoveredParticipantLocator)); + IPLocator::setLogicalPort(discoveredParticipantLocator, 7411); + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, discoveredParticipantLocator)); + ASSERT_FALSE(client_resource_list.empty()); + auto channel = clientTransportUnderTest->get_channel_resources().begin()->second; + ASSERT_TRUE(channel->is_logical_port_added(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7411)); + auto channel_pending_logical_ports = clientTransportUnderTest->get_channel_pending_logical_ports(); + ASSERT_TRUE(channel_pending_logical_ports.empty()); + + client_resource_list.clear(); + } +} + void TCPv4Tests::HELPER_SetDescriptorDefaults() { descriptor.add_listener_port(g_default_port); diff --git a/test/unittest/transport/TCPv6Tests.cpp b/test/unittest/transport/TCPv6Tests.cpp index 0d5303bd5aa..ae7d63e15d2 100644 --- a/test/unittest/transport/TCPv6Tests.cpp +++ b/test/unittest/transport/TCPv6Tests.cpp @@ -569,6 +569,105 @@ TEST_F(TCPv6Tests, remove_from_send_resource_list) ASSERT_EQ(send_resource_list.size(), 1u); } +// This test verifies the logical port passed to OpenOutputChannel is correctly added to the channel pending list or the +// trasnport's pending channel logical ports map. +TEST_F(TCPv6Tests, add_logical_port_on_send_resource_creation) +{ + eprosima::fastdds::dds::Log::SetVerbosity(eprosima::fastdds::dds::Log::Warning); + + // TCP Client + { + uint16_t port = 12345; + TCPv6TransportDescriptor clientDescriptor; + std::unique_ptr clientTransportUnderTest(new MockTCPv6Transport(clientDescriptor)); + clientTransportUnderTest->init(); + + // Add initial peer to the client + Locator_t initialPeerLocator; + IPLocator::createLocator(LOCATOR_KIND_TCPv6, "::1", port, initialPeerLocator); + IPLocator::setLogicalPort(initialPeerLocator, 7410); + + // OpenOutputChannel + SendResourceList client_resource_list; + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, initialPeerLocator)); + IPLocator::setLogicalPort(initialPeerLocator, 7411); + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, initialPeerLocator)); + ASSERT_FALSE(client_resource_list.empty()); + auto channel = clientTransportUnderTest->get_channel_resources().begin()->second; + ASSERT_TRUE(channel->is_logical_port_added(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7411)); + auto channel_pending_logical_ports = clientTransportUnderTest->get_channel_pending_logical_ports(); + ASSERT_TRUE(channel_pending_logical_ports.empty()); + + client_resource_list.clear(); + } + + // TCP Server - LARGE_DATA + { + uint16_t port = 12345; + // Discovered participant physical port has to have a lower value than the listening port to behave as a server + uint16_t participantPhysicalLocator = 12344; + // Create a TCP Server transport + TCPv6TransportDescriptor serverDescriptor; + serverDescriptor.add_listener_port(port); + std::unique_ptr serverTransportUnderTest(new MockTCPv6Transport(serverDescriptor)); + serverTransportUnderTest->init(); + + // Add participant discovered (from UDP discovery for example) + Locator_t discoveredParticipantLocator; + IPLocator::createLocator(LOCATOR_KIND_TCPv6, "::1", participantPhysicalLocator, discoveredParticipantLocator); + IPLocator::setLogicalPort(discoveredParticipantLocator, 7410); + + // OpenOutputChannel + SendResourceList server_resource_list; + ASSERT_TRUE(serverTransportUnderTest->OpenOutputChannel(server_resource_list, discoveredParticipantLocator)); + IPLocator::setLogicalPort(discoveredParticipantLocator, 7411); + ASSERT_TRUE(serverTransportUnderTest->OpenOutputChannel(server_resource_list, discoveredParticipantLocator)); + ASSERT_FALSE(server_resource_list.empty()); + ASSERT_TRUE(serverTransportUnderTest->get_channel_resources().empty()); + auto channel_pending_logical_ports = serverTransportUnderTest->get_channel_pending_logical_ports(); + ASSERT_EQ(channel_pending_logical_ports.size(), 1); + ASSERT_EQ(channel_pending_logical_ports.begin()->second.size(), 2); + ASSERT_TRUE(channel_pending_logical_ports.begin()->second.find( + 7410) != channel_pending_logical_ports.begin()->second.end()); + ASSERT_TRUE(channel_pending_logical_ports.begin()->second.find( + 7411) != channel_pending_logical_ports.begin()->second.end()); + + server_resource_list.clear(); + } + + // TCP Client - LARGE_DATA + { + uint16_t port = 12345; + // Discovered participant physical port has to have a larger value than the listening port to behave as a client + uint16_t participantPhysicalLocator = 12346; + // Create a TCP Client transport + TCPv6TransportDescriptor clientDescriptor; + clientDescriptor.add_listener_port(port); + std::unique_ptr clientTransportUnderTest(new MockTCPv6Transport(clientDescriptor)); + clientTransportUnderTest->init(); + + // Add participant discovered (from UDP discovery for example) + Locator_t discoveredParticipantLocator; + IPLocator::createLocator(LOCATOR_KIND_TCPv6, "::1", participantPhysicalLocator, discoveredParticipantLocator); + IPLocator::setLogicalPort(discoveredParticipantLocator, 7410); + + // OpenOutputChannel + SendResourceList client_resource_list; + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, discoveredParticipantLocator)); + IPLocator::setLogicalPort(discoveredParticipantLocator, 7411); + ASSERT_TRUE(clientTransportUnderTest->OpenOutputChannel(client_resource_list, discoveredParticipantLocator)); + ASSERT_FALSE(client_resource_list.empty()); + auto channel = clientTransportUnderTest->get_channel_resources().begin()->second; + ASSERT_TRUE(channel->is_logical_port_added(7410)); + ASSERT_TRUE(channel->is_logical_port_added(7411)); + auto channel_pending_logical_ports = clientTransportUnderTest->get_channel_pending_logical_ports(); + ASSERT_TRUE(channel_pending_logical_ports.empty()); + + client_resource_list.clear(); + } +} + // TODO: TEST_F(TCPv6Tests, send_and_receive_between_both_secure_ports) // TODO: TEST_F(TCPv6Tests, send_and_receive_between_ports) diff --git a/test/unittest/transport/mock/MockTCPv4Transport.h b/test/unittest/transport/mock/MockTCPv4Transport.h index 08569dbd08a..e2aa8e0e9b1 100644 --- a/test/unittest/transport/mock/MockTCPv4Transport.h +++ b/test/unittest/transport/mock/MockTCPv4Transport.h @@ -65,6 +65,11 @@ class MockTCPv4Transport : public TCPv4Transport return TCPv4Transport::send(send_buffer, send_buffer_size, send_resource_locator, remote_locator); } + const std::map>& get_channel_pending_logical_ports() const + { + return channel_pending_logical_ports_; + } + }; } // namespace rtps diff --git a/test/unittest/transport/mock/MockTCPv6Transport.h b/test/unittest/transport/mock/MockTCPv6Transport.h index 3ecbfaf3db9..cc23f039a5a 100644 --- a/test/unittest/transport/mock/MockTCPv6Transport.h +++ b/test/unittest/transport/mock/MockTCPv6Transport.h @@ -65,6 +65,11 @@ class MockTCPv6Transport : public TCPv6Transport return TCPv6Transport::send(send_buffer, send_buffer_size, send_resource_locator, remote_locator); } + const std::map>& get_channel_pending_logical_ports() const + { + return channel_pending_logical_ports_; + } + }; } // namespace rtps diff --git a/test/unittest/xmlparser/XMLParserTests.cpp b/test/unittest/xmlparser/XMLParserTests.cpp index 6757bfca11e..b298f615780 100644 --- a/test/unittest/xmlparser/XMLParserTests.cpp +++ b/test/unittest/xmlparser/XMLParserTests.cpp @@ -1021,6 +1021,7 @@ TEST_F(XMLParserTests, parseXMLTransportData) false\ false\ false\ + 100\ \ \ 12\ @@ -1088,6 +1089,7 @@ TEST_F(XMLParserTests, parseXMLTransportData) EXPECT_EQ(pTCPv4Desc->listening_ports[1], 5200u); EXPECT_EQ(pTCPv4Desc->keep_alive_thread, modified_thread_settings); EXPECT_EQ(pTCPv4Desc->accept_thread, modified_thread_settings); + EXPECT_EQ(pTCPv4Desc->tcp_negotiation_timeout, 100u); EXPECT_EQ(pTCPv4Desc->default_reception_threads(), modified_thread_settings); EXPECT_EQ(pTCPv4Desc->get_thread_config_for_port(12345), modified_thread_settings); EXPECT_EQ(pTCPv4Desc->get_thread_config_for_port(12346), modified_thread_settings); @@ -1118,6 +1120,7 @@ TEST_F(XMLParserTests, parseXMLTransportData) EXPECT_EQ(pTCPv6Desc->listening_ports[1], 5200u); EXPECT_EQ(pTCPv4Desc->keep_alive_thread, modified_thread_settings); EXPECT_EQ(pTCPv4Desc->accept_thread, modified_thread_settings); + EXPECT_EQ(pTCPv6Desc->tcp_negotiation_timeout, 100u); EXPECT_EQ(pTCPv6Desc->default_reception_threads(), modified_thread_settings); EXPECT_EQ(pTCPv6Desc->get_thread_config_for_port(12345), modified_thread_settings); EXPECT_EQ(pTCPv6Desc->get_thread_config_for_port(12346), modified_thread_settings); @@ -1240,6 +1243,7 @@ TEST_F(XMLParserTests, parseXMLTransportData_NegativeClauses) "tls", "keep_alive_thread", "accept_thread", + "tcp_negotiation_timeout", "default_reception_threads", "reception_threads", "bad_element"