Line data Source code
1 : #ifndef THINGER_HTTP_CLIENT_CONNECTION_HPP
2 : #define THINGER_HTTP_CLIENT_CONNECTION_HPP
3 :
4 : #include <atomic>
5 : #include <chrono>
6 : #include <functional>
7 : #include <memory>
8 : #include <optional>
9 : #include <string>
10 : #include <boost/noncopyable.hpp>
11 : #include <boost/asio/ssl/context.hpp>
12 : #include <boost/asio/steady_timer.hpp>
13 :
14 : #include "../common/http_request.hpp"
15 : #include "../common/http_response.hpp"
16 : #include "proxy.hpp"
17 : #include "response_factory.hpp"
18 : #include "stream_types.hpp"
19 : #include "../../asio/sockets/tcp_socket.hpp"
20 : #include "../../asio/sockets/unix_socket.hpp"
21 : #include "../../util/types.hpp"
22 :
23 : namespace thinger::http {
24 :
25 : // How a connection reaches its origin through an HTTP proxy
26 : struct proxy_route {
27 : std::shared_ptr<const proxy_endpoint> proxy;
28 : // Open a tunnel to the origin with CONNECT (https, ws, wss); otherwise the requests are
29 : // sent to the proxy, with an absolute-form target (plain http)
30 : bool tunnel = false;
31 : // TLS with the origin inside the tunnel; null for none
32 : std::shared_ptr<boost::asio::ssl::context> tls;
33 : };
34 :
35 : class client_connection : public std::enable_shared_from_this<client_connection>, public boost::noncopyable {
36 :
37 : static constexpr unsigned MAX_BUFFER_SIZE = 4096;
38 : static constexpr unsigned MAX_RETRIES = 3;
39 : static constexpr auto DEFAULT_TIMEOUT = std::chrono::seconds{60};
40 : static constexpr auto CONNECT_TIMEOUT = std::chrono::seconds{10};
41 : // Largest answer of a proxy refusing a tunnel (an error page) read whole
42 : static constexpr size_t MAX_PROXY_RESPONSE_SIZE = 1048576;
43 :
44 : public:
45 : static std::atomic<unsigned long> connections;
46 :
47 : // Constructors
48 : explicit client_connection(std::shared_ptr<thinger::asio::socket> socket,
49 : std::chrono::seconds timeout = DEFAULT_TIMEOUT);
50 :
51 : client_connection(std::shared_ptr<thinger::asio::unix_socket> socket,
52 : const std::string& path,
53 : std::chrono::seconds timeout = DEFAULT_TIMEOUT);
54 :
55 : // Through an HTTP proxy (it connects on the first request)
56 : client_connection(boost::asio::io_context& io_context, proxy_route route,
57 : std::chrono::seconds timeout = DEFAULT_TIMEOUT);
58 :
59 : virtual ~client_connection();
60 :
61 : // Main operation - send request and get response (null on error: see last_error())
62 : awaitable<std::shared_ptr<http_response>> send_request(std::shared_ptr<http_request> request);
63 :
64 : // Streaming operation - send request and stream response through callback
65 : awaitable<stream_result> send_request_streaming(
66 : std::shared_ptr<http_request> request,
67 : stream_callback callback);
68 :
69 : // Connection management
70 : void close();
71 : std::shared_ptr<thinger::asio::socket> release_socket();
72 :
73 : // Why the last request failed: connection, timeout, or an invalid response
74 : // (client_error). The connection is closed after any of them
75 3200 : const boost::system::error_code& last_error() const { return error_; }
76 :
77 : // The answer of the proxy when it refused to open a tunnel (last_error() is then
78 : // proxy_authentication_required or proxy_tunnel_failed), null otherwise
79 1452 : const std::shared_ptr<http_response>& proxy_response() const { return proxy_response_; }
80 :
81 : // Bytes received after the last response, read along with it (e.g. data the server
82 : // sent right after a protocol upgrade response): they come before anything read from
83 : // the released socket
84 : std::string take_buffered();
85 : std::shared_ptr<thinger::asio::socket> get_socket() const { return socket_; }
86 9985 : bool is_open() const { return socket_ && socket_->is_open(); }
87 :
88 : // A connection serves one request at a time: whoever hands it out claims it first (the
89 : // pool, taking it), and it is unclaimed when idle again (given back to the pool). False
90 : // while another request is using it
91 56281 : bool claim() { return !claimed_.exchange(true); }
92 49771 : void unclaim() { claimed_ = false; }
93 :
94 : // After a request: whether the connection can carry another one (RFC 9112 section 9.3):
95 : // the response was read whole, neither side asked to close it, nothing past the response
96 : // was received, and it did not switch protocols. Virtual for test doubles
97 8206 : virtual bool reusable() const { return reusable_ && is_open(); }
98 :
99 : // While idle: whether it can no longer be used: closed, or the peer closed it, reset it,
100 : // or sent something unasked. Checked without waiting before reusing it. Virtual for test
101 : // doubles
102 3459 : virtual bool stale() const { return !is_open() || socket_->readable_now(); }
103 :
104 : // How the last request failed, to tell whether it can be sent again on a new connection:
105 : // the connection was open before it (reused, not connected for it), nothing of the
106 : // request was written, or some byte of the response was received
107 1551 : bool reused() const { return reused_; }
108 81 : bool request_unsent() const { return request_unsent_; }
109 705 : bool response_started() const { return response_started_; }
110 :
111 : // The idle timeout the server announced with its last response (Keep-Alive: timeout=N)
112 56278 : std::optional<std::chrono::seconds> keep_alive_timeout() const { return keep_alive_timeout_; }
113 :
114 : // While idle in a pool: call `on_expired` (on the io_context of the connection) after
115 : // `timeout`. Calling it again replaces the previous wait (completed with an error)
116 : void expire_after(std::chrono::steady_clock::duration timeout, std::function<void()> on_expired);
117 :
118 : // Forward max response size to the underlying parser. It bounds buffered bodies once
119 : // decompressed too
120 8465 : void set_max_content_size(size_t size) { response_parser_.set_max_content_size(size); }
121 3420 : void set_timeout(std::chrono::seconds timeout) { timeout_ = timeout; }
122 : // Decode gzip and deflate bodies (the default); otherwise they are kept as received
123 8465 : void set_decompress(bool decompress) { decompress_ = decompress; }
124 : // How the requests and responses are logged (see body_logging)
125 8504 : void set_body_logging(const body_logging& options) {
126 8504 : body_logging_ = options;
127 8504 : response_parser_.set_body_logging(options);
128 8504 : }
129 :
130 : private:
131 : // Internal helpers
132 : awaitable<boost::system::error_code> ensure_connected(const http_request& request);
133 : awaitable<boost::system::error_code> connect_socket(const std::string& host, const std::string& port);
134 : awaitable<boost::system::error_code> connect_through_proxy(const http_request& request);
135 : awaitable<boost::system::error_code> open_tunnel(const http_request& request);
136 : awaitable<io_result> write_to_proxy(const http_request& request);
137 : awaitable<std::shared_ptr<http_response>> exchange(std::shared_ptr<http_request> request);
138 : awaitable<std::shared_ptr<http_response>> read_response(const http_request& request);
139 : // After a write that failed with `write_ec` once `bytes` were sent: the answer the server
140 : // may have sent before closing, if any (the connection is not reused then)
141 : awaitable<std::shared_ptr<http_response>> read_after_failed_write(const http_request& request,
142 : boost::system::error_code write_ec,
143 : size_t bytes);
144 : // The response parsed whole, with the bytes read past it; null, and why in error_, if it
145 : // cannot be used
146 : std::shared_ptr<http_response> complete_response(const http_request& request, const uint8_t* begin,
147 : const uint8_t* end);
148 :
149 : std::shared_ptr<thinger::asio::socket> socket_;
150 : std::string socket_path_;
151 : proxy_route route_;
152 : std::shared_ptr<http_response> proxy_response_;
153 : std::chrono::seconds timeout_;
154 : uint8_t buffer_[MAX_BUFFER_SIZE];
155 : std::string buffered_;
156 : response_factory response_parser_;
157 : boost::system::error_code error_;
158 : std::atomic<bool> claimed_{false};
159 : bool reused_ = false;
160 : bool request_unsent_ = false;
161 : bool response_started_ = false;
162 : bool reusable_ = false;
163 : bool decompress_ = true;
164 : body_logging body_logging_;
165 : std::optional<std::chrono::seconds> keep_alive_timeout_;
166 : boost::asio::steady_timer idle_timer_;
167 : };
168 :
169 : }
170 :
171 : #endif
|
