diff --git a/include/cpp_request/client.hpp b/include/cpp_request/client.hpp index 2628587..483d165 100644 --- a/include/cpp_request/client.hpp +++ b/include/cpp_request/client.hpp @@ -1,6 +1,8 @@ #pragma once #include +#include +#include #include #include @@ -12,6 +14,12 @@ namespace cpp_request { class Client final { public: Client() noexcept = default; + ~Client() noexcept; + + Client(const Client&) = delete; + Client& operator=(const Client&) = delete; + Client(Client&& other) noexcept; + Client& operator=(Client&& other) noexcept; [[nodiscard]] Result request(const Request& request); @@ -41,6 +49,13 @@ class Client final { } private: + void close_reusable_connection() noexcept; + + std::uintptr_t reusable_socket_token_{0}; + bool has_reusable_connection_{false}; + std::string reusable_host_; + std::uint16_t reusable_port_{0}; + std::chrono::milliseconds connect_timeout_{5000}; std::chrono::milliseconds read_timeout_{30000}; std::chrono::milliseconds write_timeout_{30000}; diff --git a/src/core/client.cpp b/src/core/client.cpp index d7aab35..794ede9 100644 --- a/src/core/client.cpp +++ b/src/core/client.cpp @@ -4,16 +4,97 @@ #include "http/response_parser.hpp" #include "net/resolver.hpp" #include "net/tcp_connection.hpp" +#include "platform/native_socket.hpp" #include #include #include +#include #include #include namespace cpp_request { namespace { +[[nodiscard]] constexpr char ascii_lower(char ch) noexcept { + return ch >= 'A' && ch <= 'Z' + ? static_cast(ch + ('a' - 'A')) + : ch; +} + +[[nodiscard]] bool ascii_iequals( + std::string_view lhs, + std::string_view rhs) noexcept { + if (lhs.size() != rhs.size()) { + return false; + } + + for (std::size_t index = 0; index < lhs.size(); ++index) { + if (ascii_lower(lhs[index]) != ascii_lower(rhs[index])) { + return false; + } + } + return true; +} + +[[nodiscard]] std::string_view trim_ows(std::string_view value) noexcept { + while (!value.empty() && (value.front() == ' ' || value.front() == '\t')) { + value.remove_prefix(1); + } + while (!value.empty() && (value.back() == ' ' || value.back() == '\t')) { + value.remove_suffix(1); + } + return value; +} + +[[nodiscard]] bool contains_token( + std::string_view value, + std::string_view expected) noexcept { + std::size_t begin = 0; + while (begin <= value.size()) { + const std::size_t comma = value.find(',', begin); + const std::size_t end = comma == std::string_view::npos + ? value.size() + : comma; + if (ascii_iequals(trim_ows(value.substr(begin, end - begin)), expected)) { + return true; + } + if (comma == std::string_view::npos) { + break; + } + begin = comma + 1; + } + return false; +} + +[[nodiscard]] bool request_wants_connection_close(const Request& request) noexcept { + for (const auto& field : request.headers()) { + if (ascii_iequals(field.name, "Connection") + && contains_token(field.value, "close")) { + return true; + } + } + return false; +} + +[[nodiscard]] bool same_origin( + std::string_view stored_host, + std::uint16_t stored_port, + std::string_view host, + std::uint16_t port) noexcept { + return stored_port == port && ascii_iequals(stored_host, host); +} + +[[nodiscard]] std::uintptr_t encode_socket_handle( + detail::platform::NativeSocketHandle handle) noexcept { + return static_cast(handle); +} + +[[nodiscard]] detail::platform::NativeSocketHandle decode_socket_handle( + std::uintptr_t token) noexcept { + return static_cast(token); +} + [[nodiscard]] std::chrono::milliseconds remaining_timeout( std::chrono::steady_clock::time_point deadline) noexcept { const auto now = std::chrono::steady_clock::now(); @@ -40,6 +121,49 @@ namespace { } // namespace +Client::~Client() noexcept { + close_reusable_connection(); +} + +Client::Client(Client&& other) noexcept + : reusable_socket_token_(std::exchange(other.reusable_socket_token_, 0)), + has_reusable_connection_(std::exchange(other.has_reusable_connection_, false)), + reusable_host_(std::move(other.reusable_host_)), + reusable_port_(std::exchange(other.reusable_port_, 0)), + connect_timeout_(other.connect_timeout_), + read_timeout_(other.read_timeout_), + write_timeout_(other.write_timeout_) {} + +Client& Client::operator=(Client&& other) noexcept { + if (this == &other) { + return *this; + } + + close_reusable_connection(); + + reusable_socket_token_ = std::exchange(other.reusable_socket_token_, 0); + has_reusable_connection_ = std::exchange(other.has_reusable_connection_, false); + reusable_host_ = std::move(other.reusable_host_); + reusable_port_ = std::exchange(other.reusable_port_, 0); + connect_timeout_ = other.connect_timeout_; + read_timeout_ = other.read_timeout_; + write_timeout_ = other.write_timeout_; + return *this; +} + +void Client::close_reusable_connection() noexcept { + if (has_reusable_connection_) { + detail::platform::NativeSocket socket{ + decode_socket_handle(reusable_socket_token_)}; + socket.close(); + } + + reusable_socket_token_ = 0; + has_reusable_connection_ = false; + reusable_host_.clear(); + reusable_port_ = 0; +} + Result Client::request(const Request& request_value) { auto serialized_result = detail::http::serialize_request(request_value); if (!serialized_result) { @@ -47,20 +171,39 @@ Result Client::request(const Request& request_value) { } auto serialized = std::move(serialized_result).value(); - auto endpoints_result = detail::net::Resolver::resolve( - serialized.url.host(), - serialized.url.port()); - if (!endpoints_result) { - return endpoints_result.error(); - } + detail::net::TcpConnection connection; + const bool reuse_existing = has_reusable_connection_ + && same_origin( + reusable_host_, + reusable_port_, + serialized.url.host(), + serialized.url.port()); + + if (reuse_existing) { + connection = detail::net::TcpConnection::adopt_native_handle( + decode_socket_handle(reusable_socket_token_)); + reusable_socket_token_ = 0; + has_reusable_connection_ = false; + reusable_host_.clear(); + reusable_port_ = 0; + } else { + close_reusable_connection(); + + auto endpoints_result = detail::net::Resolver::resolve( + serialized.url.host(), + serialized.url.port()); + if (!endpoints_result) { + return endpoints_result.error(); + } - auto connection_result = detail::net::TcpConnection::connect( - endpoints_result.value(), - connect_timeout_); - if (!connection_result) { - return connection_result.error(); + auto connection_result = detail::net::TcpConnection::connect( + endpoints_result.value(), + connect_timeout_); + if (!connection_result) { + return connection_result.error(); + } + connection = std::move(connection_result).value(); } - auto connection = std::move(connection_result).value(); const auto write_deadline = std::chrono::steady_clock::now() + write_timeout_; @@ -110,7 +253,21 @@ Result Client::request(const Request& request_value) { } } - return parser.take_response(); + const bool reusable = connection.connected() + && parser.connection_reusable() + && parser.pending_bytes().empty() + && !request_wants_connection_close(request_value); + + Response response = parser.take_response(); + + if (reusable) { + reusable_socket_token_ = encode_socket_handle(connection.release_native_handle()); + has_reusable_connection_ = true; + reusable_host_.assign(serialized.url.host()); + reusable_port_ = serialized.url.port(); + } + + return response; } Result Client::get(std::string_view url) { diff --git a/src/net/tcp_connection.hpp b/src/net/tcp_connection.hpp index fc4d196..b34a075 100644 --- a/src/net/tcp_connection.hpp +++ b/src/net/tcp_connection.hpp @@ -26,6 +26,11 @@ class TcpConnection final { const std::vector& endpoints, std::chrono::milliseconds timeout); + [[nodiscard]] static TcpConnection adopt_native_handle( + platform::NativeSocketHandle handle) noexcept { + return TcpConnection{platform::NativeSocket{handle}}; + } + [[nodiscard]] Result write_all( std::string_view data, std::chrono::milliseconds timeout); @@ -43,6 +48,10 @@ class TcpConnection final { return socket_.get(); } + [[nodiscard]] platform::NativeSocketHandle release_native_handle() noexcept { + return socket_.release(); + } + void close() noexcept { socket_.close(); } diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 13f6524..b7289a2 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -10,6 +10,7 @@ add_executable(cpp_request_tests ${CMAKE_CURRENT_SOURCE_DIR}/response_parser_test.cpp ${CMAKE_CURRENT_SOURCE_DIR}/response_upgrade_test.cpp ${CMAKE_CURRENT_SOURCE_DIR}/client_test.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/client_reuse_test.cpp ${CMAKE_CURRENT_SOURCE_DIR}/native_socket_test.cpp ${CMAKE_CURRENT_SOURCE_DIR}/resolver_test.cpp ${CMAKE_CURRENT_SOURCE_DIR}/tcp_connection_test.cpp diff --git a/tests/client_reuse_test.cpp b/tests/client_reuse_test.cpp new file mode 100644 index 0000000..d7512e8 --- /dev/null +++ b/tests/client_reuse_test.cpp @@ -0,0 +1,473 @@ +#include + +#include "platform/native_socket.hpp" +#include "platform/network_runtime.hpp" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef _WIN32 +#include +#else +#include +#include +#include +#endif + +namespace { + +using cpp_request::Client; +using cpp_request::detail::platform::NativeSocket; +using cpp_request::detail::platform::NativeSocketHandle; + +struct LoopbackServer { + NativeSocket listener; + std::uint16_t port{0}; +}; + +LoopbackServer make_loopback_server() { + const auto runtime = cpp_request::detail::platform::ensure_network_runtime(); + if (!runtime.ok) { + return {}; + } + + NativeSocket listener{::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)}; + if (!listener) { + return {}; + } + + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_port = 0; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + + if (::bind( + listener.get(), + reinterpret_cast(&address), + static_cast(sizeof(address))) != 0) { + return {}; + } + + if (::listen(listener.get(), 4) != 0) { + return {}; + } + +#ifdef _WIN32 + int address_size = sizeof(address); +#else + socklen_t address_size = sizeof(address); +#endif + if (::getsockname( + listener.get(), + reinterpret_cast(&address), + &address_size) != 0) { + return {}; + } + + return {std::move(listener), ntohs(address.sin_port)}; +} + +bool wait_readable( + NativeSocketHandle socket, + std::chrono::milliseconds timeout) { + fd_set read_set; + FD_ZERO(&read_set); + FD_SET(socket, &read_set); + + timeval value{}; + value.tv_sec = static_cast(timeout.count() / 1000); + value.tv_usec = static_cast((timeout.count() % 1000) * 1000); + +#ifdef _WIN32 + return ::select(0, &read_set, nullptr, nullptr, &value) > 0; +#else + return ::select(socket + 1, &read_set, nullptr, nullptr, &value) > 0; +#endif +} + +NativeSocket accept_one( + NativeSocketHandle listener, + std::chrono::milliseconds timeout = std::chrono::seconds{4}) { + if (!wait_readable(listener, timeout)) { + return {}; + } + return NativeSocket{::accept(listener, nullptr, nullptr)}; +} + +int receive_native( + NativeSocketHandle socket, + char* buffer, + std::size_t capacity) { +#ifdef _WIN32 + return ::recv(socket, buffer, static_cast(capacity), 0); +#else + return static_cast(::recv(socket, buffer, capacity, 0)); +#endif +} + +int send_native( + NativeSocketHandle socket, + const char* data, + std::size_t size) { +#ifdef _WIN32 + return ::send(socket, data, static_cast(size), 0); +#else + return static_cast(::send(socket, data, size, 0)); +#endif +} + +bool send_all_native(NativeSocketHandle socket, std::string_view data) { + std::size_t written = 0; + while (written < data.size()) { + const int count = send_native( + socket, + data.data() + written, + data.size() - written); + if (count <= 0) { + return false; + } + written += static_cast(count); + } + return true; +} + +std::size_t parse_content_length(std::string_view headers) { + constexpr std::string_view kName = "Content-Length:"; + const std::size_t position = headers.find(kName); + if (position == std::string_view::npos) { + return 0; + } + + std::size_t begin = position + kName.size(); + while (begin < headers.size() && headers[begin] == ' ') { + ++begin; + } + + const std::size_t end = headers.find("\r\n", begin); + const std::string_view value = headers.substr(begin, end - begin); + std::size_t length = 0; + const auto parsed = std::from_chars( + value.data(), + value.data() + value.size(), + length, + 10); + return parsed.ec == std::errc{} && parsed.ptr == value.data() + value.size() + ? length + : 0; +} + +std::string receive_request( + NativeSocketHandle socket, + std::chrono::milliseconds timeout = std::chrono::seconds{3}) { + std::string request; + std::array buffer{}; + + for (;;) { + if (!wait_readable(socket, timeout)) { + return request; + } + + const int count = receive_native(socket, buffer.data(), buffer.size()); + if (count <= 0) { + return request; + } + + request.append(buffer.data(), static_cast(count)); + const std::size_t head_end = request.find("\r\n\r\n"); + if (head_end == std::string::npos) { + continue; + } + + const std::size_t expected_size = head_end + 4 + + parse_content_length(std::string_view{request}.substr(0, head_end + 2)); + if (request.size() >= expected_size) { + return request; + } + } +} + +std::string make_url(std::uint16_t port, std::string_view target = "/") { + return "http://127.0.0.1:" + std::to_string(port) + std::string{target}; +} + +void configure_test_timeouts(Client& client) { + client.set_connect_timeout(std::chrono::seconds{2}); + client.set_write_timeout(std::chrono::seconds{2}); + client.set_read_timeout(std::chrono::seconds{2}); +} + +TEST(ClientReuseTest, ReusesOneSocketForSequentialSameOriginRequests) { + auto server = make_loopback_server(); + ASSERT_TRUE(server.listener); + + std::atomic accepted{0}; + std::string first_request; + std::string second_request; + + std::thread worker([&] { + NativeSocket peer = accept_one(server.listener.get()); + if (!peer) { + return; + } + ++accepted; + + first_request = receive_request(peer.get()); + if (first_request.empty()) { + return; + } + (void)send_all_native( + peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Length: 3\r\n" + "\r\n" + "one"); + + second_request = receive_request(peer.get()); + if (second_request.empty()) { + return; + } + (void)send_all_native( + peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Length: 3\r\n" + "\r\n" + "two"); + }); + + Client client; + configure_test_timeouts(client); + + auto first = client.get(make_url(server.port, "/one")); + auto second = client.get(make_url(server.port, "/two")); + + worker.join(); + + ASSERT_TRUE(first); + ASSERT_TRUE(second); + EXPECT_EQ(first.value().body(), "one"); + EXPECT_EQ(second.value().body(), "two"); + EXPECT_EQ(accepted.load(), 1); + EXPECT_NE(first_request.find("GET /one HTTP/1.1\r\n"), std::string::npos); + EXPECT_NE(second_request.find("GET /two HTTP/1.1\r\n"), std::string::npos); +} + +TEST(ClientReuseTest, ConnectionCloseResponseForcesReconnect) { + auto server = make_loopback_server(); + ASSERT_TRUE(server.listener); + + std::atomic accepted{0}; + std::thread worker([&] { + NativeSocket first_peer = accept_one(server.listener.get()); + if (!first_peer) { + return; + } + ++accepted; + (void)receive_request(first_peer.get()); + (void)send_all_native( + first_peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Length: 3\r\n" + "Connection: close\r\n" + "\r\n" + "one"); + first_peer.close(); + + NativeSocket second_peer = accept_one(server.listener.get()); + if (!second_peer) { + return; + } + ++accepted; + (void)receive_request(second_peer.get()); + (void)send_all_native( + second_peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Length: 3\r\n" + "\r\n" + "two"); + }); + + Client client; + configure_test_timeouts(client); + + auto first = client.get(make_url(server.port, "/one")); + auto second = client.get(make_url(server.port, "/two")); + + worker.join(); + + ASSERT_TRUE(first); + ASSERT_TRUE(second); + EXPECT_EQ(first.value().body(), "one"); + EXPECT_EQ(second.value().body(), "two"); + EXPECT_EQ(accepted.load(), 2); +} + +TEST(ClientReuseTest, CloseDelimitedResponseForcesReconnect) { + auto server = make_loopback_server(); + ASSERT_TRUE(server.listener); + + std::atomic accepted{0}; + std::thread worker([&] { + NativeSocket first_peer = accept_one(server.listener.get()); + if (!first_peer) { + return; + } + ++accepted; + (void)receive_request(first_peer.get()); + (void)send_all_native( + first_peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Type: text/plain\r\n" + "\r\n" + "close-body"); + first_peer.close(); + + NativeSocket second_peer = accept_one(server.listener.get()); + if (!second_peer) { + return; + } + ++accepted; + (void)receive_request(second_peer.get()); + (void)send_all_native( + second_peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Length: 5\r\n" + "\r\n" + "fresh"); + }); + + Client client; + configure_test_timeouts(client); + + auto first = client.get(make_url(server.port, "/close")); + auto second = client.get(make_url(server.port, "/fresh")); + + worker.join(); + + ASSERT_TRUE(first); + ASSERT_TRUE(second); + EXPECT_EQ(first.value().body(), "close-body"); + EXPECT_EQ(second.value().body(), "fresh"); + EXPECT_EQ(accepted.load(), 2); +} + +TEST(ClientReuseTest, DifferentPortReplacesRetainedConnection) { + auto first_server = make_loopback_server(); + auto second_server = make_loopback_server(); + ASSERT_TRUE(first_server.listener); + ASSERT_TRUE(second_server.listener); + + std::thread first_worker([&] { + NativeSocket peer = accept_one(first_server.listener.get()); + if (!peer) { + return; + } + (void)receive_request(peer.get()); + (void)send_all_native( + peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Length: 3\r\n" + "\r\n" + "one"); + (void)wait_readable(peer.get(), std::chrono::seconds{3}); + }); + + std::thread second_worker([&] { + NativeSocket peer = accept_one(second_server.listener.get()); + if (!peer) { + return; + } + (void)receive_request(peer.get()); + (void)send_all_native( + peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Length: 3\r\n" + "\r\n" + "two"); + }); + + Client client; + configure_test_timeouts(client); + + auto first = client.get(make_url(first_server.port)); + auto second = client.get(make_url(second_server.port)); + + first_worker.join(); + second_worker.join(); + + ASSERT_TRUE(first); + ASSERT_TRUE(second); + EXPECT_EQ(first.value().body(), "one"); + EXPECT_EQ(second.value().body(), "two"); +} + +TEST(ClientReuseTest, StaleKeepAliveFailureIsDiscardedWithoutAutomaticRetry) { + auto server = make_loopback_server(); + ASSERT_TRUE(server.listener); + + std::promise stale_closed_promise; + auto stale_closed = stale_closed_promise.get_future(); + + std::thread worker([&] { + NativeSocket stale_peer = accept_one(server.listener.get()); + if (!stale_peer) { + stale_closed_promise.set_value(); + return; + } + + (void)receive_request(stale_peer.get()); + (void)send_all_native( + stale_peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Length: 3\r\n" + "\r\n" + "one"); + stale_peer.close(); + stale_closed_promise.set_value(); + + NativeSocket fresh_peer = accept_one(server.listener.get(), std::chrono::seconds{6}); + if (!fresh_peer) { + return; + } + (void)receive_request(fresh_peer.get()); + (void)send_all_native( + fresh_peer.get(), + "HTTP/1.1 200 OK\r\n" + "Content-Length: 5\r\n" + "\r\n" + "fresh"); + }); + + Client client; + configure_test_timeouts(client); + + auto first = client.get(make_url(server.port, "/first")); + ASSERT_TRUE(first); + EXPECT_EQ(first.value().body(), "one"); + + stale_closed.wait(); + + auto stale_attempt = client.get(make_url(server.port, "/stale")); + EXPECT_FALSE(stale_attempt); + + auto fresh_attempt = client.get(make_url(server.port, "/fresh")); + + worker.join(); + + ASSERT_TRUE(fresh_attempt); + EXPECT_EQ(fresh_attempt.value().body(), "fresh"); +} + +} // namespace