1   
//
1   
//
2   
// Copyright (c) 2025 Vinnie Falco (vinnie dot falco at gmail dot com)
2   
// Copyright (c) 2025 Vinnie Falco (vinnie dot falco at gmail dot com)
3   
//
3   
//
4   
// Distributed under the Boost Software License, Version 1.0. (See accompanying
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)
5   
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6   
//
6   
//
7   
// Official repository: https://github.com/cppalliance/capy
7   
// Official repository: https://github.com/cppalliance/capy
8   
//
8   
//
9  
9  
10   
#ifndef BOOST_CAPY_IO_ANY_BUFFER_SOURCE_HPP
10   
#ifndef BOOST_CAPY_IO_ANY_BUFFER_SOURCE_HPP
11   
#define BOOST_CAPY_IO_ANY_BUFFER_SOURCE_HPP
11   
#define BOOST_CAPY_IO_ANY_BUFFER_SOURCE_HPP
12  
12  
13   
#include <boost/capy/detail/config.hpp>
13   
#include <boost/capy/detail/config.hpp>
14   
#include <boost/capy/detail/await_suspend_helper.hpp>
14   
#include <boost/capy/detail/await_suspend_helper.hpp>
15   
#include <boost/capy/buffers.hpp>
15   
#include <boost/capy/buffers.hpp>
16   
#include <boost/capy/buffers/buffer_copy.hpp>
16   
#include <boost/capy/buffers/buffer_copy.hpp>
 
