new stream test

add_test(http1_checksum_test http1_checksum_test)

add_executable(http1_nonblocking_stream_test

    http1_nonblocking_stream_test.cpp

)

if(WIN32)

    target_link_libraries (http1_nonblocking_stream_test httppp-static ws2_32)

else()

    target_link_libraries (http1_nonblocking_stream_test httppp-static)

endif()

add_test(http1_nonblocking_stream_test http1_nonblocking_stream_test)

add_executable(http2_checksum_test

    http2_checksum_test.cpp

)

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

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.

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

// Regression test for HttpClient::readBodyChunkNonBlocking()'s contract on

// STREAM_CHUNKED and STREAM_EOF bodies: "returns 0 immediately if no data

// available yet" (see http.h). Historically this fell back to the *blocking*

// readBodyChunk() for exactly these two modes, silently violating the

// contract for any event-loop-driven caller. There was previously no test

// exercising the non-blocking path at all for chunked/EOF bodies -- this

// file drives a raw hand-built HTTP/1.1 server (not the HttpEvent framework,

// so response timing is fully controlled) that sends a body in two halves

// with a deliberate delay in between, then asserts every

// readBodyChunkNonBlocking() call that finds no data returns in well under

// that delay, and that the fully-assembled body is still byte-for-byte

// correct once all data has arrived.

#include <atomic>

#include <chrono>

#include <cstdio>

#include <cstring>

#include <iostream>

#include <sstream>

#include <string>

#include <thread>

#include <vector>

#include <arpa/inet.h>

#include <netinet/in.h>

#include <sys/socket.h>

#include <unistd.h>

#include <netplus/crypto/sha.h>

#include <netplus/random.h>

#include "http.h"

using namespace libhttppp;

namespace {

int g_passed = 0, g_failed = 0;

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

    if (ok) { std::cout << "  PASS: " << name << std::endl; ++g_passed; }

    else    { std::cout << "  FAIL: " << name << std::endl; ++g_failed; }

}

std::vector<char> randomBuffer(size_t len) {

    std::vector<char> buf(len);

    if (len) netplus::fillRandomBytes(reinterpret_cast<uint8_t *>(buf.data()), len);

    return buf;

}

int listenOnEphemeralPort() {

    int fd = ::socket(AF_INET, SOCK_STREAM, 0);

    int one = 1;

    ::setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));

    sockaddr_in addr{};

    addr.sin_family = AF_INET;

    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);

    addr.sin_port = 0;

    if (::bind(fd, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) < 0)

        throw std::runtime_error("bind failed");

    ::listen(fd, 1);

    return fd;

}

int portOf(int listenFd) {

    sockaddr_in addr{};

    socklen_t len = sizeof(addr);

    ::getsockname(listenFd, reinterpret_cast<sockaddr *>(&addr), &len);

    return ntohs(addr.sin_port);

}

void sendAll(int fd, const char *data, size_t len) {

    size_t sent = 0;

    while (sent < len) {

        ssize_t n = ::send(fd, data + sent, len - sent, 0);

        if (n <= 0) return; // client may have already gotten everything it needed

        sent += static_cast<size_t>(n);

    }

}

void sendAll(int fd, const std::string &s) { sendAll(fd, s.data(), s.size()); }

// Accepts exactly one connection, discards the request, then streams `body`

// in two halves with `delay` in between. `chunked` selects

// Transfer-Encoding: chunked framing vs. a bare Connection: close/EOF body.

