body_read_stream implementation (replacement for PR 57)

include/boost/beast2.hpp

@@ -10,6 +10,7 @@
 #ifndef BOOST_BEAST2_HPP
 #define BOOST_BEAST2_HPP
 
+#include <boost/beast2/body_read_stream.hpp>
 #include <boost/beast2/read.hpp>
 #include <boost/beast2/write.hpp>
 

include/boost/beast2/body_read_stream.hpp

@@ -0,0 +1,161 @@
+//
+// Copyright (c) 2025 Mungo Gill
+//
+// 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/beast2
+//
+
+#ifndef BOOST_BEAST2_BODY_READ_STREAM_HPP
+#define BOOST_BEAST2_BODY_READ_STREAM_HPP
+
+#include <boost/asio/async_result.hpp>
+#include <boost/http_proto/parser.hpp>
+#include <boost/system/error_code.hpp>
+
+namespace boost {
+namespace beast2 {
+
+/** A body reader for HTTP/1 messages.
+
+    This type meets the requirements of asio's AsyncReadStream, and is
+    constructed with a reference to an underlying AsyncReadStream.
+
+    Any call to `async_read_some` initially triggers reads from the underlying
+    stream until all of the HTTP headers and at least one byte of the body have
+    been read and processed.  Thereafter, each subsequent call to
+    `async_read_some` processes at least one byte of the body, triggering, where
+    required, calls to the underlying stream's `async_read_some` method.  The
+    resulting body data is stored in the referenced MutableBufferSequence.
+
+    All processing depends on a beast2::parser object owned by the caller and
+    referenced in the construction of this object.
+
+    @see
+        @ref http_proto::parser.
+*/
+template<class AsyncReadStream>
+class body_read_stream
+{
+
+public:
+    /** The type of the executor associated with the stream.
+
+        This will be the type of executor used to invoke completion handlers
+        which do not have an explicit associated executor.
+    */
+    typedef typename AsyncReadStream::executor_type executor_type;
+
+    /** Get the executor associated with the object.
+
+        This function may be used to obtain the executor object that the stream
+        uses to dispatch completion handlers without an assocaited executor.
+
+        @return A copy of the executor that stream will use to dispatch
+        handlers.
+    */
+    executor_type
+    get_executor()
+    {
+        return stream_.get_executor();
+    }
+
+    /** Constructor
+
+        This constructor creates the stream by forwarding all arguments to the
+        underlying socket.  The socket then needs to be open and connected or
+        accepted before data can be sent or received on it.
+
+        @param s  The underlying stream from which the HTTP message is read.
+                  This object's executor is initialized to that of the
+                  underlying stream.
+
+        @param pr A http_proto::parser object which will perform the parsing of
+                  the HTTP message and extraction of the body.  This must be
+                  initialized by the caller and ownership of the parser is
+                  retained by the caller, which must guarantee that it remains
+                  valid until the handler is called.
+    */
+    explicit body_read_stream(AsyncReadStream& s, http_proto::parser& pr);
+
+    /** Read some data asynchronously.
+
+        This function is used to asynchronously read data from the stream.
+
+        This call always returns immediately.  The asynchronous operation will
+        continue until one of the following conditions is true:
+
+        @li The HTTP headers are read in full from the underlying stream and one
+            or more bytes of the body are read from the stream and stored in the
+            buffer `mb`.
+
+        @li An error occurs.
+
+        The algorithm, known as a <em>composed asynchronous operation</em>, is
+        implemented in terms of calls to the underlying stream's
+        `async_read_some` function. The program must ensure that no other calls
+        implemented using the underlying stream's `async_read_some` are
+        performed until this operation completes.
+
+        @param mb The buffers into which the body data will be read.  If the
+        size of the buffers is zero bytes, the operation always completes
+        immediately with no error.  Although the buffers object may be copied as
+        necessary, ownership of the underlying memory blocks is retained by the
+        caller, which must guarantee that they remain valid until the handler is
+        called.  Where the mb buffer is not of sufficient size to hold the read
+        data, the remainder may be read by subsequent calls to this function.
+
+        @param handler The completion handler to invoke when the operation
+        completes.  The implementation takes ownership of the handler by
+        performing a decay-copy.  The equivalent function signature of the
+        handler must be:
+        @code
+        void handler(
+            error_code const& error,        // result of operation
+            std::size_t bytes_transferred   // the number of bytes consumed by the parser
+        );
+        @endcode
+        Regardless of whether the asynchronous operation
+        completes immediately or not, the completion handler will not be invoked
+        from within this function. On immediate completion, invocation of the
+        handler will be performed in a manner equivalent to using
+        `asio::async_immediate`.
+
+        @note The `async_read_some` operation may not receive all of the
+        requested number of bytes.  Consider using the function
+        `asio::async_read` if you need to ensure that the requested amount of
+        data is read before the asynchronous operation completes.
+
+        @par Per-Operation Cancellation
+
+        This asynchronous operation supports cancellation for the following
+        net::cancellation_type values:
+
+        @li @c net::cancellation_type::terminal
+        @li @c net::cancellation_type::partial
+        @li @c net::cancellation_type::total
+
+        if they are also supported by the underlying stream's @c async_read_some
+        operation.
+    */
+    template<
+        class MutableBufferSequence,
+        BOOST_ASIO_COMPLETION_TOKEN_FOR(void(system::error_code, std::size_t))
+            CompletionToken>
+    BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(
+        CompletionToken,
+        void(system::error_code, std::size_t))
+    async_read_some(MutableBufferSequence mb, CompletionToken&& handler);
+
+private:
+    AsyncReadStream& stream_;
+    http_proto::parser& pr_;
+};
+
+} // beast2
+} // boost
+
+#include <boost/beast2/impl/body_read_stream.hpp>
+
+#endif // BOOST_BEAST2_BODY_READ_STREAM_HPP

