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