test

    // stale value per request handled on the connection, until a strict backend (observed

    // live: Home Assistant's aiohttp) rejects it outright as a duplicate header.

    HttpHeader::clear();

    _bodyStreamDeclined = false;

    // 2) Copy header bytes (include the first \r\n of the separator so

    //    every header line, including the last one, ends with \r\n)

    int               _RequestType = PARSEREQUEST;

    size_t            _MaxUploadSize = DEFAULT_UPLOADSIZE;

    std::shared_ptr<RequestBodyStream> _bodyStream;

    // H1: shouldStreamRequestBody() already said no for the request being parsed -- a

    // buffered body re-enters dispatch on every read, the hook needn't run each time.

    bool              _bodyStreamDeclined = false;

    // Cached strings derived from _firstHeaderData pseudo-headers.

    // Populated by parseH1/parseH2/parseH3, read by getters.

                n = sock->recvData(b, 0);

                break;

            } catch (netplus::NetException &e) {

                // A Note is "nothing usable yet", not a failure -- notably ssl::recvData()

                // with only part of a TLS record arrived, even on a blocking socket. Same

                // wait-and-retry HttpClient::_recvBlocking() does.

                // A Note is "nothing usable yet", not a failure.

                if (e.getErrorType() != netplus::NetException::Note) throw;

                if (std::chrono::steady_clock::now() >= deadline)

                    fail("request body: timed out waiting for data");

                if (!dynamic_cast<netplus::ssl *>(sock.get())) {

                    _sw.waitRead(*sock, 500);

                    continue;

                }

                // Accepted TLS sockets only decrypt what's been pushed in via prime_recv() --

                // the event loop's job while attached, ours now that the connection is

                // detached (same step as proxyplus's WebSocket relay).

                netplus::buffer raw(64 * 1024);

                size_t r = 0;

                try {

                    r = sock->prime_recv(raw, 0);

                } catch (netplus::NetException &e2) {

                    if (e2.getErrorType() != netplus::NetException::Note) throw;

                    _sw.waitRead(*sock, 500);

                    continue;

                }

                if (r == 0) fail("request body: peer closed the connection before end of body");

            }

        }

        if (n == 0) fail("request body: peer closed the connection before end of body");

} // namespace

bool libhttppp::HttpEvent::_startH1BodyStream(HttpRequest &cureq, const int tid, ULONG_PTR args) {

    if (!_h1DispatchPool) return false;

    if (!_h1DispatchPool || cureq._bodyStreamDeclined) return false;

    const bool chunked = cureq.isChunkedRequest();

    const size_t clen = chunked ? 0 : cureq.getContentLength();

    if (!chunked && clen == 0) return false; // no body to stream

    if (!shouldOffloadH1Dispatch(cureq) || !shouldStreamRequestBody(cureq)) return false;

    if (!shouldOffloadH1Dispatch(cureq) || !shouldStreamRequestBody(cureq)) {

        cureq._bodyStreamDeclined = true;

        return false;

    }

    cureq._bodyStream = std::make_shared<H1RequestBodyReader>(cureq, chunked, clen);

    _dispatchH1Request(cureq, 0, tid, args);

    return true;

                    }

                    // A streamed body the handler stopped reading early (e.g. it answered an

                    // error first): the peer may still be sending it.

                    bool bodyLeftUnread = false;

                    if (auto &bs = (*trBox)->_bodyStream) {

                        bodyLeftUnread = !bs->finished();

                        bs->abort();

                        bs.reset();

                    }

                    // error first): the peer may still be sending it. Its pipe stays live

                    // until we hold event_mutex below -- aborting it any earlier lets the event

                    // loop answer the next DATA frame with RST_STREAM ahead of the response.

                    std::shared_ptr<RequestBodyStream> bodyStream =

                        std::move((*trBox)->_bodyStream);

                    const bool bodyLeftUnread = bodyStream && !bodyStream->finished();

                    HttpRequest &cureq2 = static_cast<HttpRequest&>(*connOwner);

                    std::lock_guard<std::recursive_mutex> lk(cureq2.event_mutex);

                    if (bodyStream) bodyStream->abort();

                    if (!cureq2.slots[0].csock)

                        return; // connection torn down while we were blocked

// End-to-end test for streamed request bodies: HttpEvent::shouldStreamRequestBody() +

// HttpRequest::readBody() on the server, HttpClient::RequestStream() on the client.

