src/detail/connection_pool.cpp

98.5% Lines (66/67) 100.0% List of functions (23/23) 88.9% Branches (24/27)

[ ] prev/next file n p prev/next uncovered line

connection_pool.cpp

f(x) Functions (23)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Mohammad Nejati
3			//
4			// Distributed under the Boost Software License, Version 1.0. (See accompanying
5			// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6			//
7			// Official repository: https://github.com/cppalliance/burl
8			//
9
10			#include "connection_pool.hpp"
11
12			#include <boost/burl/error.hpp>
13
14			#include "effective_port.hpp"
15			#include "http_tunnel.hpp"
16			#include "socks5_tunnel.hpp"
17
18			#include <boost/capy/io/any_stream.hpp>
19			#include <boost/corosio/connect.hpp>
20			#include <boost/corosio/openssl_stream.hpp>
21			#include <boost/corosio/resolver.hpp>
22			#include <boost/corosio/shutdown_type.hpp>
23			#include <boost/corosio/socket_option.hpp>
24			#include <boost/corosio/tcp_socket.hpp>
25			#include <boost/corosio/timeout.hpp>
26			#include <boost/corosio/tls_stream.hpp>
27			#include <boost/url/scheme.hpp>
28			#include <boost/url/url_view.hpp>
29
30			#include <memory>
31			#include <string>
32			#include <string_view>
33			#include <utility>
34
35			namespace boost
36			{
37			namespace burl
38			{
39			namespace detail
40			{
41
42			namespace
43			{
44
45			std::string
46		69x	origin(urls::url_view url)
47			{
48	1/1 ✓ Branch 2 taken 69 times.	138x	std::string key{ url.scheme() };
49	1/1 ✓ Branch 1 taken 69 times.	69x	key += "://";
50	1/1 ✓ Branch 2 taken 69 times.	69x	key += url.encoded_host_and_port();
51		69x	return key;
52		✗	}
53
54			capy::io_task<>
55	1/1 ✓ Branch 1 taken 9 times.	9x	connect_tcp(
56			corosio::tcp_socket& socket,
57			capy::executor_ref exec,
58			const client::config& cfg,
59			urls::url_view url)
60			{
61			corosio::resolver resolver(exec);
62			auto [ec, eps] = co_await resolver.resolve(
63			url.encoded_host_address(), effective_port(url));
64			if(ec)
65			co_return ec;
66
67			std::tie(ec, std::ignore) = co_await corosio::connect(socket, eps);
68			if(ec)
69			co_return ec;
70
71			if(cfg.tcp_nodelay)
72			socket.set_option(corosio::socket_option::no_delay(true));
73
74			co_return {};
75		18x	}
76
77			class tcp_connection final : public connection
78			{
79			corosio::tcp_socket socket_;
80
81			public:
82		4x	explicit tcp_connection(corosio::tcp_socket socket)
83		4x	: socket_(std::move(socket))
84			{
85		4x	}
86
87			bool
88		12x	is_open() const noexcept override
89			{
90		12x	return socket_.is_open();
91			}
92
93			// capy::io_task<>
94			// shutdown() override
95			// {
96			// socket_.shutdown(corosio::shutdown_both);
97			// co_return {};
98			// }
99
100			private:
101			capy::io_task<std::size_t>
102	1/1 ✓ Branch 1 taken 8 times.	8x	do_read_some(std::span<capy::mutable_buffer const> buffers) override
103			{
104			co_return co_await socket_.read_some(buffers);
105		16x	}
106
107			capy::io_task<std::size_t>
108	1/1 ✓ Branch 1 taken 8 times.	8x	do_write_some(std::span<capy::const_buffer const> buffers) override
109			{
110			co_return co_await socket_.write_some(buffers);
111		16x	}
112			};
113
114			class tls_connection final : public connection
115			{
116			corosio::tcp_socket socket_;
117			corosio::openssl_stream stream_;
118
119			public:
120		2x	tls_connection(corosio::tcp_socket socket, const corosio::tls_context& ctx)
121		2x	: socket_(std::move(socket))
122	1/1 ✓ Branch 3 taken 2 times.	4x	, stream_(&socket_, ctx)
123			{
124		2x	}
125
126			capy::io_task<>
127		2x	handshake()
128			{
129		2x	return stream_.handshake(corosio::tls_role::client);
130			}
131
132			void
133		2x	set_hostname(std::string_view hostname)
134			{
135		2x	stream_.set_hostname(hostname);
136		2x	}
137
138			bool
139		3x	is_open() const noexcept override
140			{
141		3x	return socket_.is_open();
142			}
143
144			// capy::io_task<>
145			// shutdown() override
146			// {
147			// return stream_.shutdown();
148			// }
149
150			private:
151			capy::io_task<std::size_t>
152		2x	do_read_some(std::span<capy::mutable_buffer const> buffers) override
153			{
154		2x	return stream_.read_some(buffers);
155			}
156
157			capy::io_task<std::size_t>
158		2x	do_write_some(std::span<capy::const_buffer const> buffers) override
159			{
160		2x	return stream_.write_some(buffers);
161			}
162			};
163
164			class stream_connection final : public connection
165			{
166			capy::any_stream stream_;
167			bool open_ = true;
168
169			public:
170		47x	explicit stream_connection(capy::any_stream stream)
171		47x	: stream_(std::move(stream))
172			{
173		47x	}
174
175			bool
176		38x	is_open() const noexcept override
177			{
178		38x	return open_;
179			}
180
181			// capy::io_task<>
182			// shutdown() override
183			// {
184			// open_ = false;
185			// co_return {};
186			// }
187
188			private:
189			capy::io_task<std::size_t>
190	1/1 ✓ Branch 1 taken 39 times.	39x	do_read_some(std::span<capy::mutable_buffer const> buffers) override
191			{
192			auto [ec, n] = co_await stream_.read_some(buffers);
193			if(ec)
194			open_ = false;
195			co_return { ec, n };
196		78x	}
197
198			capy::io_task<std::size_t>
199	1/1 ✓ Branch 1 taken 71 times.	71x	do_write_some(std::span<capy::const_buffer const> buffers) override
200			{
201			auto [ec, n] = co_await stream_.write_some(buffers);
202			if(ec)
203			open_ = false;
204			co_return { ec, n };
205		142x	}
206			};
207
208			} // namespace
209
210			capy::io_task<std::size_t>
211	1/1 ✓ Branch 1 taken 237 times.	237x	connection::
212			read_some_impl(capy::detail::mutable_buffer_array<8> rba)
213			{
214			if(io_timeout_)
215			co_return co_await corosio::timeout(
216			do_read_some(rba), *io_timeout_);
217			co_return co_await do_read_some(rba);
218		474x	}
219
220			capy::io_task<std::size_t>
221	1/1 ✓ Branch 1 taken 82 times.	82x	connection::
222			write_some_impl(capy::detail::const_buffer_array<8> wba)
223			{
224			if(io_timeout_)
225			co_return co_await corosio::timeout(
226			do_write_some(wba), *io_timeout_);
227			co_return co_await do_write_some(wba);
228		164x	}
229
230		55x	connection_pool::connection_pool(
231			capy::executor_ref exec,
232			corosio::tls_context tls_ctx,
233		55x	config cfg)
234		55x	: exec_(exec)
235		55x	, tls_ctx_(std::move(tls_ctx))
236		110x	, config_(std::move(cfg))
237			{
238		55x	}
239
240			capy::io_task<pooled_connection>
241	1/1 ✓ Branch 1 taken 69 times.	69x	connection_pool::acquire(urls::url_view url)
242			{
243			auto key = origin(url);
244			auto [it, last] = idle_.equal_range(key);
245			while(it != last)
246			{
247			auto entry = std::move(it->second);
248			it = idle_.erase(it);
249
250			if(config::clock::now() - entry.idle_since >= config_.pool_idle_timeout)
251			continue;
252
253			if(!entry.conn->is_open())
254			continue;
255
256			entry.conn->set_io_timeout(config_.io_timeout);
257			co_return {
258			std::error_code(),
259			pooled_connection{ std::move(entry.conn),
260			weak_from_this(),
261			std::move(key) }
262			};
263			}
264
265			auto [ec, conn] =
266			co_await corosio::timeout(
267			connect(url), config_.connect_timeout);
268			if(ec)
269			co_return { ec, pooled_connection{} };
270
271			conn->set_io_timeout(config_.io_timeout);
272			co_return {
273			std::error_code(),
274			pooled_connection{ std::move(conn),
275			weak_from_this(),
276			std::move(key) }
277			};
278		138x	}
279
280			void
281		44x	connection_pool::release(pooled_connection pc)
282			{
283	5/6 ✓ Branch 1 taken 43 times. ✓ Branch 2 taken 1 time. ✗ Branch 5 not taken. ✓ Branch 6 taken 43 times. ✓ Branch 7 taken 1 time. ✓ Branch 8 taken 43 times.	44x	if(!pc.conn_ \|\| !pc.conn_->is_open())
284		1x	return;
285
286	2/2 ✓ Branch 1 taken 1 time. ✓ Branch 2 taken 42 times.	43x	if(idle_.count(pc.key_) >= config_.pool_max_idle_per_host)
287		1x	return;
288
289	1/1 ✓ Branch 1 taken 42 times.	42x	idle_.emplace(
290		42x	std::move(pc.key_),
291	0/2 ✗ Branch 6 not taken. ✗ Branch 7 not taken.	84x	idle_connection{ std::move(pc.conn_), config::clock::now() });
292			}
293
294			capy::io_task<std::unique_ptr<connection>>
295	1/1 ✓ Branch 1 taken 59 times.	59x	connection_pool::connect(urls::url_view url) const
296			{
297			using urls::scheme;
298
299			if(url.scheme_id() != scheme::http && url.scheme_id() != scheme::https)
300			co_return { error::unsupported_url_scheme, nullptr };
301
302			if(config_.connect_handler)
303			{
304			auto [ec, stream] = co_await config_.connect_handler(url);
305			if(ec)
306			co_return { ec, nullptr };
307			co_return { std::error_code(),
308			std::make_unique<stream_connection>(std::move(stream)) };
309			}
310
311			corosio::tcp_socket socket(exec_);
312
313			if(config_.proxy)
314			{
315			auto const& proxy = *config_.proxy;
316			if(effective_port(proxy).empty())
317			co_return { error::unsupported_proxy_scheme, nullptr };
318
319			if(auto [ec] = co_await connect_tcp(socket, exec_, config_, proxy); ec)
320			co_return { ec, nullptr };
321
322			if(proxy.scheme() == "http")
323			{
324			auto [ec] = co_await open_http_tunnel(
325			capy::any_stream(&socket), url, proxy);
326			if(ec)
327			co_return { ec, nullptr };
328			}
329			else if(proxy.scheme() == "socks5")
330			{
331			urls::url resolved;
332
333			corosio::resolver resolver(exec_);
334			auto [ec, eps] = co_await resolver.resolve(
335			url.encoded_host_address(), effective_port(url));
336			if(ec)
337			co_return { ec, nullptr };
338
339			auto const& ep = eps.front().get_endpoint();
340			resolved.set_port_number(ep.port());
341			if(ep.is_v4())
342			resolved.set_host_ipv4(
343			urls::ipv4_address(ep.v4_address().to_bytes()));
344			else
345			resolved.set_host_ipv6(
346			urls::ipv6_address(ep.v6_address().to_bytes()));
347
348			std::tie(ec) = co_await open_socks5_tunnel(
349			capy::any_stream(&socket), resolved, proxy);
350			if(ec)
351			co_return { ec, nullptr };
352			}
353			else if(proxy.scheme() == "socks5h")
354			{
355			auto [ec] = co_await open_socks5_tunnel(
356			capy::any_stream(&socket), url, proxy);
357			if(ec)
358			co_return { ec, nullptr };
359			}
360			else
361			{
362			co_return { error::unsupported_proxy_scheme, nullptr };
363			}
364			}
365			else
366			{
367			if(auto [ec] = co_await connect_tcp(socket, exec_, config_, url); ec)
368			co_return { ec, nullptr };
369			}
370
371			if(url.scheme_id() == scheme::https)
372			{
373			auto conn =
374			std::make_unique<tls_connection>(std::move(socket), tls_ctx_);
375			conn->set_hostname(url.encoded_host());
376			auto [ec] = co_await conn->handshake();
377			if(ec)
378			co_return { ec, nullptr };
379
380			co_return { std::error_code(), std::move(conn) };
381			}
382
383			co_return { std::error_code(),
384			std::make_unique<tcp_connection>(std::move(socket)) };
385		118x	}
386
387			void
388		65x	pooled_connection::return_to_pool()
389			{
390	2/2 ✓ Branch 2 taken 22 times. ✓ Branch 3 taken 43 times.	65x	if(auto pool = pool_.lock())
391	1/1 ✓ Branch 4 taken 22 times.	65x	pool->release(std::move(*this));
392		65x	}
393
394			} // namespace detail
395			} // namespace burl
396			} // namespace boost
397