Line data Source code
1 : #ifndef THINGER_HTTP_CLIENT_ASYNC_CLIENT_HPP
2 : #define THINGER_HTTP_CLIENT_ASYNC_CLIENT_HPP
3 :
4 : #include "http_client_base.hpp"
5 : #include "stream_types.hpp"
6 : #include "request_builder.hpp"
7 : #include "../../asio/workers.hpp"
8 : #include <atomic>
9 : #include <concepts>
10 : #include <fstream>
11 : #include <filesystem>
12 : #include <functional>
13 :
14 : namespace thinger::http {
15 :
16 : namespace detail {
17 : // What an async_client shares with its requests, which keep it alive: the HTTP client they
18 : // use, and the requests in flight (see async_client.cpp)
19 : class client_state;
20 : // A request of an async_client, from run() to its end
21 : class client_request;
22 : }
23 :
24 : /**
25 : * Asynchronous HTTP client with callback-based API.
26 : * Runs its requests and callbacks on a pool of workers, which the application owns, starts,
27 : * and keeps alive longer than the client (stop the client, or destroy it, before the workers).
28 : * Perfect for concurrent requests and non-blocking I/O.
29 : *
30 : * It is a handle: every request keeps what it uses of the client (its settings, connections and
31 : * TLS contexts) alive, so a request may outlive the client and touch no freed memory.
32 : *
33 : * stop() and the destructor cancel the requests in flight, in any phase, and close their
34 : * connections. Once they returned, from any thread, the callback of a request cancelled is not
35 : * called. A coroutine, of run() or a callback that is one, sees its requests fail with
36 : * operation_aborted, and any further one fail at once, without connecting. Called from a thread
37 : * that is not one of the workers, they also wait until no callback of the client runs and the
38 : * requests cancelled ended, so what their callbacks and coroutines captured may be destroyed then;
39 : * a coroutine ignoring the cancellation is waited for up to 1 s, and left behind. Called from a
40 : * worker, they do not wait for the requests: a coroutine, as one left behind, may resume after
41 : * they returned, and see operation_aborted or a response that completed before the cancellation
42 : * reached it. What it uses, this client too, must stay alive: hold it by shared_ptr (see README),
43 : * or stop it from a thread that is not a worker. The WebSockets handed out outlive the client,
44 : * and so does what the application spawns on the plain executor of a coroutine of run().
45 : *
46 : * Usage with callbacks:
47 : * asio::workers workers;
48 : * workers.start();
49 : * http::async_client client(workers);
50 : * client.get("https://api.example.com/data", [](auto& response) {
51 : * std::cout << response.body() << std::endl;
52 : * });
53 : * client.wait(); // Wait for all requests to complete
54 : *
55 : * Usage with coroutines:
56 : * http::async_client client(workers);
57 : * client.run([&]() -> awaitable<void> {
58 : * auto response = co_await client.get("https://api.example.com/data");
59 : * });
60 : * client.wait();
61 : */
62 : class async_client {
63 : private:
64 : // Kept by the requests too; client_ is the HTTP client in it
65 : std::shared_ptr<detail::client_state> state_;
66 : http_client_base& client_;
67 : std::atomic<bool> running_{true};
68 :
69 : template<typename> friend class request_builder;
70 :
71 : // The end of a request: the completion of its coroutine, which outlives every frame of the
72 : // request and releases it when destroyed, whether it was called or not (a frame dropped by
73 : // workers that stopped). It holds the request, and so what the frames use of the client
74 : class request_end {
75 : public:
76 2764 : explicit request_end(std::shared_ptr<detail::client_request> request) noexcept : request_(std::move(request)) {}
77 24876 : request_end(request_end&&) noexcept = default;
78 : ~request_end();
79 :
80 2764 : void operator()(const std::exception_ptr&) const noexcept {}
81 :
82 : private:
83 : std::shared_ptr<detail::client_request> request_;
84 : };
85 :
86 : // The synchronous callback of a request about to run: not to be called if the request was
87 : // cancelled; otherwise, counted as running until the end of this scope
88 : class callback_scope {
89 : public:
90 : explicit callback_scope(detail::client_request& request);
91 : ~callback_scope();
92 :
93 22159 : explicit operator bool() const { return entered_; }
94 :
95 : private:
96 : detail::client_request& request_;
97 : bool entered_;
98 : };
99 :
100 : // What the operation of a request with a callback runs with
101 : class request_context {
102 : public:
103 : explicit request_context(detail::client_request& request);
104 :
105 : http_client_base& client;
106 :
107 : // The chunks of a streamed body as callbacks of the request: not called once it was
108 : // cancelled, which stops the stream
109 : stream_callback chunks(stream_callback callback) const;
110 :
111 : private:
112 : detail::client_request& request_;
113 : };
114 :
115 : // A request about to run, registered: null while stop() or the destructor runs, when it
116 : // is not run
117 : std::shared_ptr<detail::client_request> start_request();
118 : // Where it runs (the worker of the caller, or one of the pool), and what cancels it
119 : static boost::asio::io_context::executor_type executor_of(const detail::client_request& request);
120 : static boost::asio::cancellation_slot slot_of(detail::client_request& request);
121 : // Once its coroutine is spawned: it may be cancelled from then on
122 : static void spawned(const std::shared_ptr<detail::client_request>& request);
123 : // Whether it was cancelled: it does not start, nor call its callback
124 : static bool cancelled(detail::client_request& request);
125 :
126 : // Cancel the requests in flight and wait for them as the thread allows (see stop()); with
127 : // `final`, for good
128 : void close_requests(bool final);
129 :
130 : // Run the coroutine `start` makes for a request. Its frames hold the request by reference,
131 : // and whatever else they use, so they all end before the completion of the coroutine
132 : template<typename Start>
133 2776 : void spawn(Start start) {
134 2776 : auto request = start_request();
135 2776 : if (!request) return;
136 2764 : co_spawn(executor_of(*request), start(*request),
137 5528 : boost::asio::bind_cancellation_slot(slot_of(*request), request_end(request)));
138 2764 : spawned(request);
139 2776 : }
140 :
141 : template<typename F>
142 1214 : static awaitable<void> run_factory(detail::client_request& request, F factory) {
143 : if (!cancelled(request)) (void) co_await factory();
144 2428 : }
145 :
146 : // The coroutine of a request with a callback. A synchronous callback is waited for by stop()
147 : // and the destructor until it returns; a coroutine is part of the request: the cancellation
148 : // reaches what it waits for, and it is waited for as the rest of the request is
149 : template<typename Operation, typename Callback>
150 1550 : static awaitable<void> run_callback(detail::client_request& request, Operation operation, Callback callback) {
151 : if (cancelled(request)) co_return;
152 : auto result = co_await operation(request_context(request));
153 : if constexpr (is_awaitable_v<std::invoke_result_t<Callback&, decltype(result)&>>) {
154 : if (!cancelled(request)) (void) co_await std::invoke(callback, result);
155 : } else {
156 : callback_scope running(request);
157 : if (running) std::invoke(callback, result);
158 : }
159 3100 : }
160 :
161 : // A request with a callback: every one of them goes through here. `operation` takes a
162 : // request_context and returns the awaitable of the result handed to the callback
163 : template<typename Operation, typename Callback>
164 1556 : void run_with_callback(Operation operation, Callback&& callback) {
165 2956 : spawn([&](detail::client_request& request) {
166 1550 : return run_callback(request, std::move(operation), std::forward<Callback>(callback));
167 : });
168 1556 : }
169 :
170 : // A WebSocket with a callback, by URL or with a request of its own
171 : template<typename Target, typename Callback>
172 34 : void websocket_with_callback(Target target, std::string subprotocol, Callback&& callback) {
173 65 : run_with_callback([target = std::move(target), proto = std::move(subprotocol)](auto c) mutable
174 : -> awaitable<std::shared_ptr<websocket_client>> {
175 : auto result = co_await c.client.upgrade_websocket(std::move(target), proto);
176 : if (!result) co_return nullptr;
177 : co_return std::make_shared<websocket_client>(std::move(*result));
178 : }, std::forward<Callback>(callback));
179 34 : }
180 :
181 : public:
182 : explicit async_client(asio::workers& workers);
183 : // Ends the requests in flight as stop() does, for good. Called from a worker, it does not
184 : // wait for the requests, but it waits for the callbacks running on other workers: no callback
185 : // of the client runs once it returned, from any thread
186 : ~async_client();
187 :
188 : async_client(const async_client&) = delete;
189 : async_client& operator=(const async_client&) = delete;
190 :
191 : // Count a request the application tracks itself (wait() waits for it), and its end
192 : void track_request_start();
193 : void track_request_end();
194 :
195 : // Stop: release wait(), cancel the requests in flight, and close their connections and the
196 : // idle ones. Returns whether it was running. The callbacks of the requests cancelled are not
197 : // called (one running, of a request that completed before, is not interrupted), and coroutines
198 : // see their requests fail with operation_aborted. From a thread that is not one of the
199 : // workers, it returns once none of its callbacks runs and the requests cancelled ended
200 : // (up to 1 s: one ignoring the cancellation is left behind); from one of them (e.g. in a
201 : // callback), it waits for nothing, and the coroutines cancelled may resume after it returned.
202 : // Requests started while it runs are not run; those started afterwards are
203 : bool stop();
204 33 : bool is_running() const { return running_; }
205 :
206 : // Wait until no request is active, or the client stops; with a timeout
207 : void wait();
208 : bool wait_for(std::chrono::milliseconds timeout);
209 :
210 : size_t pending_requests() const;
211 12 : bool has_pending_requests() const { return pending_requests() > 0; }
212 :
213 : // Run an awaitable on the worker pool (counted until it ends, also by an exception)
214 : template<typename T>
215 1083 : void run(awaitable<T> coro) {
216 2166 : run([coro = std::move(coro)]() mutable { return std::move(coro); });
217 1083 : }
218 :
219 : // Run a coroutine factory on the worker pool (counted until it ends, also by an exception)
220 : template<typename F>
221 1220 : void run(F&& coro_factory) {
222 2434 : spawn([&](detail::client_request& request) { return run_factory(request, std::forward<F>(coro_factory)); });
223 1220 : }
224 :
225 : // --- Settings (see http_client_base) ---
226 :
227 177 : http_client_base& timeout(std::chrono::seconds t) { return client_.timeout(t); }
228 18 : http_client_base& max_redirects(unsigned int max) { return client_.max_redirects(max); }
229 12 : http_client_base& follow_redirects(bool follow) { return client_.follow_redirects(follow); }
230 6 : http_client_base& user_agent(const std::string& agent) { return client_.user_agent(agent); }
231 6 : http_client_base& auto_decompress(bool decompress) { return client_.auto_decompress(decompress); }
232 12 : http_client_base& verify_ssl(bool verify) { return client_.verify_ssl(verify); }
233 : http_client_base& log_bodies(bool enabled = true, size_t max_length = body_logging::default_max_length) {
234 : return client_.log_bodies(enabled, max_length);
235 : }
236 3 : http_client_base& redact_parameters(util::url::redacted_parameters names) { return client_.redact_parameters(std::move(names)); }
237 21 : http_client_base& unix_socket(const std::string& path) { return client_.unix_socket(path); }
238 6 : http_client_base& max_content_size(size_t size) { return client_.max_content_size(size); }
239 : http_client_base& proxy(const std::string& url, const std::string& no_proxy = "") { return client_.proxy(url, no_proxy); }
240 : http_client_base& proxy(const proxy_config& config) { return client_.proxy(config); }
241 : http_client_base& proxy_from_environment(bool enabled) { return client_.proxy_from_environment(enabled); }
242 : http_client_base& add_ca_file(const std::string& path) { return client_.add_ca_file(path); }
243 21 : http_client_base& add_ca_pem(const std::string& pem) { return client_.add_ca_pem(pem); }
244 : http_client_base& use_system_ca(bool enabled) { return client_.use_system_ca(enabled); }
245 : boost::system::error_code client_certificate(const std::string& certificate, const std::string& private_key,
246 : const std::string& chain = "") {
247 : return client_.client_certificate(certificate, private_key, chain);
248 : }
249 : boost::system::error_code client_certificate_file(const std::string& certificate_path, const std::string& private_key_path,
250 : const std::string& chain_path = "") {
251 : return client_.client_certificate_file(certificate_path, private_key_path, chain_path);
252 : }
253 : void clear_client_certificate() { client_.clear_client_certificate(); }
254 : bool has_client_certificate() const { return client_.has_client_certificate(); }
255 : http_client_base& max_idle_connections_per_host(size_t max) { return client_.max_idle_connections_per_host(max); }
256 : http_client_base& max_idle_connections(size_t max) { return client_.max_idle_connections(max); }
257 : http_client_base& idle_timeout(std::chrono::milliseconds timeout) { return client_.idle_timeout(timeout); }
258 :
259 9 : std::chrono::seconds get_timeout() const { return client_.get_timeout(); }
260 15 : unsigned int get_max_redirects() const { return client_.get_max_redirects(); }
261 18 : bool get_follow_redirects() const { return client_.get_follow_redirects(); }
262 9 : const std::string& get_user_agent() const { return client_.get_user_agent(); }
263 9 : bool get_auto_decompress() const { return client_.get_auto_decompress(); }
264 9 : bool get_verify_ssl() const { return client_.get_verify_ssl(); }
265 : const body_logging& get_body_logging() const { return client_.get_body_logging(); }
266 : size_t get_max_content_size() const { return client_.get_max_content_size(); }
267 : size_t get_max_idle_connections_per_host() const { return client_.get_max_idle_connections_per_host(); }
268 : size_t get_max_idle_connections() const { return client_.get_max_idle_connections(); }
269 : std::chrono::milliseconds get_idle_timeout() const { return client_.get_idle_timeout(); }
270 :
271 : std::string proxy_for(const std::string& url) const { return client_.proxy_for(url); }
272 : std::string proxy_for(const http_request& request) const { return client_.proxy_for(request); }
273 :
274 33 : std::shared_ptr<http_request> create_request(method m, const std::string& url) { return client_.create_request(m, url); }
275 :
276 138 : void clear_connections() { client_.clear_connections(); }
277 36 : size_t pool_size() const { return client_.pool_size(); }
278 36 : connection_pool::stats pool_stats() const { return client_.pool_stats(); }
279 :
280 : // --- Awaitable methods (see http_client_base) ---
281 : // Inside run(), the request keeps what they use alive; elsewhere, the client must outlive
282 : // them
283 :
284 4070 : awaitable<client_response> get(const std::string& url, headers_map headers = {}) {
285 4070 : return client_.get(url, std::move(headers));
286 : }
287 27 : awaitable<client_response> post(const std::string& url, std::string body = {},
288 : std::string content_type = "application/json", headers_map headers = {}) {
289 27 : return client_.post(url, std::move(body), std::move(content_type), std::move(headers));
290 : }
291 : awaitable<client_response> post(const std::string& url, const form& form, headers_map headers = {}) {
292 : return client_.post(url, form, std::move(headers));
293 : }
294 27 : awaitable<client_response> put(const std::string& url, std::string body = {},
295 : std::string content_type = "application/json", headers_map headers = {}) {
296 27 : return client_.put(url, std::move(body), std::move(content_type), std::move(headers));
297 : }
298 27 : awaitable<client_response> patch(const std::string& url, std::string body = {},
299 : std::string content_type = "application/json", headers_map headers = {}) {
300 27 : return client_.patch(url, std::move(body), std::move(content_type), std::move(headers));
301 : }
302 27 : awaitable<client_response> del(const std::string& url, headers_map headers = {}) {
303 27 : return client_.del(url, std::move(headers));
304 : }
305 : awaitable<client_response> head(const std::string& url, headers_map headers = {}) {
306 : return client_.head(url, std::move(headers));
307 : }
308 27 : awaitable<client_response> options(const std::string& url, headers_map headers = {}) {
309 27 : return client_.options(url, std::move(headers));
310 : }
311 26 : awaitable<client_response> send(std::shared_ptr<http_request> request) { return client_.send(std::move(request)); }
312 6 : awaitable<stream_result> send_streaming(std::shared_ptr<http_request> request, stream_callback callback) {
313 6 : return client_.send_streaming(std::move(request), std::move(callback));
314 : }
315 : awaitable<std::pair<client_response, std::shared_ptr<client_connection>>>
316 9 : send_with_connection(std::shared_ptr<http_request> request) {
317 9 : return client_.send_with_connection(std::move(request));
318 : }
319 : // The set http_client_base declares (and H8 counts there), forwarded with its return type
320 : auto upgrade_websocket(const std::string& url, const std::string& subprotocol = "") {
321 : return client_.upgrade_websocket(url, subprotocol);
322 : }
323 : auto upgrade_websocket(std::shared_ptr<http_request> request, const std::string& subprotocol = "") {
324 : return client_.upgrade_websocket(std::move(request), subprotocol);
325 : }
326 :
327 : // --- Callback-based async methods ---
328 : // A simpler interface when you're not already in a coroutine: the callback is invoked with
329 : // the response when the request completes (see the usage at the top)
330 :
331 : template<typename Callback>
332 1436 : void get(const std::string& url, Callback&& callback, headers_map headers = {}) {
333 2860 : run_with_callback([url, h = std::move(headers)](auto c) mutable { return c.client.get(url, std::move(h)); },
334 : std::forward<Callback>(callback));
335 1436 : }
336 :
337 : template<typename Callback>
338 20 : void post(const std::string& url, Callback&& callback, std::string body = {},
339 : std::string content_type = "application/json", headers_map headers = {}) {
340 38 : run_with_callback([url, b = std::move(body), ct = std::move(content_type), h = std::move(headers)](auto c) mutable {
341 20 : return c.client.post(url, std::move(b), std::move(ct), std::move(h));
342 : }, std::forward<Callback>(callback));
343 20 : }
344 :
345 : template<typename Callback>
346 : void post(const std::string& url, const form& form, Callback&& callback, headers_map headers = {}) {
347 : run_with_callback([url, f = form, h = std::move(headers)](auto c) mutable { return c.client.post(url, f, std::move(h)); },
348 : std::forward<Callback>(callback));
349 : }
350 :
351 : template<typename Callback>
352 : void put(const std::string& url, Callback&& callback, std::string body = {},
353 : std::string content_type = "application/json", headers_map headers = {}) {
354 : run_with_callback([url, b = std::move(body), ct = std::move(content_type), h = std::move(headers)](auto c) mutable {
355 : return c.client.put(url, std::move(b), std::move(ct), std::move(h));
356 : }, std::forward<Callback>(callback));
357 : }
358 :
359 : template<typename Callback>
360 : void patch(const std::string& url, Callback&& callback, std::string body = {},
361 : std::string content_type = "application/json", headers_map headers = {}) {
362 : run_with_callback([url, b = std::move(body), ct = std::move(content_type), h = std::move(headers)](auto c) mutable {
363 : return c.client.patch(url, std::move(b), std::move(ct), std::move(h));
364 : }, std::forward<Callback>(callback));
365 : }
366 :
367 : template<typename Callback>
368 : void del(const std::string& url, Callback&& callback, headers_map headers = {}) {
369 : run_with_callback([url, h = std::move(headers)](auto c) mutable { return c.client.del(url, std::move(h)); },
370 : std::forward<Callback>(callback));
371 : }
372 :
373 : template<typename Callback>
374 : void head(const std::string& url, Callback&& callback, headers_map headers = {}) {
375 : run_with_callback([url, h = std::move(headers)](auto c) mutable { return c.client.head(url, std::move(h)); },
376 : std::forward<Callback>(callback));
377 : }
378 :
379 : template<typename Callback>
380 : void options(const std::string& url, Callback&& callback, headers_map headers = {}) {
381 : run_with_callback([url, h = std::move(headers)](auto c) mutable { return c.client.options(url, std::move(h)); },
382 : std::forward<Callback>(callback));
383 : }
384 :
385 : // --- Streaming methods ---
386 :
387 : // Send a streaming request with callback: stream_cb takes each chunk, result_cb the result
388 : template<typename Callback>
389 9 : void send_streaming(std::shared_ptr<http_request> request, stream_callback stream_cb, Callback&& result_cb) {
390 9 : run_with_callback([req = std::move(request), scb = std::move(stream_cb)](auto c) mutable {
391 9 : return c.client.send_streaming(std::move(req), c.chunks(std::move(scb)));
392 : }, std::forward<Callback>(result_cb));
393 9 : }
394 :
395 : // Streaming GET with callback
396 : template<typename Callback>
397 9 : void get_streaming(const std::string& url, stream_callback stream_cb, Callback&& result_cb, headers_map headers = {}) {
398 9 : auto request = std::make_shared<http_request>();
399 9 : request->set_url(url);
400 9 : request->set_method(method::GET);
401 9 : for (const auto& [key, value] : headers) request->add_header(key, value);
402 9 : send_streaming(std::move(request), std::move(stream_cb), std::forward<Callback>(result_cb));
403 9 : }
404 :
405 : // Download file with progress callback
406 : template<typename Callback>
407 : void download(const std::string& url, const std::filesystem::path& path,
408 : Callback&& result_cb, progress_callback progress = {}) {
409 : run_with_callback([url, path, prog = std::move(progress)](auto c) mutable -> awaitable<stream_result> {
410 : std::ofstream file(path, std::ios::binary);
411 : if (!file) {
412 : stream_result result;
413 : result.error = "Cannot open file for writing: " + path.string();
414 : co_return result;
415 : }
416 : auto request = std::make_shared<http_request>();
417 : request->set_url(url);
418 : request->set_method(method::GET);
419 : co_return co_await c.client.send_streaming(request, c.chunks([&file, &prog](const stream_info& info) {
420 : file.write(info.data.data(), static_cast<std::streamsize>(info.data.size()));
421 : if (prog) prog(info.downloaded, info.total);
422 : return true;
423 : }));
424 : }, std::forward<Callback>(result_cb));
425 : }
426 :
427 : // --- Request builder for fluent API ---
428 :
429 : /**
430 : * Create a request builder for fluent API.
431 : *
432 : * Usage:
433 : * auto res = co_await client.request("https://api.com/data")
434 : * .header("Authorization", "Bearer xxx")
435 : * .get();
436 : */
437 100 : request_builder<async_client> request(const std::string& url) {
438 100 : return request_builder<async_client>(this, url);
439 : }
440 :
441 : // --- WebSocket ---
442 :
443 : /**
444 : * Connect to a WebSocket server (awaitable version).
445 : *
446 : * Usage:
447 : * auto ws = co_await client.websocket("wss://server.com/ws", "subprotocol");
448 : * if (ws) {
449 : * co_await ws->send_text_async("Hello!");
450 : * }
451 : */
452 21 : awaitable<std::optional<websocket_client>> websocket(
453 : const std::string& url, const std::string& subprotocol = "") {
454 21 : return client_.upgrade_websocket(url, subprotocol);
455 : }
456 :
457 : // Connect to a WebSocket server with custom request (for builder pattern)
458 7 : awaitable<std::optional<websocket_client>> websocket(
459 : std::shared_ptr<http_request> request, const std::string& subprotocol = "") {
460 7 : return client_.upgrade_websocket(std::move(request), subprotocol);
461 : }
462 :
463 : /**
464 : * Connect to a WebSocket server (callback version).
465 : *
466 : * Usage:
467 : * client.websocket("ws://server.com/path", [](auto ws) -> awaitable<void> {
468 : * if (ws) {
469 : * co_await ws->send_text_async("Hello!");
470 : * auto [msg, binary] = co_await ws->receive_async();
471 : * co_await ws->close_async();
472 : * }
473 : * });
474 : *
475 : * The callback receives a shared_ptr to maintain ownership (null if the connection
476 : * failed). It runs on a worker, which runs the io_context of the WebSocket: there the
477 : * synchronous calls of websocket_client (send_text(), receive(), close()...) throw
478 : * std::logic_error. Use its coroutine API, as above (the callback may be a coroutine),
479 : * run() with on_message() (the loop runs while the websocket_client lives: keep the
480 : * shared_ptr), or stop() to close without waiting.
481 : */
482 : template<typename Callback>
483 : requires std::invocable<Callback&, std::shared_ptr<websocket_client>> // not a subprotocol
484 24 : void websocket(const std::string& url, Callback&& callback, const std::string& subprotocol = "") {
485 24 : websocket_with_callback(url, subprotocol, std::forward<Callback>(callback));
486 24 : }
487 : };
488 :
489 : } // namespace thinger::http
490 :
491 : #endif
|
