src/detail/connection_pool.cpp

46.1% Lines (35/76) 41.7% List of functions (10/24) 41.0% Branches (16/39)

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

connection_pool.cpp

f(x) Functions (24)

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