test

            }

        }

        netplus::buffer b(64 * 1024);

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

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

        size_t n = 0;

        for (;;) {

            b.size = 64 * 1024; // recvData() shrinks it to what arrived

            try {

                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.

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

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

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

                _sw.waitRead(*sock, 500);

            }

        }

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

        _req.RecvData.append(b.data.buf, n);

    }

    bool   _sendContinue = false;

    bool   _done = false;

    bool   _aborted = false;

    netplus::socketwait _sw;

};

} // namespace

endif()

add_test(httpd_connection_handoff_test httpd_connection_handoff_test)

add_executable(httpd_request_body_stream_test

    httpd_request_body_stream_test.cpp

)

if(WIN32)

    target_link_libraries (httpd_request_body_stream_test httppp-static ws2_32)

else()

    target_link_libraries (httpd_request_body_stream_test httppp-static)

endif()

add_test(httpd_request_body_stream_test httpd_request_body_stream_test)

/*******************************************************************************

Copyright (c) 2026, Jan Koester jan.koester@gmx.net

All rights reserved.

Redistribution and use in source and binary forms, with or without

modification, are permitted provided that the following conditions are met:

    * Redistributions of source code must retain the above copyright

      notice, this list of conditions and the following disclaimer.

    * Redistributions in binary form must reproduce the above copyright

      notice, this list of conditions and the following disclaimer in the

      documentation and/or other materials provided with the distribution.

    * Neither the name of the <organization> nor the

      names of its contributors may be used to endorse or promote products

      derived from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND

ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE

DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY

DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES

(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;

LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND

ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

*******************************************************************************/

// 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

//     (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).

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

#include <chrono>

#include <cstring>

#include <functional>

#include <iostream>

#include <map>

#include <memory>

#include <stdexcept>

#include <string>

#include <thread>

#include <vector>

#include <netplus/exception.h>

#include <netplus/socket.h>

#include <netplus/crypto/tls.h>

#include <netplus/crypto/x509.h>

#include "checksum_common.h"

#include "https_certs.h"

#include "https_ca_cert.h"

#include "raw_tcp_client.h"

using namespace checksumtest;

namespace {

int g_passed = 0, g_failed = 0;

void check(bool ok, const std::string &what) {

    if (ok) {

        ++g_passed;

        std::cout << "  PASS " << what << std::endl;

    } else {

        ++g_failed;

        std::cout << "  FAIL " << what << std::endl;

    }

}

// POST /sum      -> reads the body via readBody(); responds with its sha256 and whether it

//                   was streamed (x-streamed: 1) or buffered (0).

// POST /slowsum  -> same, sleeping between reads (forces H2 backpressure).

// POST /early    -> answers 413 without reading the body at all.

// POST /buffered -> shouldStreamRequestBody() says no: body arrives in RecvData as before.

class StreamEvent : public libhttppp::HttpEvent {

public:

    explicit StreamEvent(std::vector<netplus::socket *> socks)

        : libhttppp::HttpEvent(socks, 1000, /*h2OffloadThreads=*/4, /*idleTimeoutSeconds=*/0,

                               /*h1OffloadThreads=*/4) {}

    bool shouldOffloadH1Dispatch(libhttppp::HttpRequest &) const override { return true; }

    bool shouldOffloadH2Dispatch(libhttppp::HttpRequest &, uint32_t) const override {

        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";

    }

    void RequestEvent(libhttppp::HttpRequest &req, const int, ULONG_PTR) override {

        const std::string url = req.getRequestURL();

        libhttppp::HttpResponse res;

        res.setContentType("text/plain");

        if (url == "/early") {

            res.setState("413 Payload Too Large");

            res.send(req, std::string("too big"), 7);

            return;

        }

        std::vector<char> body;

        const bool streamed = req.hasStreamingBody();

        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; }

                if (n == 0) break;

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

                if (url == "/slowsum")

                    std::this_thread::sleep_for(std::chrono::microseconds(300));

            }

        } else {

            body.assign(req.RecvData.data(), req.RecvData.data() + req.RecvData.size());

        }

        res.setHeaderData("x-sha256")->push_back(sha256Hex(body));

        res.setHeaderData("x-streamed")->push_back(streamed ? "1" : "0");

        std::string n = std::to_string(body.size());

        res.send(req, n, static_cast<int>(n.size()));

    }

};

netplus::ssl::CertificateBundle buildTestCertBundle() {

    netplus::x509cert cert;

    if (!cert.loadFromBuffer(test_cert_der))

        throw std::runtime_error("failed to load embedded test certificate");

    netplus::ssl::CertificateBundle bundle;

    bundle.cert = cert;

    bundle.privateKeyDer = std::vector<uint8_t>(test_key_der.begin(), test_key_der.end());

    bundle.rsa_key = netplus::rsa(bundle.privateKeyDer);

    bundle.chain.push_back(std::vector<uint8_t>(

        MKCERT_ROOT_CA_DER, MKCERT_ROOT_CA_DER + MKCERT_ROOT_CA_DER_LEN));

    return bundle;

}

