. /include/boost/capy /io /pull_from.hpp
100.00% Lines (4/4) 100.00% Functions (2/2)
[ ] 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   // 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_PULL_FROM_HPP 10   #ifndef BOOST_CAPY_IO_PULL_FROM_HPP
11   #define BOOST_CAPY_IO_PULL_FROM_HPP 11   #define BOOST_CAPY_IO_PULL_FROM_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/cond.hpp> 15   #include <boost/capy/cond.hpp>
16   #include <boost/capy/concept/buffer_sink.hpp> 16   #include <boost/capy/concept/buffer_sink.hpp>
17   #include <boost/capy/concept/read_source.hpp> 17   #include <boost/capy/concept/read_source.hpp>
18   #include <boost/capy/concept/read_stream.hpp> 18   #include <boost/capy/concept/read_stream.hpp>
19   #include <boost/capy/io_task.hpp> 19   #include <boost/capy/io_task.hpp>
20   20  
21   #include <cstddef> 21   #include <cstddef>
22   #include <span> 22   #include <span>
23   23  
24   namespace boost { 24   namespace boost {
25   namespace capy { 25   namespace capy {
26   26  
27   /** Transfer data from a ReadSource to a BufferSink. 27   /** Transfer data from a ReadSource to a BufferSink.
28   28  
29   This function reads data from the source directly into the sink's 29   This function reads data from the source directly into the sink's
30   internal buffers using the callee-owns-buffers model. The sink 30   internal buffers using the callee-owns-buffers model. The sink
31   provides writable buffers via `prepare()`, the source reads into 31   provides writable buffers via `prepare()`, the source reads into
32   them, and the sink commits the data. When the source signals EOF, 32   them, and the sink commits the data. When the source signals EOF,
33   `commit_eof()` is called on the sink to finalize the transfer. 33   `commit_eof()` is called on the sink to finalize the transfer.
34   34  
35   @tparam Src The source type, must satisfy @ref ReadSource. 35   @tparam Src The source type, must satisfy @ref ReadSource.
36   @tparam Sink The sink type, must satisfy @ref BufferSink. 36   @tparam Sink The sink type, must satisfy @ref BufferSink.
37   37  
38   @param source The source to read data from. 38   @param source The source to read data from.
39   @param sink The sink to write data to. 39   @param sink The sink to write data to.
40   40  
41   @return A task that yields `(std::error_code, std::size_t)`. 41   @return A task that yields `(std::error_code, std::size_t)`.
42   On success, `ec` is default-constructed (no error) and `n` is 42   On success, `ec` is default-constructed (no error) and `n` is
43   the total number of bytes transferred. On error, `ec` contains 43   the total number of bytes transferred. On error, `ec` contains
44   the error code and `n` is the total number of bytes transferred 44   the error code and `n` is the total number of bytes transferred
45   before the error. 45   before the error.
46   46  
47   @par Example 47   @par Example
48   @code 48   @code
49   task<void> transfer_body(ReadSource auto& source, BufferSink auto& sink) 49   task<void> transfer_body(ReadSource auto& source, BufferSink auto& sink)
50   { 50   {
51   auto [ec, n] = co_await pull_from(source, sink); 51   auto [ec, n] = co_await pull_from(source, sink);
52   if (ec) 52   if (ec)
53   { 53   {
54   // Handle error 54   // Handle error
55   } 55   }
56   // n bytes were transferred 56   // n bytes were transferred
57   } 57   }
58   @endcode 58   @endcode
59   59  
60   @see ReadSource, BufferSink, push_to 60   @see ReadSource, BufferSink, push_to
61   */ 61   */
62   template<ReadSource Src, BufferSink Sink> 62   template<ReadSource Src, BufferSink Sink>
63   io_task<std::size_t> 63   io_task<std::size_t>
HITCBC 64   132 pull_from(Src& source, Sink& sink) 64   136 pull_from(Src& source, Sink& sink)
65   { 65   {
66   mutable_buffer dst_arr[detail::max_iovec_]; 66   mutable_buffer dst_arr[detail::max_iovec_];
67   std::size_t total = 0; 67   std::size_t total = 0;
68   68  
69   for(;;) 69   for(;;)
70   { 70   {
71   auto dst_bufs = sink.prepare(dst_arr); 71   auto dst_bufs = sink.prepare(dst_arr);
72   if(dst_bufs.empty()) 72   if(dst_bufs.empty())
73   { 73   {
74   // No buffer space available; commit nothing to flush 74   // No buffer space available; commit nothing to flush
75   auto [flush_ec] = co_await sink.commit(0); 75   auto [flush_ec] = co_await sink.commit(0);
76   if(flush_ec) 76   if(flush_ec)
77   co_return {flush_ec, total}; 77   co_return {flush_ec, total};
78   continue; 78   continue;
79   } 79   }
80   80  
81   auto [ec, n] = co_await source.read( 81   auto [ec, n] = co_await source.read(
82   std::span<mutable_buffer const>(dst_bufs)); 82   std::span<mutable_buffer const>(dst_bufs));
83   83  
84 - if(n > 0) 84 + auto [commit_ec] = co_await sink.commit(n);
85 - { 85 + total += n;
86 - auto [commit_ec] = co_await sink.commit(n); 86 +
87 - if(commit_ec) 87 + if(commit_ec)
88 - co_return {commit_ec, total}; 88 + co_return {commit_ec, total};
89 - total += n;  
90 - }  
91   89  
92   if(ec == cond::eof) 90   if(ec == cond::eof)
93   { 91   {
94   auto [eof_ec] = co_await sink.commit_eof(0); 92   auto [eof_ec] = co_await sink.commit_eof(0);
95   co_return {eof_ec, total}; 93   co_return {eof_ec, total};
96   } 94   }
97   95  
98   if(ec) 96   if(ec)
99   co_return {ec, total}; 97   co_return {ec, total};
100   } 98   }
HITCBC 101   264 } 99   272 }
102   100  
103   /** Transfer data from a ReadStream to a BufferSink. 101   /** Transfer data from a ReadStream to a BufferSink.
104   102  
105   This function reads data from the stream directly into the sink's 103   This function reads data from the stream directly into the sink's
106   internal buffers using the callee-owns-buffers model. The sink 104   internal buffers using the callee-owns-buffers model. The sink
107   provides writable buffers via `prepare()`, the stream reads into 105   provides writable buffers via `prepare()`, the stream reads into
108   them using `read_some()`, and the sink commits the data. When the 106   them using `read_some()`, and the sink commits the data. When the
109   stream signals EOF, `commit_eof()` is called on the sink to 107   stream signals EOF, `commit_eof()` is called on the sink to
110   finalize the transfer. 108   finalize the transfer.
111   109  
112   This overload handles partial reads from the stream, committing 110   This overload handles partial reads from the stream, committing
113   data incrementally as it arrives. It loops until EOF is encountered 111   data incrementally as it arrives. It loops until EOF is encountered
114   or an error occurs. 112   or an error occurs.
115   113  
116   @tparam Src The source type, must satisfy @ref ReadStream. 114   @tparam Src The source type, must satisfy @ref ReadStream.
117   @tparam Sink The sink type, must satisfy @ref BufferSink. 115   @tparam Sink The sink type, must satisfy @ref BufferSink.
118   116  
119   @param source The stream to read data from. 117   @param source The stream to read data from.
120   @param sink The sink to write data to. 118   @param sink The sink to write data to.
121   119  
122   @return A task that yields `(std::error_code, std::size_t)`. 120   @return A task that yields `(std::error_code, std::size_t)`.
123   On success, `ec` is default-constructed (no error) and `n` is 121   On success, `ec` is default-constructed (no error) and `n` is
124   the total number of bytes transferred. On error, `ec` contains 122   the total number of bytes transferred. On error, `ec` contains
125   the error code and `n` is the total number of bytes transferred 123   the error code and `n` is the total number of bytes transferred
126   before the error. 124   before the error.
127   125  
128   @par Example 126   @par Example
129   @code 127   @code
130   task<void> transfer_body(ReadStream auto& stream, BufferSink auto& sink) 128   task<void> transfer_body(ReadStream auto& stream, BufferSink auto& sink)
131   { 129   {
132   auto [ec, n] = co_await pull_from(stream, sink); 130   auto [ec, n] = co_await pull_from(stream, sink);
133   if (ec) 131   if (ec)
134   { 132   {
135   // Handle error 133   // Handle error
136   } 134   }
137   // n bytes were transferred 135   // n bytes were transferred
138   } 136   }
139   @endcode 137   @endcode
140   138  
141   @see ReadStream, BufferSink, push_to 139   @see ReadStream, BufferSink, push_to
142   */ 140   */
143   template<ReadStream Src, BufferSink Sink> 141   template<ReadStream Src, BufferSink Sink>
144   requires (!ReadSource<Src>) 142   requires (!ReadSource<Src>)
145   io_task<std::size_t> 143   io_task<std::size_t>
HITCBC 146   200 pull_from(Src& source, Sink& sink) 144   226 pull_from(Src& source, Sink& sink)
147   { 145   {
148   mutable_buffer dst_arr[detail::max_iovec_]; 146   mutable_buffer dst_arr[detail::max_iovec_];
149   std::size_t total = 0; 147   std::size_t total = 0;
150   148  
151   for(;;) 149   for(;;)
152   { 150   {
153   // Prepare destination buffers from the sink 151   // Prepare destination buffers from the sink
154   auto dst_bufs = sink.prepare(dst_arr); 152   auto dst_bufs = sink.prepare(dst_arr);
155   if(dst_bufs.empty()) 153   if(dst_bufs.empty())
156   { 154   {
157   // No buffer space available; commit nothing to flush 155   // No buffer space available; commit nothing to flush
158   auto [flush_ec] = co_await sink.commit(0); 156   auto [flush_ec] = co_await sink.commit(0);
159   if(flush_ec) 157   if(flush_ec)
160   co_return {flush_ec, total}; 158   co_return {flush_ec, total};
161   continue; 159   continue;
162   } 160   }
163   161  
164   // Read data from the stream into the sink's buffers 162   // Read data from the stream into the sink's buffers
165   auto [ec, n] = co_await source.read_some( 163   auto [ec, n] = co_await source.read_some(
166   std::span<mutable_buffer const>(dst_bufs)); 164   std::span<mutable_buffer const>(dst_bufs));
167   165  
168 - // Commit any data that was read 166 + auto [commit_ec] = co_await sink.commit(n);
169 - if(n > 0) 167 + total += n;
170 - { 168 +
171 - auto [commit_ec] = co_await sink.commit(n); 169 + if(commit_ec)
172 - if(commit_ec) 170 + co_return {commit_ec, total};
173 - co_return {commit_ec, total};  
174 - total += n;  
175 - }  
176 - // Check for EOF condition  
177   171  
178   if(ec == cond::eof) 172   if(ec == cond::eof)
179   { 173   {
180   auto [eof_ec] = co_await sink.commit_eof(0); 174   auto [eof_ec] = co_await sink.commit_eof(0);
181   co_return {eof_ec, total}; 175   co_return {eof_ec, total};
182   } 176   }
183   177  
184   // Check for other errors 178   // Check for other errors
185   if(ec) 179   if(ec)
186   co_return {ec, total}; 180   co_return {ec, total};
187   } 181   }
HITCBC 188   400 } 182   452 }
189   183  
190   } // namespace capy 184   } // namespace capy
191   } // namespace boost 185   } // namespace boost
192   186  
193   #endif 187   #endif