include/boost/capy/read_until.hpp

100.0% Lines (65/66) 90.5% List of functions (38/42)

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

read_until.hpp

f(x) Functions (42)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.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/capy
8		//
9
10		#ifndef BOOST_CAPY_READ_UNTIL_HPP
11		#define BOOST_CAPY_READ_UNTIL_HPP
12
13		#include <boost/capy/detail/config.hpp>
14		#include <boost/capy/buffers.hpp>
15		#include <boost/capy/cond.hpp>
16		#include <coroutine>
17		#include <boost/capy/error.hpp>
18		#include <boost/capy/io_result.hpp>
19		#include <boost/capy/io_task.hpp>
20		#include <boost/capy/concept/dynamic_buffer.hpp>
21		#include <boost/capy/concept/match_condition.hpp>
22		#include <boost/capy/concept/read_stream.hpp>
23		#include <boost/capy/ex/io_env.hpp>
24
25		#include <algorithm>
26		#include <cstddef>
27		#include <optional>
28		#include <stop_token>
29		#include <string_view>
30		#include <type_traits>
31
32		namespace boost {
33		namespace capy {
34
35		namespace detail {
36
37		// Linearize a buffer sequence into a string
38		inline
39		std::string
40	2x	linearize_buffers(ConstBufferSequence auto const& data)
41		{
42	2x	std::string linear;
43	2x	linear.reserve(buffer_size(data));
44	2x	auto const end_ = end(data);
45	6x	for(auto it = begin(data); it != end_; ++it)
46	8x	linear.append(
47	4x	static_cast<char const*>(it->data()),
48		it->size());
49	4x	return linear;
50	−	} // LCOV_EXCL_LINE gcov brace artifact (linearize_buffers is exercised)
51
52		// Search buffer using a MatchCondition, with single-buffer optimization
53		template<MatchCondition M>
54		std::size_t
55	263x	search_buffer_for_match(
56		ConstBufferSequence auto const& data,
57		M const& match,
58		std::size_t* hint = nullptr)
59		{
60		// Fast path: single buffer - no linearization needed
61	263x	if(buffer_length(data) == 1)
62		{
63	262x	auto const& buf = *begin(data);
64	786x	return match(std::string_view(
65	262x	static_cast<char const*>(buf.data()),
66	262x	buf.size()), hint);
67		}
68		// Multiple buffers - linearize
69	1x	return match(linearize_buffers(data), hint);
70		}
71
72		// Implementation coroutine for read_until with MatchCondition
73		template<class Stream, class B, MatchCondition M>
74		io_task<std::size_t>
75	136x	read_until_match_impl(
76		Stream& stream,
77		B& buffers,
78		M match,
79		std::size_t initial_amount)
80		{
81		std::size_t amount = initial_amount;
82
83		for(;;)
84		{
85		// Check max_size before preparing
86		if(buffers.size() >= buffers.max_size())
87		co_return {error::not_found, 0};
88
89		// Prepare space, respecting max_size
90		std::size_t const available = buffers.max_size() - buffers.size();
91		std::size_t const to_prepare = (std::min)(amount, available);
92		if(to_prepare == 0)
93		co_return {error::not_found, 0};
94
95		auto mb = buffers.prepare(to_prepare);
96		auto [ec, n] = co_await stream.read_some(mb);
97		buffers.commit(n);
98
99		if(!ec)
100		{
101		auto pos = search_buffer_for_match(buffers.data(), match);
102		if(pos != std::string_view::npos)
103		co_return {{}, pos};
104		}
105
106		if(ec == cond::eof)
107		co_return {error::eof, buffers.size()};
108		if(ec)
109		co_return {ec, buffers.size()};
110
111		// Grow buffer size for next iteration
112		if(n == buffer_size(mb))
113		amount = amount / 2 + amount;
114		}
115	272x	}
116
117		template<class Stream, class B, MatchCondition M, bool OwnsBuffer>
118		struct read_until_awaitable
119		{
120		Stream* stream_;
121		M match_;
122		std::size_t initial_amount_;
123		std::optional<io_result<std::size_t>> immediate_;
124		std::optional<io_task<std::size_t>> inner_;
125
126		using storage_type = std::conditional_t<OwnsBuffer, B, B*>;
127		storage_type buffers_storage_;
128
129	136x	B& buffers() noexcept
130		{
131		if constexpr(OwnsBuffer)
132	126x	return buffers_storage_;
133		else
134	10x	return *buffers_storage_;
135		}
136
137		// Constructor for lvalue (pointer storage)
138	14x	read_until_awaitable(
139		Stream& stream,
140		B* buffers,
141		M match,
142		std::size_t initial_amount)
143		requires (!OwnsBuffer)
144	14x	: stream_(std::addressof(stream))
145	14x	, match_(std::move(match))
146	14x	, initial_amount_(initial_amount)
147	14x	, buffers_storage_(buffers)
148		{
149	14x	auto pos = search_buffer_for_match(
150	14x	buffers_storage_->data(), match_);
151	14x	if(pos != std::string_view::npos)
152	4x	immediate_.emplace(io_result<std::size_t>{{}, pos});
153	14x	}
154
155		// Constructor for rvalue adapter (owned storage)
156	132x	read_until_awaitable(
157		Stream& stream,
158		B&& buffers,
159		M match,
160		std::size_t initial_amount)
161		requires OwnsBuffer
162	132x	: stream_(std::addressof(stream))
163	132x	, match_(std::move(match))
164	132x	, initial_amount_(initial_amount)
165	132x	, buffers_storage_(std::move(buffers))
166		{
167	132x	auto pos = search_buffer_for_match(
168	132x	buffers_storage_.data(), match_);
169	132x	if(pos != std::string_view::npos)
170	6x	immediate_.emplace(io_result<std::size_t>{{}, pos});
171	132x	}
172
173		bool
174	146x	await_ready() const noexcept
175		{
176	146x	return immediate_.has_value();
177		}
178
179		std::coroutine_handle<>
180	136x	await_suspend(std::coroutine_handle<> h, io_env const* env)
181		{
182	272x	inner_.emplace(read_until_match_impl(
183	136x	*stream_, buffers(), match_, initial_amount_));
184	136x	return inner_->await_suspend(h, env);
185		}
186
187		io_result<std::size_t>
188	146x	await_resume()
189		{
190	146x	if(immediate_)
191	10x	return *immediate_;
192	136x	return inner_->await_resume();
193		}
194		};
195
196		template<ReadStream Stream, class B, MatchCondition M>
197		using read_until_return_t = read_until_awaitable<
198		Stream,
199		std::remove_reference_t<B>,
200		M,
201		!std::is_lvalue_reference_v<B&&>>;
202
203		} // namespace detail
204
205		/** Match condition that searches for a delimiter string.
206
207		Satisfies @ref MatchCondition. Returns the position after the
208		delimiter when found, or `npos` otherwise. Provides an overlap
209		hint of `delim.size() - 1` to handle delimiters spanning reads.
210
211		@see MatchCondition, read_until
212		*/
213		struct match_delim
214		{
215		/** The delimiter string to search for.
216
217		@note The referenced characters must remain valid
218		for the lifetime of this object and any pending
219		read operation.
220		*/
221		std::string_view delim;
222
223		/** Search for the delimiter in `data`.
224
225		@param data The data to search.
226		@param hint If non-null, receives the overlap hint
227		on miss.
228		@return `0` if `delim` is empty; otherwise the position
229		just past the delimiter, or `npos` if not found.
230		*/
231		std::size_t
232	226x	operator()(
233		std::string_view data,
234		std::size_t* hint) const noexcept
235		{
236	226x	if(delim.empty())
237	2x	return 0;
238	224x	auto pos = data.find(delim);
239	224x	if(pos != std::string_view::npos)
240	27x	return pos + delim.size();
241	197x	if(hint)
242	1x	*hint = delim.size() > 1 ? delim.size() - 1 : 0;
243	197x	return std::string_view::npos;
244		}
245		};
246
247		/** Asynchronously read until a match condition is satisfied.
248
249		Reads data from `stream` and appends it to `dynbuf` via calling
250		`stream.read_some` zero or more times and using the prepare/commit
251		interface until:
252
253		@li either @c match returns a valid position,
254		@li or @c dynbuf.size() == @c dynbuf.max_size() ,
255		@li or a contingency on @c stream.read_some occurs.
256
257		If the match condition is satisfied by data in `dynbuf.data()` upon entry,
258		no call to `stream.read_some` is performed.
259
260
261		@par Await-returns
262
263		An object of type `io_result<std::size_t>` destructuring as `[ec, n]`.
264
265		If `!ec`, the match succeeded and `n` is the position returned by the
266		match condition (the number of bytes through the end of the
267		match, i.e. the position one past the matched delimiter).
268
269		If `bool(ec)`, the match was not found and `n` is the number of bytes
270		accumulated in `dynbuf` before the contingency arose.
271
272
273		Contingencies:
274
275		@li The first contingency, reported from awaiting @c stream.read_some .
276		@li @c cond::not_found -- when @c dynbuf.size() == @c dynbuf.max_size()
277		and the match condition is not satisfied by data in @c dynbuf.data() .
278
279		@param stream The stream to read from. The caller retains ownership.
280		@param dynbuf The dynamic buffer to append data to. Must remain
281		valid until the operation completes.
282		@param match The match condition callable. Copied into the awaitable.
283		@param initial_amount Initial bytes to read per iteration (default
284		2048). Grows by 1.5x when filled.
285
286
287
288
289		@par Await-throws
290
291		Whatever operations on @c dunbuf throw.
292
293		(Note: types modeling @c DynamicBufferParam provided by Capy throw
294		@c std::bad_alloc from member function
295		@c prepare .)
296
297		@par Remarks
298		Supports _IoAwaitable cancellation_.
299
300		@par Example
301
302		@code
303		task<> read_http_header( ReadStream auto& stream )
304		{
305		std::string header;
306		auto [ec, n] = co_await read_until(
307		stream,
308		string_dynamic_buffer( &header ),
309		[]( std::string_view data, std::size_t* hint ) {
310		auto pos = data.find( "\r\n\r\n" );
311		if( pos != std::string_view::npos )
312		return pos + 4;
313		if( hint )
314		(*hint) = 3; // partial "\r\n\r" possible
315		return std::string_view::npos;
316		} );
317		if( ec )
318		detail::throw_system_error( ec );
319		// header contains data through "\r\n\r\n"
320		}
321		@endcode
322
323		@see read_some, MatchCondition, DynamicBufferParam
324		*/
325		template<ReadStream Stream, class B, MatchCondition M>
326		requires DynamicBufferParam<B&&>
327		detail::read_until_return_t<Stream, B, M>
328	146x	read_until(
329		Stream& stream,
330		B&& dynbuf,
331		M match,
332		std::size_t initial_amount = 2048)
333		{
334	146x	constexpr bool is_lvalue = std::is_lvalue_reference_v<B&&>;
335		using BareB = std::remove_reference_t<B>;
336
337		if constexpr(is_lvalue)
338		return detail::read_until_awaitable<Stream, BareB, M, false>(
339	14x	stream, std::addressof(dynbuf), std::move(match), initial_amount);
340		else
341		return detail::read_until_awaitable<Stream, BareB, M, true>(
342	132x	stream, std::move(dynbuf), std::move(match), initial_amount);
343		}
344
345		/** Asynchronously read until a delimiter string is found.
346
347		Reads data from the stream until the delimiter is found. This is
348		a convenience overload equivalent to calling `read_until` with
349		`match_delim{delim}`. If the delimiter already exists in the
350		buffer, returns immediately without I/O.
351
352		@li The operation completes when:
353		@li The delimiter string is found
354		@li End-of-stream is reached (`cond::eof`)
355		@li The buffer's `max_size()` is reached (`cond::not_found`)
356		@li An error occurs
357		@li The operation is cancelled
358
359		@par Cancellation
360		Supports cancellation via `stop_token` propagated through the
361		IoAwaitable protocol. When cancelled, returns with `cond::canceled`.
362
363		@param stream The stream to read from. The caller retains ownership.
364		@param buffers The dynamic buffer to append data to. Must remain
365		valid until the operation completes.
366		@param delim The delimiter string to search for.
367		@param initial_amount Initial bytes to read per iteration (default
368		2048). Grows by 1.5x when filled.
369
370		@return An awaitable that await-returns `(error_code, std::size_t)`.
371		On success, `n` is the number of bytes through the end of the
372		delimiter (i.e. the position one past the delimiter).
373		Compare error codes to conditions:
374		@li `cond::eof` - EOF before delimiter; `n` is buffer size
375		@li `cond::not_found` - `max_size()` reached before delimiter
376		@li `cond::canceled` - Operation was cancelled
377
378		@par Example
379
380		@code
381		task<std::string> read_line( ReadStream auto& stream )
382		{
383		std::string line;
384		auto [ec, n] = co_await read_until(
385		stream, string_dynamic_buffer( &line ), "\r\n" );
386		if( ec == cond::eof )
387		co_return line; // partial line at EOF
388		if( ec )
389		detail::throw_system_error( ec );
390		line.resize( n - 2 ); // remove "\r\n"
391		co_return line;
392		}
393		@endcode
394
395		@see read_until, match_delim, DynamicBufferParam
396		*/
397		template<ReadStream Stream, class B>
398		requires DynamicBufferParam<B&&>
399		detail::read_until_return_t<Stream, B, match_delim>
400	118x	read_until(
401		Stream& stream,
402		B&& buffers,
403		std::string_view delim,
404		std::size_t initial_amount = 2048)
405		{
406		return read_until(
407		stream,
408		std::forward<B>(buffers),
409		match_delim{delim},
410	118x	initial_amount);
411		}
412
413		} // namespace capy
414		} // namespace boost
415
416		#endif
417