. /include/boost/capy /test /read_stream.hpp
100.00% Lines (36/36) 100.00% Functions (8/8)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com) 2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3   // Copyright (c) 2026 Michael Vandeberg 3   // Copyright (c) 2026 Michael Vandeberg
4   // 4   //
5   // Distributed under the Boost Software License, Version 1.0. (See accompanying 5   // Distributed under the Boost Software License, Version 1.0. (See accompanying
6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) 6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7   // 7   //
8   // Official repository: https://github.com/cppalliance/capy 8   // Official repository: https://github.com/cppalliance/capy
9   // 9   //
10   10  
11   #ifndef BOOST_CAPY_TEST_READ_STREAM_HPP 11   #ifndef BOOST_CAPY_TEST_READ_STREAM_HPP
12   #define BOOST_CAPY_TEST_READ_STREAM_HPP 12   #define BOOST_CAPY_TEST_READ_STREAM_HPP
13   13  
14   #include <boost/capy/detail/config.hpp> 14   #include <boost/capy/detail/config.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/make_buffer.hpp> 17   #include <boost/capy/buffers/make_buffer.hpp>
18   #include <boost/capy/cond.hpp> 18   #include <boost/capy/cond.hpp>
19   #include <coroutine> 19   #include <coroutine>
20   #include <boost/capy/ex/io_env.hpp> 20   #include <boost/capy/ex/io_env.hpp>
21   #include <boost/capy/io_result.hpp> 21   #include <boost/capy/io_result.hpp>
22   #include <boost/capy/test/fuse.hpp> 22   #include <boost/capy/test/fuse.hpp>
23   23  
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. 39   This class satisfies the @ref ReadStream concept.
40   40  
41   @par Thread Safety 41   @par Thread Safety
42   Not thread-safe. 42   Not thread-safe.
43   43  
44   @par Example 44   @par Example
45   @code 45   @code
46   fuse f; 46   fuse f;
47   read_stream rs( f ); 47   read_stream rs( f );
48   rs.provide( "Hello, " ); 48   rs.provide( "Hello, " );
49   rs.provide( "World!" ); 49   rs.provide( "World!" );
50   50  
51   auto r = f.armed( [&]( fuse& ) -> task<void> { 51   auto r = f.armed( [&]( fuse& ) -> task<void> {
52   char buf[32]; 52   char buf[32];
53   auto [ec, n] = co_await rs.read_some( 53   auto [ec, n] = co_await rs.read_some(
54   mutable_buffer( buf, sizeof( buf ) ) ); 54   mutable_buffer( buf, sizeof( buf ) ) );
55   if( ec ) 55   if( ec )
56   co_return; 56   co_return;
57   // buf contains "Hello, World!" 57   // buf contains "Hello, World!"
58   } ); 58   } );
59   @endcode 59   @endcode
60   60  
61   @see fuse, ReadStream 61   @see fuse, ReadStream
62   */ 62   */
63   class read_stream 63   class read_stream
64   { 64   {
65   fuse f_; 65   fuse f_;
66   std::string data_; 66   std::string data_;
67   std::size_t pos_ = 0; 67   std::size_t pos_ = 0;
68   std::size_t max_read_size_; 68   std::size_t max_read_size_;
69   69  
70   public: 70   public:
71   /** Construct a read stream. 71   /** Construct a read stream.
72   72  
73   @param f The fuse used to inject errors during reads. 73   @param f The fuse used to inject errors during reads.
74   74  
75   @param max_read_size Maximum bytes returned per read. 75   @param max_read_size Maximum bytes returned per read.
76   Use to simulate chunked network delivery. 76   Use to simulate chunked network delivery.
77   */ 77   */
HITCBC 78   473 explicit read_stream( 78   473 explicit read_stream(
79   fuse f = {}, 79   fuse f = {},
80   std::size_t max_read_size = std::size_t(-1)) noexcept 80   std::size_t max_read_size = std::size_t(-1)) noexcept
HITCBC 81   473 : f_(std::move(f)) 81   473 : f_(std::move(f))
HITCBC 82   473 , max_read_size_(max_read_size) 82   473 , max_read_size_(max_read_size)
83   { 83   {
HITCBC 84   473 } 84   473 }
85   85  
86   /** Append data to be returned by subsequent reads. 86   /** Append data to be returned by subsequent reads.
87   87  
88   Multiple calls accumulate data that @ref read_some returns. 88   Multiple calls accumulate data that @ref read_some returns.
89   89  
90   @param sv The data to append. 90   @param sv The data to append.
91   */ 91   */
92   void 92   void
HITCBC 93   465 provide(std::string_view sv) 93   465 provide(std::string_view sv)
94   { 94   {
HITCBC 95   465 data_.append(sv); 95   465 data_.append(sv);
HITCBC 96   465 } 96   465 }
97   97  
98   /// Clear all data and reset the read position. 98   /// Clear all data and reset the read position.
99   void 99   void
HITCBC 100   6 clear() noexcept 100   6 clear() noexcept
101   { 101   {
HITCBC 102   6 data_.clear(); 102   6 data_.clear();
HITCBC 103   6 pos_ = 0; 103   6 pos_ = 0;
HITCBC 104   6 } 104   6 }
105   105  
106   /// Return the number of bytes available for reading. 106   /// Return the number of bytes available for reading.
107   std::size_t 107   std::size_t
HITCBC 108   23 available() const noexcept 108   23 available() const noexcept
109   { 109   {
HITCBC 110   23 return data_.size() - pos_; 110   23 return data_.size() - pos_;
111   } 111   }
112   112  
113   /** Asynchronously read data from the stream. 113   /** Asynchronously read data from the stream.
114   114  
115   Transfers up to `buffer_size( buffers )` bytes from the internal 115   Transfers up to `buffer_size( buffers )` bytes from the internal
116   buffer to the provided mutable buffer sequence. If no data remains, 116   buffer to the provided mutable buffer sequence. If no data remains,
117   returns `error::eof`. Before every read, the attached @ref fuse is 117   returns `error::eof`. Before every read, the attached @ref fuse is
118   consulted to possibly inject an error for testing fault scenarios. 118   consulted to possibly inject an error for testing fault scenarios.
119   The returned `std::size_t` is the number of bytes transferred. 119   The returned `std::size_t` is the number of bytes transferred.
120   120  
121   @par Effects 121   @par Effects
122   On success, advances the internal read position by the number of 122   On success, advances the internal read position by the number of
123   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
124   remains unchanged. 124   remains unchanged.
125   125  
126   @par Exception Safety 126   @par Exception Safety
127   Injected I/O conditions are reported via the `error_code` 127   Injected I/O conditions are reported via the `error_code`
128   component of the result. Throws `std::system_error` only when 128   component of the result. Throws `std::system_error` only when
129   the attached @ref fuse is in exception mode and reaches its 129   the attached @ref fuse is in exception mode and reaches its
130   failure point; no-throw otherwise. 130   failure point; no-throw otherwise.
131   131  
132   @par Cancellation 132   @par Cancellation
133   If the environment's stop token has been requested, the read 133   If the environment's stop token has been requested, the read
134   completes immediately with `error::canceled` and transfers no 134   completes immediately with `error::canceled` and transfers no
135   data. This lets code under test exercise its cancellation paths. 135   data. This lets code under test exercise its cancellation paths.
136   An empty buffer sequence is a no-op that completes successfully 136   An empty buffer sequence is a no-op that completes successfully
137   regardless of the stop token. 137   regardless of the stop token.
138   138  
139   @param buffers The mutable buffer sequence to receive data. 139   @param buffers The mutable buffer sequence to receive data.
140   140  
141   @return An awaitable that await-returns `(error_code,std::size_t)`. 141   @return An awaitable that await-returns `(error_code,std::size_t)`.
142   142  
143   @throws std::system_error When the attached @ref fuse is in 143   @throws std::system_error When the attached @ref fuse is in
144   exception mode and reaches its failure point. 144   exception mode and reaches its failure point.
145   145  
146   @see fuse 146   @see fuse
147   */ 147   */
148   template<MutableBufferSequence MB> 148   template<MutableBufferSequence MB>
149   auto 149   auto
HITCBC 150   818 read_some(MB buffers) 150   818 read_some(MB buffers)
151   { 151   {
152   struct awaitable 152   struct awaitable
153   { 153   {
154   read_stream* self_; 154   read_stream* self_;
155   MB buffers_; 155   MB buffers_;
156   bool canceled_ = false; 156   bool canceled_ = false;
157   157  
HITCBC 158   818 bool await_ready() const noexcept { return false; } 158   818 bool await_ready() const noexcept { return false; }
159   159  
160   // The operation completes synchronously, but await_suspend 160   // The operation completes synchronously, but await_suspend
161   // is the only place io_env is delivered (the promise's 161   // is the only place io_env is delivered (the promise's
162   // transform_awaiter forwards it here). Returning false means 162   // transform_awaiter forwards it here). Returning false means
163   // the coroutine does not actually suspend — it resumes 163   // the coroutine does not actually suspend — it resumes
164   // immediately — so the read still completes synchronously 164   // immediately — so the read still completes synchronously
165   // while having observed the stop token. See io_env, IoAwaitable. 165   // while having observed the stop token. See io_env, IoAwaitable.
166   bool 166   bool
HITCBC 167   818 await_suspend( 167   818 await_suspend(
168   std::coroutine_handle<>, 168   std::coroutine_handle<>,
169   io_env const* env) noexcept 169   io_env const* env) noexcept
170   { 170   {
HITCBC 171   818 canceled_ = env->stop_token.stop_requested(); 171   818 canceled_ = env->stop_token.stop_requested();
HITCBC 172   818 return false; 172   818 return false;
173   } 173   }
174   174  
175   io_result<std::size_t> 175   io_result<std::size_t>
HITCBC 176   818 await_resume() 176   818 await_resume()
177   { 177   {
178   // Empty buffer is a no-op regardless of 178   // Empty buffer is a no-op regardless of
179   // stream state, stop token, or fuse. 179   // stream state, stop token, or fuse.
HITCBC 180   818 if(buffer_empty(buffers_)) 180   818 if(buffer_empty(buffers_))
HITCBC 181   7 return {{}, 0}; 181   7 return {{}, 0};
182   182  
HITCBC 183   811 if(canceled_) 183   811 if(canceled_)
HITCBC 184   1 return {error::canceled, 0}; 184   1 return {error::canceled, 0};
185   185  
HITCBC 186   810 auto ec = self_->f_.maybe_fail(); 186   810 auto ec = self_->f_.maybe_fail();
HITCBC 187   684 if(ec) 187   684 if(ec)
HITCBC 188   126 return {ec, 0}; 188   126 return {ec, 0};
189   189  
HITCBC 190   558 if(self_->pos_ >= self_->data_.size()) 190   558 if(self_->pos_ >= self_->data_.size())
HITCBC 191   92 return {error::eof, 0}; 191   92 return {error::eof, 0};
192   192  
HITCBC 193   466 std::size_t avail = self_->data_.size() - self_->pos_; 193   466 std::size_t avail = self_->data_.size() - self_->pos_;
HITCBC 194   466 if(avail > self_->max_read_size_) 194   466 if(avail > self_->max_read_size_)
HITCBC 195   174 avail = self_->max_read_size_; 195   174 avail = self_->max_read_size_;
HITCBC 196   466 auto src = make_buffer(self_->data_.data() + self_->pos_, avail); 196   466 auto src = make_buffer(self_->data_.data() + self_->pos_, avail);
HITCBC 197   466 std::size_t const n = buffer_copy(buffers_, src); 197   466 std::size_t const n = buffer_copy(buffers_, src);
HITCBC 198   466 self_->pos_ += n; 198   466 self_->pos_ += n;
HITCBC 199   466 return {{}, n}; 199   466 return {{}, n};
200   } 200   }
201   }; 201   };
HITCBC 202   818 return awaitable{this, buffers}; 202   818 return awaitable{this, buffers};
203   } 203   }
204   }; 204   };
205   205  
206   } // test 206   } // test
207   } // capy 207   } // capy
208   } // boost 208   } // boost
209   209  
210   #endif 210   #endif