void rawStreamingServer(int listenFd, std::vector<char> body,

                         std::chrono::milliseconds delay, bool chunked) {

    sockaddr_in peer{};

    socklen_t plen = sizeof(peer);

    int fd = ::accept(listenFd, reinterpret_cast<sockaddr *>(&peer), &plen);

    if (fd < 0) return;

    // Read (and discard) the request up through the blank line.

    std::string req;

    char tmp[4096];

    for (;;) {

        if (req.find("\r\n\r\n") != std::string::npos) break;

        ssize_t n = ::recv(fd, tmp, sizeof(tmp), 0);

        if (n <= 0) { ::close(fd); return; }

        req.append(tmp, static_cast<size_t>(n));

    }

    if (chunked) {

        sendAll(fd, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n");

    } else {

        sendAll(fd, "HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n");

    }

    size_t half = body.size() / 2;

    auto writePart = [&](const char *data, size_t len) {

        if (len == 0) return;

        if (chunked) {

            std::ostringstream sz;

            sz << std::hex << len << "\r\n";

            sendAll(fd, sz.str());

            sendAll(fd, data, len);

            sendAll(fd, "\r\n");

        } else {

            sendAll(fd, data, len);

        }

    };

    writePart(body.data(), half);

    std::this_thread::sleep_for(delay);

    writePart(body.data() + half, body.size() - half);

    if (chunked) sendAll(fd, "0\r\n\r\n");

    ::close(fd);

}

// Drives GetStream() + a readBodyChunkNonBlocking() poll loop against a

// rawStreamingServer, asserting every "no data yet" (0) return happens well

// under interChunkDelay, and that the fully-collected body is correct.

void testNonBlockingStreaming(bool chunked, std::chrono::milliseconds interChunkDelay) {

    const std::string label = chunked ? "chunked" : "eof";

    const size_t size = 65536 + 777; // odd size, spans multiple TCP segments

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

    int listenFd = listenOnEphemeralPort();

    int port = portOf(listenFd);

    std::thread server(rawStreamingServer, listenFd, body, interChunkDelay, chunked);

    bool timingOk = true;

    std::chrono::milliseconds worstBusyWait{0};

    std::vector<char> collected;

    bool completed = false;

    try {

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

        HttpClient clt(url, /*vers=*/0, /*timeoutSec=*/5);

        HttpRequest req;

        req.setRequestURL("/x");

        HttpResponse res = clt.GetStream(req);

        (void)res;

        collected.reserve(size);

        char buf[8192];

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

        int zeroReturns = 0;

        for (;;) {

            if (std::chrono::steady_clock::now() > deadline) break;

            auto t0 = std::chrono::steady_clock::now();

            size_t n = clt.readBodyChunkNonBlocking(buf, sizeof(buf));

            auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(

                std::chrono::steady_clock::now() - t0);

            if (n == static_cast<size_t>(-1)) {

                completed = true;

                break;

            }

            if (n == 0) {

                ++zeroReturns;

                // The whole point of the *non-blocking* variant: it must not

                // itself block for the server's inter-chunk delay. Allow some

                // slack for scheduling jitter, but this must stay far below

                // interChunkDelay -- if it doesn't, the call silently fell

                // back to blocking behavior.

                if (elapsed >= interChunkDelay / 2) {

                    timingOk = false;

                    if (elapsed > worstBusyWait) worstBusyWait = elapsed;

                }

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

                continue;

            }

            collected.insert(collected.end(), buf, buf + n);

        }

        (void)zeroReturns;

    } catch (std::exception &e) {

        std::cerr << "[" << label << "] exception: " << e.what() << std::endl;

    }

    if (server.joinable()) server.join();

    ::close(listenFd);

    check(completed, label + ": stream reached completion (-1) within deadline");

    check(collected == body, label + ": collected bytes match the original body exactly");

    check(timingOk, label + ": readBodyChunkNonBlocking() never blocked for the "

                     "server's inter-chunk delay (worst observed: " +

                     std::to_string(worstBusyWait.count()) + "ms, limit " +

                     std::to_string(interChunkDelay.count() / 2) + "ms)");

}

} // namespace

int main() {

    std::cout << "=== HTTP/1.1 Non-Blocking Streaming Test ===" << std::endl;

    const auto delay = std::chrono::milliseconds(400);

    testNonBlockingStreaming(/*chunked=*/true, delay);

    testNonBlockingStreaming(/*chunked=*/false, delay);

    std::cout << "==============================" << std::endl;

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

    std::cout << "==============================" << std::endl;

    return g_failed > 0 ? 1 : 0;

}