From 5e3cb02cd49684ae0f3c385764d93b7ce08da76e Mon Sep 17 00:00:00 2001 From: Mohammad Nejati Date: Sun, 5 Jul 2026 10:57:16 +0330 Subject: [PATCH 1/3] serializer: add encoder support --- src/client.cpp | 15 +- src/detail/serializer.cpp | 313 ++++++++--- src/detail/serializer.hpp | 181 +++--- test/unit/detail/serializer.cpp | 968 ++++++++++++++++++++++++++++++-- 4 files changed, 1293 insertions(+), 184 deletions(-) diff --git a/src/client.cpp b/src/client.cpp index 2cbe60e..c33a077 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -264,24 +264,17 @@ client::execute_impl( // TODO: expect100timeout + capy::any_write_stream ws(&conn); + detail::serializer sr(&ws, &headers, {}, {}); if(request.body.has_value()) { - capy::any_write_stream ws(&conn); - detail::serializer sr(ws, headers); capy::any_buffer_sink sink(&sr); if(auto [wec] = co_await request.body.write(sink); wec) co_return { wec, {} }; - if(!sr.is_done()) - { - if(auto [wec] = co_await sr.write_eof(); wec) - co_return { wec, {} }; - } } - else + if(!sr.is_done()) { - auto [wec, n] = - co_await capy::write(conn, capy::make_buffer(headers.buffer())); - if(wec) + if(auto [wec] = co_await sr.write_eof(); wec) co_return { wec, {} }; } diff --git a/src/detail/serializer.cpp b/src/detail/serializer.cpp index c6ea42a..00a0405 100644 --- a/src/detail/serializer.cpp +++ b/src/detail/serializer.cpp @@ -9,10 +9,9 @@ #include "serializer.hpp" -#include - -#include #include +#include +#include namespace boost { @@ -21,143 +20,315 @@ namespace burl namespace detail { -serializer::serializer( - capy::any_write_stream& stream, - http::request_base& req, +serializer:: +serializer( + capy::any_write_stream* stream, + http::message_base* msg, + encoder* enc, config cfg) : stream_(stream) - , req_(req) + , msg_(msg) + , enc_(enc) , cfg_(cfg) - , storage_(new unsigned char[cfg_.buffer_size + margin] + margin) + , buf_(new unsigned char[ + cfg_.buffer_size + + (enc ? enc->buffer_size : 0) + margin] + margin) +{ +} + +serializer:: +serializer(serializer&& other) noexcept + : stream_(other.stream_) + , msg_(other.msg_) + , enc_(other.enc_) + , cfg_(other.cfg_) + , buf_(other.buf_) + , out_len_(other.out_len_) + , in_len_(other.in_len_) + , total_body_(other.total_body_) + , enc_started_(other.enc_started_) + , shifted_(other.shifted_) + , hdr_sent_(other.hdr_sent_) + , done_(other.done_) +{ + other.buf_ = nullptr; +} + +serializer& +serializer:: +operator=(serializer&& other) noexcept { - BOOST_ASSERT( - req_.chunked() || req_.payload() == http::payload::size); + if(this == &other) + return *this; + this->~serializer(); + return *new(this) serializer(std::move(other)); } -serializer::~serializer() +serializer:: +~serializer() { - delete[](storage_ - margin); + if(buf_) + delete[](buf_ - + (shifted_ ? cfg_.buffer_size : 0) - margin); +} + +bool +serializer:: +is_done() const noexcept +{ + return done_; } capy::io_task<> -serializer::write_eof() +serializer:: +write_eof() { return commit_eof(0); } std::span -serializer::prepare(std::span dest) +serializer:: +prepare(std::span dest) { if(dest.empty() || capacity() == 0) return dest.first(0); - dest[0] = writable(); + dest[0] = do_prepare(); return dest.first(1); } capy::io_task<> -serializer::commit(std::size_t n) +serializer:: +commit(std::size_t n) { BOOST_ASSERT(n <= capacity()); - avail_ += n; + do_commit(n); if(capacity() >= cfg_.min_prepare) co_return {}; - auto [ec, _] = co_await drain({}, false); + auto [ec, _] = co_await process({}, false); co_return { ec }; } capy::io_task<> -serializer::commit_eof(std::size_t n) +serializer:: +commit_eof(std::size_t n) { BOOST_ASSERT(n <= capacity()); - avail_ += n; - decide_framing(0); - auto [ec, _] = co_await drain({}, true); + do_commit(n); + finalize(0); + auto [ec, _] = co_await process({}, true); co_return { ec }; } +std::size_t +serializer:: +capacity() const noexcept +{ + if(enc_) + return enc_->buffer_size - in_len_; + + return cfg_.buffer_size - out_len_; +} + +capy::mutable_buffer +serializer:: +do_prepare() noexcept +{ + if(enc_) + return { buf_ + cfg_.buffer_size + in_len_, + enc_->buffer_size - in_len_ }; + + return { buf_ + out_len_, + cfg_.buffer_size - out_len_ }; +} + +void +serializer:: +do_commit(std::size_t n) noexcept +{ + if(enc_) + in_len_ += n; + else + out_len_ += n; +} + +bool +serializer:: +can_coalesce(std::size_t avail) const noexcept +{ + if(avail > capacity()) + return false; + if(enc_) + return !enc_started_ && avail + in_len_ < enc_->threshold; + return avail < cfg_.direct_thr; +} + +void +serializer:: +finalize(std::size_t remaining) noexcept +{ + if(enc_) + { + if(enc_started_) + return; + + if(remaining + in_len_ >= enc_->threshold) + return; + + msg_->erase(http::field::content_encoding); + enc_ = nullptr; + out_len_ = in_len_; + buf_ += cfg_.buffer_size; + shifted_ = true; + } + decide_framing(remaining); +} + void -serializer::decide_framing(std::size_t remaining) noexcept +serializer:: +decide_framing(std::size_t remaining) noexcept { - if(!req_.chunked() || hdr_sent_) + if(!msg_->chunked() || hdr_sent_) return; BOOST_ASSERT(total_body_ == 0); - req_.erase(http::field::transfer_encoding); - req_.set_content_length(avail_ + remaining); + msg_->erase(http::field::transfer_encoding); + msg_->set_content_length(out_len_ + remaining); } capy::io_task -serializer::drain( +serializer:: +process( std::span tail, bool eof) { - auto const tail_size = capy::buffer_size(tail); - auto const chunked = req_.chunked(); + if(enc_) + return encode(tail, eof); + return flush(tail, eof); +} + +capy::io_task +serializer:: +encode( + std::span tail, + bool eof) +{ + std::size_t n = 0; + std::size_t consumed = 0; + capy::const_buffer in = { buf_ + cfg_.buffer_size, in_len_ }; + for(;;) + { + if(in.size() == 0) + { + if(n != tail.size()) + in = tail[n++]; + else if(!eof) + co_return { {}, consumed }; + } + + if(cfg_.buffer_size - out_len_ == 0) + { + if(auto [ec, _] = co_await flush({}, false); ec) + co_return { ec, consumed }; + } + + capy::mutable_buffer out = + { buf_ + out_len_, cfg_.buffer_size - out_len_ }; + + auto r = enc_->process( + out, in, eof && n == tail.size()); + enc_started_ = true; + out_len_ += r.produced; + in += r.consumed; + if(in_len_ != 0) + in_len_ -= r.consumed; + else + consumed += r.consumed; + + if(r.ec) + { + if(r.ec == capy::cond::eof) + { + decide_framing(0); + auto [ec, _] = co_await flush({}, true); + co_return { ec, consumed }; + } + else + { + co_return { r.ec, consumed }; + } + } - BOOST_ASSERT(eof || avail_ + tail_size != 0); + if(!eof && consumed != 0) + co_return { {}, consumed }; + } +} + +capy::io_task +serializer:: +flush( + std::span tail, + bool eof) +{ + auto const tail_size = capy::buffer_size(tail); + auto const chunked = msg_->chunked(); + BOOST_ASSERT(eof || out_len_ + tail_size != 0); + BOOST_ASSERT(tail.size() <= capy::detail::max_iovec_); capy::const_buffer vec[capy::detail::max_iovec_ + 3]; - std::size_t n = 0; + std::size_t n = 0; std::size_t sum = 0; + auto const append = [&](capy::const_buffer b) + { + vec[n++] = b; + sum += b.size(); + }; if(!hdr_sent_) - { - vec[n++] = capy::make_buffer(req_.buffer()); - sum += req_.buffer().size(); - } + append(capy::make_buffer(msg_->buffer())); if(chunked) { - auto const buf = chunk_frame(tail_size); - vec[n++] = buf; - sum += buf.size(); + append(chunk_frame(tail_size)); } else { auto const declared = - req_.payload() == http::payload::size - ? req_.payload_size() : 0; - auto const produced = total_body_ + avail_ + tail_size; + msg_->payload() == http::payload::size + ? msg_->payload_size() : 0; + auto const produced = total_body_ + out_len_ + tail_size; if(produced > declared || (eof && produced != declared)) co_return { error::body_size_mismatch, 0 }; - if(avail_ != 0) - { - vec[n++] = { storage_, avail_ }; - sum += avail_; - } + if(out_len_ != 0) + append({ buf_, out_len_ }); } auto const owned = sum; - sum += tail_size; - for(auto const& b : tail) - vec[n++] = b; + for(auto& b : tail) + append(b); if(chunked && eof) - { - std::size_t const term = avail_ || tail_size ? 7 : 2; - vec[n++] = { "\r\n0\r\n\r\n", term }; - sum += term; - } + append({ "\r\n0\r\n\r\n", out_len_ || tail_size ? 7u : 2u }); - auto const need = chunked || eof ? sum : owned + bool(tail_size); + auto const need = chunked || eof ? sum : owned + (tail_size != 0); auto [ec, written] = co_await write_at_least({ vec, n }, need); if(ec) co_return { ec, 0 }; auto const consumed = (std::min)(written - owned, tail_size); - total_body_ += avail_ + consumed; - avail_ = 0; + total_body_ += out_len_ + consumed; + out_len_ = 0; hdr_sent_ = true; done_ = eof; co_return { {}, consumed }; } capy::io_task -serializer::write_at_least( - std::span buffers, +serializer:: +write_at_least( + std::span buffers, std::size_t bytes) { BOOST_ASSERT(bytes <= capy::buffer_size(buffers)); @@ -165,7 +336,8 @@ serializer::write_at_least( std::size_t written = 0; while(written < bytes) { - auto [ec, n] = co_await stream_.write_some(slice.data()); + auto [ec, n] = co_await stream_-> + write_some(slice.data()); written += n; if(ec && written < bytes) co_return { ec, written }; @@ -175,30 +347,31 @@ serializer::write_at_least( } capy::const_buffer -serializer::chunk_frame(std::size_t tail_size) noexcept +serializer:: +chunk_frame(std::size_t tail_size) noexcept { - static constexpr char hex[] = "0123456789ABCDEF"; - - auto size = avail_ + tail_size; - auto* p = storage_; + static constexpr char hex[] = "0123456789abcdef"; + auto s = out_len_ + tail_size; + auto p = buf_; *--p = '\n'; *--p = '\r'; do { - *--p = hex[size & 0xF]; - size >>= 4; - } while(size != 0); + *--p = hex[s & 0xF]; + s >>= 4; + } while(s != 0); - // previous chunk's CRLF + // prev chunk's CRLF if(total_body_ != 0) { *--p = '\n'; *--p = '\r'; } - return { p, static_cast(storage_ - p) + avail_ }; + return { p, static_cast( + buf_ - p) + out_len_ }; } } // namespace detail diff --git a/src/detail/serializer.hpp b/src/detail/serializer.hpp index 66e57fb..dfe0827 100644 --- a/src/detail/serializer.hpp +++ b/src/detail/serializer.hpp @@ -10,15 +10,14 @@ #ifndef BOOST_BURL_SRC_DETAIL_SERIALIZER_HPP #define BOOST_BURL_SRC_DETAIL_SERIALIZER_HPP -#include +#include -#include #include #include -#include #include #include #include +#include namespace boost { @@ -30,77 +29,64 @@ namespace detail class serializer { public: - struct config + struct encoder { - std::size_t buffer_size = 64 * 1024; - std::size_t min_prepare = 32 * 1024; - std::size_t min_direct = 8 * 1024; + std::size_t buffer_size = 8 * 1024; + std::size_t threshold = 4 * 1024; + + struct result + { + std::size_t consumed; + std::size_t produced; + std::error_code ec; + }; + + virtual result + process( + capy::mutable_buffer out, + capy::const_buffer in, + bool eof) = 0; }; - serializer( - capy::any_write_stream& stream, - http::request_base& req) - : serializer(stream, req, {}) + struct config { - } + std::size_t buffer_size = 64 * 1024; + std::size_t min_prepare = 4 * 1024; + std::size_t direct_thr = 2 * 1024; + }; serializer( - capy::any_write_stream& stream, - http::request_base& req, + capy::any_write_stream* stream, + http::message_base* msg, + encoder* enc, config cfg); + serializer(serializer&& other) noexcept; + + serializer& + operator=(serializer&& other) noexcept; + + serializer(const serializer&) = delete; + + serializer& + operator=(const serializer&) = delete; + ~serializer(); bool - is_done() const noexcept - { - return done_; - } + is_done() const noexcept; template capy::io_task - write_some(Buffers buffers) - { - auto const avail = capy::buffer_size(buffers); - - if(avail < cfg_.min_direct) - { - if(avail >= capacity()) - if(auto [ec, _] = co_await drain({}, false); ec) - co_return { ec, 0 }; - auto n = capy::buffer_copy(writable(), buffers); - avail_ += n; - co_return { {}, n }; - } - - capy::const_buffer_param bp(buffers); - co_return co_await drain(bp.data(), false); - } + write_some(Buffers buffers); template capy::io_task - write(Buffers buffers) - { - return capy::write(*this, std::move(buffers)); - } + write(Buffers buffers); template capy::io_task - write_eof(Buffers buffers) - { - decide_framing(capy::buffer_size(buffers)); - - capy::const_buffer_param bp(buffers); - - if(!bp.more()) - co_return co_await drain(bp.data(), true); - - auto [ec1, n] = co_await write(buffers); - if(ec1) - co_return { ec1, n }; - auto [ec2, _] = co_await drain({}, true); - co_return { ec2, n }; - } + write_eof(Buffers buffers); capy::io_task<> write_eof(); @@ -116,29 +102,42 @@ class serializer private: std::size_t - capacity() const noexcept - { - return cfg_.buffer_size - avail_; - } + capacity() const noexcept; capy::mutable_buffer - writable() const noexcept - { - return { storage_ + avail_, capacity() }; - } + do_prepare() noexcept; + + void + do_commit(std::size_t n) noexcept; + + bool + can_coalesce(std::size_t avail) const noexcept; + + void + finalize(std::size_t remaining) noexcept; void decide_framing(std::size_t remaining) noexcept; capy::io_task - drain( + process( + std::span tail, + bool eof); + + capy::io_task + encode( + std::span tail, + bool eof); + + capy::io_task + flush( std::span tail, bool eof); // TODO: replace this with capy's version capy::io_task write_at_least( - std::span buffers, + std::span buffers, std::size_t bytes); capy::const_buffer @@ -146,16 +145,62 @@ class serializer static constexpr std::size_t margin = 24; - capy::any_write_stream& stream_; - http::request_base& req_; + capy::any_write_stream* stream_; + http::message_base* msg_; + encoder* enc_; config cfg_; - unsigned char* storage_; + unsigned char* buf_; + std::size_t out_len_ = 0; + std::size_t in_len_ = 0; std::uint64_t total_body_ = 0; - std::size_t avail_ = 0; + bool enc_started_ = false; + bool shifted_ = false; bool hdr_sent_ = false; bool done_ = false; }; +template +capy::io_task +serializer:: +write_some(Buffers buffers) +{ + auto const avail = capy::buffer_size(buffers); + if(can_coalesce(avail)) + { + do_commit(capy::buffer_copy(do_prepare(), buffers)); + co_return { {}, avail }; + } + co_return co_await process( + capy::buffer_param(buffers).data(), false); +} + +template +capy::io_task +serializer:: +write(Buffers buffers) +{ + return capy::write(*this, std::move(buffers)); +} + +template +capy::io_task +serializer:: +write_eof(Buffers buffers) +{ + finalize(capy::buffer_size(buffers)); + capy::buffer_param bp(buffers); + std::size_t sum = 0; + for(;;) + { + bool eof = !bp.more(); + auto [ec, n] = co_await process(bp.data(), eof); + sum += n; + bp.consume(n); + if(ec || eof) + co_return { ec, sum }; + } +} + } // namespace detail } // namespace burl } // namespace boost diff --git a/test/unit/detail/serializer.cpp b/test/unit/detail/serializer.cpp index 3d2caf4..6731c24 100644 --- a/test/unit/detail/serializer.cpp +++ b/test/unit/detail/serializer.cpp @@ -12,12 +12,15 @@ #include +#include #include #include #include #include #include +#include + #include "test_suite.hpp" namespace boost @@ -36,7 +39,7 @@ class serializer_test static constexpr serializer::config cfg{ .buffer_size = 64, .min_prepare = 32, - .min_direct = 16 }; + .direct_thr = 16 }; static http::request make_request() @@ -54,6 +57,86 @@ class serializer_test return req; } + // An encoder that increments every body byte by one, so + // encoded output is distinguishable from identity output, + // and appends an optional trailer once the input ends. It + // consumes and produces as much as the given buffers allow. + struct test_encoder : serializer::encoder + { + std::string trailer; + std::error_code fail; + std::size_t calls = 0; + bool finished = false; + + explicit test_encoder( + std::size_t bufsz = 32, + std::size_t thr = 8, + std::string trl = {}) + : trailer(std::move(trl)) + { + buffer_size = bufsz; + threshold = thr; + } + + result + process( + capy::mutable_buffer out, + capy::const_buffer in, + bool eof) override + { + ++calls; + if(fail) + return { 0, 0, fail }; + + auto* dst = static_cast(out.data()); + auto* src = static_cast(in.data()); + auto const n = (std::min)(out.size(), in.size()); + for(std::size_t i = 0; i != n; ++i) + dst[i] = static_cast(src[i] + 1); + + result r{ n, n, {} }; + if(eof && n == in.size()) + { + auto const t = (std::min)( + out.size() - n, trailer.size() - trailer_pos_); + std::memcpy( + dst + n, trailer.data() + trailer_pos_, t); + trailer_pos_ += t; + r.produced += t; + if(trailer_pos_ == trailer.size()) + { + finished = true; + r.ec = capy::error::eof; + } + } + return r; + } + + private: + std::size_t trailer_pos_ = 0; + }; + + // Returns a body with a varied byte pattern so reordered + // or duplicated regions cannot go unnoticed. + static std::string + make_body(std::size_t n) + { + std::string s(n, '\0'); + for(std::size_t i = 0; i != n; ++i) + s[i] = static_cast('0' + i % 64); + return s; + } + + // Returns the mock-encoded form of `s`. + static std::string + encoded(std::string_view s) + { + std::string r(s); + for(auto& c : r) + c = static_cast(c + 1); + return r; + } + public: void testContentLengthSmallBody() @@ -64,7 +147,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); BOOST_TEST(!sr.is_done()); auto [ec1, n] = co_await sr.write( @@ -87,7 +170,7 @@ class serializer_test testContentLengthLargeBody() { // A body at the direct-write threshold bypasses staging. - std::string const body(cfg.min_direct, 'x'); + std::string const body(cfg.direct_thr, 'x'); auto req = make_request(body.size()); auto [client, server] = capy::test::make_stream_pair(); @@ -95,7 +178,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); auto [ec1, n] = co_await sr.write( capy::make_buffer(body)); @@ -123,7 +206,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); std::array bufs{ capy::make_buffer(b1), @@ -151,7 +234,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); auto [ec1, n] = co_await sr.write( capy::const_buffer("hello", 5)); @@ -176,7 +259,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); auto [ec, n] = co_await sr.write_eof( capy::const_buffer("hello", 5)); @@ -197,9 +280,9 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); - // The body stays below min_direct, so it is fully + // The body stays below direct_thr, so it is fully // buffered before the header goes out. auto [ec1, n] = co_await sr.write( capy::const_buffer("hello world", 11)); @@ -232,7 +315,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); auto [ec, n] = co_await sr.write_eof( capy::const_buffer("hello", 5)); @@ -250,7 +333,7 @@ class serializer_test void testChunkedLargeBody() { - std::string const body(cfg.min_direct, 'x'); + std::string const body(cfg.direct_thr, 'x'); auto req = make_request(); auto [client, server] = capy::test::make_stream_pair(); @@ -258,7 +341,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); auto [ec1, n1] = co_await sr.write( capy::make_buffer(body)); @@ -295,7 +378,7 @@ class serializer_test // caller buffers must frame the staged bytes and the // caller's bytes as the final chunk and terminate the // body with the last-chunk in the same gather. - std::string const body(cfg.min_direct, 'x'); + std::string const body(cfg.direct_thr, 'x'); auto req = make_request(); auto [client, server] = capy::test::make_stream_pair(); @@ -303,7 +386,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); auto [ec1, n1] = co_await sr.write( capy::make_buffer(body)); @@ -335,7 +418,7 @@ class serializer_test // write_eof() with caller buffers while no bytes are // staged must still terminate the body with the // last-chunk after the final chunk. - std::string const body(cfg.min_direct, 'x'); + std::string const body(cfg.direct_thr, 'x'); auto req = make_request(); auto [client, server] = capy::test::make_stream_pair(); @@ -343,7 +426,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); auto [ec1, n1] = co_await sr.write( capy::make_buffer(body)); @@ -373,7 +456,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req); + serializer sr(&ws, &req, {}, {}); BOOST_TEST(!sr.is_done()); auto [ec] = co_await sr.write_eof(); @@ -397,7 +480,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -428,7 +511,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -462,7 +545,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -477,7 +560,7 @@ class serializer_test }()); std::string expected(req.buffer()); - expected += "3C\r\n" + body + "\r\n"; + expected += "3c\r\n" + body + "\r\n"; expected += "0\r\n\r\n"; BOOST_TEST(req.chunked()); BOOST_TEST_EQ(server.data(), expected); @@ -495,7 +578,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -525,7 +608,7 @@ class serializer_test auto [client, server] = capy::test::make_stream_pair(); capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); BOOST_TEST(sr.prepare({}).empty()); @@ -553,7 +636,7 @@ class serializer_test capy::test::write_stream ws_impl(f); capy::any_write_stream ws(&ws_impl); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -589,7 +672,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -632,8 +715,8 @@ class serializer_test // per write: bytes already on the wire may not be sent // again, and every call must consume at least one byte // of the caller's buffer. - std::string const b1(cfg.min_direct, 'x'); - std::string const b2(cfg.min_direct, 'y'); + std::string const b1(cfg.direct_thr, 'x'); + std::string const b2(cfg.direct_thr, 'y'); auto req = make_request(b1.size() + b2.size()); capy::test::write_stream ws_impl({}, 1); @@ -641,7 +724,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); // The first drain writes the header in full and // exactly one body byte. @@ -653,7 +736,7 @@ class serializer_test ws_impl.data(), std::string(req.buffer()) + "x"); - // The rest of the first buffer is below min_direct + // The rest of the first buffer is below direct_thr // and is staged without touching the wire. auto [ec2, n2] = co_await sr.write( capy::const_buffer(b1.data() + 1, b1.size() - 1)); @@ -702,7 +785,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -736,7 +819,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); auto [ec1, n] = co_await sr.write( capy::const_buffer("hello", 5)); @@ -763,7 +846,7 @@ class serializer_test // gather as one chunk with the staged bytes first, even // when the stream accepts one byte at a time; // 3 + 16 == 0x13. - std::string const body(cfg.min_direct, 'x'); + std::string const body(cfg.direct_thr, 'x'); auto req = make_request(); capy::test::write_stream ws_impl({}, 1); @@ -771,7 +854,7 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); auto [ec1, n1] = co_await sr.write( capy::const_buffer("abc", 3)); @@ -800,7 +883,7 @@ class serializer_test capy::test::fuse f; auto r = f.armed([&](capy::test::fuse&) -> capy::task<> { - std::string const body(cfg.min_direct, 'x'); + std::string const body(cfg.direct_thr, 'x'); auto req = make_request(); @@ -808,7 +891,7 @@ class serializer_test capy::test::make_stream_pair(f); capy::any_write_stream ws(&client); - serializer sr(ws, req, cfg); + serializer sr(&ws, &req, {}, cfg); if(auto [ec, n] = co_await sr.write( capy::make_buffer(body)); ec) @@ -841,6 +924,804 @@ class serializer_test BOOST_TEST(r.success); } + void + testEncoderSmallBodyDropsEncoder() + { + // A body that stays below the encoder threshold skips + // encoding entirely: the encoder is dropped at eof, the + // Content-Encoding header is removed, and the identity + // body is sent with a Content-Length. + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + // below the threshold: coalesced into the encoder's + // staging buffer without invoking the encoder + auto [ec1, n] = co_await sr.write( + capy::const_buffer("hello", 5)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n, 5u); + BOOST_TEST_EQ(enc.calls, 0u); + BOOST_TEST(server.data().empty()); + + auto [ec2] = co_await sr.write_eof(); + BOOST_TEST(!ec2); + BOOST_TEST(sr.is_done()); + }()); + + BOOST_TEST_EQ(enc.calls, 0u); + BOOST_TEST(!req.chunked()); + BOOST_TEST_EQ(req.payload_size(), 5u); + BOOST_TEST( + server.data().find("Content-Encoding") == + std::string_view::npos); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + "hello"); + } + + void + testEncoderEmptyBody() + { + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec] = co_await sr.write_eof(); + BOOST_TEST(!ec); + BOOST_TEST(sr.is_done()); + }()); + + // an empty body is below any threshold: the encoder is + // dropped and the request converts to Content-Length: 0 + BOOST_TEST_EQ(enc.calls, 0u); + BOOST_TEST(!req.chunked()); + BOOST_TEST( + server.data().find("Content-Length: 0\r\n") != + std::string_view::npos); + BOOST_TEST( + server.data().find("Content-Encoding") == + std::string_view::npos); + BOOST_TEST_EQ(server.data(), req.buffer()); + } + + void + testEncoderPrepareCommitEof() + { + // With an encoder, prepare() exposes the encoder's + // staging buffer rather than the output buffer. + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + capy::mutable_buffer tmp[2]; + auto dest = sr.prepare(tmp); + BOOST_TEST_EQ( + capy::buffer_size(dest), enc.buffer_size); + + auto n = capy::buffer_copy( + dest, capy::const_buffer("hello", 5)); + BOOST_TEST_EQ(n, 5u); + + // below the threshold at eof: the encoder is + // dropped and the staged bytes are sent as-is + auto [ec] = co_await sr.commit_eof(n); + BOOST_TEST(!ec); + BOOST_TEST(sr.is_done()); + }()); + + BOOST_TEST_EQ(enc.calls, 0u); + BOOST_TEST(!req.chunked()); + BOOST_TEST_EQ(req.payload_size(), 5u); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + "hello"); + } + + void + testEncoderThreshold() + { + // exactly at the threshold: the encoder is kept + { + std::string const body = make_body(8); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec, n] = co_await sr.write_eof( + capy::make_buffer(body)); + BOOST_TEST(!ec); + BOOST_TEST_EQ(n, body.size()); + BOOST_TEST(sr.is_done()); + }()); + + // the whole encoded output was buffered before the + // header went out, so chunked is replaced with the + // encoded length while Content-Encoding is kept + BOOST_TEST(enc.finished); + BOOST_TEST(!req.chunked()); + BOOST_TEST_EQ(req.payload_size(), 8u); + BOOST_TEST( + server.data().find("Content-Encoding: test\r\n") != + std::string_view::npos); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + encoded(body)); + } + + // one byte below the threshold: the encoder is dropped + { + std::string const body = make_body(7); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec, n] = co_await sr.write_eof( + capy::make_buffer(body)); + BOOST_TEST(!ec); + BOOST_TEST_EQ(n, body.size()); + }()); + + BOOST_TEST_EQ(enc.calls, 0u); + BOOST_TEST_EQ(req.payload_size(), 7u); + BOOST_TEST( + server.data().find("Content-Encoding") == + std::string_view::npos); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + body); + } + } + + void + testEncoderStagedAndTail() + { + // Staged bytes and the caller's tail together reach the + // threshold at eof, so the encoder is kept; it consumes + // the staged bytes first, then the tail, and the count + // returned to the caller covers only the tail. + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec1, n1] = co_await sr.write( + capy::const_buffer("abc", 3)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n1, 3u); + BOOST_TEST_EQ(enc.calls, 0u); + + auto [ec2, n2] = co_await sr.write_eof( + capy::const_buffer("defgh", 5)); + BOOST_TEST(!ec2); + BOOST_TEST_EQ(n2, 5u); + BOOST_TEST(sr.is_done()); + }()); + + BOOST_TEST(enc.finished); + BOOST_TEST_EQ(req.payload_size(), 8u); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + encoded("abcdefgh")); + } + + void + testEncoderLargeBodyChunked() + { + // A body larger than the output buffer forces the + // header out early and streams encoded chunks; once the + // encoder has started, even small writes are encoded + // rather than coalesced. Chunk payloads hold encoded + // bytes; 64 == 0x40. + std::string const body = make_body(200); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec1, n] = co_await sr.write( + capy::make_buffer(body)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n, body.size()); + BOOST_TEST(!sr.is_done()); + + auto [ec2] = co_await sr.write_eof(); + BOOST_TEST(!ec2); + BOOST_TEST(sr.is_done()); + }()); + + auto const enc_body = encoded(body); + std::string expected(req.buffer()); + expected += "40\r\n" + enc_body.substr(0, 64); + expected += "\r\n40\r\n" + enc_body.substr(64, 64); + expected += "\r\n40\r\n" + enc_body.substr(128, 64); + expected += "\r\n8\r\n" + enc_body.substr(192, 8); + expected += "\r\n0\r\n\r\n"; + BOOST_TEST(enc.finished); + BOOST_TEST(req.chunked()); + BOOST_TEST( + server.data().find("Content-Encoding: test\r\n") != + std::string_view::npos); + BOOST_TEST_EQ(server.data(), expected); + } + + void + testEncoderCommit() + { + // commit() stages input for the encoder; a drain is + // deferred until the remaining capacity drops below + // min_prepare, and encoded bytes stay buffered until + // the output buffer needs room. + std::string const body = make_body(54); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc(64, 8); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + capy::mutable_buffer tmp[2]; + + // capacity stays >= min_prepare: no drain + auto n1 = capy::buffer_copy( + sr.prepare(tmp), + capy::const_buffer(body.data(), 20)); + BOOST_TEST_EQ(n1, 20u); + auto [ec1] = co_await sr.commit(n1); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(enc.calls, 0u); + + // capacity drops below min_prepare: the staged + // bytes are encoded, but the encoded output still + // fits and nothing reaches the wire + auto n2 = capy::buffer_copy( + sr.prepare(tmp), + capy::const_buffer(body.data() + 20, 24)); + BOOST_TEST_EQ(n2, 24u); + auto [ec2] = co_await sr.commit(n2); + BOOST_TEST(!ec2); + BOOST_TEST_EQ(enc.calls, 1u); + BOOST_TEST(server.data().empty()); + + // staged bytes present at commit_eof() are encoded + // before the encoder is finished + auto n3 = capy::buffer_copy( + sr.prepare(tmp), + capy::const_buffer(body.data() + 44, 10)); + BOOST_TEST_EQ(n3, 10u); + auto [ec3] = co_await sr.commit_eof(n3); + BOOST_TEST(!ec3); + BOOST_TEST(sr.is_done()); + }()); + + // everything fit before the header went out: chunked + // is replaced with the encoded length + BOOST_TEST_EQ(enc.calls, 2u); + BOOST_TEST(enc.finished); + BOOST_TEST(!req.chunked()); + BOOST_TEST_EQ(req.payload_size(), 54u); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + encoded(body)); + } + + void + testEncoderCommitTriggersFlush() + { + // A commit-triggered drain that overflows the output + // buffer sends the header and a full chunk mid-commit; + // the remainder stays buffered for the final flush; + // 64 == 0x40, 32 == 0x20. + std::string const body = make_body(96); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc(96, 8); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + capy::mutable_buffer tmp[2]; + auto n = capy::buffer_copy( + sr.prepare(tmp), capy::make_buffer(body)); + BOOST_TEST_EQ(n, body.size()); + + auto [ec1] = co_await sr.commit(n); + BOOST_TEST(!ec1); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + + "40\r\n" + encoded(body).substr(0, 64)); + + auto [ec2] = co_await sr.write_eof(); + BOOST_TEST(!ec2); + }()); + + std::string expected(req.buffer()); + expected += "40\r\n" + encoded(body).substr(0, 64); + expected += "\r\n20\r\n" + encoded(body).substr(64, 32); + expected += "\r\n0\r\n\r\n"; + BOOST_TEST(req.chunked()); + BOOST_TEST_EQ(server.data(), expected); + } + + void + testEncoderTrailerSpansFlush() + { + // The encoder finishes over multiple eof calls when its + // trailer does not fit in the remaining output space; a + // flush runs between the calls; 64 == 0x40. + std::string const body = make_body(60); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc(32, 8, "0123456789"); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec, n] = co_await sr.write_eof( + capy::make_buffer(body)); + BOOST_TEST(!ec); + BOOST_TEST_EQ(n, body.size()); + BOOST_TEST(sr.is_done()); + }()); + + std::string expected(req.buffer()); + expected += "40\r\n" + encoded(body) + "0123"; + expected += "\r\n6\r\n456789"; + expected += "\r\n0\r\n\r\n"; + BOOST_TEST_EQ(enc.calls, 2u); + BOOST_TEST(enc.finished); + BOOST_TEST(req.chunked()); + BOOST_TEST_EQ(server.data(), expected); + } + + void + testEncoderContentLength() + { + // A sized request works with an encoder when the + // declared length matches the encoded output (body + // plus trailer). + std::string const body = make_body(30); + + auto req = make_request(34); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc(32, 8, "eof!"); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec, n] = co_await sr.write_eof( + capy::make_buffer(body)); + BOOST_TEST(!ec); + BOOST_TEST_EQ(n, body.size()); + BOOST_TEST(sr.is_done()); + }()); + + BOOST_TEST(enc.finished); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + + encoded(body) + "eof!"); + } + + void + testEncoderContentLengthMismatch() + { + // encoded output exceeds the declared length + { + std::string const body = make_body(30); + + auto req = make_request(30); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc(32, 8, "eof!"); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec, n] = co_await sr.write_eof( + capy::make_buffer(body)); + BOOST_TEST(ec == error::body_size_mismatch); + BOOST_TEST(!sr.is_done()); + }()); + + BOOST_TEST(server.data().empty()); + } + + // encoded output falls short of the declared length + { + std::string const body = make_body(30); + + auto req = make_request(40); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc(32, 8, "eof!"); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec, n] = co_await sr.write_eof( + capy::make_buffer(body)); + BOOST_TEST(ec == error::body_size_mismatch); + BOOST_TEST(!sr.is_done()); + }()); + + BOOST_TEST(server.data().empty()); + } + } + + void + testEncoderFailure() + { + // the encoder fails before anything is flushed + { + std::string const body = make_body(20); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + enc.fail = make_error_code(std::errc::io_error); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec, n] = co_await sr.write_eof( + capy::make_buffer(body)); + BOOST_TEST(ec == std::errc::io_error); + BOOST_TEST_EQ(n, 0u); + BOOST_TEST(!sr.is_done()); + }()); + + BOOST_TEST_EQ(enc.calls, 1u); + BOOST_TEST(server.data().empty()); + } + + // the encoder fails mid-stream, after encoded output + // has already been buffered + { + std::string const body = make_body(64); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + auto [ec1, n1] = co_await sr.write( + capy::make_buffer(body)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n1, body.size()); + BOOST_TEST(server.data().empty()); + + enc.fail = make_error_code(std::errc::io_error); + auto [ec2, n2] = co_await sr.write( + capy::const_buffer("more", 4)); + BOOST_TEST(ec2 == std::errc::io_error); + BOOST_TEST_EQ(n2, 0u); + BOOST_TEST(!sr.is_done()); + }()); + + // the bytes encoded before the failure were flushed + // while making room for more output; 64 == 0x40 + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + + "40\r\n" + encoded(body)); + } + } + + void + testEncoderMultipleBuffers() + { + // the encoder walks every caller buffer in order, + // including empty ones + std::string const b1 = make_body(6); + std::string const b2 = make_body(6); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + std::array bufs{ + capy::make_buffer(b1), + capy::const_buffer{}, + capy::make_buffer(b2) }; + + auto [ec, n] = co_await sr.write_eof(bufs); + BOOST_TEST(!ec); + BOOST_TEST_EQ(n, b1.size() + b2.size()); + BOOST_TEST(sr.is_done()); + }()); + + BOOST_TEST(enc.finished); + BOOST_TEST_EQ(req.payload_size(), 12u); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + + encoded(b1) + encoded(b2)); + } + + void + testEncoderWriteEofManyBuffers() + { + // More caller buffers than fit in one gather window: + // write_eof() feeds the encoder window by window, with + // eof deferred until the last window. + std::string const body = make_body(24); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + std::array< + capy::const_buffer, 24> bufs; + for(std::size_t i = 0; i != bufs.size(); ++i) + bufs[i] = { body.data() + i, 1 }; + + auto [ec, n] = co_await sr.write_eof(bufs); + BOOST_TEST(!ec); + BOOST_TEST_EQ(n, body.size()); + BOOST_TEST(sr.is_done()); + }()); + + BOOST_TEST(enc.finished); + BOOST_TEST(!req.chunked()); + BOOST_TEST_EQ(req.payload_size(), 24u); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + encoded(body)); + } + + void + testEncoderWriteSome() + { + // A write_some() that would cross the encoder threshold + // is not coalesced: the staged bytes are encoded first, + // then the caller's bytes, and the returned count covers + // only the caller's bytes. + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + // stays below the threshold: staged, encoder not + // started + auto [ec1, n1] = co_await sr.write_some( + capy::const_buffer("abcd", 4)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n1, 4u); + BOOST_TEST_EQ(enc.calls, 0u); + + // reaches the threshold: everything goes through + // the encoder, and the encoded bytes stay buffered + auto [ec2, n2] = co_await sr.write_some( + capy::const_buffer("efgh", 4)); + BOOST_TEST(!ec2); + BOOST_TEST_EQ(n2, 4u); + BOOST_TEST_EQ(enc.calls, 2u); + BOOST_TEST(server.data().empty()); + + auto [ec3] = co_await sr.write_eof(); + BOOST_TEST(!ec3); + }()); + + BOOST_TEST(enc.finished); + BOOST_TEST_EQ(req.payload_size(), 8u); + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + encoded("abcdefgh")); + } + + void + testEncoderMove() + { + // moving mid-stream transfers staged input and encoder + // state + { + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr1(&ws, &req, &enc, cfg); + + auto [ec1, n1] = co_await sr1.write( + capy::const_buffer("abc", 3)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n1, 3u); + + serializer sr2(std::move(sr1)); + auto [ec2, n2] = co_await sr2.write_eof( + capy::const_buffer("defgh", 5)); + BOOST_TEST(!ec2); + BOOST_TEST_EQ(n2, 5u); + BOOST_TEST(sr2.is_done()); + + // move-assignment into the moved-from shell + sr1 = std::move(sr2); + BOOST_TEST(sr1.is_done()); + }()); + + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + + encoded("abcdefgh")); + } + + // moving a serializer whose encoder was dropped must + // transfer ownership of the shifted staging buffer; + // deallocating at the wrong offset corrupts the heap + { + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr1(&ws, &req, &enc, cfg); + + auto [ec1, n1] = co_await sr1.write( + capy::const_buffer("hello", 5)); + BOOST_TEST(!ec1); + + // dropping the encoder shifts the internal + // buffer + auto [ec2] = co_await sr1.write_eof(); + BOOST_TEST(!ec2); + + serializer sr2(std::move(sr1)); + BOOST_TEST(sr2.is_done()); + + sr1 = std::move(sr2); + BOOST_TEST(sr1.is_done()); + + // self-move is a no-op + auto* self = &sr1; + sr1 = std::move(*self); + BOOST_TEST(sr1.is_done()); + }()); + + BOOST_TEST_EQ( + server.data(), + std::string(req.buffer()) + "hello"); + } + } + + void + testEncoderErrorInjection() + { + capy::test::fuse f; + auto r = f.armed([&](capy::test::fuse&) -> capy::task<> + { + std::string const body = make_body(100); + + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + test_encoder enc; + + auto [client, server] = + capy::test::make_stream_pair(f); + + capy::any_write_stream ws(&client); + serializer sr(&ws, &req, &enc, cfg); + + if(auto [ec, n] = co_await sr.write( + capy::make_buffer(body)); ec) + { + BOOST_TEST(!sr.is_done()); + co_return; + } + + if(auto [ec] = co_await sr.write_eof(); ec) + { + BOOST_TEST(!sr.is_done()); + co_return; + } + + BOOST_TEST(sr.is_done()); + BOOST_TEST(enc.finished); + + auto const enc_body = encoded(body); + std::string expected(req.buffer()); + expected += "40\r\n" + enc_body.substr(0, 64); + expected += "\r\n24\r\n" + enc_body.substr(64, 36); + expected += "\r\n0\r\n\r\n"; + BOOST_TEST_EQ(server.data(), expected); + }); + BOOST_TEST(r.success); + } + void run() { @@ -866,6 +1747,23 @@ class serializer_test testShortWriteEofDoesNotTruncate(); testChunkedPreservesWriteOrder(); testErrorInjection(); + testEncoderSmallBodyDropsEncoder(); + testEncoderEmptyBody(); + testEncoderPrepareCommitEof(); + testEncoderThreshold(); + testEncoderStagedAndTail(); + testEncoderLargeBodyChunked(); + testEncoderCommit(); + testEncoderCommitTriggersFlush(); + testEncoderTrailerSpansFlush(); + testEncoderContentLength(); + testEncoderContentLengthMismatch(); + testEncoderFailure(); + testEncoderMultipleBuffers(); + testEncoderWriteEofManyBuffers(); + testEncoderWriteSome(); + testEncoderMove(); + testEncoderErrorInjection(); } }; From ea57c0290e01342e33f67290d47aac80fdedf265 Mon Sep 17 00:00:00 2001 From: Mohammad Nejati Date: Mon, 6 Jul 2026 16:54:10 +0330 Subject: [PATCH 2/3] serializer is reusable across messages --- src/client.cpp | 4 +- src/detail/serializer.cpp | 118 ++++++++------ src/detail/serializer.hpp | 78 +++++++--- test/unit/detail/serializer.cpp | 263 +++++++++++++++++++++++--------- 4 files changed, 327 insertions(+), 136 deletions(-) diff --git a/src/client.cpp b/src/client.cpp index c33a077..f0d6e52 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -234,6 +234,8 @@ client::execute_impl( http::response_parser parser( http::make_parser_config(parser_cfg)); + detail::serializer sr({}); + auto url = request.url; auto trusted = true; auto followlocation = request.options.followlocation.value_or(config_.followlocation); @@ -265,7 +267,7 @@ client::execute_impl( // TODO: expect100timeout capy::any_write_stream ws(&conn); - detail::serializer sr(&ws, &headers, {}, {}); + sr.reset(&ws, &headers); if(request.body.has_value()) { capy::any_buffer_sink sink(&sr); diff --git a/src/detail/serializer.cpp b/src/detail/serializer.cpp index 00a0405..1dfc4d6 100644 --- a/src/detail/serializer.cpp +++ b/src/detail/serializer.cpp @@ -13,6 +13,9 @@ #include #include +#include +#include + namespace boost { namespace burl @@ -21,18 +24,16 @@ namespace detail { serializer:: -serializer( - capy::any_write_stream* stream, - http::message_base* msg, - encoder* enc, - config cfg) +serializer(config const& cfg, capy::any_write_stream* stream) : stream_(stream) - , msg_(msg) - , enc_(enc) - , cfg_(cfg) - , buf_(new unsigned char[ - cfg_.buffer_size + - (enc ? enc->buffer_size : 0) + margin] + margin) + , min_prepare_(cfg.min_prepare) + , direct_thr_(cfg.direct_thr) + , enc_thr_(cfg.enc_thr) + , in_(new unsigned char[ + cfg.out_buffer + cfg.enc_buffer + margin] + margin) + , in_cap_(cfg.out_buffer) + , out_(in_) + , out_cap_(cfg.enc_buffer) { } @@ -41,17 +42,22 @@ serializer(serializer&& other) noexcept : stream_(other.stream_) , msg_(other.msg_) , enc_(other.enc_) - , cfg_(other.cfg_) - , buf_(other.buf_) + , min_prepare_(other.min_prepare_) + , direct_thr_(other.direct_thr_) + , enc_thr_(other.enc_thr_) + , out_(other.out_) + , out_cap_(other.out_cap_) , out_len_(other.out_len_) + , in_(other.in_) + , in_cap_(other.in_cap_) , in_len_(other.in_len_) , total_body_(other.total_body_) + , head_(other.head_) , enc_started_(other.enc_started_) - , shifted_(other.shifted_) , hdr_sent_(other.hdr_sent_) , done_(other.done_) { - other.buf_ = nullptr; + other.out_ = nullptr; } serializer& @@ -67,16 +73,37 @@ operator=(serializer&& other) noexcept serializer:: ~serializer() { - if(buf_) - delete[](buf_ - - (shifted_ ? cfg_.buffer_size : 0) - margin); + if(out_) + delete[]((out_ > in_ ? in_ : out_) - margin); } -bool +void serializer:: -is_done() const noexcept +reset( + capy::any_write_stream* stream, + http::message_base* msg, + encoder_base* enc, + bool head) noexcept { - return done_; + if(out_ > in_) + out_ = in_; + + if(!enc_ && enc) + { + std::swap(out_cap_, in_cap_); + in_ += out_cap_; + } + + stream_ = stream; + msg_ = msg; + enc_ = enc; + in_len_ = 0; + out_len_ = 0; + total_body_ = 0; + head_ = head; + enc_started_ = false; + hdr_sent_ = false; + done_ = false; } capy::io_task<> @@ -102,7 +129,7 @@ commit(std::size_t n) { BOOST_ASSERT(n <= capacity()); do_commit(n); - if(capacity() >= cfg_.min_prepare) + if(capacity() >= min_prepare_) co_return {}; auto [ec, _] = co_await process({}, false); co_return { ec }; @@ -123,32 +150,21 @@ std::size_t serializer:: capacity() const noexcept { - if(enc_) - return enc_->buffer_size - in_len_; - - return cfg_.buffer_size - out_len_; + return in_cap_ - in_len_; } capy::mutable_buffer serializer:: do_prepare() noexcept { - if(enc_) - return { buf_ + cfg_.buffer_size + in_len_, - enc_->buffer_size - in_len_ }; - - return { buf_ + out_len_, - cfg_.buffer_size - out_len_ }; + return { in_ + in_len_, in_cap_ - in_len_ }; } void serializer:: do_commit(std::size_t n) noexcept { - if(enc_) - in_len_ += n; - else - out_len_ += n; + in_len_ += n; } bool @@ -158,8 +174,8 @@ can_coalesce(std::size_t avail) const noexcept if(avail > capacity()) return false; if(enc_) - return !enc_started_ && avail + in_len_ < enc_->threshold; - return avail < cfg_.direct_thr; + return !enc_started_ && avail + in_len_ < enc_thr_; + return avail < direct_thr_; } void @@ -171,14 +187,13 @@ finalize(std::size_t remaining) noexcept if(enc_started_) return; - if(remaining + in_len_ >= enc_->threshold) + if(remaining + in_len_ >= enc_thr_) return; msg_->erase(http::field::content_encoding); enc_ = nullptr; - out_len_ = in_len_; - buf_ += cfg_.buffer_size; - shifted_ = true; + std::swap(out_, in_); + std::swap(out_cap_, in_cap_); } decide_framing(remaining); } @@ -192,7 +207,7 @@ decide_framing(std::size_t remaining) noexcept BOOST_ASSERT(total_body_ == 0); msg_->erase(http::field::transfer_encoding); - msg_->set_content_length(out_len_ + remaining); + msg_->set_content_length((enc_ ? out_len_ : in_len_) + remaining); } capy::io_task @@ -214,7 +229,7 @@ encode( { std::size_t n = 0; std::size_t consumed = 0; - capy::const_buffer in = { buf_ + cfg_.buffer_size, in_len_ }; + capy::const_buffer in = { in_, in_len_ }; for(;;) { if(in.size() == 0) @@ -225,14 +240,14 @@ encode( co_return { {}, consumed }; } - if(cfg_.buffer_size - out_len_ == 0) + if(out_len_ == out_cap_) { if(auto [ec, _] = co_await flush({}, false); ec) co_return { ec, consumed }; } capy::mutable_buffer out = - { buf_ + out_len_, cfg_.buffer_size - out_len_ }; + { out_ + out_len_, out_cap_ - out_len_ }; auto r = enc_->process( out, in, eof && n == tail.size()); @@ -269,6 +284,9 @@ flush( std::span tail, bool eof) { + if(!enc_) + out_len_ = in_len_; + auto const tail_size = capy::buffer_size(tail); auto const chunked = msg_->chunked(); BOOST_ASSERT(eof || out_len_ + tail_size != 0); @@ -301,7 +319,7 @@ flush( co_return { error::body_size_mismatch, 0 }; if(out_len_ != 0) - append({ buf_, out_len_ }); + append({ out_, out_len_ }); } auto const owned = sum; @@ -320,6 +338,8 @@ flush( auto const consumed = (std::min)(written - owned, tail_size); total_body_ += out_len_ + consumed; out_len_ = 0; + if(!enc_) + in_len_ = 0; hdr_sent_ = true; done_ = eof; co_return { {}, consumed }; @@ -352,7 +372,7 @@ chunk_frame(std::size_t tail_size) noexcept { static constexpr char hex[] = "0123456789abcdef"; auto s = out_len_ + tail_size; - auto p = buf_; + auto p = out_; *--p = '\n'; *--p = '\r'; @@ -371,7 +391,7 @@ chunk_frame(std::size_t tail_size) noexcept } return { p, static_cast( - buf_ - p) + out_len_ }; + out_ - p) + out_len_ }; } } // namespace detail diff --git a/src/detail/serializer.hpp b/src/detail/serializer.hpp index dfe0827..3b13006 100644 --- a/src/detail/serializer.hpp +++ b/src/detail/serializer.hpp @@ -29,11 +29,8 @@ namespace detail class serializer { public: - struct encoder + struct encoder_base { - std::size_t buffer_size = 8 * 1024; - std::size_t threshold = 4 * 1024; - struct result { std::size_t consumed; @@ -41,6 +38,8 @@ class serializer std::error_code ec; }; + virtual ~encoder_base() = default; + virtual result process( capy::mutable_buffer out, @@ -50,16 +49,16 @@ class serializer struct config { - std::size_t buffer_size = 64 * 1024; - std::size_t min_prepare = 4 * 1024; - std::size_t direct_thr = 2 * 1024; + std::size_t out_buffer = 64 * 1024; + std::size_t min_prepare = 4 * 1024; + std::size_t direct_thr = 2 * 1024; + std::size_t enc_buffer = 8 * 1024; + std::size_t enc_thr = 4 * 1024; }; serializer( - capy::any_write_stream* stream, - http::message_base* msg, - encoder* enc, - config cfg); + config const& cfg, + capy::any_write_stream* stream = nullptr); serializer(serializer&& other) noexcept; @@ -74,7 +73,45 @@ class serializer ~serializer(); bool - is_done() const noexcept; + is_done() const noexcept + { + return done_; + } + + capy::any_write_stream* + stream() const noexcept + { + return stream_; + } + + http::message_base* + message() const noexcept + { + return msg_; + } + + encoder_base* + encoder() const noexcept + { + return enc_; + } + + void + reset( + capy::any_write_stream* stream, + http::message_base* msg, + encoder_base* enc = nullptr, + bool head = false) noexcept; + + void + reset( + http::message_base* msg, + encoder_base* enc = nullptr, + bool head = false) noexcept + { + BOOST_ASSERT(stream_); + reset(stream_, msg, enc, head); + } template capy::io_task @@ -146,15 +183,20 @@ class serializer static constexpr std::size_t margin = 24; capy::any_write_stream* stream_; - http::message_base* msg_; - encoder* enc_; - config cfg_; - unsigned char* buf_; - std::size_t out_len_ = 0; + http::message_base* msg_ = nullptr; + encoder_base* enc_ = nullptr; + std::size_t min_prepare_; + std::size_t direct_thr_; + std::size_t enc_thr_; + unsigned char* in_; + std::size_t in_cap_; std::size_t in_len_ = 0; + unsigned char* out_; + std::size_t out_cap_; + std::size_t out_len_ = 0; std::uint64_t total_body_ = 0; + bool head_ = false; bool enc_started_ = false; - bool shifted_ = false; bool hdr_sent_ = false; bool done_ = false; }; diff --git a/test/unit/detail/serializer.cpp b/test/unit/detail/serializer.cpp index 6731c24..2f7cf4a 100644 --- a/test/unit/detail/serializer.cpp +++ b/test/unit/detail/serializer.cpp @@ -32,14 +32,17 @@ namespace detail class serializer_test { - // A scaled-down config so that framing thresholds (staged - // vs. direct writes, commit-triggered flushes) are crossed - // with tiny bodies, and so tests keep exercising the same - // paths if the default config values change. + // A scaled-down config so that framing and encoder + // thresholds (staged vs. direct writes, commit-triggered + // flushes, encoder drop) are crossed with tiny bodies, and + // so tests keep exercising the same paths if the default + // config values change. static constexpr serializer::config cfg{ - .buffer_size = 64, + .out_buffer = 64, .min_prepare = 32, - .direct_thr = 16 }; + .direct_thr = 16, + .enc_buffer = 32, + .enc_thr = 8 }; static http::request make_request() @@ -61,21 +64,16 @@ class serializer_test // encoded output is distinguishable from identity output, // and appends an optional trailer once the input ends. It // consumes and produces as much as the given buffers allow. - struct test_encoder : serializer::encoder + struct test_encoder : serializer::encoder_base { std::string trailer; std::error_code fail; std::size_t calls = 0; bool finished = false; - explicit test_encoder( - std::size_t bufsz = 32, - std::size_t thr = 8, - std::string trl = {}) + explicit test_encoder(std::string trl = {}) : trailer(std::move(trl)) { - buffer_size = bufsz; - threshold = thr; } result @@ -147,7 +145,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); BOOST_TEST(!sr.is_done()); auto [ec1, n] = co_await sr.write( @@ -178,7 +177,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); auto [ec1, n] = co_await sr.write( capy::make_buffer(body)); @@ -206,7 +206,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); std::array bufs{ capy::make_buffer(b1), @@ -234,7 +235,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); auto [ec1, n] = co_await sr.write( capy::const_buffer("hello", 5)); @@ -259,7 +261,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); auto [ec, n] = co_await sr.write_eof( capy::const_buffer("hello", 5)); @@ -280,7 +283,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); // The body stays below direct_thr, so it is fully // buffered before the header goes out. @@ -315,7 +319,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); auto [ec, n] = co_await sr.write_eof( capy::const_buffer("hello", 5)); @@ -341,7 +346,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); auto [ec1, n1] = co_await sr.write( capy::make_buffer(body)); @@ -386,7 +392,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); auto [ec1, n1] = co_await sr.write( capy::make_buffer(body)); @@ -426,7 +433,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); auto [ec1, n1] = co_await sr.write( capy::make_buffer(body)); @@ -456,7 +464,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, {}); + serializer sr({}, &ws); + sr.reset(&req); BOOST_TEST(!sr.is_done()); auto [ec] = co_await sr.write_eof(); @@ -480,7 +489,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -511,7 +521,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -545,7 +556,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -578,7 +590,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -608,7 +621,8 @@ class serializer_test auto [client, server] = capy::test::make_stream_pair(); capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); BOOST_TEST(sr.prepare({}).empty()); @@ -617,7 +631,7 @@ class serializer_test capy::mutable_buffer tmp[2]; auto dest = sr.prepare(tmp); BOOST_TEST_EQ(dest.size(), 1u); - BOOST_TEST_EQ(capy::buffer_size(dest), cfg.buffer_size); + BOOST_TEST_EQ(capy::buffer_size(dest), cfg.out_buffer); } void @@ -630,13 +644,14 @@ class serializer_test capy::test::fuse f; auto r = f.armed([&](capy::test::fuse&) -> capy::task<> { - std::string const body(cfg.buffer_size, 'z'); + std::string const body(cfg.out_buffer, 'z'); auto req = make_request(body.size()); capy::test::write_stream ws_impl(f); capy::any_write_stream ws(&ws_impl); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -654,7 +669,7 @@ class serializer_test auto dest = sr.prepare(tmp); BOOST_TEST_EQ(dest.size(), 1u); BOOST_TEST_EQ( - capy::buffer_size(dest), cfg.buffer_size); + capy::buffer_size(dest), cfg.out_buffer); }); BOOST_TEST(r.success); } @@ -664,7 +679,7 @@ class serializer_test { // The staging buffer holds raw body bytes only, so the // full capacity is stageable as one chunk; 64 == 0x40. - std::string const body(cfg.buffer_size, 'z'); + std::string const body(cfg.out_buffer, 'z'); auto req = make_request(); capy::test::write_stream ws_impl({}, 1); @@ -672,7 +687,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -724,7 +740,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); // The first drain writes the header in full and // exactly one body byte. @@ -785,7 +802,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -819,7 +837,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); auto [ec1, n] = co_await sr.write( capy::const_buffer("hello", 5)); @@ -854,7 +873,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&ws_impl); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); auto [ec1, n1] = co_await sr.write( capy::const_buffer("abc", 3)); @@ -891,7 +911,8 @@ class serializer_test capy::test::make_stream_pair(f); capy::any_write_stream ws(&client); - serializer sr(&ws, &req, {}, cfg); + serializer sr(cfg, &ws); + sr.reset(&req); if(auto [ec, n] = co_await sr.write( capy::make_buffer(body)); ec) @@ -939,7 +960,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); // below the threshold: coalesced into the encoder's // staging buffer without invoking the encoder @@ -977,7 +999,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec] = co_await sr.write_eof(); BOOST_TEST(!ec); @@ -1010,12 +1033,13 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); capy::mutable_buffer tmp[2]; auto dest = sr.prepare(tmp); BOOST_TEST_EQ( - capy::buffer_size(dest), enc.buffer_size); + capy::buffer_size(dest), cfg.enc_buffer); auto n = capy::buffer_copy( dest, capy::const_buffer("hello", 5)); @@ -1051,7 +1075,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec, n] = co_await sr.write_eof( capy::make_buffer(body)); @@ -1086,7 +1111,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec, n] = co_await sr.write_eof( capy::make_buffer(body)); @@ -1120,7 +1146,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec1, n1] = co_await sr.write( capy::const_buffer("abc", 3)); @@ -1160,7 +1187,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec1, n] = co_await sr.write( capy::make_buffer(body)); @@ -1200,12 +1228,17 @@ class serializer_test auto req = make_request(); req.set(http::field::content_encoding, "test"); auto [client, server] = capy::test::make_stream_pair(); - test_encoder enc(64, 8); + test_encoder enc; + + // a staging buffer large enough to hold all commits + auto ecfg = cfg; + ecfg.enc_buffer = 64; capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(ecfg, &ws); + sr.reset(&req, &enc); capy::mutable_buffer tmp[2]; @@ -1264,12 +1297,18 @@ class serializer_test auto req = make_request(); req.set(http::field::content_encoding, "test"); auto [client, server] = capy::test::make_stream_pair(); - test_encoder enc(96, 8); + test_encoder enc; + + // a staging buffer larger than the output buffer, so a + // single commit overflows it + auto ecfg = cfg; + ecfg.enc_buffer = 96; capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(ecfg, &ws); + sr.reset(&req, &enc); capy::mutable_buffer tmp[2]; auto n = capy::buffer_copy( @@ -1306,12 +1345,13 @@ class serializer_test auto req = make_request(); req.set(http::field::content_encoding, "test"); auto [client, server] = capy::test::make_stream_pair(); - test_encoder enc(32, 8, "0123456789"); + test_encoder enc("0123456789"); capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec, n] = co_await sr.write_eof( capy::make_buffer(body)); @@ -1341,12 +1381,13 @@ class serializer_test auto req = make_request(34); req.set(http::field::content_encoding, "test"); auto [client, server] = capy::test::make_stream_pair(); - test_encoder enc(32, 8, "eof!"); + test_encoder enc("eof!"); capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec, n] = co_await sr.write_eof( capy::make_buffer(body)); @@ -1372,12 +1413,13 @@ class serializer_test auto req = make_request(30); req.set(http::field::content_encoding, "test"); auto [client, server] = capy::test::make_stream_pair(); - test_encoder enc(32, 8, "eof!"); + test_encoder enc("eof!"); capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec, n] = co_await sr.write_eof( capy::make_buffer(body)); @@ -1395,12 +1437,13 @@ class serializer_test auto req = make_request(40); req.set(http::field::content_encoding, "test"); auto [client, server] = capy::test::make_stream_pair(); - test_encoder enc(32, 8, "eof!"); + test_encoder enc("eof!"); capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec, n] = co_await sr.write_eof( capy::make_buffer(body)); @@ -1428,7 +1471,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec, n] = co_await sr.write_eof( capy::make_buffer(body)); @@ -1454,7 +1498,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); auto [ec1, n1] = co_await sr.write( capy::make_buffer(body)); @@ -1495,7 +1540,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); std::array bufs{ capy::make_buffer(b1), @@ -1532,7 +1578,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); std::array< capy::const_buffer, 24> bufs; @@ -1568,7 +1615,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); // stays below the threshold: staged, encoder not // started @@ -1612,7 +1660,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr1(&ws, &req, &enc, cfg); + serializer sr1(cfg); + sr1.reset(&ws, &req, &enc); auto [ec1, n1] = co_await sr1.write( capy::const_buffer("abc", 3)); @@ -1649,7 +1698,8 @@ class serializer_test capy::test::run_blocking()([&]() -> capy::task<> { capy::any_write_stream ws(&client); - serializer sr1(&ws, &req, &enc, cfg); + serializer sr1(cfg); + sr1.reset(&ws, &req, &enc); auto [ec1, n1] = co_await sr1.write( capy::const_buffer("hello", 5)); @@ -1678,6 +1728,81 @@ class serializer_test } } + void + testReset() + { + // reset() rebinds the serializer to a new stream, + // message, and encoder for reuse. The first request + // drops its encoder, which swaps the internal buffers; + // reset() must restore the original layout so the + // second request encodes with full capacity. + auto req1 = make_request(); + req1.set(http::field::content_encoding, "test"); + auto req2 = make_request(); + req2.set(http::field::content_encoding, "test"); + std::string const body = make_body(200); + + auto [client1, server1] = capy::test::make_stream_pair(); + auto [client2, server2] = capy::test::make_stream_pair(); + test_encoder enc1; + test_encoder enc2; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws1(&client1); + capy::any_write_stream ws2(&client2); + + serializer sr(cfg, &ws1); + sr.reset(&req1, &enc1); + BOOST_TEST(sr.stream() == &ws1); + BOOST_TEST(sr.message() == &req1); + BOOST_TEST(sr.encoder() == &enc1); + + // the small body drops the encoder + auto [ec1, n1] = co_await sr.write_eof( + capy::const_buffer("hello", 5)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n1, 5u); + BOOST_TEST(sr.is_done()); + + sr.reset(&ws2, &req2, &enc2); + BOOST_TEST(sr.stream() == &ws2); + BOOST_TEST(sr.message() == &req2); + BOOST_TEST(sr.encoder() == &enc2); + BOOST_TEST(!sr.is_done()); + + auto [ec2, n2] = co_await sr.write( + capy::make_buffer(body)); + BOOST_TEST(!ec2); + BOOST_TEST_EQ(n2, body.size()); + + auto [ec3] = co_await sr.write_eof(); + BOOST_TEST(!ec3); + BOOST_TEST(sr.is_done()); + }()); + + BOOST_TEST_EQ(enc1.calls, 0u); + BOOST_TEST( + server1.data().find("Content-Encoding") == + std::string_view::npos); + BOOST_TEST_EQ( + server1.data(), + std::string(req1.buffer()) + "hello"); + + // the second request streams encoded chunks sized by + // the restored output buffer; 64 == 0x40 + auto const enc_body = encoded(body); + std::string expected(req2.buffer()); + expected += "40\r\n" + enc_body.substr(0, 64); + expected += "\r\n40\r\n" + enc_body.substr(64, 64); + expected += "\r\n40\r\n" + enc_body.substr(128, 64); + expected += "\r\n8\r\n" + enc_body.substr(192, 8); + expected += "\r\n0\r\n\r\n"; + BOOST_TEST(enc2.finished); + BOOST_TEST(req2.chunked()); + BOOST_TEST_EQ(server2.data(), expected); + } + void testEncoderErrorInjection() { @@ -1694,7 +1819,8 @@ class serializer_test capy::test::make_stream_pair(f); capy::any_write_stream ws(&client); - serializer sr(&ws, &req, &enc, cfg); + serializer sr(cfg, &ws); + sr.reset(&req, &enc); if(auto [ec, n] = co_await sr.write( capy::make_buffer(body)); ec) @@ -1763,6 +1889,7 @@ class serializer_test testEncoderWriteEofManyBuffers(); testEncoderWriteSome(); testEncoderMove(); + testReset(); testEncoderErrorInjection(); } }; From 1f1d8cdaf7bf041442e5cb2fe1b5725535aa343b Mon Sep 17 00:00:00 2001 From: Mohammad Nejati Date: Mon, 6 Jul 2026 23:16:09 +0330 Subject: [PATCH 3/3] serializer handles head responses --- src/detail/serializer.cpp | 125 ++++++----- src/detail/serializer.hpp | 29 +-- test/unit/detail/serializer.cpp | 363 +++++++++++++++++++++++++++++++- 3 files changed, 427 insertions(+), 90 deletions(-) diff --git a/src/detail/serializer.cpp b/src/detail/serializer.cpp index 1dfc4d6..5e93ef4 100644 --- a/src/detail/serializer.cpp +++ b/src/detail/serializer.cpp @@ -29,11 +29,11 @@ serializer(config const& cfg, capy::any_write_stream* stream) , min_prepare_(cfg.min_prepare) , direct_thr_(cfg.direct_thr) , enc_thr_(cfg.enc_thr) - , in_(new unsigned char[ + , out_(new unsigned char[ cfg.out_buffer + cfg.enc_buffer + margin] + margin) - , in_cap_(cfg.out_buffer) - , out_(in_) , out_cap_(cfg.enc_buffer) + , in_(out_) + , in_cap_(cfg.out_buffer) { } @@ -74,7 +74,7 @@ serializer:: ~serializer() { if(out_) - delete[]((out_ > in_ ? in_ : out_) - margin); + delete[](out_ - margin); } void @@ -82,16 +82,16 @@ serializer:: reset( capy::any_write_stream* stream, http::message_base* msg, - encoder_base* enc, + encoder* enc, bool head) noexcept { - if(out_ > in_) - out_ = in_; + if(head) + enc = nullptr; - if(!enc_ && enc) + if(!enc == (in_ != out_)) { - std::swap(out_cap_, in_cap_); - in_ += out_cap_; + std::swap(in_cap_, out_cap_); + in_ = enc ? out_ + out_cap_ : out_; } stream_ = stream; @@ -169,7 +169,7 @@ do_commit(std::size_t n) noexcept bool serializer:: -can_coalesce(std::size_t avail) const noexcept +should_coalesce(std::size_t avail) const noexcept { if(avail > capacity()) return false; @@ -182,6 +182,9 @@ void serializer:: finalize(std::size_t remaining) noexcept { + if(head_) + return; + if(enc_) { if(enc_started_) @@ -192,8 +195,6 @@ finalize(std::size_t remaining) noexcept msg_->erase(http::field::content_encoding); enc_ = nullptr; - std::swap(out_, in_); - std::swap(out_cap_, in_cap_); } decide_framing(remaining); } @@ -246,11 +247,11 @@ encode( co_return { ec, consumed }; } - capy::mutable_buffer out = - { out_ + out_len_, out_cap_ - out_len_ }; - auto r = enc_->process( - out, in, eof && n == tail.size()); + { out_ + out_len_, out_cap_ - out_len_ }, + in, + eof && n == tail.size()); + enc_started_ = true; out_len_ += r.produced; in += r.consumed; @@ -284,12 +285,11 @@ flush( std::span tail, bool eof) { - if(!enc_) - out_len_ = in_len_; - - auto const tail_size = capy::buffer_size(tail); - auto const chunked = msg_->chunked(); - BOOST_ASSERT(eof || out_len_ + tail_size != 0); + auto const buf = enc_ ? out_ : in_; + auto& len = enc_ ? out_len_ : in_len_; + auto const tail_len = capy::buffer_size(tail); + auto const chunked = msg_->chunked() && !head_; + BOOST_ASSERT(eof || len + tail_len != 0); BOOST_ASSERT(tail.size() <= capy::detail::max_iovec_); capy::const_buffer vec[capy::detail::max_iovec_ + 3]; @@ -306,20 +306,43 @@ flush( if(chunked) { - append(chunk_frame(tail_size)); + static constexpr char hex[] = "0123456789abcdef"; + auto s = len + tail_len; + auto p = buf; + + *--p = '\n'; + *--p = '\r'; + + do + { + *--p = hex[s & 0xF]; + s >>= 4; + } while(s != 0); + + // prev chunk's CRLF + if(total_body_) + { + *--p = '\n'; + *--p = '\r'; + } + + append({ p, static_cast(buf - p) + len }); } else { - auto const declared = - msg_->payload() == http::payload::size - ? msg_->payload_size() : 0; - auto const produced = total_body_ + out_len_ + tail_size; + auto const decl = [&]()-> std::uint64_t + { + if(head_ || msg_->payload() != http::payload::size) + return 0; + return msg_->payload_size(); + }(); + auto const prod = total_body_ + len + tail_len; - if(produced > declared || (eof && produced != declared)) + if(prod > decl || (eof && prod != decl)) co_return { error::body_size_mismatch, 0 }; - if(out_len_ != 0) - append({ out_, out_len_ }); + if(len != 0) + append({ buf, len }); } auto const owned = sum; @@ -328,18 +351,16 @@ flush( append(b); if(chunked && eof) - append({ "\r\n0\r\n\r\n", out_len_ || tail_size ? 7u : 2u }); + append({ "\r\n0\r\n\r\n", len || tail_len ? 7u : 2u }); - auto const need = chunked || eof ? sum : owned + (tail_size != 0); + auto const need = chunked || eof ? sum : owned + !!tail_len; auto [ec, written] = co_await write_at_least({ vec, n }, need); if(ec) co_return { ec, 0 }; - auto const consumed = (std::min)(written - owned, tail_size); - total_body_ += out_len_ + consumed; - out_len_ = 0; - if(!enc_) - in_len_ = 0; + auto const consumed = (std::min)(written - owned, tail_len); + total_body_ += len + consumed; + len = 0; hdr_sent_ = true; done_ = eof; co_return { {}, consumed }; @@ -366,34 +387,6 @@ write_at_least( co_return { {}, written }; } -capy::const_buffer -serializer:: -chunk_frame(std::size_t tail_size) noexcept -{ - static constexpr char hex[] = "0123456789abcdef"; - auto s = out_len_ + tail_size; - auto p = out_; - - *--p = '\n'; - *--p = '\r'; - - do - { - *--p = hex[s & 0xF]; - s >>= 4; - } while(s != 0); - - // prev chunk's CRLF - if(total_body_ != 0) - { - *--p = '\n'; - *--p = '\r'; - } - - return { p, static_cast( - out_ - p) + out_len_ }; -} - } // namespace detail } // namespace burl } // namespace boost diff --git a/src/detail/serializer.hpp b/src/detail/serializer.hpp index 3b13006..0d0b887 100644 --- a/src/detail/serializer.hpp +++ b/src/detail/serializer.hpp @@ -29,7 +29,7 @@ namespace detail class serializer { public: - struct encoder_base + struct encoder { struct result { @@ -38,7 +38,7 @@ class serializer std::error_code ec; }; - virtual ~encoder_base() = default; + virtual ~encoder() = default; virtual result process( @@ -90,23 +90,17 @@ class serializer return msg_; } - encoder_base* - encoder() const noexcept - { - return enc_; - } - void reset( capy::any_write_stream* stream, http::message_base* msg, - encoder_base* enc = nullptr, + encoder* enc = nullptr, bool head = false) noexcept; void reset( http::message_base* msg, - encoder_base* enc = nullptr, + encoder* enc = nullptr, bool head = false) noexcept { BOOST_ASSERT(stream_); @@ -148,7 +142,7 @@ class serializer do_commit(std::size_t n) noexcept; bool - can_coalesce(std::size_t avail) const noexcept; + should_coalesce(std::size_t avail) const noexcept; void finalize(std::size_t remaining) noexcept; @@ -177,23 +171,20 @@ class serializer std::span buffers, std::size_t bytes); - capy::const_buffer - chunk_frame(std::size_t tail_size) noexcept; - static constexpr std::size_t margin = 24; capy::any_write_stream* stream_; http::message_base* msg_ = nullptr; - encoder_base* enc_ = nullptr; + encoder* enc_ = nullptr; std::size_t min_prepare_; std::size_t direct_thr_; std::size_t enc_thr_; - unsigned char* in_; - std::size_t in_cap_; - std::size_t in_len_ = 0; unsigned char* out_; std::size_t out_cap_; std::size_t out_len_ = 0; + unsigned char* in_; + std::size_t in_cap_; + std::size_t in_len_ = 0; std::uint64_t total_body_ = 0; bool head_ = false; bool enc_started_ = false; @@ -207,7 +198,7 @@ serializer:: write_some(Buffers buffers) { auto const avail = capy::buffer_size(buffers); - if(can_coalesce(avail)) + if(should_coalesce(avail)) { do_commit(capy::buffer_copy(do_prepare(), buffers)); co_return { {}, avail }; diff --git a/test/unit/detail/serializer.cpp b/test/unit/detail/serializer.cpp index 2f7cf4a..6fbfe34 100644 --- a/test/unit/detail/serializer.cpp +++ b/test/unit/detail/serializer.cpp @@ -64,7 +64,7 @@ class serializer_test // encoded output is distinguishable from identity output, // and appends an optional trailer once the input ends. It // consumes and produces as much as the given buffers allow. - struct test_encoder : serializer::encoder_base + struct test_encoder : serializer::encoder { std::string trailer; std::error_code fail; @@ -236,7 +236,7 @@ class serializer_test { capy::any_write_stream ws(&client); serializer sr(cfg, &ws); - sr.reset(&req); + sr.reset(&req); auto [ec1, n] = co_await sr.write( capy::const_buffer("hello", 5)); @@ -262,7 +262,7 @@ class serializer_test { capy::any_write_stream ws(&client); serializer sr(cfg, &ws); - sr.reset(&req); + sr.reset(&req); auto [ec, n] = co_await sr.write_eof( capy::const_buffer("hello", 5)); @@ -1756,7 +1756,6 @@ class serializer_test sr.reset(&req1, &enc1); BOOST_TEST(sr.stream() == &ws1); BOOST_TEST(sr.message() == &req1); - BOOST_TEST(sr.encoder() == &enc1); // the small body drops the encoder auto [ec1, n1] = co_await sr.write_eof( @@ -1768,7 +1767,6 @@ class serializer_test sr.reset(&ws2, &req2, &enc2); BOOST_TEST(sr.stream() == &ws2); BOOST_TEST(sr.message() == &req2); - BOOST_TEST(sr.encoder() == &enc2); BOOST_TEST(!sr.is_done()); auto [ec2, n2] = co_await sr.write( @@ -1803,6 +1801,356 @@ class serializer_test BOOST_TEST_EQ(server2.data(), expected); } + void + testResetAfterEncoderKept() + { + // reset() must also restore the layout when the + // previous message kept its encoder: a following plain + // message must send its own staged body, not the stale + // contents of the encoder's output buffer, and + // re-adding an encoder afterwards must not shift the + // staging buffer out of the allocation. + auto req1 = make_request(); + req1.set(http::field::content_encoding, "test"); + auto req2 = make_request(5); + auto req3 = make_request(); + req3.set(http::field::content_encoding, "test"); + std::string const body1 = make_body(20); + std::string const body3 = make_body(200); + + auto [client1, server1] = capy::test::make_stream_pair(); + auto [client2, server2] = capy::test::make_stream_pair(); + auto [client3, server3] = capy::test::make_stream_pair(); + test_encoder enc1; + test_encoder enc3; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws1(&client1); + capy::any_write_stream ws2(&client2); + capy::any_write_stream ws3(&client3); + + serializer sr(cfg, &ws1); + sr.reset(&req1, &enc1); + + // the body crosses the threshold: the encoder is + // kept + auto [ec1, n1] = co_await sr.write_eof( + capy::make_buffer(body1)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n1, body1.size()); + BOOST_TEST(sr.is_done()); + + sr.reset(&ws2, &req2); + auto [ec2, n2] = co_await sr.write( + capy::const_buffer("hello", 5)); + BOOST_TEST(!ec2); + auto [ec3] = co_await sr.write_eof(); + BOOST_TEST(!ec3); + BOOST_TEST(sr.is_done()); + + sr.reset(&ws3, &req3, &enc3); + auto [ec4, n4] = co_await sr.write( + capy::make_buffer(body3)); + BOOST_TEST(!ec4); + BOOST_TEST_EQ(n4, body3.size()); + auto [ec5] = co_await sr.write_eof(); + BOOST_TEST(!ec5); + BOOST_TEST(sr.is_done()); + }()); + + BOOST_TEST_EQ( + server1.data(), + std::string(req1.buffer()) + encoded(body1)); + + BOOST_TEST_EQ( + server2.data(), + std::string(req2.buffer()) + "hello"); + + // the third request streams encoded chunks sized by the + // restored output buffer; 64 == 0x40 + auto const enc_body = encoded(body3); + std::string expected(req3.buffer()); + expected += "40\r\n" + enc_body.substr(0, 64); + expected += "\r\n40\r\n" + enc_body.substr(64, 64); + expected += "\r\n40\r\n" + enc_body.substr(128, 64); + expected += "\r\n8\r\n" + enc_body.substr(192, 8); + expected += "\r\n0\r\n\r\n"; + BOOST_TEST(req3.chunked()); + BOOST_TEST_EQ(server3.data(), expected); + } + + void + testHeadContentLength() + { + // A head message sends only the header; the declared + // Content-Length describes the body a non-head message + // would have carried, and stays untouched. + auto req = make_request(5); + auto req2 = make_request(5); + + auto [client1, server1] = capy::test::make_stream_pair(); + auto [client2, server2] = capy::test::make_stream_pair(); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws1(&client1); + capy::any_write_stream ws2(&client2); + + serializer sr(cfg, &ws1); + sr.reset(&req, nullptr, true); + BOOST_TEST(!sr.is_done()); + + auto [ec1] = co_await sr.write_eof(); + BOOST_TEST(!ec1); + BOOST_TEST(sr.is_done()); + + // the head flag does not stick across reset() + sr.reset(&ws2, &req2); + auto [ec2, n] = co_await sr.write( + capy::const_buffer("hello", 5)); + BOOST_TEST(!ec2); + auto [ec3] = co_await sr.write_eof(); + BOOST_TEST(!ec3); + }()); + + BOOST_TEST_EQ(req.payload_size(), 5u); + BOOST_TEST_EQ(server1.data(), std::string(req.buffer())); + BOOST_TEST_EQ( + server2.data(), + std::string(req2.buffer()) + "hello"); + } + + void + testHeadChunked() + { + // A chunked head message keeps Transfer-Encoding in the + // header to describe the framing a non-head message + // would have used; no framing bytes reach the wire. + auto req = make_request(); + auto [client, server] = capy::test::make_stream_pair(); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(cfg, &ws); + sr.reset(&req, nullptr, true); + + auto [ec] = co_await sr.write_eof(); + BOOST_TEST(!ec); + BOOST_TEST(sr.is_done()); + }()); + + BOOST_TEST(req.chunked()); + BOOST_TEST( + server.data().find("Content-Length") == + std::string_view::npos); + BOOST_TEST_EQ(server.data(), std::string(req.buffer())); + } + + void + testHeadBodyMismatch() + { + // buffered body bytes are rejected at eof, even when + // they match the declared Content-Length + { + auto req = make_request(5); + auto [client, server] = capy::test::make_stream_pair(); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(cfg, &ws); + sr.reset(&req, nullptr, true); + + auto [ec1, n] = co_await sr.write( + capy::const_buffer("hello", 5)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n, 5u); + + auto [ec2] = co_await sr.write_eof(); + BOOST_TEST(ec2 == error::body_size_mismatch); + BOOST_TEST(!sr.is_done()); + }()); + + BOOST_TEST(server.data().empty()); + } + + // a direct write is rejected before the header goes out + { + std::string const body(cfg.direct_thr, 'x'); + + auto req = make_request(); + auto [client, server] = capy::test::make_stream_pair(); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(cfg, &ws); + sr.reset(&req, nullptr, true); + + auto [ec, n] = co_await sr.write( + capy::make_buffer(body)); + BOOST_TEST(ec == error::body_size_mismatch); + BOOST_TEST(!sr.is_done()); + }()); + + BOOST_TEST(server.data().empty()); + } + + // write_eof() with caller buffers + { + auto req = make_request(); + auto [client, server] = capy::test::make_stream_pair(); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(cfg, &ws); + sr.reset(&req, nullptr, true); + + auto [ec, n] = co_await sr.write_eof( + capy::const_buffer("hello", 5)); + BOOST_TEST(ec == error::body_size_mismatch); + BOOST_TEST(!sr.is_done()); + }()); + + BOOST_TEST(server.data().empty()); + } + + // prepare()/commit_eof() + { + auto req = make_request(5); + auto [client, server] = capy::test::make_stream_pair(); + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(cfg, &ws); + sr.reset(&req, nullptr, true); + + capy::mutable_buffer tmp[2]; + auto n = capy::buffer_copy( + sr.prepare(tmp), + capy::const_buffer("hello", 5)); + BOOST_TEST_EQ(n, 5u); + + auto [ec] = co_await sr.commit_eof(n); + BOOST_TEST(ec == error::body_size_mismatch); + BOOST_TEST(!sr.is_done()); + }()); + + BOOST_TEST(server.data().empty()); + } + } + + void + testHeadEncoder() + { + // A head message never invokes its encoder: the header + // goes out as-is, with Content-Encoding intact, and the + // serializer remains reusable for a plain message. + { + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto req2 = make_request(5); + + auto [client1, server1] = capy::test::make_stream_pair(); + auto [client2, server2] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws1(&client1); + capy::any_write_stream ws2(&client2); + + serializer sr(cfg, &ws1); + sr.reset(&req, &enc, true); + + auto [ec1] = co_await sr.write_eof(); + BOOST_TEST(!ec1); + BOOST_TEST(sr.is_done()); + + sr.reset(&ws2, &req2); + auto [ec2, n] = co_await sr.write( + capy::const_buffer("hello", 5)); + BOOST_TEST(!ec2); + auto [ec3] = co_await sr.write_eof(); + BOOST_TEST(!ec3); + }()); + + BOOST_TEST_EQ(enc.calls, 0u); + BOOST_TEST(req.chunked()); + BOOST_TEST( + server1.data().find("Content-Encoding: test") != + std::string_view::npos); + BOOST_TEST_EQ( + server1.data(), std::string(req.buffer())); + BOOST_TEST_EQ( + server2.data(), + std::string(req2.buffer()) + "hello"); + } + + // body bytes staged for the encoder are rejected, and + // the encoder is never invoked + { + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(cfg, &ws); + sr.reset(&req, &enc, true); + + // below enc_thr: staged without invoking the + // encoder + auto [ec1, n] = co_await sr.write( + capy::const_buffer("hello", 5)); + BOOST_TEST(!ec1); + BOOST_TEST_EQ(n, 5u); + + auto [ec2] = co_await sr.write_eof(); + BOOST_TEST(ec2 == error::body_size_mismatch); + BOOST_TEST(!sr.is_done()); + }()); + + BOOST_TEST_EQ(enc.calls, 0u); + BOOST_TEST(server.data().empty()); + } + + // commit() surfaces the mismatch when it triggers a + // flush + { + auto req = make_request(); + req.set(http::field::content_encoding, "test"); + auto [client, server] = capy::test::make_stream_pair(); + test_encoder enc; + + capy::test::run_blocking()([&]() -> capy::task<> + { + capy::any_write_stream ws(&client); + serializer sr(cfg, &ws); + sr.reset(&req, &enc, true); + + capy::mutable_buffer tmp[2]; + auto n = capy::buffer_copy( + sr.prepare(tmp), + capy::const_buffer("x", 1)); + BOOST_TEST_EQ(n, 1u); + + auto [ec] = co_await sr.commit_eof(n); + BOOST_TEST(ec == error::body_size_mismatch); + BOOST_TEST(!sr.is_done()); + }()); + + BOOST_TEST_EQ(enc.calls, 0u); + BOOST_TEST(server.data().empty()); + } + } + void testEncoderErrorInjection() { @@ -1890,6 +2238,11 @@ class serializer_test testEncoderWriteSome(); testEncoderMove(); testReset(); + testResetAfterEncoderKept(); + testHeadContentLength(); + testHeadChunked(); + testHeadBodyMismatch(); + testHeadEncoder(); testEncoderErrorInjection(); } };