GCC Code Coverage Report

Line	Branch	Exec	Source
1			//
2			// Copyright (c) 2025 Vinnie Falco (vinnie dot falco at gmail dot com)
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/beast2
8			//
9
10			#ifndef BOOST_BEAST2_SERVER_HTTP_STREAM_HPP
11			#define BOOST_BEAST2_SERVER_HTTP_STREAM_HPP
12
13			#include <boost/beast2/detail/config.hpp>
14			#include <boost/beast2/log_service.hpp>
15			#include <boost/beast2/error.hpp>
16			#include <boost/beast2/format.hpp>
17			#include <boost/beast2/server/route_handler_corosio.hpp>
18			#include <boost/beast2/server/router_corosio.hpp>
19			#include <boost/beast2/detail/except.hpp>
20			#include <boost/capy/application.hpp>
21			#include <boost/capy/task.hpp>
22			#include <boost/capy/buffers.hpp>
23			#include <boost/corosio/socket.hpp>
24			#include <boost/corosio/io_result.hpp>
25			#include <boost/http/request_parser.hpp>
26			#include <boost/http/response.hpp>
27			#include <boost/http/serializer.hpp>
28			#include <boost/http/string_body.hpp>
29			#include <boost/http/server/basic_router.hpp>
30			#include <boost/http/error.hpp>
31			#include <boost/url/parse.hpp>
32
33			namespace boost {
34			namespace beast2 {
35
36			//------------------------------------------------
37
38			/** An HTTP server stream which routes requests to handlers and sends responses.
39
40			This class provides a coroutine-based HTTP server session that reads
41			HTTP requests, routes them to handlers installed in a router, and
42			sends the HTTP response.
43
44			The session runs as a coroutine task and uses Corosio for async I/O.
45			*/
46			class http_stream
47			{
48			public:
49			/** Constructor.
50
51			Initializes a new HTTP connection object that operates on
52			the given socket and uses the specified router to dispatch
53			incoming requests.
54
55			@param app The owning application, used to access shared services
56			such as logging and protocol objects.
57			@param sock The socket to read from and write to.
58			@param routes The router used to dispatch incoming HTTP requests.
59			*/
60			http_stream(
61			capy::application& app,
62			corosio::socket& sock,
63			router_corosio routes);
64
65			/** Run the HTTP session as a coroutine.
66
67			Reads HTTP requests, dispatches them through the router,
68			and writes responses until the connection closes or an
69			error occurs.
70
71			@param config The acceptor configuration for this connection.
72
73			@return A task that completes when the session ends.
74			*/
75			capy::task<void>
76			run(http::acceptor_config const& config);
77
78			private:
79			capy::task<corosio::io_result<std::size_t>>
80			read_header();
81
82			capy::task<corosio::io_result<std::size_t>>
83			read_body();
84
85			capy::task<corosio::io_result<std::size_t>>
86			write_response();
87
88			void on_headers();
89			http::route_result do_dispatch();
90			void do_respond(http::route_result rv);
91
92		✗	std::string id() const
93			{
94		✗	return std::string("[") + std::to_string(id_) + "] ";
95			}
96
97			void clear() noexcept;
98
99			section sect_;
100			std::size_t id_ = 0;
101			corosio::socket& sock_;
102			router_corosio routes_;
103			http::acceptor_config const* pconfig_ = nullptr;
104			corosio_route_params rp_;
105			};
106
107			//------------------------------------------------
108
109			inline
110		✗	http_stream::
111			http_stream(
112			capy::application& app,
113			corosio::socket& sock,
114		✗	router_corosio routes)
115		✗	: sect_(use_log_service(app).get_section("http_stream"))
116		✗	, id_(
117		✗	[]() noexcept
118			{
119			static std::size_t n = 0;
120		✗	return ++n;
121		✗	}())
122		✗	, sock_(sock)
123		✗	, routes_(std::move(routes))
124		✗	, rp_(sock_)
125			{
126		✗	rp_.parser = http::request_parser(app);
127		✗	rp_.serializer = http::serializer(app);
128			// Note: suspend mechanism removed - handlers must complete synchronously
129		✗	}
130
131			inline
132			capy::task<void>
133		✗	http_stream::
134			run(http::acceptor_config const& config)
135			{
136			pconfig_ = &config;
137
138			for (;;)
139			{
140			// Reset parser for new request
141			rp_.parser.reset();
142			rp_.session_data.clear();
143			rp_.parser.start();
144
145			// Read HTTP request header
146			auto [ec, n] = co_await read_header();
147			if (ec)
148			{
149			LOG_TRC(sect_)("{} read_header: {}", id(), ec.message());
150			break;
151			}
152
153			LOG_TRC(sect_)("{} read_header bytes={}", id(), n);
154
155			// Process headers and dispatch
156			on_headers();
157
158			auto rv = do_dispatch();
159			do_respond(rv);
160
161			// Write response
162			if (!rp_.serializer.is_done())
163			{
164			auto [wec, wn] = co_await write_response();
165			if (wec)
166			{
167			LOG_TRC(sect_)("{} write_response: {}", id(), wec.message());
168			break;
169			}
170			LOG_TRC(sect_)("{} write_response bytes={}", id(), wn);
171			}
172
173			// Check keep-alive
174			if (!rp_.res.keep_alive())
175			break;
176			}
177
178			clear();
179		✗	}
180
181			inline
182			capy::task<corosio::io_result<std::size_t>>
183		✗	http_stream::
184			read_header()
185			{
186			std::size_t total_bytes = 0;
187			system::error_code ec;
188
189			for (;;)
190			{
191			// Try to parse what we have
192			rp_.parser.parse(ec);
193
194			if (ec == http::condition::need_more_input)
195			{
196			// Need to read more data
197			auto buf = rp_.parser.prepare();
198			auto [read_ec, n] = co_await sock_.read_some(buf);
199
200			if (read_ec)
201			{
202			co_return {read_ec, total_bytes};
203			}
204
205			if (n == 0)
206			{
207			// EOF
208			rp_.parser.commit_eof();
209			ec = {};
210			}
211			else
212			{
213			rp_.parser.commit(n);
214			total_bytes += n;
215			}
216			continue;
217			}
218
219			if (ec.failed())
220			{
221			co_return {ec, total_bytes};
222			}
223
224			// Header complete
225			if (rp_.parser.got_header())
226			{
227			co_return {{}, total_bytes};
228			}
229			}
230		✗	}
231
232			inline
233			capy::task<corosio::io_result<std::size_t>>
234			http_stream::
235			read_body()
236			{
237			std::size_t total_bytes = 0;
238			system::error_code ec;
239
240			while (!rp_.parser.is_complete())
241			{
242			rp_.parser.parse(ec);
243
244			if (ec == http::condition::need_more_input)
245			{
246			auto buf = rp_.parser.prepare();
247			auto [read_ec, n] = co_await sock_.read_some(buf);
248
249			if (read_ec)
250			{
251			co_return {read_ec, total_bytes};
252			}
253
254			if (n == 0)
255			{
256			rp_.parser.commit_eof();
257			}
258			else
259			{
260			rp_.parser.commit(n);
261			total_bytes += n;
262			}
263			continue;
264			}
265
266			if (ec.failed())
267			{
268			co_return {ec, total_bytes};
269			}
270			}
271
272			co_return {{}, total_bytes};
273			}
274
275			inline
276			capy::task<corosio::io_result<std::size_t>>
277		✗	http_stream::
278			write_response()
279			{
280			std::size_t total_bytes = 0;
281
282			while (!rp_.serializer.is_done())
283			{
284			auto rv = rp_.serializer.prepare();
285			if (!rv)
286			{
287			co_return {rv.error(), total_bytes};
288			}
289
290			auto bufs = *rv;
291			std::size_t buf_size = 0;
292			for (auto const& buf : bufs)
293			buf_size += buf.size();
294
295			if (buf_size == 0)
296			{
297			// Serializer done
298			break;
299			}
300
301			// Write buffers - we need to write them all
302			std::size_t written = 0;
303			for (auto const& buf : bufs)
304			{
305			auto [ec, n] = co_await sock_.write_some(
306			capy::const_buffer(buf.data(), buf.size()));
307			if (ec)
308			{
309			co_return {ec, total_bytes + written};
310			}
311			written += n;
312			}
313
314			rp_.serializer.consume(written);
315			total_bytes += written;
316			}
317
318			co_return {{}, total_bytes};
319		✗	}
320
321			inline
322			void
323		✗	http_stream::
324			on_headers()
325			{
326			// Set up Request and Response objects
327		✗	rp_.req = rp_.parser.get();
328		✗	rp_.route_data.clear();
329		✗	rp_.res.set_start_line(
330			http::status::ok, rp_.req.version());
331		✗	rp_.res.set_keep_alive(rp_.req.keep_alive());
332		✗	rp_.serializer.reset();
333
334			// Parse the URL
335		✗	auto rv = urls::parse_uri_reference(rp_.req.target());
336		✗	if (rv.has_error())
337			{
338		✗	rp_.status(http::status::bad_request);
339		✗	rp_.set_body("Bad Request: " + rv.error().message());
340		✗	return;
341			}
342
343		✗	rp_.url = rv.value();
344			}
345
346			inline
347			http::route_result
348		✗	http_stream::
349			do_dispatch()
350			{
351		✗	return routes_.dispatch(
352		✗	rp_.req.method(), rp_.url, rp_);
353			}
354
355			inline
356			void
357		✗	http_stream::
358			do_respond(http::route_result rv)
359			{
360		✗	if (rv == http::route::close)
361			{
362		✗	rp_.res.set_keep_alive(false);
363		✗	return;
364			}
365
366		✗	if (rv == http::route::complete)
367			{
368			// Handler sent the response directly
369		✗	return;
370			}
371
372		✗	if (rv == http::route::suspend)
373			{
374			// Suspend not supported - treat as internal error
375		✗	rp_.status(http::status::internal_server_error);
376		✗	rp_.set_body("Handler suspend not supported");
377		✗	rp_.res.set_keep_alive(false);
378		✗	return;
379			}
380
381		✗	if (rv == http::route::next)
382			{
383			// Unhandled request
384		✗	auto const status = http::status::not_found;
385		✗	rp_.status(status);
386		✗	rp_.set_body(http::to_string(status));
387		✗	return;
388			}
389
390		✗	if (rv != http::route::send)
391			{
392			// Error message of last resort
393		✗	BOOST_ASSERT(rv.failed());
394		✗	BOOST_ASSERT(!http::is_route_result(rv));
395		✗	rp_.status(http::status::internal_server_error);
396		✗	std::string s;
397		✗	format_to(s, "An internal server error occurred: {}", rv.message());
398		✗	rp_.res.set_keep_alive(false);
399		✗	rp_.set_body(s);
400		✗	}
401			}
402
403			inline
404			void
405		✗	http_stream::
406			clear() noexcept
407			{
408		✗	rp_.parser.reset();
409		✗	rp_.serializer.reset();
410		✗	rp_.res.clear();
411		✗	}
412
413			} // beast2
414			} // boost
415
416			#endif
417