//

//   - H1: Content-Length and chunked bodies, keep-alive reuse after a streamed request

//     (the reader must consume exactly the body), and Expect: 100-continue.

//   - H2 (TLS+ALPN): known and unknown length, a body larger than the 16MB stream window

//   - H1, cleartext and TLS (HttpClient::RequestStream): Content-Length and chunked bodies,

//     keep-alive reuse after a streamed request (the reader must consume exactly the body),

//     and Expect: 100-continue. TLS matters separately: a detached TLS socket has to be fed

//     through prime_recv() by the reader itself.

//   - H2 (h2c, a raw client honoring flow control -- HttpClient doesn't negotiate h2 against

//     this server): known and unknown length, a body larger than the 16MB stream window

//     (only possible if stream credit really is returned as the handler consumes), a slow

//     consumer (backpressure instead of overrun), and a handler answering before reading

//     the body, followed by more requests on the same connection (connection-level credit

//     must still be returned for DATA of an abandoned stream).

//     the body, followed by more uploads on the same connection than one connection window

//     holds (credit must still be returned for DATA of the abandoned stream).

//   - A buffered POST still works when shouldStreamRequestBody() says no.

#include <chrono>

#include <netplus/crypto/x509.h>

#include "checksum_common.h"

#include "hpack.h"

#include "https_certs.h"

#include "https_ca_cert.h"

#include "raw_tcp_client.h"

        return true;

    }

    bool shouldStreamRequestBody(libhttppp::HttpRequest &req) const override {

        std::cerr << "[DBG] stream? url=" << req.getRequestURL() << " ver=" << req.getRequestVersion() << std::endl;

        return req.getRequestURL() != "/buffered";

    }

        if (streamed) {

            std::vector<char> buf(7777); // odd size: reads never line up with frames/chunks

            for (;;) {

                size_t n = 0;

                try { n = req.readBody(buf.data(), buf.size()); }

                catch (std::exception &e) { std::cerr << "[DBG] readBody threw: " << e.what() << std::endl; throw; }

                size_t n = req.readBody(buf.data(), buf.size());

                if (n == 0) break;

                body.insert(body.end(), buf.data(), buf.data() + n);

                if (url == "/slowsum")

    }

}

// Minimal h2c client that uploads with real flow control: DATA only within the window the

// server granted (SETTINGS_INITIAL_WINDOW_SIZE, WINDOW_UPDATE) -- so a server that stops

// returning stream credit stalls it instead of being overrun.

class H2Uploader {

public:

    struct Response {

        int status = 0;

        bool reset = false;      // server sent RST_STREAM

        uint32_t resetCode = 0;

        std::map<std::string, std::string> headers;

        std::string body;

    };

    explicit H2Uploader(int port) {

        rawtcp::connectLocal(_sock, port);

        rawtcp::sendAllOrThrow(_sock, std::string("PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"));

        sendFrame(0x4, 0, 0, ""); // empty SETTINGS

    }

    ~H2Uploader() { _sock.close(); }

    Response upload(const std::string &path, const std::vector<char> &data, bool sendLength) {

        uint32_t sid = _nextSid;

        _nextSid += 2;

        std::vector<libhttppp::hpack::HeaderField> extra;

        if (sendLength) extra.push_back({"content-length", std::to_string(data.size()), false});

        sendFrame(0x1, 0x4 /*END_HEADERS*/, sid,

                  libhttppp::hpack::Encoder::encodeRequestHeaders("POST", path, "http",

                                                                  "127.0.0.1", extra));

        _streamWindow = _initialWindow;

        Response res;

        bool done = false;

        size_t off = 0;

        while (off < data.size() && !done) {

            int64_t allowed = std::min(_connWindow, _streamWindow);

            if (allowed <= 0) {

                done = handle(readFrame(), sid, res);

                continue;

            }

            size_t take = std::min({size_t(16384), data.size() - off, size_t(allowed)});

            bool last = off + take == data.size();

            sendFrame(0x0, last ? 0x1 : 0, sid, std::string(data.data() + off, take));

            _connWindow -= take;

            _streamWindow -= take;

            off += take;

        }

        if (data.empty()) sendFrame(0x0, 0x1, sid, "");

        while (!done) done = handle(readFrame(), sid, res);

        return res;

    }

private:

