diff --git a/include/fastdds/rtps/transport/TCPTransportDescriptor.h b/include/fastdds/rtps/transport/TCPTransportDescriptor.h index 9ceeff84714..99b24a08337 100644 --- a/include/fastdds/rtps/transport/TCPTransportDescriptor.h +++ b/include/fastdds/rtps/transport/TCPTransportDescriptor.h @@ -48,6 +48,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 @@ -242,7 +244,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 07e8b0747de..7ffd2e9133a 100644 --- a/include/fastrtps/xmlparser/XMLParserCommon.h +++ b/include/fastrtps/xmlparser/XMLParserCommon.h @@ -71,6 +71,7 @@ extern const char* METADATA_LOGICAL_PORT; extern const char* LISTENING_PORTS; extern const char* CALCULATE_CRC; extern const char* CHECK_CRC; +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 3ebb04c95aa..dc36d33a52c 100644 --- a/resources/xsd/fastRTPS_profiles.xsd +++ b/resources/xsd/fastRTPS_profiles.xsd @@ -854,6 +854,7 @@ + @@ -1138,4 +1139,3 @@ - diff --git a/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp b/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp index 59610758b10..32f015cad01 100644 --- a/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp +++ b/src/cpp/rtps/attributes/RTPSParticipantAttributes.cpp @@ -98,6 +98,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; return descriptor; } @@ -114,6 +115,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; return descriptor; } diff --git a/src/cpp/rtps/transport/TCPChannelResource.cpp b/src/cpp/rtps/transport/TCPChannelResource.cpp index 8581534c85f..ac6eea25b96 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_) @@ -181,6 +232,7 @@ void TCPChannelResource::add_logical_port_response( pending_logical_output_ports_.erase(portIt); logical_output_ports_.push_back(port); logInfo(RTCP, "OpenedLogicalPort: " << port); + logical_output_ports_updated_cv.notify_all(); } 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 4bceb696bf2..172f99e8ddc 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; @@ -201,6 +215,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 1c2f4227e07..1ce807907a5 100644 --- a/src/cpp/rtps/transport/TCPTransportInterface.cpp +++ b/src/cpp/rtps/transport/TCPTransportInterface.cpp @@ -55,10 +55,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) @@ -66,7 +62,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) @@ -638,6 +634,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. @@ -650,7 +648,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; } @@ -663,7 +680,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? @@ -753,6 +769,8 @@ bool TCPTransportInterface::OpenOutputChannel( logInfo(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); } } @@ -1225,7 +1243,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()) { @@ -1253,31 +1271,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) - { - logWarning(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) + { + logWarning(DEBUG, "Failed to send RTCP message (" << sent << " of " << + TCPHeader::size() + send_buffer_size << " b): " << ec.message()); + success = false; + } + else + { + success = true; } } } @@ -1749,6 +1778,21 @@ void TCPTransportInterface::fill_local_physical_port( } } +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 c56fe0245d3..cc09d73d557 100644 --- a/src/cpp/rtps/transport/TCPTransportInterface.h +++ b/src/cpp/rtps/transport/TCPTransportInterface.h @@ -118,6 +118,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); @@ -459,6 +465,13 @@ class TCPTransportInterface : public TransportInterface return non_blocking_send_; } + /** + * 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 2252c9d594f..d48ea73641e 100644 --- a/src/cpp/rtps/transport/tcp/RTCPMessageManager.cpp +++ b/src/cpp/rtps/transport/tcp/RTCPMessageManager.cpp @@ -467,6 +467,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 8735d56895a..8ba659db413 100644 --- a/src/cpp/rtps/xmlparser/XMLParser.cpp +++ b/src/cpp/rtps/xmlparser/XMLParser.cpp @@ -239,6 +239,7 @@ XMLP_ret XMLParser::parseXMLTransportData( + @@ -498,6 +499,7 @@ XMLP_ret XMLParser::parseXMLCommonTransportData( strcmp(name, LOGICAL_PORT_INCREMENT) == 0 || strcmp(name, LISTENING_PORTS) == 0 || strcmp(name, CALCULATE_CRC) == 0 || strcmp(name, CHECK_CRC) == 0 || strcmp(name, ENABLE_TCP_NODELAY) == 0 || strcmp(name, TLS) == 0 || + strcmp(name, TCP_NEGOTIATION_TIMEOUT) == 0 || strcmp(name, NON_BLOCKING_SEND) == 0 || strcmp(name, SEGMENT_SIZE) == 0 || strcmp(name, PORT_QUEUE_CAPACITY) == 0 || strcmp(name, PORT_OVERFLOW_POLICY) == 0 || strcmp(name, SEGMENT_OVERFLOW_POLICY) == 0 || @@ -653,6 +655,16 @@ XMLP_ret XMLParser::parseXMLCommonTCPTransportData( { // Parsed Outside of this method } + 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 { logError(XMLPARSER, "Invalid element found into 'rtpsTransportDescriptorType'. Name: " << name); diff --git a/src/cpp/rtps/xmlparser/XMLParserCommon.cpp b/src/cpp/rtps/xmlparser/XMLParserCommon.cpp index dd23443a1c0..d4339a436bc 100644 --- a/src/cpp/rtps/xmlparser/XMLParserCommon.cpp +++ b/src/cpp/rtps/xmlparser/XMLParserCommon.cpp @@ -58,6 +58,7 @@ const char* METADATA_LOGICAL_PORT = "metadata_logical_port"; const char* LISTENING_PORTS = "listening_ports"; const char* CALCULATE_CRC = "calculate_crc"; const char* CHECK_CRC = "check_crc"; +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 ce7e695b5f0..51d72642ca3 100644 --- a/test/blackbox/api/dds-pim/PubSubReader.hpp +++ b/test/blackbox/api/dds-pim/PubSubReader.hpp @@ -942,7 +942,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; @@ -958,6 +959,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 @@ -967,6 +973,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 c4b6d948fad..0a0d29395bb 100644 --- a/test/blackbox/api/dds-pim/PubSubWriter.hpp +++ b/test/blackbox/api/dds-pim/PubSubWriter.hpp @@ -907,7 +907,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; @@ -923,6 +924,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 @@ -932,6 +938,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 8c1ae7c5a70..0ad40825f7a 100644 --- a/test/blackbox/api/fastrtps_deprecated/PubSubReader.hpp +++ b/test/blackbox/api/fastrtps_deprecated/PubSubReader.hpp @@ -744,7 +744,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; @@ -760,6 +761,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 @@ -769,6 +775,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 63fcc8c5148..59e7047eb6a 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 7ad0f22238b..50709b4176b 100644 --- a/test/blackbox/common/BlackboxTestsTransportTCP.cpp +++ b/test/blackbox/common/BlackboxTestsTransportTCP.cpp @@ -774,6 +774,111 @@ TEST_P(TransportTCP, large_data_topology) writers.clear(); } +// 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 02a22f32c00..f0044ad2299 100644 --- a/test/mock/rtps/TCPTransportDescriptor/fastrtps/transport/TCPTransportDescriptor.h +++ b/test/mock/rtps/TCPTransportDescriptor/fastrtps/transport/TCPTransportDescriptor.h @@ -171,6 +171,8 @@ typedef struct TCPTransportDescriptor : public SocketTransportDescriptor TLSConfig tls_config; + uint32_t tcp_negotiation_timeout; + void add_listener_port( uint16_t port) { diff --git a/test/unittest/transport/TCPv4Tests.cpp b/test/unittest/transport/TCPv4Tests.cpp index d4edcb9a2c7..5ba5cc5fa83 100644 --- a/test/unittest/transport/TCPv4Tests.cpp +++ b/test/unittest/transport/TCPv4Tests.cpp @@ -1966,6 +1966,107 @@ TEST_F(TCPv4Tests, opening_output_channel_with_same_locator_as_local_listening_p ASSERT_EQ(send_resource_list.size(), 2); } +// 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 e5f74fde5da..6d24f842975 100644 --- a/test/unittest/transport/TCPv6Tests.cpp +++ b/test/unittest/transport/TCPv6Tests.cpp @@ -439,186 +439,108 @@ TEST_F(TCPv6Tests, opening_output_channel_with_same_locator_as_local_listening_p ASSERT_EQ(send_resource_list.size(), 2); } -/* - TEST_F(TCPv6Tests, send_and_receive_between_both_secure_ports) - { - eprosima::fastdds::dds::Log::SetVerbosity(eprosima::fastdds::dds::Log::Kind::Info); - - using TLSOptions = TCPTransportDescriptor::TLSConfig::TLSOptions; - using TLSVerifyMode = TCPTransportDescriptor::TLSConfig::TLSVerifyMode; - - TCPv6TransportDescriptor recvDescriptor; - recvDescriptor.add_listener_port(g_default_port); - recvDescriptor.apply_security = true; - recvDescriptor.tls_config.password = "testkey"; - recvDescriptor.tls_config.cert_chain_file = "mainpubcert.pem"; - recvDescriptor.tls_config.private_key_file = "mainpubkey.pem"; - recvDescriptor.tls_config.verify_file = "maincacert.pem"; - // Server doesn't accept clients without certs - recvDescriptor.tls_config.verify_mode = TLSVerifyMode::VERIFY_PEER | TLSVerifyMode::VERIFY_FAIL_IF_NO_PEER_CERT; - recvDescriptor.tls_config.add_option(TLSOptions::DEFAULT_WORKAROUNDS); - recvDescriptor.tls_config.add_option(TLSOptions::SINGLE_DH_USE); - recvDescriptor.tls_config.add_option(TLSOptions::NO_COMPRESSION); - recvDescriptor.tls_config.add_option(TLSOptions::NO_SSLV2); - recvDescriptor.tls_config.add_option(TLSOptions::NO_SSLV3); - TCPv6Transport receiveTransportUnderTest(recvDescriptor); - receiveTransportUnderTest.init(); - - TCPv6TransportDescriptor sendDescriptor; - sendDescriptor.apply_security = true; - sendDescriptor.tls_config.password = "testkey"; - sendDescriptor.tls_config.cert_chain_file = "mainsubcert.pem"; - sendDescriptor.tls_config.private_key_file = "mainsubkey.pem"; - sendDescriptor.tls_config.verify_file = "maincacert.pem"; - sendDescriptor.tls_config.verify_mode = TLSVerifyMode::VERIFY_PEER; - sendDescriptor.tls_config.add_option(TLSOptions::DEFAULT_WORKAROUNDS); - sendDescriptor.tls_config.add_option(TLSOptions::SINGLE_DH_USE); - sendDescriptor.tls_config.add_option(TLSOptions::NO_COMPRESSION); - sendDescriptor.tls_config.add_option(TLSOptions::NO_SSLV2); - sendDescriptor.tls_config.add_option(TLSOptions::NO_SSLV3); - TCPv6Transport sendTransportUnderTest(sendDescriptor); - sendTransportUnderTest.init(); - - Locator_t inputLocator; - inputLocator.kind = LOCATOR_KIND_TCPv6; - inputLocator.port = g_default_port; - IPLocator::setIPv4(inputLocator, "::1"); - IPLocator::setLogicalPort(inputLocator, 7410); - - Locator_t outputLocator; - outputLocator.kind = LOCATOR_KIND_TCPv6; - IPLocator::setIPv4(outputLocator, "::1"); - outputLocator.port = g_default_port; - IPLocator::setLogicalPort(outputLocator, 7410); +// 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 { - MockReceiverResource receiver(receiveTransportUnderTest, inputLocator); - MockMessageReceiver *msg_recv = dynamic_cast(receiver.CreateMessageReceiver()); - ASSERT_TRUE(receiveTransportUnderTest.IsInputChannelOpen(inputLocator)); - - ASSERT_TRUE(sendTransportUnderTest.OpenOutputChannel(outputLocator)); - octet message[5] = { 'H','e','l','l','o' }; - - Semaphore sem; - std::function recCallback = [&]() - { - EXPECT_EQ(memcmp(message, msg_recv->data, 5), 0); - sem.post(); - }; - - msg_recv->setCallback(recCallback); - - auto sendThreadFunction = [&]() - { - bool sent = sendTransportUnderTest.send(message, 5, outputLocator, inputLocator); - while (!sent) - { - sent = sendTransportUnderTest.send(message, 5, outputLocator, inputLocator); - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - } - EXPECT_TRUE(sent); - //EXPECT_TRUE(transportUnderTest.send(message, 5, outputLocator, inputLocator)); - }; - - senderThread.reset(new std::thread(sendThreadFunction)); - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - senderThread->join(); - sem.wait(); + 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(); } - ASSERT_TRUE(sendTransportUnderTest.CloseOutputChannel(outputLocator)); - } - */ -// TODO SKIP AT THIS MOMENT -/* - TEST_F(TCPv6Tests, send_and_receive_between_ports) - { - descriptor.listening_ports.push_back(g_default_port); - TCPv6Transport transportUnderTest(descriptor); - transportUnderTest.init(); - - Locator_t localLocator; - localLocator.port = g_default_port; - localLocator.kind = LOCATOR_KIND_TCPv6; - IPLocator::setIPv6(localLocator, "::1"); - - Locator_t outputChannelLocator; - outputChannelLocator = g_default_port; - outputChannelLocator.kind = LOCATOR_KIND_TCPv6; - IPLocator::setIPv6(outputChannelLocator, "::1"); - - MockReceiverResource receiver(transportUnderTest, localLocator); - MockMessageReceiver *msg_recv = dynamic_cast(receiver.CreateMessageReceiver()); - - ASSERT_TRUE(transportUnderTest.OpenOutputChannel(outputChannelLocator)); // Includes loopback - ASSERT_TRUE(transportUnderTest.IsInputChannelOpen(localLocator)); - octet message[5] = { 'H','e','l','l','o' }; - std::this_thread::sleep_for(std::chrono::seconds(5)); - - Semaphore sem; - std::function recCallback = [&]() + // TCP Server - LARGE_DATA { - EXPECT_EQ(memcmp(message,msg_recv->data,5), 0); - sem.post(); - }; - - msg_recv->setCallback(recCallback); - - auto sendThreadFunction = [&]() - { - EXPECT_TRUE(transportUnderTest.send(message, 5, outputChannelLocator, localLocator)); - }; - - senderThread.reset(new std::thread(sendThreadFunction)); - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - senderThread->join(); - sem.wait(); - ASSERT_TRUE(transportUnderTest.CloseOutputChannel(outputChannelLocator)); - } - - TEST_F(TCPv6Tests, send_to_loopback) - { - TCPv6Transport transportUnderTest(descriptor); - transportUnderTest.init(); - - Locator_t multicastLocator; - multicastLocator.set_port(g_default_port); - multicastLocator.kind = LOCATOR_KIND_TCPv6; - IPLocator::setIPv6(multicastLocator, 0xff31, 0, 0, 0, 0, 0, 0, 0); - - Locator_t outputChannelLocator; - outputChannelLocator.set_port(g_default_port + 1); - outputChannelLocator.kind = LOCATOR_KIND_TCPv6; - IPLocator::setIPv6(outputChannelLocator, 0,0,0,0,0,0,0,1); // Loopback - - MockReceiverResource receiver(transportUnderTest, multicastLocator); - MockMessageReceiver *msg_recv = dynamic_cast(receiver.CreateMessageReceiver()); - - ASSERT_TRUE(transportUnderTest.OpenOutputChannel(outputChannelLocator)); - ASSERT_TRUE(transportUnderTest.IsInputChannelOpen(multicastLocator)); - octet message[5] = { 'H','e','l','l','o' }; + 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(); + } - Semaphore sem; - std::function recCallback = [&]() + // TCP Client - LARGE_DATA { - EXPECT_EQ(memcmp(message,msg_recv->data,5), 0); - sem.post(); - }; + 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(); + } +} - msg_recv->setCallback(recCallback); +// TODO: TEST_F(TCPv6Tests, send_and_receive_between_both_secure_ports) +// TODO: TEST_F(TCPv6Tests, send_and_receive_between_ports) - auto sendThreadFunction = [&]() - { - EXPECT_TRUE(transportUnderTest.send(message, 5, outputChannelLocator, multicastLocator)); - }; - - senderThread.reset(new std::thread(sendThreadFunction)); - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - senderThread->join(); - sem.wait(); - ASSERT_TRUE(transportUnderTest.CloseOutputChannel(outputChannelLocator)); - } - */ #endif // ifndef __APPLE__ void TCPv6Tests::HELPER_SetDescriptorDefaults() 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 4a3c7bac43b..8ac9da73b9d 100644 --- a/test/unittest/transport/mock/MockTCPv6Transport.h +++ b/test/unittest/transport/mock/MockTCPv6Transport.h @@ -55,6 +55,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 b27b5764bff..0decf6a160e 100644 --- a/test/unittest/xmlparser/XMLParserTests.cpp +++ b/test/unittest/xmlparser/XMLParserTests.cpp @@ -681,6 +681,7 @@ TEST_F(XMLParserTests, parseXMLTransportData) false\ false\ false\ + 100\ \ \ "; @@ -714,6 +715,7 @@ TEST_F(XMLParserTests, parseXMLTransportData) EXPECT_EQ(pTCPv4Desc->logical_port_increment, 2u); EXPECT_EQ(pTCPv4Desc->listening_ports[0], 5100u); EXPECT_EQ(pTCPv4Desc->listening_ports[1], 5200u); + EXPECT_EQ(pTCPv4Desc->tcp_negotiation_timeout, 100u); xmlparser::XMLProfileManager::DeleteInstance(); // TCPv6 @@ -739,6 +741,7 @@ TEST_F(XMLParserTests, parseXMLTransportData) EXPECT_EQ(pTCPv6Desc->logical_port_increment, 2u); EXPECT_EQ(pTCPv6Desc->listening_ports[0], 5100u); EXPECT_EQ(pTCPv6Desc->listening_ports[1], 5200u); + EXPECT_EQ(pTCPv6Desc->tcp_negotiation_timeout, 100u); xmlparser::XMLProfileManager::DeleteInstance(); } @@ -825,6 +828,7 @@ TEST_F(XMLParserTests, parseXMLTransportData_NegativeClauses) "check_crc", "enable_tcp_nodelay", "tls", + "tcp_negotiation_timeout", "bad_element" };