include/boost/capy/io/any_buffer_sink.hpp

89.0% Lines (252/283) 88.7% List of functions (63/71) 90.3% Branches (28/31)

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

f(x) Functions (71)

Function Calls Lines Branches Blocks

Line	Branch	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_IO_ANY_BUFFER_SINK_HPP
11			#define BOOST_CAPY_IO_ANY_BUFFER_SINK_HPP
12
13			#include <boost/capy/detail/config.hpp>
14			#include <boost/capy/detail/await_suspend_helper.hpp>
15			#include <boost/capy/buffers.hpp>
16			#include <boost/capy/buffers/buffer_copy.hpp>
17			#include <boost/capy/buffers/buffer_param.hpp>
18			#include <boost/capy/concept/buffer_sink.hpp>
19			#include <boost/capy/concept/io_awaitable.hpp>
20			#include <boost/capy/concept/write_sink.hpp>
21			#include <boost/capy/ex/io_env.hpp>
22			#include <boost/capy/io_result.hpp>
23			#include <boost/capy/io_task.hpp>
24
25			#include <concepts>
26			#include <coroutine>
27			#include <cstddef>
28			#include <exception>
29			#include <new>
30			#include <span>
31			#include <stop_token>
32			#include <system_error>
33			#include <utility>
34
35			namespace boost {
36			namespace capy {
37
38			/** Type-erased wrapper for any BufferSink.
39
40			This class provides type erasure for any type satisfying the
41			@ref BufferSink concept, enabling runtime polymorphism for
42			buffer sink operations. It uses cached awaitable storage to achieve
43			zero steady-state allocation after construction.
44
45			The wrapper exposes two interfaces for producing data:
46			the @ref BufferSink interface (`prepare`, `commit`, `commit_eof`)
47			and the @ref WriteSink interface (`write_some`, `write`,
48			`write_eof`). Choose the interface that matches how your data
49			is produced:
50
51			@par Choosing an Interface
52
53			Use the BufferSink interface when you are a generator that
54			produces data into externally-provided buffers. The sink owns
55			the memory; you call @ref prepare to obtain writable buffers,
56			fill them, then call @ref commit or @ref commit_eof.
57
58			Use the WriteSink interface when you already have buffers
59			containing the data to write:
60			- If the entire body is available up front, call
61			@ref write_eof(buffers) to send everything atomically.
62			- If data arrives incrementally, call @ref write or
63			@ref write_some in a loop, then @ref write_eof() when done.
64			Prefer `write` (complete) unless your streaming pattern
65			benefits from partial writes via `write_some`.
66
67			If the wrapped type only satisfies @ref BufferSink, the
68			@ref WriteSink operations are provided automatically.
69
70			@par Construction Modes
71
72			- Owning: Pass by value to transfer ownership. The wrapper
73			allocates storage and owns the sink.
74			- Reference: Pass a pointer to wrap without ownership. The
75			pointed-to sink must outlive this wrapper.
76
77			@par Awaitable Preallocation
78			The constructor preallocates storage for the type-erased awaitable.
79			This reserves all virtual address space at server startup
80			so memory usage can be measured up front, rather than
81			allocating piecemeal as traffic arrives.
82
83			@par Thread Safety
84			Not thread-safe. Concurrent operations on the same wrapper
85			are undefined behavior.
86
87			@par Example
88			@code
89			// Owning - takes ownership of the sink
90			any_buffer_sink abs(some_buffer_sink{args...});
91
92			// Reference - wraps without ownership
93			some_buffer_sink sink;
94			any_buffer_sink abs(&sink);
95
96			// BufferSink interface: generate into callee-owned buffers
97			mutable_buffer arr[16];
98			auto bufs = abs.prepare(arr);
99			// Write data into bufs[0..bufs.size())
100			auto [ec] = co_await abs.commit(bytes_written);
101			auto [ec2] = co_await abs.commit_eof(0);
102
103			// WriteSink interface: send caller-owned buffers
104			auto [ec3, n] = co_await abs.write(make_buffer("hello", 5));
105			auto [ec4] = co_await abs.write_eof();
106
107			// Or send everything at once
108			auto [ec5, n2] = co_await abs.write_eof(
109			make_buffer(body_data));
110			@endcode
111
112			@see any_buffer_source, BufferSink, WriteSink
113			*/
114			class any_buffer_sink
115			{
116			struct vtable;
117			struct awaitable_ops;
118			struct write_awaitable_ops;
119
120			template<BufferSink S>
121			struct vtable_for_impl;
122
123			// hot-path members first for cache locality
124			void* sink_ = nullptr;
125			vtable const* vt_ = nullptr;
126			void* cached_awaitable_ = nullptr;
127			awaitable_ops const* active_ops_ = nullptr;
128			write_awaitable_ops const* active_write_ops_ = nullptr;
129			void* storage_ = nullptr;
130
131			public:
132			/** Destructor.
133
134			Destroys the owned sink (if any) and releases the cached
135			awaitable storage.
136			*/
137			~any_buffer_sink();
138
139			/** Default constructor.
140
141			Constructs an empty wrapper. Operations on a default-constructed
142			wrapper result in undefined behavior.
143			*/
144			any_buffer_sink() = default;
145
146			/** Non-copyable.
147
148			The awaitable cache is per-instance and cannot be shared.
149			*/
150			any_buffer_sink(any_buffer_sink const&) = delete;
151			any_buffer_sink& operator=(any_buffer_sink const&) = delete;
152
153			/** Move constructor.
154
155			Transfers ownership of the wrapped sink (if owned) and
156			cached awaitable storage from `other`. After the move, `other` is
157			in a default-constructed state.
158
159			@param other The wrapper to move from.
160			*/
161		2x	any_buffer_sink(any_buffer_sink&& other) noexcept
162		2x	: sink_(std::exchange(other.sink_, nullptr))
163		2x	, vt_(std::exchange(other.vt_, nullptr))
164		2x	, cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
165		2x	, active_ops_(std::exchange(other.active_ops_, nullptr))
166		2x	, active_write_ops_(std::exchange(other.active_write_ops_, nullptr))
167		2x	, storage_(std::exchange(other.storage_, nullptr))
168			{
169		2x	}
170
171			/** Move assignment operator.
172
173			Destroys any owned sink and releases existing resources,
174			then transfers ownership from `other`.
175
176			@param other The wrapper to move from.
177			@return Reference to this wrapper.
178			*/
179			any_buffer_sink&
180			operator=(any_buffer_sink&& other) noexcept;
181
182			/** Construct by taking ownership of a BufferSink.
183
184			Allocates storage and moves the sink into this wrapper.
185			The wrapper owns the sink and will destroy it. If `S` also
186			satisfies @ref WriteSink, native write operations are
187			forwarded through the virtual boundary.
188
189			@param s The sink to take ownership of.
190			*/
191			template<BufferSink S>
192			requires (!std::same_as<std::decay_t<S>, any_buffer_sink>)
193			any_buffer_sink(S s);
194
195			/** Construct by wrapping a BufferSink without ownership.
196
197			Wraps the given sink by pointer. The sink must remain
198			valid for the lifetime of this wrapper. If `S` also
199			satisfies @ref WriteSink, native write operations are
200			forwarded through the virtual boundary.
201
202			@param s Pointer to the sink to wrap.
203			*/
204			template<BufferSink S>
205			any_buffer_sink(S* s);
206
207			/** Check if the wrapper contains a valid sink.
208
209			@return `true` if wrapping a sink, `false` if default-constructed
210			or moved-from.
211			*/
212			bool
213		26x	has_value() const noexcept
214			{
215		26x	return sink_ != nullptr;
216			}
217
218			/** Check if the wrapper contains a valid sink.
219
220			@return `true` if wrapping a sink, `false` if default-constructed
221			or moved-from.
222			*/
223			explicit
224		3x	operator bool() const noexcept
225			{
226		3x	return has_value();
227			}
228
229			/** Prepare writable buffers.
230
231			Fills the provided span with mutable buffer descriptors
232			pointing to the underlying sink's internal storage. This
233			operation is synchronous.
234
235			@param dest Span of mutable_buffer to fill.
236
237			@return A span of filled buffers.
238
239			@par Preconditions
240			The wrapper must contain a valid sink (`has_value() == true`).
241			*/
242			std::span<mutable_buffer>
243			prepare(std::span<mutable_buffer> dest);
244
245			/** Commit bytes written to the prepared buffers.
246
247			Commits `n` bytes written to the buffers returned by the
248			most recent call to @ref prepare. The operation may trigger
249			underlying I/O.
250
251			@param n The number of bytes to commit.
252
253			@return An awaitable yielding `(error_code)`.
254
255			@par Preconditions
256			The wrapper must contain a valid sink (`has_value() == true`).
257			*/
258			auto
259			commit(std::size_t n);
260
261			/** Commit final bytes and signal end-of-stream.
262
263			Commits `n` bytes written to the buffers returned by the
264			most recent call to @ref prepare and finalizes the sink.
265			After success, no further operations are permitted.
266
267			@param n The number of bytes to commit.
268
269			@return An awaitable yielding `(error_code)`.
270
271			@par Preconditions
272			The wrapper must contain a valid sink (`has_value() == true`).
273			*/
274			auto
275			commit_eof(std::size_t n);
276
277			/** Write some data from a buffer sequence.
278
279			Writes one or more bytes from the buffer sequence to the
280			underlying sink. May consume less than the full sequence.
281
282			When the wrapped type provides native @ref WriteSink support,
283			the operation forwards directly. Otherwise it is synthesized
284			from @ref prepare and @ref commit with a buffer copy.
285
286			@param buffers The buffer sequence to write.
287
288			@return An awaitable yielding `(error_code,std::size_t)`.
289
290			@par Preconditions
291			The wrapper must contain a valid sink (`has_value() == true`).
292			*/
293			template<ConstBufferSequence CB>
294			io_task<std::size_t>
295			write_some(CB buffers);
296
297			/** Write all data from a buffer sequence.
298
299			Writes all data from the buffer sequence to the underlying
300			sink. This method satisfies the @ref WriteSink concept.
301
302			When the wrapped type provides native @ref WriteSink support,
303			each window is forwarded directly. Otherwise the data is
304			copied into the sink via @ref prepare and @ref commit.
305
306			@param buffers The buffer sequence to write.
307
308			@return An awaitable yielding `(error_code,std::size_t)`.
309
310			@par Preconditions
311			The wrapper must contain a valid sink (`has_value() == true`).
312			*/
313			template<ConstBufferSequence CB>
314			io_task<std::size_t>
315			write(CB buffers);
316
317			/** Atomically write data and signal end-of-stream.
318
319			Writes all data from the buffer sequence to the underlying
320			sink and then signals end-of-stream.
321
322			When the wrapped type provides native @ref WriteSink support,
323			the final window is sent atomically via the underlying
324			`write_eof(buffers)`. Otherwise the data is synthesized
325			through @ref prepare, @ref commit, and @ref commit_eof.
326
327			@param buffers The buffer sequence to write.
328
329			@return An awaitable yielding `(error_code,std::size_t)`.
330
331			@par Preconditions
332			The wrapper must contain a valid sink (`has_value() == true`).
333			*/
334			template<ConstBufferSequence CB>
335			io_task<std::size_t>
336			write_eof(CB buffers);
337
338			/** Signal end-of-stream.
339
340			Indicates that no more data will be written to the sink.
341			This method satisfies the @ref WriteSink concept.
342
343			When the wrapped type provides native @ref WriteSink support,
344			the underlying `write_eof()` is called. Otherwise the
345			operation is implemented as `commit_eof(0)`.
346
347			@return An awaitable yielding `(error_code)`.
348
349			@par Preconditions
350			The wrapper must contain a valid sink (`has_value() == true`).
351			*/
352			auto
353			write_eof();
354
355			protected:
356			/** Rebind to a new sink after move.
357
358			Updates the internal pointer to reference a new sink object.
359			Used by owning wrappers after move assignment when the owned
360			object has moved to a new location.
361
362			@param new_sink The new sink to bind to. Must be the same
363			type as the original sink.
364
365			@note Terminates if called with a sink of different type
366			than the original.
367			*/
368			template<BufferSink S>
369			void
370			rebind(S& new_sink) noexcept
371			{
372			if(vt_ != &vtable_for_impl<S>::value)
373			std::terminate();
374			sink_ = &new_sink;
375			}
376
377			private:
378			/** Forward a partial write through the vtable.
379
380			Constructs the underlying `write_some` awaitable in
381			cached storage and returns a type-erased awaitable.
382			*/
383			auto
384			write_some_(std::span<const_buffer const> buffers);
385
386			/** Forward a complete write through the vtable.
387
388			Constructs the underlying `write` awaitable in
389			cached storage and returns a type-erased awaitable.
390			*/
391			auto
392			write_(std::span<const_buffer const> buffers);
393
394			/** Forward an atomic write-with-EOF through the vtable.
395
396			Constructs the underlying `write_eof(buffers)` awaitable
397			in cached storage and returns a type-erased awaitable.
398			*/
399			auto
400			write_eof_buffers_(std::span<const_buffer const> buffers);
401			};
402
403			//----------------------------------------------------------
404
405			/** Type-erased ops for awaitables yielding `io_result<>`. */
406			struct any_buffer_sink::awaitable_ops
407			{
408			bool (await_ready)(void);
409			std::coroutine_handle<> (await_suspend)(void, std::coroutine_handle<>, io_env const*);
410			io_result<> (await_resume)(void);
411			void (destroy)(void) noexcept;
412			};
413
414			/** Type-erased ops for awaitables yielding `io_result<std::size_t>`. */
415			struct any_buffer_sink::write_awaitable_ops
416			{
417			bool (await_ready)(void);
418			std::coroutine_handle<> (await_suspend)(void, std::coroutine_handle<>, io_env const*);
419			io_result<std::size_t> (await_resume)(void);
420			void (destroy)(void) noexcept;
421			};
422
423			struct any_buffer_sink::vtable
424			{
425			void (destroy)(void) noexcept;
426			std::span<mutable_buffer> (*do_prepare)(
427			void* sink,
428			std::span<mutable_buffer> dest);
429			std::size_t awaitable_size;
430			std::size_t awaitable_align;
431			awaitable_ops const* (*construct_commit_awaitable)(
432			void* sink,
433			void* storage,
434			std::size_t n);
435			awaitable_ops const* (*construct_commit_eof_awaitable)(
436			void* sink,
437			void* storage,
438			std::size_t n);
439
440			// WriteSink forwarding (null when wrapped type is BufferSink-only)
441			write_awaitable_ops const* (*construct_write_some_awaitable)(
442			void* sink,
443			void* storage,
444			std::span<const_buffer const> buffers);
445			write_awaitable_ops const* (*construct_write_awaitable)(
446			void* sink,
447			void* storage,
448			std::span<const_buffer const> buffers);
449			write_awaitable_ops const* (*construct_write_eof_buffers_awaitable)(
450			void* sink,
451			void* storage,
452			std::span<const_buffer const> buffers);
453			awaitable_ops const* (*construct_write_eof_awaitable)(
454			void* sink,
455			void* storage);
456			};
457
458			template<BufferSink S>
459			struct any_buffer_sink::vtable_for_impl
460			{
461			using CommitAwaitable = decltype(std::declval<S&>().commit(
462			std::size_t{}));
463			using CommitEofAwaitable = decltype(std::declval<S&>().commit_eof(
464			std::size_t{}));
465
466			static void
467		18x	do_destroy_impl(void* sink) noexcept
468			{
469		18x	static_cast<S*>(sink)->~S();
470		18x	}
471
472			static std::span<mutable_buffer>
473		126x	do_prepare_impl(
474			void* sink,
475			std::span<mutable_buffer> dest)
476			{
477		126x	auto& s = static_cast<S>(sink);
478		126x	return s.prepare(dest);
479			}
480
481			static awaitable_ops const*
482		96x	construct_commit_awaitable_impl(
483			void* sink,
484			void* storage,
485			std::size_t n)
486			{
487		96x	auto& s = static_cast<S>(sink);
488		96x	::new(storage) CommitAwaitable(s.commit(n));
489
490			static constexpr awaitable_ops ops = {
491			+[](void* p) {
492			return static_cast<CommitAwaitable*>(p)->await_ready();
493			},
494			+[](void* p, std::coroutine_handle<> h, io_env const* env) {
495			return detail::call_await_suspend(
496			static_cast<CommitAwaitable*>(p), h, env);
497			},
498			+[](void* p) {
499			return static_cast<CommitAwaitable*>(p)->await_resume();
500			},
501			+[](void* p) noexcept {
502			static_cast<CommitAwaitable*>(p)->~CommitAwaitable();
503			}
504			};
505		96x	return &ops;
506			}
507
508			static awaitable_ops const*
509		70x	construct_commit_eof_awaitable_impl(
510			void* sink,
511			void* storage,
512			std::size_t n)
513			{
514		70x	auto& s = static_cast<S>(sink);
515		70x	::new(storage) CommitEofAwaitable(s.commit_eof(n));
516
517			static constexpr awaitable_ops ops = {
518			+[](void* p) {
519			return static_cast<CommitEofAwaitable*>(p)->await_ready();
520			},
521			+[](void* p, std::coroutine_handle<> h, io_env const* env) {
522			return detail::call_await_suspend(
523			static_cast<CommitEofAwaitable*>(p), h, env);
524			},
525			+[](void* p) {
526			return static_cast<CommitEofAwaitable*>(p)->await_resume();
527			},
528			+[](void* p) noexcept {
529			static_cast<CommitEofAwaitable*>(p)->~CommitEofAwaitable();
530			}
531			};
532		70x	return &ops;
533			}
534
535			//------------------------------------------------------
536			// WriteSink forwarding (only instantiated when WriteSink<S>)
537
538			static write_awaitable_ops const*
539		6x	construct_write_some_awaitable_impl(
540			void* sink,
541			void* storage,
542			std::span<const_buffer const> buffers)
543			requires WriteSink<S>
544			{
545			using Aw = decltype(std::declval<S&>().write_some(
546			std::span<const_buffer const>{}));
547		6x	auto& s = static_cast<S>(sink);
548		6x	::new(storage) Aw(s.write_some(buffers));
549
550			static constexpr write_awaitable_ops ops = {
551		6x	+[](void* p) {
552		6x	return static_cast<Aw*>(p)->await_ready();
553			},
554		✗	+[](void* p, std::coroutine_handle<> h, io_env const* env) {
555		✗	return detail::call_await_suspend(
556		✗	static_cast<Aw*>(p), h, env);
557			},
558		6x	+[](void* p) {
559		6x	return static_cast<Aw*>(p)->await_resume();
560			},
561		6x	+[](void* p) noexcept {
562		6x	static_cast<Aw*>(p)->~Aw();
563			}
564			};
565		6x	return &ops;
566			}
567
568			static write_awaitable_ops const*
569		14x	construct_write_awaitable_impl(
570			void* sink,
571			void* storage,
572			std::span<const_buffer const> buffers)
573			requires WriteSink<S>
574			{
575			using Aw = decltype(std::declval<S&>().write(
576			std::span<const_buffer const>{}));
577		14x	auto& s = static_cast<S>(sink);
578		14x	::new(storage) Aw(s.write(buffers));
579
580			static constexpr write_awaitable_ops ops = {
581		14x	+[](void* p) {
582		14x	return static_cast<Aw*>(p)->await_ready();
583			},
584		✗	+[](void* p, std::coroutine_handle<> h, io_env const* env) {
585		✗	return detail::call_await_suspend(
586		✗	static_cast<Aw*>(p), h, env);
587			},
588		14x	+[](void* p) {
589		14x	return static_cast<Aw*>(p)->await_resume();
590			},
591		14x	+[](void* p) noexcept {
592		14x	static_cast<Aw*>(p)->~Aw();
593			}
594			};
595		14x	return &ops;
596			}
597
598			static write_awaitable_ops const*
599		12x	construct_write_eof_buffers_awaitable_impl(
600			void* sink,
601			void* storage,
602			std::span<const_buffer const> buffers)
603			requires WriteSink<S>
604			{
605			using Aw = decltype(std::declval<S&>().write_eof(
606			std::span<const_buffer const>{}));
607		12x	auto& s = static_cast<S>(sink);
608		12x	::new(storage) Aw(s.write_eof(buffers));
609
610			static constexpr write_awaitable_ops ops = {
611		12x	+[](void* p) {
612		12x	return static_cast<Aw*>(p)->await_ready();
613			},
614		✗	+[](void* p, std::coroutine_handle<> h, io_env const* env) {
615		✗	return detail::call_await_suspend(
616		✗	static_cast<Aw*>(p), h, env);
617			},
618		12x	+[](void* p) {
619		12x	return static_cast<Aw*>(p)->await_resume();
620			},
621		12x	+[](void* p) noexcept {
622		12x	static_cast<Aw*>(p)->~Aw();
623			}
624			};
625		12x	return &ops;
626			}
627
628			static awaitable_ops const*
629		16x	construct_write_eof_awaitable_impl(
630			void* sink,
631			void* storage)
632			requires WriteSink<S>
633			{
634			using Aw = decltype(std::declval<S&>().write_eof());
635		16x	auto& s = static_cast<S>(sink);
636		16x	::new(storage) Aw(s.write_eof());
637
638			static constexpr awaitable_ops ops = {
639		16x	+[](void* p) {
640		16x	return static_cast<Aw*>(p)->await_ready();
641			},
642		✗	+[](void* p, std::coroutine_handle<> h, io_env const* env) {
643		✗	return detail::call_await_suspend(
644		✗	static_cast<Aw*>(p), h, env);
645			},
646		16x	+[](void* p) {
647		16x	return static_cast<Aw*>(p)->await_resume();
648			},
649		16x	+[](void* p) noexcept {
650		16x	static_cast<Aw*>(p)->~Aw();
651			}
652			};
653		16x	return &ops;
654			}
655
656			//------------------------------------------------------
657
658			static consteval std::size_t
659			compute_max_size() noexcept
660			{
661			std::size_t s = sizeof(CommitAwaitable) > sizeof(CommitEofAwaitable)
662			? sizeof(CommitAwaitable)
663			: sizeof(CommitEofAwaitable);
664			if constexpr (WriteSink<S>)
665			{
666			using WS = decltype(std::declval<S&>().write_some(
667			std::span<const_buffer const>{}));
668			using W = decltype(std::declval<S&>().write(
669			std::span<const_buffer const>{}));
670			using WEB = decltype(std::declval<S&>().write_eof(
671			std::span<const_buffer const>{}));
672			using WE = decltype(std::declval<S&>().write_eof());
673
674			if(sizeof(WS) > s) s = sizeof(WS);
675			if(sizeof(W) > s) s = sizeof(W);
676			if(sizeof(WEB) > s) s = sizeof(WEB);
677			if(sizeof(WE) > s) s = sizeof(WE);
678			}
679			return s;
680			}
681
682			static consteval std::size_t
683			compute_max_align() noexcept
684			{
685			std::size_t a = alignof(CommitAwaitable) > alignof(CommitEofAwaitable)
686			? alignof(CommitAwaitable)
687			: alignof(CommitEofAwaitable);
688			if constexpr (WriteSink<S>)
689			{
690			using WS = decltype(std::declval<S&>().write_some(
691			std::span<const_buffer const>{}));
692			using W = decltype(std::declval<S&>().write(
693			std::span<const_buffer const>{}));
694			using WEB = decltype(std::declval<S&>().write_eof(
695			std::span<const_buffer const>{}));
696			using WE = decltype(std::declval<S&>().write_eof());
697
698			if(alignof(WS) > a) a = alignof(WS);
699			if(alignof(W) > a) a = alignof(W);
700			if(alignof(WEB) > a) a = alignof(WEB);
701			if(alignof(WE) > a) a = alignof(WE);
702			}
703			return a;
704			}
705
706			static consteval vtable
707			make_vtable() noexcept
708			{
709			vtable v{};
710			v.destroy = &do_destroy_impl;
711			v.do_prepare = &do_prepare_impl;
712			v.awaitable_size = compute_max_size();
713			v.awaitable_align = compute_max_align();
714			v.construct_commit_awaitable = &construct_commit_awaitable_impl;
715			v.construct_commit_eof_awaitable = &construct_commit_eof_awaitable_impl;
716			v.construct_write_some_awaitable = nullptr;
717			v.construct_write_awaitable = nullptr;
718			v.construct_write_eof_buffers_awaitable = nullptr;
719			v.construct_write_eof_awaitable = nullptr;
720
721			if constexpr (WriteSink<S>)
722			{
723			v.construct_write_some_awaitable =
724			&construct_write_some_awaitable_impl;
725			v.construct_write_awaitable =
726			&construct_write_awaitable_impl;
727			v.construct_write_eof_buffers_awaitable =
728			&construct_write_eof_buffers_awaitable_impl;
729			v.construct_write_eof_awaitable =
730			&construct_write_eof_awaitable_impl;
731			}
732			return v;
733			}
734
735			static constexpr vtable value = make_vtable();
736			};
737
738			//----------------------------------------------------------
739
740			inline
741		215x	any_buffer_sink::~any_buffer_sink()
742			{
743	2/2 ✓ Branch 0 taken 17 times. ✓ Branch 1 taken 198 times.	215x	if(storage_)
744			{
745		17x	vt_->destroy(sink_);
746		17x	::operator delete(storage_);
747			}
748	2/2 ✓ Branch 0 taken 208 times. ✓ Branch 1 taken 7 times.	215x	if(cached_awaitable_)
749		208x	::operator delete(cached_awaitable_);
750		215x	}
751
752			inline any_buffer_sink&
753		5x	any_buffer_sink::operator=(any_buffer_sink&& other) noexcept
754			{
755	2/2 ✓ Branch 0 taken 4 times. ✓ Branch 1 taken 1 time.	5x	if(this != &other)
756			{
757	2/2 ✓ Branch 0 taken 1 time. ✓ Branch 1 taken 3 times.	4x	if(storage_)
758			{
759		1x	vt_->destroy(sink_);
760		1x	::operator delete(storage_);
761			}
762	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 2 times.	4x	if(cached_awaitable_)
763		2x	::operator delete(cached_awaitable_);
764		4x	sink_ = std::exchange(other.sink_, nullptr);
765		4x	vt_ = std::exchange(other.vt_, nullptr);
766		4x	cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
767		4x	storage_ = std::exchange(other.storage_, nullptr);
768		4x	active_ops_ = std::exchange(other.active_ops_, nullptr);
769		4x	active_write_ops_ = std::exchange(other.active_write_ops_, nullptr);
770			}
771		5x	return *this;
772			}
773
774			template<BufferSink S>
775			requires (!std::same_as<std::decay_t<S>, any_buffer_sink>)
776		18x	any_buffer_sink::any_buffer_sink(S s)
777		18x	: vt_(&vtable_for_impl<S>::value)
778			{
779			struct guard {
780			any_buffer_sink* self;
781			bool committed = false;
782			~guard() {
783			if(!committed && self->storage_) {
784			self->vt_->destroy(self->sink_);
785			::operator delete(self->storage_);
786			self->storage_ = nullptr;
787			self->sink_ = nullptr;
788			}
789			}
790		18x	} g{this};
791
792	2/2 boost::capy::any_buffer_sink::any_buffer_sink<boost::capy::(anonymous namespace)::buffer_write_sink>(boost::capy::(anonymous namespace)::buffer_write_sink): ✓ Branch 1 taken 7 times. boost::capy::any_buffer_sink::any_buffer_sink<boost::capy::test::buffer_sink>(boost::capy::test::buffer_sink): ✓ Branch 1 taken 11 times.	18x	storage_ = ::operator new(sizeof(S));
793		18x	sink_ = ::new(storage_) S(std::move(s));
794
795	2/2 boost::capy::any_buffer_sink::any_buffer_sink<boost::capy::(anonymous namespace)::buffer_write_sink>(boost::capy::(anonymous namespace)::buffer_write_sink): ✓ Branch 1 taken 7 times. boost::capy::any_buffer_sink::any_buffer_sink<boost::capy::test::buffer_sink>(boost::capy::test::buffer_sink): ✓ Branch 1 taken 11 times.	18x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
796
797		18x	g.committed = true;
798		18x	}
799
800			template<BufferSink S>
801		192x	any_buffer_sink::any_buffer_sink(S* s)
802		192x	: sink_(s)
803		192x	, vt_(&vtable_for_impl<S>::value)
804			{
805		192x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
806		192x	}
807
808			//----------------------------------------------------------
809
810			inline std::span<mutable_buffer>
811		126x	any_buffer_sink::prepare(std::span<mutable_buffer> dest)
812			{
813		126x	return vt_->do_prepare(sink_, dest);
814			}
815
816			inline auto
817		96x	any_buffer_sink::commit(std::size_t n)
818			{
819			struct awaitable
820			{
821			any_buffer_sink* self_;
822			std::size_t n_;
823
824			bool
825		96x	await_ready()
826			{
827		192x	self_->active_ops_ = self_->vt_->construct_commit_awaitable(
828		96x	self_->sink_,
829		96x	self_->cached_awaitable_,
830			n_);
831		96x	return self_->active_ops_->await_ready(self_->cached_awaitable_);
832			}
833
834			std::coroutine_handle<>
835		✗	await_suspend(std::coroutine_handle<> h, io_env const* env)
836			{
837		✗	return self_->active_ops_->await_suspend(
838		✗	self_->cached_awaitable_, h, env);
839			}
840
841			io_result<>
842		96x	await_resume()
843			{
844			struct guard {
845			any_buffer_sink* self;
846		96x	~guard() {
847		96x	self->active_ops_->destroy(self->cached_awaitable_);
848		96x	self->active_ops_ = nullptr;
849		96x	}
850		96x	} g{self_};
851		96x	return self_->active_ops_->await_resume(
852	1/1 ✓ Branch 1 taken 70 times.	166x	self_->cached_awaitable_);
853		96x	}
854			};
855		96x	return awaitable{this, n};
856			}
857
858			inline auto
859		54x	any_buffer_sink::commit_eof(std::size_t n)
860			{
861			struct awaitable
862			{
863			any_buffer_sink* self_;
864			std::size_t n_;
865
866			bool
867		54x	await_ready()
868			{
869		108x	self_->active_ops_ = self_->vt_->construct_commit_eof_awaitable(
870		54x	self_->sink_,
871		54x	self_->cached_awaitable_,
872			n_);
873		54x	return self_->active_ops_->await_ready(self_->cached_awaitable_);
874			}
875
876			std::coroutine_handle<>
877		✗	await_suspend(std::coroutine_handle<> h, io_env const* env)
878			{
879		✗	return self_->active_ops_->await_suspend(
880		✗	self_->cached_awaitable_, h, env);
881			}
882
883			io_result<>
884		54x	await_resume()
885			{
886			struct guard {
887			any_buffer_sink* self;
888		54x	~guard() {
889		54x	self->active_ops_->destroy(self->cached_awaitable_);
890		54x	self->active_ops_ = nullptr;
891		54x	}
892		54x	} g{self_};
893		54x	return self_->active_ops_->await_resume(
894	1/1 ✓ Branch 1 taken 38 times.	92x	self_->cached_awaitable_);
895		54x	}
896			};
897		54x	return awaitable{this, n};
898			}
899
900			//----------------------------------------------------------
901			// Private helpers for native WriteSink forwarding
902
903			inline auto
904		6x	any_buffer_sink::write_some_(
905			std::span<const_buffer const> buffers)
906			{
907			struct awaitable
908			{
909			any_buffer_sink* self_;
910			std::span<const_buffer const> buffers_;
911
912			bool
913		6x	await_ready() const noexcept
914			{
915		6x	return false;
916			}
917
918			std::coroutine_handle<>
919		6x	await_suspend(std::coroutine_handle<> h, io_env const* env)
920			{
921		12x	self_->active_write_ops_ =
922		12x	self_->vt_->construct_write_some_awaitable(
923		6x	self_->sink_,
924		6x	self_->cached_awaitable_,
925			buffers_);
926
927	1/2 ✓ Branch 0 taken 6 times. ✗ Branch 1 not taken.	6x	if(self_->active_write_ops_->await_ready(
928		6x	self_->cached_awaitable_))
929		6x	return h;
930
931		✗	return self_->active_write_ops_->await_suspend(
932		✗	self_->cached_awaitable_, h, env);
933			}
934
935			io_result<std::size_t>
936		6x	await_resume()
937			{
938			struct guard {
939			any_buffer_sink* self;
940		6x	~guard() {
941		6x	self->active_write_ops_->destroy(
942		6x	self->cached_awaitable_);
943		6x	self->active_write_ops_ = nullptr;
944		6x	}
945		6x	} g{self_};
946		6x	return self_->active_write_ops_->await_resume(
947	1/1 ✓ Branch 1 taken 4 times.	10x	self_->cached_awaitable_);
948		6x	}
949			};
950		6x	return awaitable{this, buffers};
951			}
952
953			inline auto
954		14x	any_buffer_sink::write_(
955			std::span<const_buffer const> buffers)
956			{
957			struct awaitable
958			{
959			any_buffer_sink* self_;
960			std::span<const_buffer const> buffers_;
961
962			bool
963		14x	await_ready() const noexcept
964			{
965		14x	return false;
966			}
967
968			std::coroutine_handle<>
969		14x	await_suspend(std::coroutine_handle<> h, io_env const* env)
970			{
971		28x	self_->active_write_ops_ =
972		28x	self_->vt_->construct_write_awaitable(
973		14x	self_->sink_,
974		14x	self_->cached_awaitable_,
975			buffers_);
976
977	1/2 ✓ Branch 0 taken 14 times. ✗ Branch 1 not taken.	14x	if(self_->active_write_ops_->await_ready(
978		14x	self_->cached_awaitable_))
979		14x	return h;
980
981		✗	return self_->active_write_ops_->await_suspend(
982		✗	self_->cached_awaitable_, h, env);
983			}
984
985			io_result<std::size_t>
986		14x	await_resume()
987			{
988			struct guard {
989			any_buffer_sink* self;
990		14x	~guard() {
991		14x	self->active_write_ops_->destroy(
992		14x	self->cached_awaitable_);
993		14x	self->active_write_ops_ = nullptr;
994		14x	}
995		14x	} g{self_};
996		14x	return self_->active_write_ops_->await_resume(
997	1/1 ✓ Branch 1 taken 10 times.	24x	self_->cached_awaitable_);
998		14x	}
999			};
1000		14x	return awaitable{this, buffers};
1001			}
1002
1003			inline auto
1004		12x	any_buffer_sink::write_eof_buffers_(
1005			std::span<const_buffer const> buffers)
1006			{
1007			struct awaitable
1008			{
1009			any_buffer_sink* self_;
1010			std::span<const_buffer const> buffers_;
1011
1012			bool
1013		12x	await_ready() const noexcept
1014			{
1015		12x	return false;
1016			}
1017
1018			std::coroutine_handle<>
1019		12x	await_suspend(std::coroutine_handle<> h, io_env const* env)
1020			{
1021		24x	self_->active_write_ops_ =
1022		24x	self_->vt_->construct_write_eof_buffers_awaitable(
1023		12x	self_->sink_,
1024		12x	self_->cached_awaitable_,
1025			buffers_);
1026
1027	1/2 ✓ Branch 0 taken 12 times. ✗ Branch 1 not taken.	12x	if(self_->active_write_ops_->await_ready(
1028		12x	self_->cached_awaitable_))
1029		12x	return h;
1030
1031		✗	return self_->active_write_ops_->await_suspend(
1032		✗	self_->cached_awaitable_, h, env);
1033			}
1034
1035			io_result<std::size_t>
1036		12x	await_resume()
1037			{
1038			struct guard {
1039			any_buffer_sink* self;
1040		12x	~guard() {
1041		12x	self->active_write_ops_->destroy(
1042		12x	self->cached_awaitable_);
1043		12x	self->active_write_ops_ = nullptr;
1044		12x	}
1045		12x	} g{self_};
1046		12x	return self_->active_write_ops_->await_resume(
1047	1/1 ✓ Branch 1 taken 8 times.	20x	self_->cached_awaitable_);
1048		12x	}
1049			};
1050		12x	return awaitable{this, buffers};
1051			}
1052
1053			//----------------------------------------------------------
1054			// Public WriteSink methods
1055
1056			template<ConstBufferSequence CB>
1057			io_task<std::size_t>
1058	1/1 ✓ Branch 1 taken 22 times.	22x	any_buffer_sink::write_some(CB buffers)
1059			{
1060			buffer_param<CB> bp(buffers);
1061			auto src = bp.data();
1062			if(src.empty())
1063			co_return {{}, 0};
1064
1065			// Native WriteSink path
1066			if(vt_->construct_write_some_awaitable)
1067			co_return co_await write_some_(src);
1068
1069			// Synthesized path: prepare + buffer_copy + commit
1070			mutable_buffer arr[detail::max_iovec_];
1071			auto dst_bufs = prepare(arr);
1072			if(dst_bufs.empty())
1073			{
1074			auto [ec] = co_await commit(0);
1075			if(ec)
1076			co_return {ec, 0};
1077			dst_bufs = prepare(arr);
1078			if(dst_bufs.empty())
1079			co_return {{}, 0};
1080			}
1081
1082			auto n = buffer_copy(dst_bufs, src);
1083			auto [ec] = co_await commit(n);
1084			if(ec)
1085			co_return {ec, 0};
1086			co_return {{}, n};
1087		44x	}
1088
1089			template<ConstBufferSequence CB>
1090			io_task<std::size_t>
1091	1/1 ✓ Branch 1 taken 38 times.	38x	any_buffer_sink::write(CB buffers)
1092			{
1093			buffer_param<CB> bp(buffers);
1094			std::size_t total = 0;
1095
1096			// Native WriteSink path
1097			if(vt_->construct_write_awaitable)
1098			{
1099			for(;;)
1100			{
1101			auto bufs = bp.data();
1102			if(bufs.empty())
1103			break;
1104
1105			auto [ec, n] = co_await write_(bufs);
1106			total += n;
1107			if(ec)
1108			co_return {ec, total};
1109			bp.consume(n);
1110			}
1111			co_return {{}, total};
1112			}
1113
1114			// Synthesized path: prepare + buffer_copy + commit
1115			for(;;)
1116			{
1117			auto src = bp.data();
1118			if(src.empty())
1119			break;
1120
1121			mutable_buffer arr[detail::max_iovec_];
1122			auto dst_bufs = prepare(arr);
1123			if(dst_bufs.empty())
1124			{
1125			auto [ec] = co_await commit(0);
1126			if(ec)
1127			co_return {ec, total};
1128			continue;
1129			}
1130
1131			auto n = buffer_copy(dst_bufs, src);
1132			auto [ec] = co_await commit(n);
1133			if(ec)
1134			co_return {ec, total};
1135			bp.consume(n);
1136			total += n;
1137			}
1138
1139			co_return {{}, total};
1140		76x	}
1141
1142			inline auto
1143		32x	any_buffer_sink::write_eof()
1144			{
1145			struct awaitable
1146			{
1147			any_buffer_sink* self_;
1148
1149			bool
1150		32x	await_ready()
1151			{
1152	2/2 ✓ Branch 0 taken 16 times. ✓ Branch 1 taken 16 times.	32x	if(self_->vt_->construct_write_eof_awaitable)
1153			{
1154			// Native WriteSink: forward to underlying write_eof()
1155		32x	self_->active_ops_ =
1156		16x	self_->vt_->construct_write_eof_awaitable(
1157		16x	self_->sink_,
1158		16x	self_->cached_awaitable_);
1159			}
1160			else
1161			{
1162			// Synthesized: commit_eof(0)
1163		32x	self_->active_ops_ =
1164		16x	self_->vt_->construct_commit_eof_awaitable(
1165		16x	self_->sink_,
1166		16x	self_->cached_awaitable_,
1167			0);
1168			}
1169		64x	return self_->active_ops_->await_ready(
1170		32x	self_->cached_awaitable_);
1171			}
1172
1173			std::coroutine_handle<>
1174		✗	await_suspend(std::coroutine_handle<> h, io_env const* env)
1175			{
1176		✗	return self_->active_ops_->await_suspend(
1177		✗	self_->cached_awaitable_, h, env);
1178			}
1179
1180			io_result<>
1181		32x	await_resume()
1182			{
1183			struct guard {
1184			any_buffer_sink* self;
1185		32x	~guard() {
1186		32x	self->active_ops_->destroy(self->cached_awaitable_);
1187		32x	self->active_ops_ = nullptr;
1188		32x	}
1189		32x	} g{self_};
1190		32x	return self_->active_ops_->await_resume(
1191	1/1 ✓ Branch 1 taken 22 times.	54x	self_->cached_awaitable_);
1192		32x	}
1193			};
1194		32x	return awaitable{this};
1195			}
1196
1197			template<ConstBufferSequence CB>
1198			io_task<std::size_t>
1199	1/1 ✓ Branch 1 taken 40 times.	40x	any_buffer_sink::write_eof(CB buffers)
1200			{
1201			// Native WriteSink path
1202			if(vt_->construct_write_eof_buffers_awaitable)
1203			{
1204			const_buffer_param<CB> bp(buffers);
1205			std::size_t total = 0;
1206
1207			for(;;)
1208			{
1209			auto bufs = bp.data();
1210			if(bufs.empty())
1211			{
1212			auto [ec] = co_await write_eof();
1213			co_return {ec, total};
1214			}
1215
1216			if(!bp.more())
1217			{
1218			// Last window: send atomically with EOF
1219			auto [ec, n] = co_await write_eof_buffers_(bufs);
1220			total += n;
1221			co_return {ec, total};
1222			}
1223
1224			auto [ec, n] = co_await write_(bufs);
1225			total += n;
1226			if(ec)
1227			co_return {ec, total};
1228			bp.consume(n);
1229			}
1230			}
1231
1232			// Synthesized path: prepare + buffer_copy + commit + commit_eof
1233			buffer_param<CB> bp(buffers);
1234			std::size_t total = 0;
1235
1236			for(;;)
1237			{
1238			auto src = bp.data();
1239			if(src.empty())
1240			break;
1241
1242			mutable_buffer arr[detail::max_iovec_];
1243			auto dst_bufs = prepare(arr);
1244			if(dst_bufs.empty())
1245			{
1246			auto [ec] = co_await commit(0);
1247			if(ec)
1248			co_return {ec, total};
1249			continue;
1250			}
1251
1252			auto n = buffer_copy(dst_bufs, src);
1253			auto [ec] = co_await commit(n);
1254			if(ec)
1255			co_return {ec, total};
1256			bp.consume(n);
1257			total += n;
1258			}
1259
1260			auto [ec] = co_await commit_eof(0);
1261			if(ec)
1262			co_return {ec, total};
1263
1264			co_return {{}, total};
1265		80x	}
1266
1267			//----------------------------------------------------------
1268
1269			static_assert(BufferSink<any_buffer_sink>);
1270			static_assert(WriteSink<any_buffer_sink>);
1271
1272			} // namespace capy
1273			} // namespace boost
1274
1275			#endif
1276