    struct Frame { uint8_t type = 0, flags = 0; uint32_t sid = 0; std::string payload; };

    static uint32_t be32(const std::string &p, size_t at) {

        return (uint32_t(uint8_t(p[at])) << 24) | (uint32_t(uint8_t(p[at + 1])) << 16) |

               (uint32_t(uint8_t(p[at + 2])) << 8) | uint32_t(uint8_t(p[at + 3]));

    }

    // Returns true once stream `sid` is finished (response complete or reset).

    bool handle(const Frame &f, uint32_t sid, Response &res) {

        switch (f.type) {

        case 0x4: // SETTINGS

            if (f.flags & 0x1) return false;

            for (size_t i = 0; i + 6 <= f.payload.size(); i += 6) {

                uint16_t id = (uint16_t(uint8_t(f.payload[i])) << 8) | uint8_t(f.payload[i + 1]);

                if (id == 0x4) {

                    int64_t v = be32(f.payload, i + 2);

                    _streamWindow += v - _initialWindow;

                    _initialWindow = v;

                }

            }

            sendFrame(0x4, 0x1, 0, "");

            return false;

        case 0x8: // WINDOW_UPDATE

            if (f.sid == 0) _connWindow += be32(f.payload, 0) & 0x7fffffff;

            else if (f.sid == sid) _streamWindow += be32(f.payload, 0) & 0x7fffffff;

            return false;

        case 0x6: // PING

            if (!(f.flags & 0x1)) sendFrame(0x6, 0x1, 0, f.payload);

            return false;

        case 0x7:

            throw std::runtime_error("H2Uploader: GOAWAY");

        case 0x3: // RST_STREAM

            if (f.sid != sid) return false;

            res.reset = true;

            res.resetCode = be32(f.payload, 0);

            return true;

        case 0x1: { // HEADERS

            if (f.sid != sid) return false;

            for (auto &h : _decoder.decode(reinterpret_cast<const uint8_t *>(f.payload.data()),

                                           f.payload.size())) {

                if (h.name == ":status") res.status = std::stoi(h.value);

                else res.headers[h.name] = h.value;

            }

            return (f.flags & 0x1) != 0;

        }

        case 0x0: // DATA

            if (f.sid != sid) return false;

            res.body += f.payload;

            if (!f.payload.empty()) {

                sendFrame(0x8, 0, 0, win(f.payload.size()));

                if (!(f.flags & 0x1)) sendFrame(0x8, 0, sid, win(f.payload.size()));

            }

            return (f.flags & 0x1) != 0;

        }

        return false;

    }

    static std::string win(size_t n) {

        std::string p(4, '\0');

        for (int i = 0; i < 4; ++i) p[i] = char((n >> (24 - 8 * i)) & 0xff);

        return p;

    }

    void sendFrame(uint8_t type, uint8_t flags, uint32_t sid, const std::string &payload) {

        std::string out;

        size_t len = payload.size();

        out += char((len >> 16) & 0xff);

        out += char((len >> 8) & 0xff);

        out += char(len & 0xff);

        out += char(type);

        out += char(flags);

        for (int i = 0; i < 4; ++i) out += char((sid >> (24 - 8 * i)) & (i == 0 ? 0x7f : 0xff));

        out += payload;

        rawtcp::sendAllOrThrow(_sock, out);

    }

    Frame readFrame() {

        auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(30);

        auto need = [&](size_t n) {

            while (_in.size() < n) {

                if (std::chrono::steady_clock::now() >= deadline)

                    throw std::runtime_error("H2Uploader: timed out (stalled flow control?)");

                netplus::buffer b(65536);

                size_t r = _sock.recvData(b, 0);

                if (r == 0) throw std::runtime_error("H2Uploader: connection closed");

                _in.append(b.data.buf, r);

            }

        };

        need(9);

        size_t len = (size_t(uint8_t(_in[0])) << 16) | (size_t(uint8_t(_in[1])) << 8) |

                     uint8_t(_in[2]);

        need(9 + len);

        Frame f;

        f.type = uint8_t(_in[3]);

        f.flags = uint8_t(_in[4]);

        f.sid = be32(_in, 5) & 0x7fffffff;

        f.payload = _in.substr(9, len);

        _in.erase(0, 9 + len);

        return f;

    }

    netplus::tcp _sock;

    std::string _in;

    uint32_t _nextSid = 1;

    int64_t _connWindow = 65535;