17 + 
#include <boost/capy/buffers/buffer_param.hpp>
17   
#include <boost/capy/buffers/slice.hpp>
18   
#include <boost/capy/buffers/slice.hpp>
18   
#include <boost/capy/concept/buffer_source.hpp>
19   
#include <boost/capy/concept/buffer_source.hpp>
19   
#include <boost/capy/concept/io_awaitable.hpp>
20   
#include <boost/capy/concept/io_awaitable.hpp>
20   
#include <boost/capy/concept/read_source.hpp>
21   
#include <boost/capy/concept/read_source.hpp>
21   
#include <boost/capy/coro.hpp>
22   
#include <boost/capy/coro.hpp>
22   
#include <boost/capy/error.hpp>
23   
#include <boost/capy/error.hpp>
23   
#include <boost/capy/ex/executor_ref.hpp>
24   
#include <boost/capy/ex/executor_ref.hpp>
24   
#include <boost/capy/io_result.hpp>
25   
#include <boost/capy/io_result.hpp>
25 - 
#include <boost/capy/task.hpp>
26 + 
#include <boost/capy/io_task.hpp>
26  
27  
27   
#include <concepts>
28   
#include <concepts>
28   
#include <coroutine>
29   
#include <coroutine>
29   
#include <cstddef>
30   
#include <cstddef>
30   
#include <exception>
31   
#include <exception>
31   
#include <new>
32   
#include <new>
32   
#include <span>
33   
#include <span>
33   
#include <stop_token>
34   
#include <stop_token>
34   
#include <system_error>
35   
#include <system_error>
35   
#include <utility>
36   
#include <utility>
36  
37  
37   
namespace boost {
38   
namespace boost {
38   
namespace capy {
39   
namespace capy {
39  
40  
40   
/** Type-erased wrapper for any BufferSource.
41   
/** Type-erased wrapper for any BufferSource.
41  
42  
42   
    This class provides type erasure for any type satisfying the
43   
    This class provides type erasure for any type satisfying the
43   
    @ref BufferSource concept, enabling runtime polymorphism for
44   
    @ref BufferSource concept, enabling runtime polymorphism for
44 - 
    buffer pull operations. The wrapper also satisfies @ref ReadSource,
45 + 
    buffer pull operations. It uses cached awaitable storage to achieve
45 - 
    allowing it to be used with code expecting either interface.
46 + 
    zero steady-state allocation after construction.
46 - 
    It uses cached awaitable storage to achieve zero steady-state
 
47 - 
    allocation after construction.
 
48  
47  
49 - 
    The wrapper also satisfies @ref ReadSource through the templated
48 + 
    The wrapper also satisfies @ref ReadSource. When the wrapped type
50 - 
    @ref read method. This method copies data from the source's
49 + 
    satisfies only @ref BufferSource, the read operations are
51 - 
    internal buffers into the caller's buffers, incurring one extra
50 + 
    synthesized using @ref pull and @ref consume with an extra
52 - 
    buffer copy compared to using @ref pull and @ref consume directly.
51 + 
    buffer copy. When the wrapped type satisfies both @ref BufferSource
 
52 + 
    and @ref ReadSource, the native read operations are forwarded
 
53 + 
    directly across the virtual boundary, avoiding the copy.
53  
54  
54   
    The wrapper supports two construction modes:
55   
    The wrapper supports two construction modes:
55   
    - **Owning**: Pass by value to transfer ownership. The wrapper
56   
    - **Owning**: Pass by value to transfer ownership. The wrapper
56   
      allocates storage and owns the source.
57   
      allocates storage and owns the source.
57   
    - **Reference**: Pass a pointer to wrap without ownership. The
58   
    - **Reference**: Pass a pointer to wrap without ownership. The
58   
      pointed-to source must outlive this wrapper.
59   
      pointed-to source must outlive this wrapper.
59  
60  
 
61 + 
    Within each mode, the vtable is populated at compile time based
 
62 + 
    on whether the wrapped type also satisfies @ref ReadSource:
 
63 + 
    - **BufferSource only**: @ref read_some and @ref read are
 
64 + 
      synthesized from @ref pull and @ref consume, incurring one
 
65 + 
      buffer copy per operation.
 
66 + 
    - **BufferSource + ReadSource**: All read operations are
 
67 + 
      forwarded natively through the type-erased boundary with
 
68 + 
      no extra copy.
 
69 +
60   
    @par Awaitable Preallocation
70   
    @par Awaitable Preallocation
61   
    The constructor preallocates storage for the type-erased awaitable.
71   
    The constructor preallocates storage for the type-erased awaitable.
62   
    This reserves all virtual address space at server startup
72   
    This reserves all virtual address space at server startup
63   
    so memory usage can be measured up front, rather than
73   
    so memory usage can be measured up front, rather than
64   
    allocating piecemeal as traffic arrives.
74   
    allocating piecemeal as traffic arrives.
65  
75  
66   
    @par Thread Safety
76   
    @par Thread Safety
67   
    Not thread-safe. Concurrent operations on the same wrapper
77   
    Not thread-safe. Concurrent operations on the same wrapper
68   
    are undefined behavior.
78   
    are undefined behavior.
69  
79  
70   
    @par Example
80   
    @par Example
71   
    @code
81   
    @code
72   
    // Owning - takes ownership of the source
82   
    // Owning - takes ownership of the source
73   
    any_buffer_source abs(some_buffer_source{args...});
83   
    any_buffer_source abs(some_buffer_source{args...});
74  
84  
75   
    // Reference - wraps without ownership
85   
    // Reference - wraps without ownership
76   
    some_buffer_source src;
86   
    some_buffer_source src;
77   
    any_buffer_source abs(&src);
87   
    any_buffer_source abs(&src);
78  
88  
79   
    const_buffer arr[16];
89   
    const_buffer arr[16];
80   
    auto [ec, bufs] = co_await abs.pull(arr);
90   
    auto [ec, bufs] = co_await abs.pull(arr);
 
91 +
 
92 + 
    // ReadSource interface also available
 
93 + 
    char buf[64];
 
94 + 
    auto [ec2, n] = co_await abs.read_some(mutable_buffer(buf, 64));
81   
    @endcode
95   
    @endcode
82  
96  
83   
    @see any_buffer_sink, BufferSource, ReadSource
97   
    @see any_buffer_sink, BufferSource, ReadSource
84   
*/
98   
*/
85   
class any_buffer_source
99   
class any_buffer_source
86   
{
100   
{
87   
    struct vtable;
101   
    struct vtable;
88   
    struct awaitable_ops;
102   
    struct awaitable_ops;
 
103 + 
    struct read_awaitable_ops;
89  
104  
90   
    template<BufferSource S>
105   
    template<BufferSource S>
91   
    struct vtable_for_impl;
106   
    struct vtable_for_impl;
92  
107  
 
108 + 
    // hot-path members first for cache locality
93   
    void* source_ = nullptr;
109   
    void* source_ = nullptr;
94   
    vtable const* vt_ = nullptr;
110   
    vtable const* vt_ = nullptr;
95 - 
    void* storage_ = nullptr;
 
96   
    void* cached_awaitable_ = nullptr;
111   
    void* cached_awaitable_ = nullptr;
97   
    awaitable_ops const* active_ops_ = nullptr;
112   
    awaitable_ops const* active_ops_ = nullptr;
 
113 + 
    read_awaitable_ops const* active_read_ops_ = nullptr;
 
114 + 
    void* storage_ = nullptr;
98  
115  
99   
public:
116   
public:
100   
    /** Destructor.
117   
    /** Destructor.
101  
118  
102   
        Destroys the owned source (if any) and releases the cached
119   
        Destroys the owned source (if any) and releases the cached
103   
        awaitable storage.
120   
        awaitable storage.
104   
    */
121   
    */
105   
    ~any_buffer_source();
122   
    ~any_buffer_source();
106  
123  
107   
    /** Default constructor.
124   
    /** Default constructor.
108  
125  
109   
        Constructs an empty wrapper. Operations on a default-constructed
126   
        Constructs an empty wrapper. Operations on a default-constructed
110   
        wrapper result in undefined behavior.
127   
        wrapper result in undefined behavior.
111   
    */
128   
    */
112   
    any_buffer_source() = default;
129   
    any_buffer_source() = default;
113  
130  
114   
    /** Non-copyable.
131   
    /** Non-copyable.
115  
132  
116   
        The awaitable cache is per-instance and cannot be shared.
133   
        The awaitable cache is per-instance and cannot be shared.
117   
    */
134   
    */
118   
    any_buffer_source(any_buffer_source const&) = delete;
135   
    any_buffer_source(any_buffer_source const&) = delete;
119   
    any_buffer_source& operator=(any_buffer_source const&) = delete;
136   
    any_buffer_source& operator=(any_buffer_source const&) = delete;
120  
137  
121   
    /** Move constructor.
138   
    /** Move constructor.
122  
139  
123   
        Transfers ownership of the wrapped source (if owned) and
140   
        Transfers ownership of the wrapped source (if owned) and
124   
        cached awaitable storage from `other`. After the move, `other` is
141   
        cached awaitable storage from `other`. After the move, `other` is
125   
        in a default-constructed state.
142   
        in a default-constructed state.
126  
143  
127   
        @param other The wrapper to move from.
144   
        @param other The wrapper to move from.
128   
    */
145   
    */
129   
    any_buffer_source(any_buffer_source&& other) noexcept
146   
    any_buffer_source(any_buffer_source&& other) noexcept
130   
        : source_(std::exchange(other.source_, nullptr))
147   
        : source_(std::exchange(other.source_, nullptr))
131   
        , vt_(std::exchange(other.vt_, nullptr))
148   
        , vt_(std::exchange(other.vt_, nullptr))
132 - 
        , storage_(std::exchange(other.storage_, nullptr))
 
133   
        , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
149   
        , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
134   
        , active_ops_(std::exchange(other.active_ops_, nullptr))
150   
        , active_ops_(std::exchange(other.active_ops_, nullptr))
 
151 + 
        , active_read_ops_(std::exchange(other.active_read_ops_, nullptr))
 
152 + 
        , storage_(std::exchange(other.storage_, nullptr))
135   
    {
153   
    {
136   
    }
154   
    }
137  
155  
138   
    /** Move assignment operator.
156   
    /** Move assignment operator.
139  
157  
140   
        Destroys any owned source and releases existing resources,
158   
        Destroys any owned source and releases existing resources,
141   
        then transfers ownership from `other`.
159   
        then transfers ownership from `other`.
142  
160  
143   
        @param other The wrapper to move from.
161   
        @param other The wrapper to move from.
144   
        @return Reference to this wrapper.
162   
        @return Reference to this wrapper.
145   
    */
163   
    */
146   
    any_buffer_source&
164   
    any_buffer_source&
147   
    operator=(any_buffer_source&& other) noexcept;
165   
    operator=(any_buffer_source&& other) noexcept;
148  
166  
149   
    /** Construct by taking ownership of a BufferSource.
167   
    /** Construct by taking ownership of a BufferSource.
150  
168  
151   
        Allocates storage and moves the source into this wrapper.
169   
        Allocates storage and moves the source into this wrapper.
152 - 
        The wrapper owns the source and will destroy it.
170 + 
        The wrapper owns the source and will destroy it. If `S` also
 
171 + 
        satisfies @ref ReadSource, native read operations are
 
172 + 
        forwarded through the virtual boundary.
153  
173  
154   
        @param s The source to take ownership of.
174   
        @param s The source to take ownership of.
155   
    */
175   
    */
156   
    template<BufferSource S>
176   
    template<BufferSource S>
157   
        requires (!std::same_as<std::decay_t<S>, any_buffer_source>)
177   
        requires (!std::same_as<std::decay_t<S>, any_buffer_source>)
158   
    any_buffer_source(S s);
178   
    any_buffer_source(S s);
159  
179  
160   
    /** Construct by wrapping a BufferSource without ownership.
180   
    /** Construct by wrapping a BufferSource without ownership.
161  
181  
162   
        Wraps the given source by pointer. The source must remain
182   
        Wraps the given source by pointer. The source must remain
163 - 
        valid for the lifetime of this wrapper.
183 + 
        valid for the lifetime of this wrapper. If `S` also
 
184 + 
        satisfies @ref ReadSource, native read operations are
 
185 + 
        forwarded through the virtual boundary.
164  
186  
165   
        @param s Pointer to the source to wrap.
187   
        @param s Pointer to the source to wrap.
166   
    */
188   
    */
167   
    template<BufferSource S>
189   
    template<BufferSource S>
168   
    any_buffer_source(S* s);
190   
    any_buffer_source(S* s);
169  
191  
170   
    /** Check if the wrapper contains a valid source.
192   
    /** Check if the wrapper contains a valid source.
171  
193  
172   
        @return `true` if wrapping a source, `false` if default-constructed
194   
        @return `true` if wrapping a source, `false` if default-constructed
173   
            or moved-from.
195   
            or moved-from.
174   
    */
196   
    */
175   
    bool
197   
    bool
176   
    has_value() const noexcept
198   
    has_value() const noexcept
177   
    {
199   
    {
178   
        return source_ != nullptr;
200   
        return source_ != nullptr;
179   
    }
201   
    }
180  
202  
181   
    /** Check if the wrapper contains a valid source.
203   
    /** Check if the wrapper contains a valid source.
182  
204  
183   
        @return `true` if wrapping a source, `false` if default-constructed
205   
        @return `true` if wrapping a source, `false` if default-constructed
184   
            or moved-from.
206   
            or moved-from.
185   
    */
207   
    */
186   
    explicit
208   
    explicit
187   
    operator bool() const noexcept
209   
    operator bool() const noexcept
188   
    {
210   
    {
189   
        return has_value();
211   
        return has_value();
190   
    }
212   
    }
191  
213  
192   
    /** Consume bytes from the source.
214   
    /** Consume bytes from the source.
193  
215  
194   
        Advances the internal read position of the underlying source
216   
        Advances the internal read position of the underlying source
195   
        by the specified number of bytes. The next call to @ref pull
217   
        by the specified number of bytes. The next call to @ref pull
196   
        returns data starting after the consumed bytes.
218   
        returns data starting after the consumed bytes.
197  
219  
198   
        @param n The number of bytes to consume. Must not exceed the
220   
        @param n The number of bytes to consume. Must not exceed the
199   
        total size of buffers returned by the previous @ref pull.
221   
        total size of buffers returned by the previous @ref pull.
200  
222  
201   
        @par Preconditions
223   
        @par Preconditions
202   
        The wrapper must contain a valid source (`has_value() == true`).
224   
        The wrapper must contain a valid source (`has_value() == true`).
203   
    */
225   
    */
204   
    void
226   
    void
205   
    consume(std::size_t n) noexcept;
227   
    consume(std::size_t n) noexcept;
206  
228  
207   
    /** Pull buffer data from the source.
229   
    /** Pull buffer data from the source.
208  
230  
209   
        Fills the provided span with buffer descriptors from the
231   
        Fills the provided span with buffer descriptors from the
210   
        underlying source. The operation completes when data is
232   
        underlying source. The operation completes when data is
211   
        available, the source is exhausted, or an error occurs.
233   
        available, the source is exhausted, or an error occurs.
212  
234  
213   
        @param dest Span of const_buffer to fill.
235   
        @param dest Span of const_buffer to fill.
214  
236  
215   
        @return An awaitable yielding `(error_code,std::span<const_buffer>)`.
237   
        @return An awaitable yielding `(error_code,std::span<const_buffer>)`.
216   
            On success with data, a non-empty span of filled buffers.
238   
            On success with data, a non-empty span of filled buffers.
217 - 
            On success with empty span, source is exhausted.
239 + 
            On EOF, `ec == cond::eof` and span is empty.
218  
240  
219   
        @par Preconditions
241   
        @par Preconditions
220   
        The wrapper must contain a valid source (`has_value() == true`).
242   
        The wrapper must contain a valid source (`has_value() == true`).
 
243 + 
        The caller must not call this function again after a prior
 
244 + 
        call returned an error.
221   
    */
245   
    */
222   
    auto
246   
    auto
223   
    pull(std::span<const_buffer> dest);
247   
    pull(std::span<const_buffer> dest);
224  
248  
225 - 
    /** Read data into a mutable buffer sequence.
249 + 
    /** Read some data into a mutable buffer sequence.
226  
250  
227 - 
        Fills the provided buffer sequence by pulling data from the
251 + 
        Reads one or more bytes into the caller's buffers. May fill
228 - 
        underlying source and copying it into the caller's buffers.
252 + 
        less than the full sequence.
229 - 
        This satisfies @ref ReadSource but incurs a copy; for zero-copy
 
230 - 
        access, use @ref pull and @ref consume instead.
 
231  
253  
232 - 
        @note This operation copies data from the source's internal
254 + 
        When the wrapped type provides native @ref ReadSource support,
233 - 
        buffers into the caller's buffers. For zero-copy reads,
255 + 
        the operation forwards directly. Otherwise it is synthesized
234 - 
        use @ref pull and @ref consume directly.
256 + 
        from @ref pull, @ref buffer_copy, and @ref consume.
 
257 +
 
258 + 
        @param buffers The buffer sequence to fill.
 
259 +
 
260 + 
        @return An awaitable yielding `(error_code,std::size_t)`.
 
261 +
 
262 + 
        @par Preconditions
 
263 + 
        The wrapper must contain a valid source (`has_value() == true`).
 
264 + 
        The caller must not call this function again after a prior
 
265 + 
        call returned an error (including EOF).
 
266 +
 
267 + 
        @see pull, consume
 
268 + 
    */
 
269 + 
    template<MutableBufferSequence MB>
 
270 + 
    io_task<std::size_t>
 
271 + 
    read_some(MB buffers);
 
272 +
 
273 + 
    /** Read data into a mutable buffer sequence.
 
274 +
 
275 + 
        Fills the provided buffer sequence completely. When the
 
276 + 
        wrapped type provides native @ref ReadSource support, each
 
277 + 
        window is forwarded directly. Otherwise the data is
 
278 + 
        synthesized from @ref pull, @ref buffer_copy, and @ref consume.
235  
279  
236   
        @param buffers The buffer sequence to fill.
280   
        @param buffers The buffer sequence to fill.
237  
281  
238   
        @return An awaitable yielding `(error_code,std::size_t)`.
282   
        @return An awaitable yielding `(error_code,std::size_t)`.
239   
            On success, `n == buffer_size(buffers)`.
283   
            On success, `n == buffer_size(buffers)`.
240   
            On EOF, `ec == error::eof` and `n` is bytes transferred.
284   
            On EOF, `ec == error::eof` and `n` is bytes transferred.
241  
285  
242   
        @par Preconditions
286   
        @par Preconditions
243   
        The wrapper must contain a valid source (`has_value() == true`).
287   
        The wrapper must contain a valid source (`has_value() == true`).
 
288 + 
        The caller must not call this function again after a prior
 
289 + 
        call returned an error (including EOF).
244  
290  
245   
        @see pull, consume
291   
        @see pull, consume
246   
    */
292   
    */
247   
    template<MutableBufferSequence MB>
293   
    template<MutableBufferSequence MB>
248 - 
    task<io_result<std::size_t>>
294 + 
    io_task<std::size_t>
249   
    read(MB buffers);
295   
    read(MB buffers);
250  
296  
251   
protected:
297   
protected:
252   
    /** Rebind to a new source after move.
298   
    /** Rebind to a new source after move.
253  
299  
254   
        Updates the internal pointer to reference a new source object.
300   
        Updates the internal pointer to reference a new source object.
255   
        Used by owning wrappers after move assignment when the owned
301   
        Used by owning wrappers after move assignment when the owned
256   
        object has moved to a new location.
302   
        object has moved to a new location.
257  
303  
258   
        @param new_source The new source to bind to. Must be the same
304   
        @param new_source The new source to bind to. Must be the same
259   
            type as the original source.
305   
            type as the original source.
260  
306  
261   
        @note Terminates if called with a source of different type
307   
        @note Terminates if called with a source of different type
262   
            than the original.
308   
            than the original.
263   
    */
309   
    */
264   
    template<BufferSource S>
310   
    template<BufferSource S>
265   
    void
311   
    void
266   
    rebind(S& new_source) noexcept
312   
    rebind(S& new_source) noexcept
267   
    {
313   
    {
268   
        if(vt_ != &vtable_for_impl<S>::value)
314   
        if(vt_ != &vtable_for_impl<S>::value)
269   
            std::terminate();
315   
            std::terminate();
270   
        source_ = &new_source;
316   
        source_ = &new_source;
271   
    }
317   
    }
 
318 +
 
319 + 
private:
 
320 + 
    /** Forward a partial read through the vtable.
 
321 +
 
322 + 
        Constructs the underlying `read_some` awaitable in
 
323 + 
        cached storage and returns a type-erased awaitable.
 
324 + 
    */
 
325 + 
    auto
 
326 + 
    read_some_(std::span<mutable_buffer const> buffers);
 
327 +
 
328 + 
    /** Forward a complete read through the vtable.
 
329 +
 
330 + 
        Constructs the underlying `read` awaitable in
 
331 + 
        cached storage and returns a type-erased awaitable.
 
332 + 
    */
 
333 + 
    auto
 
334 + 
    read_(std::span<mutable_buffer const> buffers);
272   
};
335   
};
273  
336  
274   
//----------------------------------------------------------
337   
//----------------------------------------------------------
275  
338  
 
339 + 
/** Type-erased ops for awaitables yielding `io_result<std::span<const_buffer>>`. */
276   
struct any_buffer_source::awaitable_ops
340   
struct any_buffer_source::awaitable_ops
277   
{
341   
{
278   
    bool (*await_ready)(void*);
342   
    bool (*await_ready)(void*);
279   
    coro (*await_suspend)(void*, coro, executor_ref, std::stop_token);
343   
    coro (*await_suspend)(void*, coro, executor_ref, std::stop_token);
280   
    io_result<std::span<const_buffer>> (*await_resume)(void*);
344   
    io_result<std::span<const_buffer>> (*await_resume)(void*);
281   
    void (*destroy)(void*) noexcept;
345   
    void (*destroy)(void*) noexcept;
282   
};
346   
};
283  
347  
 
348 + 
/** Type-erased ops for awaitables yielding `io_result<std::size_t>`. */
 
349 + 
struct any_buffer_source::read_awaitable_ops
 
350 + 
{
 
351 + 
    bool (*await_ready)(void*);
 
352 + 
    coro (*await_suspend)(void*, coro, executor_ref, std::stop_token);
 
353 + 
    io_result<std::size_t> (*await_resume)(void*);
 
354 + 
    void (*destroy)(void*) noexcept;
 
355 + 
};
 
356 +
284   
struct any_buffer_source::vtable
357   
struct any_buffer_source::vtable
285   
{
358   
{
 
359 + 
    // BufferSource ops (always populated)
286   
    void (*destroy)(void*) noexcept;
360   
    void (*destroy)(void*) noexcept;
287   
    void (*do_consume)(void* source, std::size_t n) noexcept;
361   
    void (*do_consume)(void* source, std::size_t n) noexcept;
288   
    std::size_t awaitable_size;
362   
    std::size_t awaitable_size;
289   
    std::size_t awaitable_align;
363   
    std::size_t awaitable_align;
290   
    awaitable_ops const* (*construct_awaitable)(
364   
    awaitable_ops const* (*construct_awaitable)(
291   
        void* source,
365   
        void* source,
292   
        void* storage,
366   
        void* storage,
293   
        std::span<const_buffer> dest);
367   
        std::span<const_buffer> dest);
 
368 +
 
369 + 
    // ReadSource forwarding (null when wrapped type is BufferSource-only)
 
370 + 
    read_awaitable_ops const* (*construct_read_some_awaitable)(
 
371 + 
        void* source,
 
372 + 
        void* storage,
 
373 + 
        std::span<mutable_buffer const> buffers);
 
374 + 
    read_awaitable_ops const* (*construct_read_awaitable)(
 
375 + 
        void* source,
 
376 + 
        void* storage,
 
377 + 
        std::span<mutable_buffer const> buffers);
294   
};
378   
};
295  
379  
296   
template<BufferSource S>
380   
template<BufferSource S>
297   
struct any_buffer_source::vtable_for_impl
381   
struct any_buffer_source::vtable_for_impl
298   
{
382   
{
299 - 
    using Awaitable = decltype(std::declval<S&>().pull(
383 + 
    using PullAwaitable = decltype(std::declval<S&>().pull(
300   
        std::declval<std::span<const_buffer>>()));
384   
        std::declval<std::span<const_buffer>>()));
301  
385  
302   
    static void
386   
    static void
303   
    do_destroy_impl(void* source) noexcept
387   
    do_destroy_impl(void* source) noexcept
304   
    {
388   
    {
305   
        static_cast<S*>(source)->~S();
389   
        static_cast<S*>(source)->~S();
306   
    }
390   
    }
307  
391  
308   
    static void
392   
    static void
309   
    do_consume_impl(void* source, std::size_t n) noexcept
393   
    do_consume_impl(void* source, std::size_t n) noexcept
310   
    {
394   
    {
311   
        static_cast<S*>(source)->consume(n);
395   
        static_cast<S*>(source)->consume(n);
312   
    }
396   
    }
313  
397  
314   
    static awaitable_ops const*
398   
    static awaitable_ops const*
315   
    construct_awaitable_impl(
399   
    construct_awaitable_impl(
316   
        void* source,
400   
        void* source,
317   
        void* storage,
401   
        void* storage,
318   
        std::span<const_buffer> dest)
402   
        std::span<const_buffer> dest)
319   
    {
403   
    {
320   
        auto& s = *static_cast<S*>(source);
404   
        auto& s = *static_cast<S*>(source);
321 - 
        ::new(storage) Awaitable(s.pull(dest));
405 + 
        ::new(storage) PullAwaitable(s.pull(dest));
322  
406  
323   
        static constexpr awaitable_ops ops = {
407   
        static constexpr awaitable_ops ops = {
324   
            +[](void* p) {
408   
            +[](void* p) {
325 - 
                return static_cast<Awaitable*>(p)->await_ready();
409 + 
                return static_cast<PullAwaitable*>(p)->await_ready();
326   
            },
410   
            },
327   
            +[](void* p, coro h, executor_ref ex, std::stop_token token) {
411   
            +[](void* p, coro h, executor_ref ex, std::stop_token token) {
328   
                return detail::call_await_suspend(
412   
                return detail::call_await_suspend(
329 - 
                    static_cast<Awaitable*>(p), h, ex, token);
413 + 
                    static_cast<PullAwaitable*>(p), h, ex, token);
330   
            },
414   
            },
331   
            +[](void* p) {
415   
            +[](void* p) {
332 - 
                return static_cast<Awaitable*>(p)->await_resume();
416 + 
                return static_cast<PullAwaitable*>(p)->await_resume();
333   
            },
417   
            },
334   
            +[](void* p) noexcept {
418   
            +[](void* p) noexcept {
335 - 
                static_cast<Awaitable*>(p)->~Awaitable();
419 + 
                static_cast<PullAwaitable*>(p)->~PullAwaitable();
336   
            }
420   
            }
337   
        };
421   
        };
338   
        return &ops;
422   
        return &ops;
339   
    }
423   
    }
340  
424  
341 - 
    static constexpr vtable value = {
425 + 
    //------------------------------------------------------
342 - 
        &do_destroy_impl,
426 + 
    // ReadSource forwarding (only instantiated when ReadSource<S>)
343 - 
        &do_consume_impl,
427 +
344 - 
        sizeof(Awaitable),
428 + 
    static read_awaitable_ops const*
345 - 
        alignof(Awaitable),
429 + 
    construct_read_some_awaitable_impl(
346 - 
        &construct_awaitable_impl
430 + 
        void* source,
347 - 
    };
431 + 
        void* storage,
 
432 + 
        std::span<mutable_buffer const> buffers)
 
433 + 
        requires ReadSource<S>
 
434 + 
    {
 
435 + 
        using Aw = decltype(std::declval<S&>().read_some(
 
436 + 
            std::span<mutable_buffer const>{}));
 
437 + 
        auto& s = *static_cast<S*>(source);
 
438 + 
        ::new(storage) Aw(s.read_some(buffers));
 
439 +
 
440 + 
        static constexpr read_awaitable_ops ops = {
 
441 + 
            +[](void* p) {
 
442 + 
                return static_cast<Aw*>(p)->await_ready();
 
443 + 
            },
 
444 + 
            +[](void* p, coro h, executor_ref ex, std::stop_token token) {
 
445 + 
                return detail::call_await_suspend(
 
446 + 
                    static_cast<Aw*>(p), h, ex, token);
 
447 + 
            },
 
448 + 
            +[](void* p) {
 
449 + 
                return static_cast<Aw*>(p)->await_resume();
 
450 + 
            },
 
451 + 
            +[](void* p) noexcept {
 
452 + 
                static_cast<Aw*>(p)->~Aw();
 
453 + 
            }
 
454 + 
        };
 
455 + 
        return &ops;
 
456 + 
    }
 
457 +
 
458 + 
    static read_awaitable_ops const*
 
459 + 
    construct_read_awaitable_impl(
 
460 + 
        void* source,
 
461 + 
        void* storage,
 
462 + 
        std::span<mutable_buffer const> buffers)
 
463 + 
        requires ReadSource<S>
 
464 + 
    {
 
465 + 
        using Aw = decltype(std::declval<S&>().read(
 
466 + 
            std::span<mutable_buffer const>{}));
 
467 + 
        auto& s = *static_cast<S*>(source);
 
468 + 
        ::new(storage) Aw(s.read(buffers));
 
469 +
 
470 + 
        static constexpr read_awaitable_ops ops = {
 
471 + 
            +[](void* p) {
 
472 + 
                return static_cast<Aw*>(p)->await_ready();
 
473 + 
            },
 
474 + 
            +[](void* p, coro h, executor_ref ex, std::stop_token token) {
 
475 + 
                return detail::call_await_suspend(
 
476 + 
                    static_cast<Aw*>(p), h, ex, token);
 
477 + 
            },
 
478 + 
            +[](void* p) {
 
479 + 
                return static_cast<Aw*>(p)->await_resume();
 
480 + 
            },
 
481 + 
            +[](void* p) noexcept {
 
482 + 
                static_cast<Aw*>(p)->~Aw();
 
483 + 
            }
 
484 + 
        };
 
485 + 
        return &ops;
 
486 + 
    }
 
487 +
 
488 + 
    //------------------------------------------------------
 
489 +
 
490 + 
    static consteval std::size_t
 
491 + 
    compute_max_size() noexcept
 
492 + 
    {
 
493 + 
        std::size_t s = sizeof(PullAwaitable);
 
494 + 
        if constexpr (ReadSource<S>)
 
495 + 
        {
 
496 + 
            using RS = decltype(std::declval<S&>().read_some(
 
497 + 
                std::span<mutable_buffer const>{}));
 
498 + 
            using R = decltype(std::declval<S&>().read(
 
499 + 
                std::span<mutable_buffer const>{}));
 
500 +
 
501 + 
            if(sizeof(RS) > s) s = sizeof(RS);
 
502 + 
            if(sizeof(R) > s) s = sizeof(R);
 
503 + 
        }
 
504 + 
        return s;
 
505 + 
    }
 
506 +
 
507 + 
    static consteval std::size_t
 
508 + 
    compute_max_align() noexcept
 
509 + 
    {
 
510 + 
        std::size_t a = alignof(PullAwaitable);
 
511 + 
        if constexpr (ReadSource<S>)
 
512 + 
        {
 
513 + 
            using RS = decltype(std::declval<S&>().read_some(
 
514 + 
                std::span<mutable_buffer const>{}));
 
515 + 
            using R = decltype(std::declval<S&>().read(
 
516 + 
                std::span<mutable_buffer const>{}));
 
517 +
 
518 + 
            if(alignof(RS) > a) a = alignof(RS);
 
519 + 
            if(alignof(R) > a) a = alignof(R);
 
520 + 
        }
 
521 + 
        return a;
 
522 + 
    }
 
523 +
 
524 + 
    static consteval vtable
 
525 + 
    make_vtable() noexcept
 
526 + 
    {
 
527 + 
        vtable v{};
 
528 + 
        v.destroy = &do_destroy_impl;
 
529 + 
        v.do_consume = &do_consume_impl;
 
530 + 
        v.awaitable_size = compute_max_size();
 
531 + 
        v.awaitable_align = compute_max_align();
 
532 + 
        v.construct_awaitable = &construct_awaitable_impl;
 
533 + 
        v.construct_read_some_awaitable = nullptr;
 
534 + 
        v.construct_read_awaitable = nullptr;
 
535 +
 
536 + 
        if constexpr (ReadSource<S>)
 
537 + 
        {
 
538 + 
            v.construct_read_some_awaitable =
 
539 + 
                &construct_read_some_awaitable_impl;
 
540 + 
            v.construct_read_awaitable =
 
541 + 
                &construct_read_awaitable_impl;
 
542 + 
        }
 
543 + 
        return v;
 
544 + 
    }
 
545 +
 
546 + 
    static constexpr vtable value = make_vtable();
348   
};
547   
};
349  
548  
350   
//----------------------------------------------------------
549   
//----------------------------------------------------------
351  
550  
352   
inline
551   
inline
353   
any_buffer_source::~any_buffer_source()
552   
any_buffer_source::~any_buffer_source()
354   
{
553   
{
355   
    if(storage_)
554   
    if(storage_)
356   
    {
555   
    {
357   
        vt_->destroy(source_);
556   
        vt_->destroy(source_);
358   
        ::operator delete(storage_);
557   
        ::operator delete(storage_);
359   
    }
558   
    }
360   
    if(cached_awaitable_)
559   
    if(cached_awaitable_)
361   
        ::operator delete(cached_awaitable_);
560   
        ::operator delete(cached_awaitable_);
362   
}
561   
}
363  
562  
364   
inline any_buffer_source&
563   
inline any_buffer_source&
365   
any_buffer_source::operator=(any_buffer_source&& other) noexcept
564   
any_buffer_source::operator=(any_buffer_source&& other) noexcept
366   
{
565   
{
367   
    if(this != &other)
566   
    if(this != &other)
368   
    {
567   
    {
369   
        if(storage_)
568   
        if(storage_)
370   
        {
569   
        {
371   
            vt_->destroy(source_);
570   
            vt_->destroy(source_);
372   
            ::operator delete(storage_);
571   
            ::operator delete(storage_);
373   
        }
572   
        }
374   
        if(cached_awaitable_)
573   
        if(cached_awaitable_)
375   
            ::operator delete(cached_awaitable_);
574   
            ::operator delete(cached_awaitable_);
376   
        source_ = std::exchange(other.source_, nullptr);
575   
        source_ = std::exchange(other.source_, nullptr);
377   
        vt_ = std::exchange(other.vt_, nullptr);
576   
        vt_ = std::exchange(other.vt_, nullptr);
378   
        cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
577   
        cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
379   
        storage_ = std::exchange(other.storage_, nullptr);
578   
        storage_ = std::exchange(other.storage_, nullptr);
380   
        active_ops_ = std::exchange(other.active_ops_, nullptr);
579   
        active_ops_ = std::exchange(other.active_ops_, nullptr);
 
580 + 
        active_read_ops_ = std::exchange(other.active_read_ops_, nullptr);
381   
    }
581   
    }
382   
    return *this;
582   
    return *this;
383   
}
583   
}
384  
584  
385   
template<BufferSource S>
585   
template<BufferSource S>
386   
    requires (!std::same_as<std::decay_t<S>, any_buffer_source>)
586   
    requires (!std::same_as<std::decay_t<S>, any_buffer_source>)
387   
any_buffer_source::any_buffer_source(S s)
587   
any_buffer_source::any_buffer_source(S s)
388   
    : vt_(&vtable_for_impl<S>::value)
588   
    : vt_(&vtable_for_impl<S>::value)
389   
{
589   
{
390   
    struct guard {
590   
    struct guard {
391   
        any_buffer_source* self;
591   
        any_buffer_source* self;
392   
        bool committed = false;
592   
        bool committed = false;
393   
        ~guard() {
593   
        ~guard() {
394   
            if(!committed && self->storage_) {
594   
            if(!committed && self->storage_) {
395   
                self->vt_->destroy(self->source_);
595   
                self->vt_->destroy(self->source_);
396   
                ::operator delete(self->storage_);
596   
                ::operator delete(self->storage_);
397   
                self->storage_ = nullptr;
597   
                self->storage_ = nullptr;
398   
                self->source_ = nullptr;
598   
                self->source_ = nullptr;
399   
            }
599   
            }
400   
        }
600   
        }
401   
    } g{this};
601   
    } g{this};
402  
602  
403   
    storage_ = ::operator new(sizeof(S));
603   
    storage_ = ::operator new(sizeof(S));
404   
    source_ = ::new(storage_) S(std::move(s));
604   
    source_ = ::new(storage_) S(std::move(s));
405 - 
    // Preallocate the awaitable storage
 
406  
605  
407   
    cached_awaitable_ = ::operator new(vt_->awaitable_size);
606   
    cached_awaitable_ = ::operator new(vt_->awaitable_size);
408  
607  
409   
    g.committed = true;
608   
    g.committed = true;
410   
}
609   
}
411  
610  
412   
template<BufferSource S>
611   
template<BufferSource S>
413   
any_buffer_source::any_buffer_source(S* s)
612   
any_buffer_source::any_buffer_source(S* s)
414   
    : source_(s)
613   
    : source_(s)
415   
    , vt_(&vtable_for_impl<S>::value)
614   
    , vt_(&vtable_for_impl<S>::value)
416 - 
    // Preallocate the awaitable storage
 
417   
{
615   
{
418   
    cached_awaitable_ = ::operator new(vt_->awaitable_size);
616   
    cached_awaitable_ = ::operator new(vt_->awaitable_size);
419   
}
617   
}
420  
618  
421   
//----------------------------------------------------------
619   
//----------------------------------------------------------
422  
620  
423   
inline void
621   
inline void
424   
any_buffer_source::consume(std::size_t n) noexcept
622   
any_buffer_source::consume(std::size_t n) noexcept
425   
{
623   
{
426   
    vt_->do_consume(source_, n);
624   
    vt_->do_consume(source_, n);
427   
}
625   
}
428  
626  
429   
inline auto
627   
inline auto
430   
any_buffer_source::pull(std::span<const_buffer> dest)
628   
any_buffer_source::pull(std::span<const_buffer> dest)
431   
{
629   
{
432   
    struct awaitable
630   
    struct awaitable
433   
    {
631   
    {
434   
        any_buffer_source* self_;
632   
        any_buffer_source* self_;
435   
        std::span<const_buffer> dest_;
633   
        std::span<const_buffer> dest_;
436  
634  
437   
        bool
635   
        bool
438 - 
        await_ready() const noexcept
636 + 
        await_ready()
439 - 
        {
 
440 - 
            return false;
 
441 - 
        }
 
442 -
 
443 - 
        coro
 
444 - 
        await_suspend(coro h, executor_ref ex, std::stop_token token)
 
445 - 
            // Construct the underlying awaitable into cached storage
 
446   
        {
637   
        {
447   
            self_->active_ops_ = self_->vt_->construct_awaitable(
638   
            self_->active_ops_ = self_->vt_->construct_awaitable(
448   
                self_->source_,
639   
                self_->source_,
449   
                self_->cached_awaitable_,
640   
                self_->cached_awaitable_,
450   
                dest_);
641   
                dest_);
 
642 + 
            return self_->active_ops_->await_ready(self_->cached_awaitable_);
 
643 + 
        }
451  
644  
452 - 
            // Check if underlying is immediately ready
645 + 
        coro
453 - 
            if(self_->active_ops_->await_ready(self_->cached_awaitable_))
646 + 
        await_suspend(coro h, executor_ref ex, std::stop_token token)
454 - 
                return h;
647 + 
        {
455 -
 
456 - 
            // Forward to underlying awaitable
 
457   
            return self_->active_ops_->await_suspend(
648   
            return self_->active_ops_->await_suspend(
458   
                self_->cached_awaitable_, h, ex, token);
649   
                self_->cached_awaitable_, h, ex, token);
459   
        }
650   
        }
460  
651  
461   
        io_result<std::span<const_buffer>>
652   
        io_result<std::span<const_buffer>>
462   
        await_resume()
653   
        await_resume()
463   
        {
654   
        {
464   
            struct guard {
655   
            struct guard {
465   
                any_buffer_source* self;
656   
                any_buffer_source* self;
466   
                ~guard() {
657   
                ~guard() {
467   
                    self->active_ops_->destroy(self->cached_awaitable_);
658   
                    self->active_ops_->destroy(self->cached_awaitable_);
468   
                    self->active_ops_ = nullptr;
659   
                    self->active_ops_ = nullptr;
469   
                }
660   
                }
470   
            } g{self_};
661   
            } g{self_};
471   
            return self_->active_ops_->await_resume(
662   
            return self_->active_ops_->await_resume(
472   
                self_->cached_awaitable_);
663   
                self_->cached_awaitable_);
473   
        }
664   
        }
474   
    };
665   
    };
475   
    return awaitable{this, dest};
666   
    return awaitable{this, dest};
476   
}
667   
}
477  
668  
 
669 + 
//----------------------------------------------------------
 
670 + 
// Private helpers for native ReadSource forwarding
 
671 +
 
672 + 
inline auto
 
673 + 
any_buffer_source::read_some_(
 
674 + 
    std::span<mutable_buffer const> buffers)
 
675 + 
{
 
676 + 
    struct awaitable
 
677 + 
    {
 
678 + 
        any_buffer_source* self_;
 
679 + 
        std::span<mutable_buffer const> buffers_;
 
680 +
 
681 + 
        bool
 
682 + 
        await_ready() const noexcept
 
683 + 
        {
 
684 + 
            return false;
 
685 + 
        }
 
686 +
 
687 + 
        coro
 
688 + 
        await_suspend(coro h, executor_ref ex, std::stop_token token)
 
689 + 
        {
 
690 + 
            self_->active_read_ops_ =
 
691 + 
                self_->vt_->construct_read_some_awaitable(
 
692 + 
                    self_->source_,
 
693 + 
                    self_->cached_awaitable_,
 
694 + 
                    buffers_);
 
695 +
 
696 + 
            if(self_->active_read_ops_->await_ready(
 
697 + 
                self_->cached_awaitable_))
 
698 + 
                return h;
 
699 +
 
700 + 
            return self_->active_read_ops_->await_suspend(
 
701 + 
                self_->cached_awaitable_, h, ex, token);
 
702 + 
        }
 
703 +
 
704 + 
        io_result<std::size_t>
 
705 + 
        await_resume()
 
706 + 
        {
 
707 + 
            struct guard {
 
708 + 
                any_buffer_source* self;
 
709 + 
                ~guard() {
 
710 + 
                    self->active_read_ops_->destroy(
 
711 + 
                        self->cached_awaitable_);
 
712 + 
                    self->active_read_ops_ = nullptr;
 
713 + 
                }
 
714 + 
            } g{self_};
 
715 + 
            return self_->active_read_ops_->await_resume(
 
716 + 
                self_->cached_awaitable_);
 
717 + 
        }
 
718 + 
    };
 
719 + 
    return awaitable{this, buffers};
 
720 + 
}
 
721 +
 
722 + 
inline auto
 
723 + 
any_buffer_source::read_(
 
724 + 
    std::span<mutable_buffer const> buffers)
 
725 + 
{
 
726 + 
    struct awaitable
 
727 + 
    {
 
728 + 
        any_buffer_source* self_;
 
729 + 
        std::span<mutable_buffer const> buffers_;
 
730 +
 
731 + 
        bool
 
732 + 
        await_ready() const noexcept
 
733 + 
        {
 
734 + 
            return false;
 
735 + 
        }
 
736 +
 
737 + 
        coro
 
738 + 
        await_suspend(coro h, executor_ref ex, std::stop_token token)
 
739 + 
        {
 
740 + 
            self_->active_read_ops_ =
 
741 + 
                self_->vt_->construct_read_awaitable(
 
742 + 
                    self_->source_,
 
743 + 
                    self_->cached_awaitable_,
 
744 + 
                    buffers_);
 
745 +
 
746 + 
            if(self_->active_read_ops_->await_ready(
 
747 + 
                self_->cached_awaitable_))
 
748 + 
                return h;
 
749 +
 
750 + 
            return self_->active_read_ops_->await_suspend(
 
751 + 
                self_->cached_awaitable_, h, ex, token);
 
752 + 
        }
 
753 +
 
754 + 
        io_result<std::size_t>
 
755 + 
        await_resume()
 
756 + 
        {
 
757 + 
            struct guard {
 
758 + 
                any_buffer_source* self;
 
759 + 
                ~guard() {
 
760 + 
                    self->active_read_ops_->destroy(
 
761 + 
                        self->cached_awaitable_);
 
762 + 
                    self->active_read_ops_ = nullptr;
 
763 + 
                }
 
764 + 
            } g{self_};
 
765 + 
            return self_->active_read_ops_->await_resume(
 
766 + 
                self_->cached_awaitable_);
 
767 + 
        }
 
768 + 
    };
 
769 + 
    return awaitable{this, buffers};
 
770 + 
}
 
771 +
 
772 + 
//----------------------------------------------------------
 
773 + 
// Public ReadSource methods
 
774 +
478   
template<MutableBufferSequence MB>
775   
template<MutableBufferSequence MB>
479 - 
task<io_result<std::size_t>>
776 + 
io_task<std::size_t>
 
777 + 
any_buffer_source::read_some(MB buffers)
 
778 + 
{
 
779 + 
    buffer_param<MB> bp(buffers);
 
780 + 
    auto dest = bp.data();
 
781 + 
    if(dest.empty())
 
782 + 
        co_return {{}, 0};
 
783 +
 
784 + 
    // Native ReadSource path
 
785 + 
    if(vt_->construct_read_some_awaitable)
 
786 + 
        co_return co_await read_some_(dest);
 
787 +
 
788 + 
    // Synthesized path: pull + buffer_copy + consume
 
789 + 
    const_buffer arr[detail::max_iovec_];
 
790 + 
    auto [ec, bufs] = co_await pull(arr);
 
791 + 
    if(ec)
 
792 + 
        co_return {ec, 0};
 
793 +
 
794 + 
    auto n = buffer_copy(dest, bufs);
 
795 + 
    consume(n);
 
796 + 
    co_return {{}, n};
 
797 + 
}
 
798 +
 
799 + 
template<MutableBufferSequence MB>
 
800 + 
io_task<std::size_t>
480   
any_buffer_source::read(MB buffers)
801   
any_buffer_source::read(MB buffers)
481   
{
802   
{
 
803 + 
    buffer_param<MB> bp(buffers);
482 - 
    auto dest = sans_prefix(buffers, 0);
 
483   
    std::size_t total = 0;
804   
    std::size_t total = 0;
484  
805  
485 - 
    while(!buffer_empty(dest))
806 + 
    // Native ReadSource path
 
807 + 
    if(vt_->construct_read_awaitable)
 
808 + 
    {
 
809 + 
        for(;;)
 
810 + 
        {
 
811 + 
            auto dest = bp.data();
 
812 + 
            if(dest.empty())
 
813 + 
                break;
 
814 +
 
815 + 
            auto [ec, n] = co_await read_(dest);
 
816 + 
            total += n;
 
817 + 
            if(ec)
 
818 + 
                co_return {ec, total};
 
819 + 
            bp.consume(n);
 
820 + 
        }
 
821 + 
        co_return {{}, total};
 
822 + 
    }
 
823 +
 
824 + 
    // Synthesized path: pull + buffer_copy + consume
 
825 + 
    for(;;)
486   
    {
826   
    {
 
827 + 
        auto dest = bp.data();
 
828 + 
        if(dest.empty())
 
829 + 
            break;
 
830 +
487   
        const_buffer arr[detail::max_iovec_];
831   
        const_buffer arr[detail::max_iovec_];
488   
        auto [ec, bufs] = co_await pull(arr);
832   
        auto [ec, bufs] = co_await pull(arr);
489  
833  
490   
        if(ec)
834   
        if(ec)
491   
            co_return {ec, total};
835   
            co_return {ec, total};
492 - 
        if(bufs.empty())
 
493 - 
            co_return {error::eof, total};
 
494 -
 
495  
836  
496   
        auto n = buffer_copy(dest, bufs);
837   
        auto n = buffer_copy(dest, bufs);
497   
        consume(n);
838   
        consume(n);
498   
        total += n;
839   
        total += n;
499 - 
        dest = sans_prefix(dest, n);
840 + 
        bp.consume(n);
500   
    }
841   
    }
501  
842  
502   
    co_return {{}, total};
843   
    co_return {{}, total};
503   
}
844   
}
504  
845  
505   
//----------------------------------------------------------
846   
//----------------------------------------------------------
506  
847  
507   
static_assert(BufferSource<any_buffer_source>);
848   
static_assert(BufferSource<any_buffer_source>);
508   
static_assert(ReadSource<any_buffer_source>);
849   
static_assert(ReadSource<any_buffer_source>);
509  
850  
510   
} // namespace capy
851   
} // namespace capy
511   
} // namespace boost
852   
} // namespace boost
512  
853  
513   
#endif
854   
#endif