std::string headerValue(const libhttppp::HttpResponse *res, const std::string &key) {

    if (!res) return {};

    auto *hd = res->getHeaderData(key);

    if (hd && hd->getfirstValue()) return hd->getfirstValue()->getvalue();

    return {};

}

// Body source handing out `data` in pieces of `piece` bytes.

libhttppp::HttpClient::BodySource sourceOf(const std::vector<char> &data, size_t piece) {

    auto off = std::make_shared<size_t>(0);

    return [&data, piece, off](char *buf, size_t len) -> size_t {

        size_t n = std::min({len, piece, data.size() - *off});

        std::memcpy(buf, data.data() + *off, n);

        *off += n;

        return n;

    };

}

// One streamed upload through `client`; checks status, echoed size/hash and that the server

// really streamed it.

void streamedUpload(libhttppp::HttpClient &client, const std::string &path,

                    const std::vector<char> &data, bool knownLength, bool expectStreamed,

                    const std::string &label) {

    try {

        libhttppp::HttpRequest req;

        req.setRequestURL(path);

        std::vector<char> resp = client.RequestStream(

            POSTREQUEST, req, sourceOf(data, 50000),

            knownLength ? static_cast<int64_t>(data.size()) : -1);

        const libhttppp::HttpResponse *last = client.lastResponse();

        check(client.lastStatusCode() == 200, label + ": 200");

        check(std::string(resp.begin(), resp.end()) == std::to_string(data.size()),

              label + ": server read every byte");

        check(headerValue(last, "x-sha256") == sha256Hex(data), label + ": sha256 matches");

        check(headerValue(last, "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);

    return std::string(b.data.buf, n);

}

} // namespace

int main() {

    std::cout << "=== Streamed request body test ===" << std::endl;

    const int kTlsPort = 19651;

    const size_t kBig = 20u * 1024 * 1024; // > the 16MB H2 stream window

    try {

        auto tcpSock = std::make_unique<netplus::tcp>("127.0.0.1", 0, 16, -1);

        tcpSock->bind();

        const int port = rawtcp::boundPort(*tcpSock);

        tcpSock->listen();

        std::map<std::string, netplus::ssl::CertificateBundle> certs;

        netplus::ssl::CertificateBundle bundle = buildTestCertBundle();

        certs["localhost"] = bundle;

        certs["127.0.0.1"] = bundle;

        auto sslSock = std::make_unique<netplus::ssl>(certs, "127.0.0.1", kTlsPort, 16, -1);

        auto server = std::make_unique<StreamEvent>(

            std::vector<netplus::socket *>{tcpSock.get(), sslSock.get()});

        std::thread serverThread([s = server.get()] { s->runEventloop(); });

        ServerThreadGuard guard{&serverThread};

        std::this_thread::sleep_for(std::chrono::milliseconds(50));

        // ---------------- HTTP/1.1 ----------------

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

        {

            libhttppp::HttpUrl url("http://127.0.0.1:" + std::to_string(port) + "/");

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

            // Same client throughout: every request reuses the keep-alive connection, so a

            // reader that over- or under-consumed would corrupt the next request.

            for (size_t size : {size_t(1), size_t(4096), 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, "H1 content-length size=" + s);

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

            }

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

                           "H1 buffered fallback");

        }

        {

            // Expect: 100-continue -- the client holds the body back until the interim 100.

            netplus::tcp sock;

            rawtcp::connectLocal(sock, port);

            rawtcp::sendAllOrThrow(sock,

                "POST /sum HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 5\r\n"

                "Expect: 100-continue\r\n\r\n");

            std::string interim = recvSome(sock);

            check(interim.rfind("HTTP/1.1 100 Continue\r\n\r\n", 0) == 0,

                  "H1 Expect: 100-continue gets interim 100");

            rawtcp::sendAllOrThrow(sock, "hello");

            std::string resp;

            for (int i = 0; i < 20 && resp.find("\r\n\r\n5") == std::string::npos; ++i)

                resp += recvSome(sock);

            check(resp.find("200") != std::string::npos &&

                      resp.find("\r\n\r\n5") != std::string::npos,

                  "H1 Expect: 100-continue body read after the 100");

            sock.close();

        }

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

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

        {

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

            libhttppp::HttpClient client(url, /*vers=*/2, /*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, "H2 known length size=" + s);

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

            }

            streamedUpload(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");

            } 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,

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

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

                           "H2 buffered fallback");

        }

    } catch (std::exception &e) {

        std::cout << "FATAL: " << e.what() << std::endl;

        return 1;

    }

    std::cout << "=== " << g_passed << " passed, " << g_failed << " failed ===" << std::endl;

    return g_failed == 0 ? 0 : 1;

}