    int64_t _initialWindow = 65535;

    int64_t _streamWindow = 65535;

    libhttppp::hpack::Decoder _decoder;

};

void h2Upload(H2Uploader &client, const std::string &path, const std::vector<char> &data,

              bool sendLength, bool expectStreamed, const std::string &label) {

    try {

        H2Uploader::Response r = client.upload(path, data, sendLength);

        check(r.status == 200, label + ": 200");

        check(r.body == std::to_string(data.size()), label + ": server read every byte");

        check(r.headers["x-sha256"] == sha256Hex(data), label + ": sha256 matches");

        check(r.headers["x-streamed"] == (expectStreamed ? "1" : "0"),

              label + (expectStreamed ? ": streamed" : ": buffered"));

    } catch (std::exception &e) {

        check(false, label + ": threw " + e.what());

    }

}

std::string recvSome(netplus::tcp &sock) {

    netplus::buffer b(4096);

    size_t n = sock.recvData(b, 0);

            sock.close();

        }

        // ---------------- HTTP/2 ----------------

        std::cout << "-- HTTP/2 (TLS) --" << std::endl;

        // ---------------- HTTP/1.1 over TLS ----------------

        std::cout << "-- HTTP/1.1 over TLS --" << std::endl;

        {

            libhttppp::HttpUrl url("https://127.0.0.1:" + std::to_string(kTlsPort) + "/");

            libhttppp::HttpClient client(url, /*vers=*/2, /*timeoutSec=*/10);

            libhttppp::HttpClient client(url, /*vers=*/1, /*timeoutSec=*/10);

            for (size_t size : {size_t(1), size_t(65536), size_t(1048576), kBig}) {

                std::vector<char> data = randomBuffer(size);

                std::string s = std::to_string(size);

                streamedUpload(client, "/sum", data, true, true, "TLS content-length size=" + s);

                streamedUpload(client, "/sum", data, false, true, "TLS chunked size=" + s);

            }

            streamedUpload(client, "/buffered", randomBuffer(100000), true, false,

                           "TLS buffered fallback");

        }

        // ---------------- HTTP/2 (h2c) ----------------

        std::cout << "-- HTTP/2 (h2c) --" << std::endl;

        {

            H2Uploader client(port);

            for (size_t size : {size_t(1), size_t(65536), size_t(1048576), kBig}) {

                std::vector<char> data = randomBuffer(size);

                std::string s = std::to_string(size);

                streamedUpload(client, "/sum", data, true, true, "H2 known length size=" + s);

                streamedUpload(client, "/sum", data, false, true, "H2 unknown length size=" + s);

                h2Upload(client, "/sum", data, true, true, "H2 known length size=" + s);

                h2Upload(client, "/sum", data, false, true, "H2 unknown length size=" + s);

            }

            streamedUpload(client, "/slowsum", randomBuffer(kBig), true, true,

            h2Upload(client, "/slowsum", randomBuffer(kBig), true, true,

                     "H2 slow consumer (backpressure) 20MB");

            try {

                std::vector<char> data = randomBuffer(1048576);

                libhttppp::HttpRequest req;

                req.setRequestURL("/early");

                client.RequestStream(POSTREQUEST, req, sourceOf(data, 50000),

                                     static_cast<int64_t>(data.size()));

                check(client.lastStatusCode() == 413, "H2 early response before body read");

                // Bigger than the stream window: the client can only finish sending if the

                // server tells it to stop (RST_STREAM NO_ERROR) after its early answer.

                H2Uploader::Response r = client.upload("/early", randomBuffer(kBig), true);

                check(r.status == 413, "H2 early response before body read");

            } catch (std::exception &e) {

                check(false, std::string("H2 early response: threw ") + e.what());

            }

            // More traffic on the same connection than one connection window holds: only

            // works if the abandoned stream's DATA still had its connection credit returned.

            for (int i = 0; i < 3; ++i)

                streamedUpload(client, "/sum", randomBuffer(8u * 1024 * 1024), true, true,

                h2Upload(client, "/sum", randomBuffer(8u * 1024 * 1024), true, true,

                         "H2 same connection after early response #" + std::to_string(i));

            streamedUpload(client, "/buffered", randomBuffer(100000), true, false,

            h2Upload(client, "/buffered", randomBuffer(100000), true, false,

                     "H2 buffered fallback");

        }

    } catch (std::exception &e) {