include/boost/beast2/impl/body_read_stream.hpp

@@ -0,0 +1,165 @@
+//
+// Copyright (c) 2025 Mungo Gill
+//
+// 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/beast2
+//
+
+#ifndef BOOST_BEAST2_IMPL_BODY_READ_STREAM_HPP
+#define BOOST_BEAST2_IMPL_BODY_READ_STREAM_HPP
+
+#include <boost/beast2/detail/config.hpp>
+#include <boost/beast2/read.hpp>
+
+#include <boost/asio/buffer.hpp>
+#include <boost/asio/buffers_iterator.hpp>
+#include <boost/asio/coroutine.hpp>
+#include <boost/core/ignore_unused.hpp>
+
+#include <iostream>
+
+namespace boost {
+namespace beast2 {
+
+namespace detail {
+
+template<class MutableBufferSequence, class AsyncReadStream>
+class body_read_stream_op : public asio::coroutine
+{
+
+    AsyncReadStream& stream_;
+    MutableBufferSequence mb_;
+    http_proto::parser& pr_;
+
+public:
+    body_read_stream_op(
+        AsyncReadStream& s,
+        MutableBufferSequence&& mb,
+        http_proto::parser& pr) noexcept
+        : stream_(s)
+        , mb_(std::move(mb))
+        , pr_(pr)
+    {
+    }
+
+    template<class Self>
+    void
+    operator()(
+        Self& self,
+        system::error_code ec = {},
+        std::size_t bytes_transferred = 0)
+    {
+        boost::ignore_unused(bytes_transferred);
+
+        BOOST_ASIO_CORO_REENTER(*this)
+        {
+            self.reset_cancellation_state(asio::enable_total_cancellation());
+
+            if(buffers::size(mb_) == 0)
+            {
+                BOOST_ASIO_CORO_YIELD
+                {
+                    BOOST_ASIO_HANDLER_LOCATION(
+                        (__FILE__, __LINE__, "immediate"));
+                    auto io_ex = self.get_io_executor();
+                    asio::async_immediate(
+                        io_ex,
+                        asio::append(
+                            std::move(self), system::error_code{}));
+                }
+                goto upcall;
+            }
+
+            if(!pr_.got_header())
+            {
+                BOOST_ASIO_CORO_YIELD
+                {
+                    BOOST_ASIO_HANDLER_LOCATION(
+                        (__FILE__, __LINE__, "async_read_header"));
+                    beast2::async_read_header<AsyncReadStream, Self>(
+                        stream_, pr_, std::move(self));
+                }
+                if(ec.failed())
+                    goto upcall;
+            }
+
+            if(!!self.cancelled())
+            {
+                ec = asio::error::operation_aborted;
+                goto upcall;
+            }
+
+            BOOST_ASIO_CORO_YIELD
+            {
+                BOOST_ASIO_HANDLER_LOCATION(
+                    (__FILE__, __LINE__, "async_read_some"));
+                beast2::async_read_some(stream_, pr_, std::move(self));
+            }
+
+        upcall:
+            std::size_t n = 0;
+
+            if(!ec.failed())
+            {
+                auto dbs = asio::buffer_size(mb_);
+
+                if(dbs > 0)
+                {
+                    auto source_buf = pr_.pull_body();
+
+                    n = asio::buffer_copy(mb_, source_buf);
+
+                    pr_.consume_body(n);
+
+                    ec = (n != 0) ? system::error_code{}
+                                  : asio::stream_errc::eof;
+                }
+            }
+
+            self.complete(ec, n);
+        }
+    }
+};
+
+} // detail
+
+//------------------------------------------------
+
+// TODO: copy in Beast's stream traits to check if AsyncReadStream
+// is an AsyncReadStream, and also static_assert that body_read_stream is too.
+
+template<class AsyncReadStream>
+body_read_stream<AsyncReadStream>::body_read_stream(
+    AsyncReadStream& s,
+    http_proto::parser& pr)
+    : stream_(s)
+    , pr_(pr)
+{
+}
+
+template<class AsyncReadStream>
+template<
+    class MutableBufferSequence,
+    BOOST_ASIO_COMPLETION_TOKEN_FOR(void(system::error_code, std::size_t))
+        CompletionToken>
+BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(
+    CompletionToken,
+    void(system::error_code, std::size_t))
+body_read_stream<AsyncReadStream>::async_read_some(
+    MutableBufferSequence mb,
+    CompletionToken&& token)
+{
+    return asio::
+        async_compose<CompletionToken, void(system::error_code, std::size_t)>(
+            detail::body_read_stream_op<MutableBufferSequence, AsyncReadStream>{
+                stream_, std::move(mb), pr_ },
+            token,
+            stream_);
+}
+
+} // beast2
+} // boost
+
+#endif // BOOST_BEAST2_IMPL_BODY_READ_STREAM_HPP

src/asio_io_context.cpp

@@ -12,7 +12,6 @@
 #include <boost/beast2/server/call_mf.hpp>
 #include <boost/asio/executor_work_guard.hpp>
 #include <boost/asio/signal_set.hpp>
-#include <boost/asio/strand.hpp>
 #include <thread>
 #include <vector>