feat: add read_at_least and write_at_least algorithms (#260)

doc/modules/ROOT/pages/6.streams/6e.algorithms.adoc

@@ -38,6 +38,41 @@ auto [ec, n] = co_await read(stream, make_buffer(buf));
 // n == 1024, or ec indicates why not
 ----
 
+=== read_at_least
+
+A straightforward extension of `read`: instead of filling `buffers`
+completely, it stops as soon as at least `n` bytes have been read:
+
+[source,cpp]
+----
+#include <boost/capy/read_at_least.hpp>
+
+template<ReadStream Stream, MutableBufferSequence Buffers>
+io_task<std::size_t>
+read_at_least(Stream& stream, Buffers buffers, std::size_t n);
+----
+
+The required amount `n` must be met or exceeded; the remaining capacity of
+`buffers` is optional, so a single `read_some` that delivers more than `n`
+bytes keeps the extra without looping again. Keeps reading until:
+
+* At least `n` bytes have been read (`n \<= return value \<= buffer_size(buffers)`)
+* The underlying `read_some` reports a condition before `n` bytes are read (the condition is propagated with the partial count)
+
+If `n` exceeds `buffer_size(buffers)` the request is impossible to satisfy and
+the operation fails immediately with `std::errc::invalid_argument` and a count
+of `0`, without reading.
+
+Example:
+
+[source,cpp]
+----
+char buf[4096];
+// Require 16 bytes; opportunistically take whatever else is ready.
+auto [ec, n] = co_await read_at_least(stream, make_buffer(buf), 16);
+// n >= 16 on success, possibly up to 4096
+----
+
 === write
 
 Writes all data by looping `write_some`:
@@ -63,6 +98,28 @@ Example:
 co_await write(stream, make_buffer("Hello, World!"));
 ----
 
+=== write_at_least
+
+The mirror of `read_at_least`, provided for symmetry: it stops as soon as at
+least `n` bytes have been written, rather than draining `buffers` entirely:
+
+[source,cpp]
+----
+#include <boost/capy/write_at_least.hpp>
+
+template<WriteStream Stream, ConstBufferSequence Buffers>
+io_task<std::size_t>
+write_at_least(Stream& stream, Buffers buffers, std::size_t n);
+----
+
+Keeps writing until:
+
+* At least `n` bytes have been written (`n \<= return value \<= buffer_size(buffers)`)
+* The underlying `write_some` reports a condition before `n` bytes are written (the condition is propagated with the partial count)
+
+If `n` exceeds `buffer_size(buffers)` the operation fails immediately with
+`std::errc::invalid_argument` and a count of `0`, without writing.
+
 === write_now
 
 `write_now` eagerly writes a complete buffer sequence, attempting to finish
@@ -250,9 +307,15 @@ else if (ec)
 | `<boost/capy/read.hpp>`
 | Composed read operations
 
+| `<boost/capy/read_at_least.hpp>`
+| Read at least a minimum number of bytes
+
 | `<boost/capy/write.hpp>`
 | Composed write operations
 
+| `<boost/capy/write_at_least.hpp>`
+| Write at least a minimum number of bytes
+
 | `<boost/capy/io/write_now.hpp>`
 | Eager write with frame caching
 

include/boost/capy.hpp

@@ -28,10 +28,12 @@
 // Algorithms
 #include <boost/capy/delay.hpp>
 #include <boost/capy/read.hpp>
+#include <boost/capy/read_at_least.hpp>
 #include <boost/capy/timeout.hpp>
 #include <boost/capy/when_all.hpp>
 #include <boost/capy/when_any.hpp>
 #include <boost/capy/write.hpp>
+#include <boost/capy/write_at_least.hpp>
 
 // Buffers
 #include <boost/capy/buffers.hpp>

include/boost/capy/read_at_least.hpp

@@ -0,0 +1,141 @@
+//
+// Copyright (c) 2026 Michael Vandeberg
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+// Official repository: https://github.com/cppalliance/capy
+//
+
+#ifndef BOOST_CAPY_READ_AT_LEAST_HPP
+#define BOOST_CAPY_READ_AT_LEAST_HPP
+
+#include <boost/capy/detail/config.hpp>
+#include <boost/capy/cond.hpp>
+#include <boost/capy/io_task.hpp>
+#include <boost/capy/buffers.hpp>
+#include <boost/capy/buffers/consuming_buffers.hpp>
+#include <boost/capy/concept/read_stream.hpp>
+
+#include <cstddef>
+#include <system_error>
+
+namespace boost {
+namespace capy {
+
+/** Read at least a minimum number of bytes from a stream.
+
+    This is a straightforward extension of @ref read. While @ref read
+    transfers exactly `buffer_size(buffers)` bytes, `read_at_least`
+    transfers at least `n` bytes: the loop stops as soon as `n` bytes
+    have been read, even if `buffers` is not yet full. Any bytes beyond
+    `n` that a single `stream.read_some` happens to deliver (up to the
+    capacity of `buffers`) are kept, but no further awaiting is performed
+    to fill the remainder.
+
+    This is useful when a caller has a required amount of data `n` that
+    must be met or exceeded, while the subsequent capacity of `buffers`
+    is optional and should not block.
+
+    @par Await-effects
+
+    If `n > buffer_size(buffers)` the request is impossible to satisfy
+    and the operation fails immediately with
+    `{std::errc::invalid_argument, 0}` without awaiting `stream.read_some`.
+
+    Otherwise reads data from `stream` via awaiting `stream.read_some`
+    repeatedly until:
+
+    @li either at least `n` bytes have been read,
+    @li or a contingency occurs on `stream.read_some`.
+
+    If `n == 0` then no awaiting `stream.read_some` is performed. This is
+    not a contingency.
+
+    @par Await-returns
+    An object of type `io_result<std::size_t>` destructuring as `[ec, n]`.
+
+    Upon a contingency, the count represents the number of bytes read so
+    far, inclusive of the last partial read.
+
+    Contingencies:
+
+    @li The first contingency reported from awaiting @c stream.read_some
+        while fewer than `n` bytes have been read. A contingency that
+        accompanies the read which reaches `n` is not reported: a
+        satisfied request is a success.
+
+    Notable conditions:
+
+    @li @c std::errc::invalid_argument — `n` exceeds `buffer_size(buffers)`,
+    @li @c cond::canceled — Operation was cancelled,
+    @li @c cond::eof — Stream reached end before `n` bytes were read.
+
+    @par Await-postcondition
+    On success the returned count is greater than or equal to `n` and
+    less than or equal to `buffer_size(buffers)`, and `ec` is success;
+    otherwise `ec` is set.
+
+    @param stream The stream to read from. If the lifetime of `stream` ends
+    before the coroutine finishes, the behavior is undefined.
+
+    @param buffers The buffer sequence to read into. If the lifetime of the
+    buffer sequence represented by `buffers` ends before the coroutine
+    finishes, the behavior is undefined.
+
+    @param n The minimum number of bytes to read. Must not exceed
+    `buffer_size(buffers)`.
+
+    @par Remarks
+    Supports _IoAwaitable cancellation_.
+
+    @par Example
+
+    @code
+    capy::task<> fill_buffer(capy::ReadStream auto& stream)
+    {
+        std::vector<char> storage(4096);  // generous capacity
+        // Require 16 header bytes; opportunistically take more.
+        auto [ec, n] = co_await capy::read_at_least(
+            stream, capy::make_buffer(storage), 16);
+        if(ec)
+            throw std::system_error(ec);
+
+        // at least 16 bytes are available; n may be larger
+    }
+    @endcode
+
+    @see read, ReadStream, MutableBufferSequence
+*/
+template <typename S, typename MB>
+  requires ReadStream<S> && MutableBufferSequence<MB>
+auto
+read_at_least(S& stream, MB buffers, std::size_t n) ->
+        io_task<std::size_t>
+{
+    consuming_buffers consuming(buffers);
+    std::size_t const total_size = buffer_size(buffers);
+
+    if(n > total_size)
+        co_return {make_error_code(std::errc::invalid_argument), 0};
+
+    std::size_t total_read = 0;
+
+    while(total_read < n)
+    {
+        auto [ec, m] = co_await stream.read_some(consuming.data());
+        consuming.consume(m);
+        total_read += m;
+        // A contingency that still satisfied the request is a success:
+        // report it only when fewer than n bytes were read.
+        if(ec && total_read < n)
+            co_return {ec, total_read};
+    }
+
+    co_return {{}, total_read};
+}
+
+} // namespace capy
+} // namespace boost
+
+#endif

include/boost/capy/write_at_least.hpp

@@ -0,0 +1,135 @@
+//
+// Copyright (c) 2026 Michael Vandeberg
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+// Official repository: https://github.com/cppalliance/capy
+//
+
+#ifndef BOOST_CAPY_WRITE_AT_LEAST_HPP
+#define BOOST_CAPY_WRITE_AT_LEAST_HPP
+
+#include <boost/capy/detail/config.hpp>
+#include <boost/capy/io_task.hpp>
+#include <boost/capy/buffers.hpp>
+#include <boost/capy/buffers/consuming_buffers.hpp>
+#include <boost/capy/concept/write_stream.hpp>
+
+#include <cstddef>
+#include <system_error>
+
+namespace boost {
+namespace capy {
+
+/** Write at least a minimum number of bytes to a stream.
+
+    This is a straightforward extension of @ref write. While @ref write
+    transfers exactly `buffer_size(buffers)` bytes, `write_at_least`
+    transfers at least `n` bytes: the loop stops as soon as `n` bytes
+    have been written, even if `buffers` has not been fully consumed.
+    Any bytes beyond `n` that a single `stream.write_some` happens to
+    transfer are counted, but no further awaiting is performed to write
+    the remainder.
+
+    Provided for symmetry with @ref read_at_least.
+
+    @par Await-effects
+
+    If `n > buffer_size(buffers)` the request is impossible to satisfy
+    and the operation fails immediately with
+    `{std::errc::invalid_argument, 0}` without awaiting `stream.write_some`.
+
+    Otherwise writes the contents of `buffers` to `stream` via awaiting
+    `stream.write_some` with consecutive portions of data from `buffers`
+    until:
+
+    @li either at least `n` bytes have been written,
+    @li or a contingency in `stream.write_some` occurs.
+
+    If `n == 0` then no awaiting `stream.write_some` is performed. This is
+    not a contingency.
+
+    @par Await-returns
+    An object of type `io_result<std::size_t>` destructuring as `[ec, n]`.
+
+    Upon a contingency, the count represents the number of bytes written
+    so far.
+
+    Contingencies:
+
+    @li The first contingency reported from awaiting @c stream.write_some
+        while fewer than `n` bytes have been written. A contingency that
+        accompanies the write which reaches `n` is not reported: a
+        satisfied request is a success.
+
+    Notable conditions:
+
+    @li @c std::errc::invalid_argument — `n` exceeds `buffer_size(buffers)`,
+    @li @c cond::canceled — Operation was cancelled,
+    @li @c std::errc::broken_pipe — Peer closed connection.
+
+    @par Await-postcondition
+    On success the returned count is greater than or equal to `n` and
+    less than or equal to `buffer_size(buffers)`, and `ec` is success;
+    otherwise `ec` is set.
+
+    @param stream The stream to write to. If the lifetime of `stream` ends
+    before the coroutine finishes, the behavior is undefined.
+
+    @param buffers The buffer sequence to write. If the lifetime of the
+    buffer sequence represented by `buffers` ends before the coroutine
+    finishes, the behavior is undefined.
+
+    @param n The minimum number of bytes to write. Must not exceed
+    `buffer_size(buffers)`.
+
+    @par Remarks
+    Supports _IoAwaitable cancellation_.
+
+    @par Example
+
+    @code
+    capy::task<> flush_at_least(capy::WriteStream auto& stream, std::string_view data)
+    {
+        auto [ec, n] = co_await capy::write_at_least(
+            stream, capy::make_buffer(data), 8);
+        if(ec)
+            throw std::system_error(ec);
+
+        // at least 8 bytes written; n may be larger
+    }
+    @endcode
+
+    @see write, WriteStream, ConstBufferSequence
+*/
+template <WriteStream S, ConstBufferSequence CB>
+auto
+write_at_least(S& stream, CB buffers, std::size_t n) -> io_task<std::size_t>
+{
+    consuming_buffers consuming(buffers);
+    std::size_t const total_size = buffer_size(buffers);
+
+    if(n > total_size)
+        co_return {make_error_code(std::errc::invalid_argument), 0};
+
+    std::size_t total_written = 0;
+
+    while(total_written < n)
+    {
+        auto [ec, m] = co_await stream.write_some(consuming.data());
+        consuming.consume(m);
+        total_written += m;
+        // A contingency that still satisfied the request is a success:
+        // report it only when fewer than n bytes were written.
+        if(ec && total_written < n)
+            co_return {ec, total_written};
+    }
+
+    co_return {{}, total_written};
+}
+
+} // namespace capy
+} // namespace boost
+
+#endif