Add parser::read(stream)

Author: ashtum

CMakeLists.txt

@@ -60,7 +60,6 @@ set(BOOST_HTTP_DEPENDENCIES
     Boost::core
     Boost::json
     Boost::mp11
-    Boost::static_assert
     Boost::system
     Boost::throw_exception
     Boost::type_traits

CMakePresets.json

@@ -340,6 +340,31 @@ class parser
     capy::io_task<>
     read_header(Stream& stream);
 
+    /** Asynchronously read a complete HTTP message.
+
+        Reads from the stream until the message is fully
+        parsed or an error occurs. The body is accumulated
+        in the parser's internal buffer and can be retrieved
+        via @ref body after completion.
+
+        If the parser's internal buffer fills before the
+        message is complete, the operation completes with
+        @ref error::in_place_overflow.
+
+        @par Preconditions
+        @li @ref reset has been called
+        @li @ref start has been called
+
+        @param stream The stream to read from.
+
+        @return An awaitable yielding `(error_code)`.
+
+        @see @ref body, @ref read_header.
+    */
+    template<capy::ReadStream Stream>
+    capy::io_task<>
+    read(Stream& stream);
+
     /** Asynchronously read body data into buffers.
 
         Reads from the stream and copies body data into
@@ -517,6 +542,37 @@ read_header(Stream& stream)
     }
 }
 
+template<capy::ReadStream Stream>
+capy::io_task<>
+parser::
+read(Stream& stream)
+{
+    system::error_code ec;
+    for(;;)
+    {
+        parse(ec);
+
+        if(is_complete())
+            co_return {};
+
+        if(ec && ec != condition::need_more_input)
+            co_return {ec};
+
+        if(ec == condition::need_more_input)
+        {
+            auto mbs = prepare();
+
+            auto [read_ec, n] = co_await stream.read_some(mbs);
+            if(read_ec == capy::cond::eof)
+                commit_eof();
+            else if(!read_ec)
+                commit(n);
+            else
+                co_return {read_ec};
+        }
+    }
+}
+
 template<capy::ReadStream Stream, capy::MutableBufferSequence MB>
 capy::io_task<std::size_t>
 parser::

include/boost/http/parser.hpp

@@ -43,7 +43,7 @@ message(
     case error::expect_100_continue: return "expect 100 continue";
     case error::end_of_message: return "end of message";
     case error::end_of_stream: return "end of stream";
-    case error::in_place_overflow: return "in place overflow";
+    case error::in_place_overflow: return "in-place overflow";
     case error::need_data: return "need data";
 
     case error::bad_connection: return "bad Connection";

src/error.cpp

@@ -1273,6 +1273,10 @@ class parser::impl
                 (is_plain() ? cb0_ : cb1_).data(),
                 body_avail_);
             return detail::make_span(cbp_);
+        case state::reset:
+            if(got_header_)
+                return {};
+            BOOST_FALLTHROUGH;
         default:
             detail::throw_logic_error();
         }
@@ -1291,6 +1295,10 @@ class parser::impl
             (is_plain() ? cb0_ : cb1_).consume(n);
             body_avail_ -= n;
             return;
+        case state::reset:
+            if(got_header_)
+                return;
+            BOOST_FALLTHROUGH;
         default:
             detail::throw_logic_error();
         }

src/parser.cpp

@@ -1974,6 +1974,192 @@ struct parser_coro_test
         BOOST_TEST(r.success);
     }
 
+    void
+    testSourceForLargeBody()
+    {
+        capy::test::fuse f;
+        auto r = f.armed([&](capy::test::fuse&) -> capy::task<>
+        {
+            std::string const expected(40000, 'x');
+
+            capy::test::read_stream rs(f);
+            rs.provide(
+                "HTTP/1.1 200 OK\r\n"
+                "Content-Length: 40000\r\n"
+                "\r\n");
+            rs.provide(expected);
+
+            response_parser pr(res_cfg_);
+            pr.reset();
+            pr.start();
+
+            auto source = pr.source_for(rs);
+
+            std::string body;
+            capy::const_buffer arr[16];
+
+            for(;;)
+            {
+                auto [ec, bufs] = co_await source.pull(arr);
+                if(ec == capy::cond::eof)
+                    break;
+                BOOST_TEST(ec != error::in_place_overflow);
+                if(ec)
+                    co_return;
+                std::size_t n = 0;
+                for(auto const& buf : bufs)
+                {
+                    body.append(
+                        static_cast<char const*>(buf.data()),
+                        buf.size());
+                    n += buf.size();
+                }
+                source.consume(n);
+            }
+
+            BOOST_TEST(body == expected);
+            BOOST_TEST(pr.is_complete());
+        });
+        BOOST_TEST(r.success);
+    }
+
+    void
+    testSourceForBodyTooLarge()
+    {
+        capy::test::fuse f;
+        auto r = f.armed([&](capy::test::fuse&) -> capy::task<>
+        {
+            parser_config cfg{false};
+            cfg.body_limit = 5;
+            auto small_cfg = make_parser_config(cfg);
+
+            capy::test::read_stream rs(f, 1);
+            rs.provide(
+                "HTTP/1.1 200 OK\r\n"
+                "Content-Length: 100\r\n"
+                "\r\n"
+                "this body is way over the configured limit");
+
+            response_parser pr(small_cfg);
+            pr.reset();
+            pr.start();
+
+            auto source = pr.source_for(rs);
+
+            capy::const_buffer arr[16];
+            auto [ec, bufs] = co_await source.pull(arr);
+
+            BOOST_TEST(static_cast<bool>(ec));
+            BOOST_TEST(ec != capy::cond::eof);
+            BOOST_TEST(capy::buffer_size(bufs) == 0);
+            BOOST_TEST(! pr.is_complete());
+        });
+        BOOST_TEST(r.success);
+    }
+
+    void
+    testRead()
+    {
+        capy::test::fuse f;
+        auto r = f.armed([&](capy::test::fuse&) -> capy::task<>
+        {
+            capy::test::read_stream rs(f, 1);
+            rs.provide(
+                "HTTP/1.1 200 OK\r\n"
+                "Content-Length: 13\r\n"
+                "\r\n"
+                "Hello, World!");
+
+            response_parser pr(res_cfg_);
+            pr.reset();
+            pr.start();
+
+            auto [ec] = co_await pr.read(rs);
+            if(ec)
+                co_return;
+
+            BOOST_TEST(pr.is_complete());
+            BOOST_TEST(pr.body() == "Hello, World!");
+        });
+        BOOST_TEST(r.success);
+    }
+
+    void
+    testReadChunked()
+    {
+        capy::test::fuse f;
+        auto r = f.armed([&](capy::test::fuse&) -> capy::task<>
+        {
+            capy::test::read_stream rs(f, 1);
+            rs.provide(
+                "HTTP/1.1 200 OK\r\n"
+                "Transfer-Encoding: chunked\r\n"
+                "\r\n"
+                "5\r\nHello\r\n"
+                "7\r\n, World\r\n"
+                "0\r\n\r\n");
+
+            response_parser pr(res_cfg_);
+            pr.reset();
+            pr.start();
+
+            auto [ec] = co_await pr.read(rs);
+            if(ec)
+                co_return;
+
+            BOOST_TEST(pr.is_complete());
+            BOOST_TEST(pr.body() == "Hello, World");
+        });
+        BOOST_TEST(r.success);
+    }
+
+    void
+    testReadEof()
+    {
+        capy::test::fuse f;
+        auto r = f.armed([&](capy::test::fuse&) -> capy::task<>
+        {
+            capy::test::read_stream rs(f, 1);
+
+            response_parser pr(res_cfg_);
+            pr.reset();
+            pr.start();
+
+            auto [ec] = co_await pr.read(rs);
+
+            BOOST_TEST(static_cast<bool>(ec));
+            BOOST_TEST(! pr.is_complete());
+        });
+        BOOST_TEST(r.success);
+    }
+
+    void
+    testReadOverflow()
+    {
+        capy::test::fuse f;
+        auto r = f.armed([&](capy::test::fuse&) -> capy::task<>
+        {
+            std::string const big(40000, 'x');
+
+            capy::test::read_stream rs(f);
+            rs.provide(
+                "HTTP/1.1 200 OK\r\n"
+                "Content-Length: 40000\r\n"
+                "\r\n");
+            rs.provide(big);
+
+            response_parser pr(res_cfg_);
+            pr.reset();
+            pr.start();
+
+            auto [ec] = co_await pr.read(rs);
+
+            BOOST_TEST(static_cast<bool>(ec));
+            BOOST_TEST(! pr.is_complete());
+        });
+        BOOST_TEST(r.success);
+    }
+
     void
     run()
     {
@@ -1982,6 +2168,12 @@ struct parser_coro_test
         testReadWriteSinkChunked();
         testSourceFor();
         testSourceForChunked();
+        testSourceForLargeBody();
+        testSourceForBodyTooLarge();
+        testRead();
+        testReadChunked();
+        testReadEof();
+        testReadOverflow();
     }
 };