. /include/boost/capy /test /write_stream.hpp
100.00% Lines (46/46) 100.00% Functions (9/9)
[ ] 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_WRITE_STREAM_HPP 11   #ifndef BOOST_CAPY_TEST_WRITE_STREAM_HPP
12   #define BOOST_CAPY_TEST_WRITE_STREAM_HPP 12   #define BOOST_CAPY_TEST_WRITE_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 <coroutine> 18   #include <coroutine>
19   #include <boost/capy/ex/io_env.hpp> 19   #include <boost/capy/ex/io_env.hpp>
20   #include <boost/capy/io_result.hpp> 20   #include <boost/capy/io_result.hpp>
21   #include <boost/capy/error.hpp> 21   #include <boost/capy/error.hpp>
22   #include <boost/capy/test/fuse.hpp> 22   #include <boost/capy/test/fuse.hpp>
23   23  
24   #include <algorithm> 24   #include <algorithm>
25   #include <string> 25   #include <string>
26   #include <string_view> 26   #include <string_view>
27   27  
28   namespace boost { 28   namespace boost {
29   namespace capy { 29   namespace capy {
30   namespace test { 30   namespace test {
31   31  
32   /** Captures bytes passed to `write_some`, retrievable afterward through `data`. 32   /** Captures bytes passed to `write_some`, retrievable afterward through `data`.
33   33  
34   Use this to verify code that performs writes without needing 34   Use this to verify code that performs writes without needing
35   real I/O. Call @ref write_some to write data, then @ref data 35   real I/O. Call @ref write_some to write data, then @ref data
36   to retrieve what was written. The associated @ref fuse enables 36   to retrieve what was written. The associated @ref fuse enables
37   error injection at controlled points. An optional 37   error injection at controlled points. An optional
38   `max_write_size` constructor parameter limits bytes per write 38   `max_write_size` constructor parameter limits bytes per write
39   to simulate chunked delivery. 39   to simulate chunked delivery.
40   40  
41   This class satisfies the @ref WriteStream concept. 41   This class satisfies the @ref WriteStream concept.
42   42  
43   @par Thread Safety 43   @par Thread Safety
44   Not thread-safe. 44   Not thread-safe.
45   45  
46   @par Example 46   @par Example
47   @code 47   @code
48   fuse f; 48   fuse f;
49   49  
50   auto r = f.armed( [&]( fuse& ) -> task<void> { 50   auto r = f.armed( [&]( fuse& ) -> task<void> {
51   // Constructed inside the lambda: armed() re-invokes this 51   // Constructed inside the lambda: armed() re-invokes this
52   // function once per injected failure point, and a write_stream 52   // function once per injected failure point, and a write_stream
53   // constructed outside would carry accumulated data across 53   // constructed outside would carry accumulated data across
54   // those rounds. 54   // those rounds.
55   write_stream ws( f ); 55   write_stream ws( f );
56   56  
57   auto [ec, n] = co_await ws.write_some( 57   auto [ec, n] = co_await ws.write_some(
58   const_buffer( "Hello", 5 ) ); 58   const_buffer( "Hello", 5 ) );
59   if( ec ) 59   if( ec )
60   co_return; 60   co_return;
61   // ws.data() returns "Hello" 61   // ws.data() returns "Hello"
62   } ); 62   } );
63   @endcode 63   @endcode
64   64  
65   @see fuse, WriteStream 65   @see fuse, WriteStream
66   */ 66   */
67   class write_stream 67   class write_stream
68   { 68   {
69   fuse f_; 69   fuse f_;
70   std::string data_; 70   std::string data_;
71   std::string expect_; 71   std::string expect_;
72   std::size_t max_write_size_; 72   std::size_t max_write_size_;
73   73  
74   std::error_code 74   std::error_code
HITCBC 75   364 consume_match_() noexcept 75   364 consume_match_() noexcept
76   { 76   {
HITCBC 77   364 if(data_.empty() || expect_.empty()) 77   364 if(data_.empty() || expect_.empty())
HITCBC 78   347 return {}; 78   347 return {};
HITCBC 79   17 std::size_t const n = (std::min)(data_.size(), expect_.size()); 79   17 std::size_t const n = (std::min)(data_.size(), expect_.size());
HITCBC 80   17 if(std::string_view(data_.data(), n) != 80   17 if(std::string_view(data_.data(), n) !=
HITCBC 81   34 std::string_view(expect_.data(), n)) 81   34 std::string_view(expect_.data(), n))
HITCBC 82   4 return error::test_failure; 82   4 return error::test_failure;
HITCBC 83   13 data_.erase(0, n); 83   13 data_.erase(0, n);
HITCBC 84   13 expect_.erase(0, n); 84   13 expect_.erase(0, n);
HITCBC 85   13 return {}; 85   13 return {};
86   } 86   }
87   87  
88   public: 88   public:
89   /** Construct a write stream. 89   /** Construct a write stream.
90   90  
91   @param f The fuse used to inject errors during writes. 91   @param f The fuse used to inject errors during writes.
92   92  
93   @param max_write_size Maximum bytes transferred per write. 93   @param max_write_size Maximum bytes transferred per write.
94   Use to simulate chunked network delivery. 94   Use to simulate chunked network delivery.
95   */ 95   */
HITCBC 96   417 explicit write_stream( 96   417 explicit write_stream(
97   fuse f = {}, 97   fuse f = {},
98   std::size_t max_write_size = std::size_t(-1)) noexcept 98   std::size_t max_write_size = std::size_t(-1)) noexcept
HITCBC 99   417 : f_(std::move(f)) 99   417 : f_(std::move(f))
HITCBC 100   417 , max_write_size_(max_write_size) 100   417 , max_write_size_(max_write_size)
101   { 101   {
HITCBC 102   417 } 102   417 }
103   103  
104   /** Return the written data as a string view. 104   /** Return the written data as a string view.
105   105  
106   @return A view of bytes written but not yet matched by @ref expect. 106   @return A view of bytes written but not yet matched by @ref expect.
107   */ 107   */
108   std::string_view 108   std::string_view
HITCBC 109   314 data() const noexcept 109   314 data() const noexcept
110   { 110   {
HITCBC 111   314 return data_; 111   314 return data_;
112   } 112   }
113   113  
114   /** Set the expected data for subsequent writes. 114   /** Set the expected data for subsequent writes.
115   115  
116   Stores the expected data and immediately tries to match 116   Stores the expected data and immediately tries to match
117   against any data already written. Matched data is consumed 117   against any data already written. Matched data is consumed
118   from both buffers. 118   from both buffers.
119   119  
120   @param sv The expected data. 120   @param sv The expected data.
121   121  
122   @return An error if existing data does not match. 122   @return An error if existing data does not match.
123   */ 123   */
124   std::error_code 124   std::error_code
HITCBC 125   31 expect(std::string_view sv) 125   31 expect(std::string_view sv)
126   { 126   {
HITCBC 127   31 expect_.assign(sv); 127   31 expect_.assign(sv);
HITCBC 128   31 return consume_match_(); 128   31 return consume_match_();
129   } 129   }
130   130  
131   /** Return the number of bytes written. 131   /** Return the number of bytes written.
132   132  
133   @return The number of bytes written but not yet matched by @ref expect. 133   @return The number of bytes written but not yet matched by @ref expect.
134   */ 134   */
135   std::size_t 135   std::size_t
HITCBC 136   5 size() const noexcept 136   5 size() const noexcept
137   { 137   {
HITCBC 138   5 return data_.size(); 138   5 return data_.size();
139   } 139   }
140   140  
141   /** Asynchronously write data to the stream. 141   /** Asynchronously write data to the stream.
142   142  
143   Transfers up to `buffer_size( buffers )` bytes from the provided 143   Transfers up to `buffer_size( buffers )` bytes from the provided
144   const buffer sequence to the internal buffer. Before every write, 144   const buffer sequence to the internal buffer. Before every write,
145   the attached @ref fuse is consulted to possibly inject an error 145   the attached @ref fuse is consulted to possibly inject an error
146   for testing fault scenarios. The returned `std::size_t` is the 146   for testing fault scenarios. The returned `std::size_t` is the
147   number of bytes transferred. 147   number of bytes transferred.
148   148  
149   @par Effects 149   @par Effects
150   On success, appends the written bytes to the internal buffer. 150   On success, appends the written bytes to the internal buffer.
151   If an error is injected by the fuse, the internal buffer remains 151   If an error is injected by the fuse, the internal buffer remains
152   unchanged. 152   unchanged.
153   153  
154   @par Exception Safety 154   @par Exception Safety
155   Injected I/O conditions are reported via the `error_code` 155   Injected I/O conditions are reported via the `error_code`
156   component of the result. Throws `std::system_error` only when 156   component of the result. Throws `std::system_error` only when
157   the attached @ref fuse is in exception mode and reaches its 157   the attached @ref fuse is in exception mode and reaches its
158   failure point; no-throw otherwise. 158   failure point; no-throw otherwise.
159   159  
160   @par Cancellation 160   @par Cancellation
161   If the environment's stop token is requested, the write 161   If the environment's stop token is requested, the write
162   completes immediately with `error::canceled` and transfers no 162   completes immediately with `error::canceled` and transfers no
163   data. An empty buffer sequence is a no-op that completes 163   data. An empty buffer sequence is a no-op that completes
164   successfully regardless of the stop token. 164   successfully regardless of the stop token.
165   165  
166   @param buffers The const buffer sequence containing data to write. 166   @param buffers The const buffer sequence containing data to write.
167   167  
168   @return An awaitable that await-returns `(error_code,std::size_t)`. 168   @return An awaitable that await-returns `(error_code,std::size_t)`.
169   169  
170   @throws std::system_error When the attached @ref fuse is in 170   @throws std::system_error When the attached @ref fuse is in
171   exception mode and reaches its failure point. 171   exception mode and reaches its failure point.
172   172  
173   @see fuse 173   @see fuse
174   */ 174   */
175   template<ConstBufferSequence CB> 175   template<ConstBufferSequence CB>
176   auto 176   auto
HITCBC 177   531 write_some(CB buffers) 177   531 write_some(CB buffers)
178   { 178   {
179   struct awaitable 179   struct awaitable
180   { 180   {
181   write_stream* self_; 181   write_stream* self_;
182   CB buffers_; 182   CB buffers_;
183   bool canceled_ = false; 183   bool canceled_ = false;
184   184  
HITCBC 185   531 bool await_ready() const noexcept { return false; } 185   531 bool await_ready() const noexcept { return false; }
186   186  
187   // The operation completes synchronously, but await_suspend is 187   // The operation completes synchronously, but await_suspend is
188   // the only place io_env is delivered (the promise's 188   // the only place io_env is delivered (the promise's
189   // transform_awaiter forwards it here). Returning false means 189   // transform_awaiter forwards it here). Returning false means
190   // the coroutine does not actually suspend; it resumes 190   // the coroutine does not actually suspend; it resumes
191   // immediately, having observed the stop token. See io_env, 191   // immediately, having observed the stop token. See io_env,
192   // IoAwaitable. 192   // IoAwaitable.
193   bool 193   bool
HITCBC 194   531 await_suspend( 194   531 await_suspend(
195   std::coroutine_handle<>, 195   std::coroutine_handle<>,
196   io_env const* env) noexcept 196   io_env const* env) noexcept
197   { 197   {
HITCBC 198   531 canceled_ = env->stop_token.stop_requested(); 198   531 canceled_ = env->stop_token.stop_requested();
HITCBC 199   531 return false; 199   531 return false;
200   } 200   }
201   201  
202   [[nodiscard]] io_result<std::size_t> 202   [[nodiscard]] io_result<std::size_t>
HITCBC 203   531 await_resume() 203   531 await_resume()
204   { 204   {
HITCBC 205   531 if(buffer_empty(buffers_)) 205   531 if(buffer_empty(buffers_))
HITCBC 206   3 return {std::error_code(), 0}; 206   3 return {std::error_code(), 0};
207   207  
HITCBC 208   528 if(canceled_) 208   528 if(canceled_)
HITCBC 209   1 return {error::canceled, 0}; 209   1 return {error::canceled, 0};
210   210  
HITCBC 211   527 auto ec = self_->f_.maybe_fail(); 211   527 auto ec = self_->f_.maybe_fail();
HITCBC 212   430 if(ec) 212   430 if(ec)
HITCBC 213   97 return {ec, 0}; 213   97 return {ec, 0};
214   214  
HITCBC 215   333 std::size_t n = buffer_size(buffers_); 215   333 std::size_t n = buffer_size(buffers_);
HITCBC 216   333 n = (std::min)(n, self_->max_write_size_); 216   333 n = (std::min)(n, self_->max_write_size_);
217   217  
HITCBC 218   333 std::size_t const old_size = self_->data_.size(); 218   333 std::size_t const old_size = self_->data_.size();
HITCBC 219   333 self_->data_.resize(old_size + n); 219   333 self_->data_.resize(old_size + n);
HITCBC 220   333 buffer_copy(make_buffer( 220   333 buffer_copy(make_buffer(
HITCBC 221   333 self_->data_.data() + old_size, n), buffers_, n); 221   333 self_->data_.data() + old_size, n), buffers_, n);
222   222  
HITCBC 223   333 ec = self_->consume_match_(); 223   333 ec = self_->consume_match_();
HITCBC 224   333 if(ec) 224   333 if(ec)
225   { 225   {
HITCBC 226   2 self_->data_.resize(old_size); 226   2 self_->data_.resize(old_size);
HITCBC 227   2 return {ec, 0}; 227   2 return {ec, 0};
228   } 228   }
229   229  
HITCBC 230   331 return {std::error_code(), n}; 230   331 return {std::error_code(), n};
231   } 231   }
232   }; 232   };
HITCBC 233   531 return awaitable{this, buffers}; 233   531 return awaitable{this, buffers};
234   } 234   }
235   }; 235   };
236   236  
237   } // test 237   } // test
238   } // capy 238   } // capy
239   } // boost 239   } // boost
240   240  
241   #endif 241   #endif