From 3be67623fa530ee2677c72cadd40ade3424eedeb Mon Sep 17 00:00:00 2001 From: eidheim Date: Tue, 14 Jan 2020 14:32:36 +0100 Subject: [PATCH] Client: async request handlers are now called multiple times if response content is larger than Config::max_response_streambuf_size. Also improved streambuf to streambuf copy, and streambuf to string copy. Finally, string() functions no longer consume streambuf. --- client_http.hpp | 309 ++++++++++++++++++++++++++----------------- client_https.hpp | 2 +- server_http.hpp | 34 ++--- tests/io_test.cpp | 84 ++++++++++++ tests/parse_test.cpp | 2 +- 5 files changed, 285 insertions(+), 146 deletions(-) diff --git a/client_http.hpp b/client_http.hpp index bcae379..c2591ca 100644 --- a/client_http.hpp +++ b/client_http.hpp @@ -73,39 +73,59 @@ namespace SimpleWeb { std::size_t size() noexcept { return streambuf.size(); } - /// Convenience function to return content as a string. The stream buffer is consumed. + /// Convenience function to return content as a string. std::string string() noexcept { - try { - std::string str; - auto size = streambuf.size(); - str.resize(size); - read(&str[0], static_cast(size)); - return str; - } - catch(...) { - return std::string(); - } + return std::string(asio::buffers_begin(streambuf.data()), asio::buffers_end(streambuf.data())); } + /// When true, this is the last response content part from server for the current request. + bool end = true; + private: asio::streambuf &streambuf; Content(asio::streambuf &streambuf) noexcept : std::istream(&streambuf), streambuf(streambuf) {} }; + protected: + class Connection; + + public: class Response { friend class ClientBase; friend class Client; + class Shared { + public: + std::string http_version, status_code; + + CaseInsensitiveMultimap header; + }; + asio::streambuf streambuf; - Response(std::size_t max_response_streambuf_size) noexcept : streambuf(max_response_streambuf_size), content(streambuf) {} + std::shared_ptr shared; + + std::weak_ptr connection_weak; + + Response(std::size_t max_response_streambuf_size, const std::shared_ptr &connection_) noexcept + : streambuf(max_response_streambuf_size), shared(new Shared()), connection_weak(connection_), http_version(shared->http_version), status_code(shared->status_code), header(shared->header), content(streambuf) {} + + /// Constructs a response object that has empty content, but otherwise is equal to the response parameter + Response(const Response &response) noexcept + : streambuf(response.streambuf.max_size()), shared(response.shared), connection_weak(response.connection_weak), http_version(shared->http_version), status_code(shared->status_code), header(shared->header), content(streambuf) {} public: - std::string http_version, status_code; + std::string &http_version, &status_code; + + CaseInsensitiveMultimap &header; Content content; - CaseInsensitiveMultimap header; + /// Closes the connection to the server, preventing further response content parts from server. + void close() noexcept { + if(auto connection = this->connection_weak.lock()) + connection->close(); + } }; class Config { @@ -138,7 +158,6 @@ namespace SimpleWeb { std::unique_ptr socket; // Socket must be unique_ptr since asio::ssl::stream is not movable bool in_use = false; - bool event_stream = false; bool attempt_reconnect = true; std::unique_ptr timer; @@ -180,7 +199,7 @@ namespace SimpleWeb { class Session { public: Session(std::size_t max_response_streambuf_size, std::shared_ptr connection_, std::unique_ptr request_streambuf_) noexcept - : connection(std::move(connection_)), request_streambuf(std::move(request_streambuf_)), response(new Response(max_response_streambuf_size)) {} + : connection(std::move(connection_)), request_streambuf(std::move(request_streambuf_)), response(new Response(max_response_streambuf_size, connection)) {} std::shared_ptr connection; std::unique_ptr request_streambuf; @@ -203,7 +222,17 @@ namespace SimpleWeb { std::shared_ptr response; error_code ec; request(method, path, content, header, [&response, &ec](std::shared_ptr response_, const error_code &ec_) { - response = response_; + if(!response) + response = response_; + else { + if(response_->streambuf.size() + response->streambuf.size() > response->streambuf.max_size()) + throw make_error_code::make_error_code(errc::message_size); + // Move partial response_ content to response: + auto &source = response_->streambuf; + auto &target = response->streambuf; + target.commit(asio::buffer_copy(target.prepare(source.size()), source.data())); + source.consume(source.size()); + } ec = ec_; }); @@ -211,7 +240,12 @@ namespace SimpleWeb { LockGuard lock(concurrent_synchronous_requests_mutex); ++concurrent_synchronous_requests; } - io_service->run(); + try { + io_service->run(); + } + catch(const error_code &ec_) { + ec = ec_; + } { LockGuard lock(concurrent_synchronous_requests_mutex); --concurrent_synchronous_requests; @@ -233,7 +267,17 @@ namespace SimpleWeb { std::shared_ptr response; error_code ec; request(method, path, content, header, [&response, &ec](std::shared_ptr response_, const error_code &ec_) { - response = response_; + if(!response) + response = response_; + else { + if(response_->streambuf.size() + response->streambuf.size() > response->streambuf.max_size()) + throw make_error_code::make_error_code(errc::message_size); + // Move partial response_ content to response: + auto &source = response_->streambuf; + auto &target = response->streambuf; + target.commit(asio::buffer_copy(target.prepare(source.size()), source.data())); + source.consume(source.size()); + } ec = ec_; }); @@ -241,7 +285,12 @@ namespace SimpleWeb { LockGuard lock(concurrent_synchronous_requests_mutex); ++concurrent_synchronous_requests; } - io_service->run(); + try { + io_service->run(); + } + catch(const error_code &ec_) { + ec = ec_; + } { LockGuard lock(concurrent_synchronous_requests_mutex); --concurrent_synchronous_requests; @@ -267,7 +316,7 @@ namespace SimpleWeb { if(auto session = session_weak.lock()) { { LockGuard lock(this->connections_mutex); - if(!session->connection->event_stream) + if(session->response->content.end) session->connection->in_use = false; // Remove unused connections, but keep one open for HTTP persistent connection: @@ -342,7 +391,7 @@ namespace SimpleWeb { if(auto session = session_weak.lock()) { { LockGuard lock(this->connections_mutex); - if(!session->connection->event_stream) + if(session->response->content.end) session->connection->in_use = false; // Remove unused connections, but keep one open for HTTP persistent connection: @@ -522,10 +571,6 @@ namespace SimpleWeb { auto lock = session->connection->handler_runner->continue_lock(); if(!lock) return; - if(session->response->streambuf.size() == session->response->streambuf.max_size()) { - session->callback(make_error_code::make_error_code(errc::message_size)); - return; - } if(!ec) { session->connection->attempt_reconnect = true; @@ -539,24 +584,8 @@ namespace SimpleWeb { auto header_it = session->response->header.find("Content-Length"); if(header_it != session->response->header.end()) { auto content_length = stoull(header_it->second); - if(content_length > num_additional_bytes) { - session->connection->set_timeout(); - asio::async_read(*session->connection->socket, session->response->streambuf, asio::transfer_exactly(content_length - num_additional_bytes), [session](const error_code &ec, std::size_t /*bytes_transferred*/) { - session->connection->cancel_timeout(); - auto lock = session->connection->handler_runner->continue_lock(); - if(!lock) - return; - if(session->response->streambuf.size() == session->response->streambuf.max_size()) { - session->callback(make_error_code::make_error_code(errc::message_size)); - return; - } - - if(!ec) - session->callback(ec); - else - session->callback(ec); - }); - } + if(content_length > num_additional_bytes) + this->read_content(session, content_length - num_additional_bytes); else session->callback(ec); } @@ -564,49 +593,24 @@ namespace SimpleWeb { // Expect hex number to not exceed 16 bytes (64-bit number), but take into account previous additional read bytes auto chunk_size_streambuf = std::make_shared(std::max(16 + 2, session->response->streambuf.size())); - // Copy leftover bytes - std::ostream ostream(chunk_size_streambuf.get()); - auto size = session->response->streambuf.size(); - std::unique_ptr buffer(new char[size]); - session->response->content.read(buffer.get(), static_cast(size)); - ostream.write(buffer.get(), static_cast(size)); + // Move leftover bytes + auto &source = session->response->streambuf; + auto &target = *chunk_size_streambuf; + target.commit(asio::buffer_copy(target.prepare(source.size()), source.data())); + source.consume(source.size()); this->read_chunked_transfer_encoded(session, chunk_size_streambuf); } - else if(session->response->http_version < "1.1" || ((header_it = session->response->header.find("Session")) != session->response->header.end() && header_it->second == "close")) { - session->connection->set_timeout(); - asio::async_read(*session->connection->socket, session->response->streambuf, [this, session](const error_code &ec, std::size_t /*bytes_transferred*/) { - session->connection->cancel_timeout(); - auto lock = session->connection->handler_runner->continue_lock(); - if(!lock) - return; - if(session->response->streambuf.size() == session->response->streambuf.max_size()) { - session->callback(make_error_code::make_error_code(errc::message_size)); - return; - } - - if(!ec) { - { - LockGuard lock(this->connections_mutex); - this->connections.erase(session->connection); - } - session->callback(ec); - } - else - session->callback(ec == error::eof ? error_code() : ec); - }); - } + else if(session->response->http_version < "1.1" || ((header_it = session->response->header.find("Session")) != session->response->header.end() && header_it->second == "close")) + read_content(session); else if(((header_it = session->response->header.find("Content-Type")) != session->response->header.end() && header_it->second == "text/event-stream")) { - session->connection->event_stream = true; - auto events_streambuf = std::make_shared(this->config.max_response_streambuf_size); - // Copy leftover bytes - std::ostream ostream(events_streambuf.get()); - auto size = session->response->streambuf.size(); - std::unique_ptr buffer(new char[size]); - session->response->content.read(buffer.get(), static_cast(size)); - ostream.write(buffer.get(), static_cast(size)); + // Move leftover bytes + auto &source = session->response->streambuf; + auto &target = *events_streambuf; + target.commit(asio::buffer_copy(target.prepare(source.size()), source.data())); + source.consume(source.size()); session->callback(ec); // Connection to a Server-Sent Events resource is opened @@ -624,7 +628,7 @@ namespace SimpleWeb { session->connection = create_connection(); session->connection->attempt_reconnect = false; session->connection->in_use = true; - session->response = std::shared_ptr(new Response(this->config.max_response_streambuf_size)); + session->response = std::shared_ptr(new Response(this->config.max_response_streambuf_size, session->connection)); connections.emplace(session->connection); lock.unlock(); this->connect(session); @@ -640,6 +644,58 @@ namespace SimpleWeb { }); } + void read_content(const std::shared_ptr &session, std::size_t remaining_length) { + session->connection->set_timeout(); + asio::async_read(*session->connection->socket, session->response->streambuf, asio::transfer_exactly(remaining_length), [this, session, remaining_length](const error_code &ec, std::size_t bytes_transferred) { + session->connection->cancel_timeout(); + auto lock = session->connection->handler_runner->continue_lock(); + if(!lock) + return; + + if(!ec) { + if(session->response->streambuf.size() == session->response->streambuf.max_size() && remaining_length > bytes_transferred) { + session->response->content.end = false; + session->callback(ec); + session->response = std::shared_ptr(new Response(*session->response)); + this->read_content(session, remaining_length - bytes_transferred); + } + else + session->callback(ec); + } + else + session->callback(ec); + }); + } + + void read_content(const std::shared_ptr &session) { + session->connection->set_timeout(); + asio::async_read(*session->connection->socket, session->response->streambuf, [this, session](const error_code &ec_, std::size_t /*bytes_transferred*/) { + session->connection->cancel_timeout(); + auto lock = session->connection->handler_runner->continue_lock(); + if(!lock) + return; + + auto ec = ec_ == error::eof ? error_code() : ec_; + + if(!ec) { + { + LockGuard lock(this->connections_mutex); + this->connections.erase(session->connection); + } + if(session->response->streambuf.size() == session->response->streambuf.max_size()) { + session->response->content.end = false; + session->callback(ec); + session->response = std::shared_ptr(new Response(*session->response)); + this->read_content(session); + } + else + session->callback(ec); + } + else + session->callback(ec); + }); + } + void read_chunked_transfer_encoded(const std::shared_ptr &session, const std::shared_ptr &chunk_size_streambuf) { session->connection->set_timeout(); asio::async_read_until(*session->connection->socket, *chunk_size_streambuf, "\r\n", [this, session, chunk_size_streambuf](const error_code &ec, size_t bytes_transferred) { @@ -662,57 +718,66 @@ namespace SimpleWeb { return; } + if(chunk_size == 0) { + session->callback(error_code()); + return; + } + auto num_additional_bytes = chunk_size_streambuf->size() - bytes_transferred; auto bytes_to_move = std::min(chunk_size, num_additional_bytes); if(bytes_to_move > 0) { - std::ostream ostream(&session->response->streambuf); - std::unique_ptr buffer(new char[bytes_to_move]); - istream.read(buffer.get(), static_cast(bytes_to_move)); - ostream.write(buffer.get(), static_cast(bytes_to_move)); - if(session->response->streambuf.size() == session->response->streambuf.max_size()) { - session->callback(make_error_code::make_error_code(errc::message_size)); - return; + if(bytes_to_move + session->response->streambuf.size() >= session->response->streambuf.max_size()) { + session->response->content.end = false; + session->callback(ec); + session->response = std::shared_ptr(new Response(*session->response)); } + + auto &source = *chunk_size_streambuf; + auto &target = session->response->streambuf; + target.commit(asio::buffer_copy(target.prepare(bytes_to_move), source.data(), bytes_to_move)); + source.consume(bytes_to_move); } if((2 + chunk_size) > num_additional_bytes) { - session->connection->set_timeout(); - asio::async_read(*session->connection->socket, session->response->streambuf, asio::transfer_exactly(2 + chunk_size - num_additional_bytes), [this, session, chunk_size_streambuf, chunk_size](const error_code &ec, size_t /*bytes_transferred*/) { - session->connection->cancel_timeout(); - auto lock = session->connection->handler_runner->continue_lock(); - if(!lock) - return; - if(session->response->streambuf.size() == session->response->streambuf.max_size()) { - session->callback(make_error_code::make_error_code(errc::message_size)); - return; - } - - if(!ec) { - std::istream istream(&session->response->streambuf); - - // Remove "\r\n" - istream.get(); - istream.get(); - - if(chunk_size > 0) - read_chunked_transfer_encoded(session, chunk_size_streambuf); - else - session->callback(error_code()); - } - else - session->callback(ec); - }); + read_chunk(session, chunk_size_streambuf, 2 + chunk_size - num_additional_bytes); } else { // Remove "\r\n" istream.get(); istream.get(); - if(chunk_size > 0) - read_chunked_transfer_encoded(session, chunk_size_streambuf); - else - session->callback(error_code()); + read_chunked_transfer_encoded(session, chunk_size_streambuf); + } + } + else + session->callback(ec); + }); + } + + void read_chunk(const std::shared_ptr &session, const std::shared_ptr &chunk_size_streambuf, std::size_t remaining_length) { + session->connection->set_timeout(); + asio::async_read(*session->connection->socket, session->response->streambuf, asio::transfer_exactly(remaining_length), [this, session, chunk_size_streambuf, remaining_length](const error_code &ec, size_t bytes_transferred) { + session->connection->cancel_timeout(); + auto lock = session->connection->handler_runner->continue_lock(); + if(!lock) + return; + + if(!ec) { + if(session->response->streambuf.size() == session->response->streambuf.max_size() && remaining_length > bytes_transferred) { + session->response->content.end = false; + session->callback(ec); + session->response = std::shared_ptr(new Response(*session->response)); + this->read_chunk(session, chunk_size_streambuf, remaining_length - bytes_transferred); + } + else { + std::istream istream(&session->response->streambuf); + + // Remove "\r\n" + istream.get(); + istream.get(); + + read_chunked_transfer_encoded(session, chunk_size_streambuf); } } else @@ -727,12 +792,9 @@ namespace SimpleWeb { auto lock = session->connection->handler_runner->continue_lock(); if(!lock) return; - if(events_streambuf->size() == events_streambuf->max_size()) { - session->callback(make_error_code::make_error_code(errc::message_size)); - return; - } if(!ec) { + session->response->content.end = false; std::istream istream(events_streambuf.get()); std::ostream ostream(&session->response->streambuf); std::string line; @@ -742,6 +804,7 @@ namespace SimpleWeb { } session->callback(ec); + session->response = std::shared_ptr(new Response(*session->response)); read_server_sent_event(session, events_streambuf); } else diff --git a/client_https.hpp b/client_https.hpp index b26144f..2ff9330 100644 --- a/client_https.hpp +++ b/client_https.hpp @@ -85,7 +85,7 @@ namespace SimpleWeb { if(!lock) return; if(!ec) { - std::shared_ptr response(new Response(this->config.max_response_streambuf_size)); + std::shared_ptr response(new Response(this->config.max_response_streambuf_size, session->connection)); session->connection->set_timeout(this->config.timeout_connect); asio::async_read_until(session->connection->socket->next_layer(), response->streambuf, "\r\n\r\n", [this, session, response](const error_code &ec, std::size_t /*bytes_transferred*/) { session->connection->cancel_timeout(); diff --git a/server_http.hpp b/server_http.hpp index 27a5f25..0b49ae9 100644 --- a/server_http.hpp +++ b/server_http.hpp @@ -201,18 +201,9 @@ namespace SimpleWeb { std::size_t size() noexcept { return streambuf.size(); } - /// Convenience function to return content as std::string. The stream buffer is consumed. + /// Convenience function to return content as std::string. std::string string() noexcept { - try { - std::string str; - auto size = streambuf.size(); - str.resize(size); - read(&str[0], static_cast(size)); - return str; - } - catch(...) { - return std::string(); - } + return std::string(asio::buffers_begin(streambuf.data()), asio::buffers_end(streambuf.data())); } private: @@ -598,12 +589,11 @@ namespace SimpleWeb { // Expect hex number to not exceed 16 bytes (64-bit number), but take into account previous additional read bytes auto chunk_size_streambuf = std::make_shared(std::max(16 + 2, session->request->streambuf.size())); - // Copy leftover bytes - std::ostream ostream(chunk_size_streambuf.get()); - auto size = session->request->streambuf.size(); - std::unique_ptr buffer(new char[size]); - session->request->content.read(buffer.get(), static_cast(size)); - ostream.write(buffer.get(), static_cast(size)); + // Move leftover bytes + auto &source = session->request->streambuf; + auto &target = *chunk_size_streambuf; + target.commit(asio::buffer_copy(target.prepare(source.size()), source.data())); + source.consume(source.size()); this->read_chunked_transfer_encoded(session, chunk_size_streambuf); } @@ -642,10 +632,12 @@ namespace SimpleWeb { auto bytes_to_move = std::min(chunk_size, num_additional_bytes); if(bytes_to_move > 0) { - std::ostream ostream(&session->request->streambuf); - std::unique_ptr buffer(new char[bytes_to_move]); - istream.read(buffer.get(), static_cast(bytes_to_move)); - ostream.write(buffer.get(), static_cast(bytes_to_move)); + // Move leftover bytes + auto &source = *chunk_size_streambuf; + auto &target = session->request->streambuf; + target.commit(asio::buffer_copy(target.prepare(bytes_to_move), source.data(), bytes_to_move)); + source.consume(bytes_to_move); + if(session->request->streambuf.size() == session->request->streambuf.max_size()) { auto response = std::shared_ptr(new Response(session, this->config.timeout_content)); response->write(StatusCode::client_error_payload_too_large); diff --git a/tests/io_test.cpp b/tests/io_test.cpp index afa0f2d..9a96315 100644 --- a/tests/io_test.cpp +++ b/tests/io_test.cpp @@ -57,6 +57,7 @@ int main() { server.resource["^/string$"]["POST"] = [](shared_ptr response, shared_ptr request) { auto content = request->content.string(); + ASSERT(content == request->content.string()); *response << "HTTP/1.1 200 OK\r\nContent-Length: " << content.length() << "\r\n\r\n" << content; @@ -209,6 +210,13 @@ int main() { *response << "HTTP/1.1 200 OK\nname: value\n\n"; }; + std::string long_response; + for(int c = 0; c < 1000; ++c) + long_response += to_string(c); + server.resource["^/long-response$"]["GET"] = [&long_response](shared_ptr response, shared_ptr /*request*/) { + response->write(long_response, {{"name", "value"}}); + }; + thread server_thread([&server]() { // Start server server.start(); @@ -241,6 +249,7 @@ int main() { auto r = client.request("POST", "/string", "A string"); ASSERT(SimpleWeb::status_code(r->status_code) == SimpleWeb::StatusCode::success_ok); ASSERT(r->content.string() == "A string"); + ASSERT(r->content.string() == "A string"); } { @@ -378,6 +387,81 @@ int main() { } } + // Test large responses + { + HttpClient client("localhost:8080"); + client.config.max_response_streambuf_size = 400; + { + bool thrown = false; + try { + auto r = client.request("GET", "/long-response"); + } + catch(...) { + thrown = true; + } + ASSERT(thrown); + } + { + size_t calls = 0; + bool end = false; + std::string content; + client.request("GET", "/long-response", [&calls, &content, &end](shared_ptr response, const SimpleWeb::error_code &ec) { + ASSERT(!ec); + content += response->content.string(); + calls++; + if(calls == 1) + ASSERT(response->content.end == false); + end = response->content.end; + }); + SimpleWeb::restart(*client.io_service); + client.io_service->run(); + ASSERT(content == long_response); + ASSERT(calls > 2); + ASSERT(end == true); + } + { + size_t calls = 0; + std::string content; + client.request("GET", "/long-response", [&calls, &content](shared_ptr response, const SimpleWeb::error_code &ec) { + if(calls == 0) + ASSERT(!ec); + content += response->content.string(); + calls++; + response->close(); + }); + SimpleWeb::restart(*client.io_service); + client.io_service->run(); + ASSERT(!content.empty()); + ASSERT(calls >= 2); + } + } + + // Test client timeout + { + HttpClient client("localhost:8080"); + client.config.timeout = 2; + { + bool thrown = false; + try { + auto r = client.request("GET", "/work"); + } + catch(...) { + thrown = true; + } + ASSERT(thrown); + } + { + bool call = false; + client.request("GET", "/work", [&call](shared_ptr /*response*/, const SimpleWeb::error_code &ec) { + ASSERT(ec); + call = true; + }); + SimpleWeb::restart(*client.io_service); + client.io_service->run(); + ASSERT(call); + } + } + // Test asynchronous requests { HttpClient client("localhost:8080"); diff --git a/tests/parse_test.cpp b/tests/parse_test.cpp index 8d7abcd..cb9fde6 100644 --- a/tests/parse_test.cpp +++ b/tests/parse_test.cpp @@ -72,7 +72,7 @@ public: } void parse_response_header_test() { - std::shared_ptr response(new Response(static_cast(-1))); + std::shared_ptr response(new Response(static_cast(-1), nullptr)); ostream stream(&response->streambuf); stream << "HTTP/1.1 200 OK\r\n";