1   
//
1   
//
2   
// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
2   
// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.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_TEST_READ_STREAM_HPP
10   
#ifndef BOOST_CAPY_TEST_READ_STREAM_HPP
11   
#define BOOST_CAPY_TEST_READ_STREAM_HPP
11   
#define BOOST_CAPY_TEST_READ_STREAM_HPP
12  
12  
13   
#include <boost/capy/detail/config.hpp>
13   
#include <boost/capy/detail/config.hpp>
14   
#include <boost/capy/buffers.hpp>
14   
#include <boost/capy/buffers.hpp>
15   
#include <boost/capy/buffers/buffer_copy.hpp>
15   
#include <boost/capy/buffers/buffer_copy.hpp>
16   
#include <boost/capy/buffers/make_buffer.hpp>
16   
#include <boost/capy/buffers/make_buffer.hpp>
17   
#include <boost/capy/cond.hpp>
17   
#include <boost/capy/cond.hpp>
18   
#include <boost/capy/coro.hpp>
18   
#include <boost/capy/coro.hpp>
19   
#include <boost/capy/ex/executor_ref.hpp>
19   
#include <boost/capy/ex/executor_ref.hpp>
20   
#include <boost/capy/io_result.hpp>
20   
#include <boost/capy/io_result.hpp>
21   
#include <boost/capy/test/fuse.hpp>
21   
#include <boost/capy/test/fuse.hpp>
22  
22  
23   
#include <stop_token>
23   
#include <stop_token>
24   
#include <string>
24   
#include <string>
25   
#include <string_view>
25   
#include <string_view>
26  
26  
27   
namespace boost {
27   
namespace boost {
28   
namespace capy {
28   
namespace capy {
29   
namespace test {
29   
namespace test {
30  
30  
31   
/** A mock stream for testing read operations.
31   
/** A mock stream for testing read operations.
32  
32  
33   
    Use this to verify code that performs reads without needing
33   
    Use this to verify code that performs reads without needing
34   
    real I/O. Call @ref provide to supply data, then @ref read_some
34   
    real I/O. Call @ref provide to supply data, then @ref read_some
35   
    to consume it. The associated @ref fuse enables error injection
35   
    to consume it. The associated @ref fuse enables error injection
36   
    at controlled points. An optional `max_read_size` constructor
36   
    at controlled points. An optional `max_read_size` constructor
37   
    parameter limits bytes per read to simulate chunked delivery.
37   
    parameter limits bytes per read to simulate chunked delivery.
38  
38  
 
39 + 
    This class satisfies the @ref ReadStream concept.
 
40 +
39   
    @par Thread Safety
41   
    @par Thread Safety
40   
    Not thread-safe.
42   
    Not thread-safe.
41  
43  
42   
    @par Example
44   
    @par Example
43   
    @code
45   
    @code
44   
    fuse f;
46   
    fuse f;
45   
    read_stream rs( f );
47   
    read_stream rs( f );
46   
    rs.provide( "Hello, " );
48   
    rs.provide( "Hello, " );
47   
    rs.provide( "World!" );
49   
    rs.provide( "World!" );
48  
50  
49   
    auto r = f.armed( [&]( fuse& ) -> task<void> {
51   
    auto r = f.armed( [&]( fuse& ) -> task<void> {
50   
        char buf[32];
52   
        char buf[32];
51   
        auto [ec, n] = co_await rs.read_some(
53   
        auto [ec, n] = co_await rs.read_some(
52   
            mutable_buffer( buf, sizeof( buf ) ) );
54   
            mutable_buffer( buf, sizeof( buf ) ) );
53   
        if( ec )
55   
        if( ec )
54   
            co_return;
56   
            co_return;
55   
        // buf contains "Hello, World!"
57   
        // buf contains "Hello, World!"
56   
    } );
58   
    } );
57   
    @endcode
59   
    @endcode
58  
60  
59 - 
    @see fuse
61 + 
    @see fuse, ReadStream
60   
*/
62   
*/
61   
class read_stream
63   
class read_stream
62   
{
64   
{
63 - 
    fuse* f_;
65 + 
    fuse f_;
64   
    std::string data_;
66   
    std::string data_;
65   
    std::size_t pos_ = 0;
67   
    std::size_t pos_ = 0;
66   
    std::size_t max_read_size_;
68   
    std::size_t max_read_size_;
67  
69  
68   
public:
70   
public:
69   
    /** Construct a read stream.
71   
    /** Construct a read stream.
70  
72  
71   
        @param f The fuse used to inject errors during reads.
73   
        @param f The fuse used to inject errors during reads.
72  
74  
73   
        @param max_read_size Maximum bytes returned per read.
75   
        @param max_read_size Maximum bytes returned per read.
74   
        Use to simulate chunked network delivery.
76   
        Use to simulate chunked network delivery.
75   
    */
77   
    */
76   
    explicit read_stream(
78   
    explicit read_stream(
77 - 
        fuse& f,
79 + 
        fuse f = {},
78   
        std::size_t max_read_size = std::size_t(-1)) noexcept
80   
        std::size_t max_read_size = std::size_t(-1)) noexcept
79 - 
        : f_(&f)
81 + 
        : f_(std::move(f))
80   
        , max_read_size_(max_read_size)
82   
        , max_read_size_(max_read_size)
81   
    {
83   
    {
82   
    }
84   
    }
83  
85  
84   
    /** Append data to be returned by subsequent reads.
86   
    /** Append data to be returned by subsequent reads.
85  
87  
86   
        Multiple calls accumulate data that @ref read_some returns.
88   
        Multiple calls accumulate data that @ref read_some returns.
87  
89  
88   
        @param sv The data to append.
90   
        @param sv The data to append.
89   
    */
91   
    */
90   
    void
92   
    void
91   
    provide(std::string_view sv)
93   
    provide(std::string_view sv)
92   
    {
94   
    {
93   
        data_.append(sv);
95   
        data_.append(sv);
94   
    }
96   
    }
95  
97  
96   
    /// Clear all data and reset the read position.
98   
    /// Clear all data and reset the read position.
97   
    void
99   
    void
98   
    clear() noexcept
100   
    clear() noexcept
99   
    {
101   
    {
100   
        data_.clear();
102   
        data_.clear();
101   
        pos_ = 0;
103   
        pos_ = 0;
102   
    }
104   
    }
103  
105  
104   
    /// Return the number of bytes available for reading.
106   
    /// Return the number of bytes available for reading.
105   
    std::size_t
107   
    std::size_t
106   
    available() const noexcept
108   
    available() const noexcept
107   
    {
109   
    {
108   
        return data_.size() - pos_;
110   
        return data_.size() - pos_;
109   
    }
111   
    }
110  
112  
111   
    /** Asynchronously read data from the stream.
113   
    /** Asynchronously read data from the stream.
112  
114  
113   
        Transfers up to `buffer_size( buffers )` bytes from the internal
115   
        Transfers up to `buffer_size( buffers )` bytes from the internal
114   
        buffer to the provided mutable buffer sequence. If no data remains,
116   
        buffer to the provided mutable buffer sequence. If no data remains,
115   
        returns `error::eof`. Before every read, the attached @ref fuse is
117   
        returns `error::eof`. Before every read, the attached @ref fuse is
116   
        consulted to possibly inject an error for testing fault scenarios.
118   
        consulted to possibly inject an error for testing fault scenarios.
117   
        The returned `std::size_t` is the number of bytes transferred.
119   
        The returned `std::size_t` is the number of bytes transferred.
118  
120  
119   
        @par Effects
121   
        @par Effects
120   
        On success, advances the internal read position by the number of
122   
        On success, advances the internal read position by the number of
121   
        bytes copied. If an error is injected by the fuse, the read position
123   
        bytes copied. If an error is injected by the fuse, the read position
122   
        remains unchanged.
124   
        remains unchanged.
123  
125  
124   
        @par Exception Safety
126   
        @par Exception Safety
125   
        No-throw guarantee.
127   
        No-throw guarantee.
126  
128  
127   
        @param buffers The mutable buffer sequence to receive data.
129   
        @param buffers The mutable buffer sequence to receive data.
128  
130  
129   
        @return An awaitable yielding `(error_code,std::size_t)`.
131   
        @return An awaitable yielding `(error_code,std::size_t)`.
130  
132  
131   
        @see fuse
133   
        @see fuse
132   
    */
134   
    */
133   
    template<MutableBufferSequence MB>
135   
    template<MutableBufferSequence MB>
134   
    auto
136   
    auto
135   
    read_some(MB buffers)
137   
    read_some(MB buffers)
136   
    {
138   
    {
137   
        struct awaitable
139   
        struct awaitable
138   
        {
140   
        {
139   
            read_stream* self_;
141   
            read_stream* self_;
140   
            MB buffers_;
142   
            MB buffers_;
141  
143  
142   
            bool await_ready() const noexcept { return true; }
144   
            bool await_ready() const noexcept { return true; }
143  
145  
144   
            // This method is required to satisfy Capy's IoAwaitable concept,
146   
            // This method is required to satisfy Capy's IoAwaitable concept,
145   
            // but is never called because await_ready() returns true.
147   
            // but is never called because await_ready() returns true.
146   
            //
148   
            //
147   
            // Capy uses a two-layer awaitable system: the promise's
149   
            // Capy uses a two-layer awaitable system: the promise's
148   
            // await_transform wraps awaitables in a transform_awaiter whose
150   
            // await_transform wraps awaitables in a transform_awaiter whose
149   
            // standard await_suspend(coroutine_handle) calls this custom
151   
            // standard await_suspend(coroutine_handle) calls this custom
150   
            // 3-argument overload, passing the executor and stop_token from
152   
            // 3-argument overload, passing the executor and stop_token from
151   
            // the coroutine's context. For synchronous test awaitables like
153   
            // the coroutine's context. For synchronous test awaitables like
152   
            // this one, the coroutine never suspends, so this is not invoked.
154   
            // this one, the coroutine never suspends, so this is not invoked.
153   
            // The signature exists to allow the same awaitable type to work
155   
            // The signature exists to allow the same awaitable type to work
154   
            // with both synchronous (test) and asynchronous (real I/O) code.
156   
            // with both synchronous (test) and asynchronous (real I/O) code.
155   
            void await_suspend(
157   
            void await_suspend(
156   
                coro,
158   
                coro,
157   
                executor_ref,
159   
                executor_ref,
158   
                std::stop_token) const noexcept
160   
                std::stop_token) const noexcept
159   
            {
161   
            {
160   
            }
162   
            }
161  
163  
162   
            io_result<std::size_t>
164   
            io_result<std::size_t>
163   
            await_resume()
165   
            await_resume()
164   
            {
166   
            {
165 - 
                auto ec = self_->f_->maybe_fail();
167 + 
                // Empty buffer is a no-op regardless of
 
168 + 
                // stream state or fuse.
 
169 + 
                if(buffer_empty(buffers_))
 
170 + 
                    return {{}, 0};
 
171 +
 
172 + 
                auto ec = self_->f_.maybe_fail();
166   
                if(ec)
173   
                if(ec)
167   
                    return {ec, 0};
174   
                    return {ec, 0};
168  
175  
169   
                if(self_->pos_ >= self_->data_.size())
176   
                if(self_->pos_ >= self_->data_.size())
170   
                    return {error::eof, 0};
177   
                    return {error::eof, 0};
171  
178  
172   
                std::size_t avail = self_->data_.size() - self_->pos_;
179   
                std::size_t avail = self_->data_.size() - self_->pos_;
173   
                if(avail > self_->max_read_size_)
180   
                if(avail > self_->max_read_size_)
174   
                    avail = self_->max_read_size_;
181   
                    avail = self_->max_read_size_;
175   
                auto src = make_buffer(self_->data_.data() + self_->pos_, avail);
182   
                auto src = make_buffer(self_->data_.data() + self_->pos_, avail);
176   
                std::size_t const n = buffer_copy(buffers_, src);
183   
                std::size_t const n = buffer_copy(buffers_, src);
177   
                self_->pos_ += n;
184   
                self_->pos_ += n;
178   
                return {{}, n};
185   
                return {{}, n};
179   
            }
186   
            }
180   
        };
187   
        };
181   
        return awaitable{this, buffers};
188   
        return awaitable{this, buffers};
182   
    }
189   
    }
183   
};
190   
};
184  
191  
185   
} // test
192   
} // test
186   
} // capy
193   
} // capy
187   
} // boost
194   
} // boost
188  
195  
189   
#endif
196   
#endif