Commit ecda13bf authored by jan.koester's avatar jan.koester
Browse files

msvc compatible unit tests

parent be5acfc3
Loading
Loading
Loading
Loading
+31 −41
Changes for test/http1_checksum_test.cpp: 31 added lines, 41 removed lines.
Original line number Diff line number Diff line
@@ -47,15 +47,11 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include <stdexcept>
#include <thread>

#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

#include <netplus/exception.h>
#include <netplus/socket.h>

#include "checksum_common.h"
#include "raw_tcp_client.h"

using namespace checksumtest;

@@ -111,10 +107,7 @@ int startServer(std::unique_ptr<netplus::tcp> &sockOwner,
    sockOwner = std::make_unique<netplus::tcp>("127.0.0.1", 0, 16, -1);
    sockOwner->bind();

    sockaddr_in bound{};
    socklen_t blen = sizeof(bound);
    ::getsockname(sockOwner->fd(), reinterpret_cast<sockaddr *>(&bound), &blen);
    int port = ntohs(bound.sin_port);
    int port = rawtcp::boundPort(*sockOwner);

    sockOwner->listen();

@@ -133,35 +126,36 @@ std::string headerValue(const libhttppp::HttpResponse &res, const std::string &k
}

// ---- Raw-socket helpers for cases HttpClient doesn't cover directly ----
// (built on netplus::tcp so this compiles/links under MSVC too, not just
// POSIX -- see raw_tcp_client.h).

int rawConnect(int port) {
    int fd = ::socket(AF_INET, SOCK_STREAM, 0);
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(static_cast<uint16_t>(port));
    inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr);
    if (::connect(fd, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) < 0) {
        ::close(fd);
        throw std::runtime_error("raw connect failed");
    }
    return fd;
}

void sendAll(int fd, const std::string &data) {
    size_t sent = 0;
    while (sent < data.size()) {
        ssize_t n = ::send(fd, data.data() + sent, data.size() - sent, 0);
        if (n <= 0) throw std::runtime_error("raw send failed");
        sent += static_cast<size_t>(n);
std::unique_ptr<netplus::tcp> rawConnect(int port) {
    auto sock = std::make_unique<netplus::tcp>();
    try {
        rawtcp::connectLocal(*sock, port);
    } catch (netplus::NetException &e) {
        throw std::runtime_error(std::string("raw connect failed: ") + e.what());
    }
    return sock;
}

// Reads one non-chunked HTTP/1.1 response (headers + Content-Length body)
// off a raw fd, reusing the library's own HttpResponse::parse() so the
// off a raw socket, reusing the library's own HttpResponse::parse() so the
// header-parsing logic under test is the real thing, not a re-implementation.
std::vector<char> rawReadResponse(int fd, libhttppp::HttpResponse &res) {
std::vector<char> rawReadResponse(netplus::socket &sock, libhttppp::HttpResponse &res) {
    auto recvSome = [&](std::vector<char> &dst) -> size_t {
        netplus::buffer buf(8192);
        size_t n;
        try {
            n = sock.recvData(buf);
        } catch (netplus::NetException &) {
            return 0;
        }
        dst.insert(dst.end(), buf.data.buf, buf.data.buf + n);
        return n;
    };

    std::vector<char> raw;
    char tmp[8192];
    size_t header_end = std::string::npos;

    for (;;) {
@@ -175,9 +169,7 @@ std::vector<char> rawReadResponse(int fd, libhttppp::HttpResponse &res) {
            }
        }
        if (header_end != std::string::npos) break;
        ssize_t n = ::recv(fd, tmp, sizeof(tmp), 0);
        if (n <= 0) throw std::runtime_error("raw recv: EOF before headers complete");
        raw.insert(raw.end(), tmp, tmp + n);
        if (recvSome(raw) == 0) throw std::runtime_error("raw recv: EOF before headers complete");
    }

    size_t parsed = res.parse(raw.data(), raw.size());
@@ -185,9 +177,7 @@ std::vector<char> rawReadResponse(int fd, libhttppp::HttpResponse &res) {

    std::vector<char> body(raw.begin() + static_cast<ptrdiff_t>(parsed), raw.end());
    while (body.size() < contentLength) {
        ssize_t n = ::recv(fd, tmp, sizeof(tmp), 0);
        if (n <= 0) throw std::runtime_error("raw recv: EOF before body complete");
        body.insert(body.end(), tmp, tmp + n);
        if (recvSome(body) == 0) throw std::runtime_error("raw recv: EOF before body complete");
    }
    body.resize(contentLength);
    return body;
@@ -217,12 +207,12 @@ void testRawChunkedUpload(int port) {
    }
    req << "0\r\n\r\n";

    int fd = rawConnect(port);
    sendAll(fd, req.str());
    auto sock = rawConnect(port);
    rawtcp::sendAllOrThrow(*sock, req.str());

    libhttppp::HttpResponse res;
    std::vector<char> respBody = rawReadResponse(fd, res);
    ::close(fd);
    std::vector<char> respBody = rawReadResponse(*sock, res);
    sock->close();

    check(res.getStatusCode() == 200, "chunked upload: 200 response");
    check(headerValue(res, "x-sha256") == expectedHash,
+56 −54
Changes for test/http1_nonblocking_stream_test.cpp: 56 added lines, 54 removed lines.
Original line number Diff line number Diff line
@@ -49,15 +49,11 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#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"
#include "raw_tcp_client.h"

using namespace libhttppp;

@@ -76,61 +72,67 @@ std::vector<char> randomBuffer(size_t 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);
std::unique_ptr<netplus::tcp> listenOnEphemeralPort() {
    // sockopts=-1 -> SO_REUSEADDR (see netplus::tcp's addr,port ctor), same
    // as the raw ::setsockopt(SO_REUSEADDR) this replaces.
    auto sock = std::make_unique<netplus::tcp>("127.0.0.1", 0, 1, -1);
    sock->bind();
    sock->listen();
    return sock;
}

void sendAll(int fd, const char *data, size_t len) {
// Silent-give-up send: the raw hand-rolled server below may write to a
// client that already stopped reading (e.g. it hit its own deadline), so a
// failed write here just gives up quietly rather than treating it as a test
// failure -- unlike raw_tcp_client.h's sendAllOrThrow(), which is for client
// requests that must reach a server this test itself just started.
void sendAll(netplus::socket &sock, 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);
        netplus::buffer buf(data + sent, len - sent);
        size_t n;
        try {
            n = sock.sendData(buf);
        } catch (netplus::NetException &) {
            return; // client may have already gotten everything it needed
        }
        if (n == 0) return;
        sent += n;
    }
}
void sendAll(int fd, const std::string &s) { sendAll(fd, s.data(), s.size()); }
void sendAll(netplus::socket &sock, const std::string &s) { sendAll(sock, 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,
void rawStreamingServer(netplus::tcp *listenSock, 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;
    std::unique_ptr<netplus::socket> csock = std::make_unique<netplus::tcp>(-1);
    try {
        listenSock->accept(csock, /*nonblock=*/false);
    } catch (netplus::NetException &) {
        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));
        netplus::buffer buf(4096);
        size_t n;
        try {
            n = csock->recvData(buf);
        } catch (netplus::NetException &) {
            csock->close();
            return;
        }
        req.append(buf.data.buf, n);
    }

    if (chunked) {
        sendAll(fd, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n");
        sendAll(*csock, "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");
        sendAll(*csock, "HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n");
    }

    size_t half = body.size() / 2;
@@ -139,11 +141,11 @@ void rawStreamingServer(int listenFd, std::vector<char> body,
        if (chunked) {
            std::ostringstream sz;
            sz << std::hex << len << "\r\n";
            sendAll(fd, sz.str());
            sendAll(fd, data, len);
            sendAll(fd, "\r\n");
            sendAll(*csock, sz.str());
            sendAll(*csock, data, len);
            sendAll(*csock, "\r\n");
        } else {
            sendAll(fd, data, len);
            sendAll(*csock, data, len);
        }
    };

@@ -151,8 +153,8 @@ void rawStreamingServer(int listenFd, std::vector<char> body,
    std::this_thread::sleep_for(delay);
    writePart(body.data() + half, body.size() - half);

    if (chunked) sendAll(fd, "0\r\n\r\n");
    ::close(fd);
    if (chunked) sendAll(*csock, "0\r\n\r\n");
    csock->close();
}

// Drives GetStream() + a readBodyChunkNonBlocking() poll loop against a
@@ -163,9 +165,9 @@ void testNonBlockingStreaming(bool chunked, std::chrono::milliseconds interChunk
    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);
    auto listenSock = listenOnEphemeralPort();
    int port = rawtcp::boundPort(*listenSock);
    std::thread server(rawStreamingServer, listenSock.get(), body, interChunkDelay, chunked);

    bool timingOk = true;
    std::chrono::milliseconds worstBusyWait{0};
@@ -218,7 +220,7 @@ void testNonBlockingStreaming(bool chunked, std::chrono::milliseconds interChunk
    }

    if (server.joinable()) server.join();
    ::close(listenFd);
    listenSock->close();

    check(completed, label + ": stream reached completion (-1) within deadline");
    check(collected == body, label + ": collected bytes match the original body exactly");
@@ -242,9 +244,9 @@ void testBlockingStreaming(bool chunked) {
    const size_t size = 65536 + 1234;
    std::vector<char> body = randomBuffer(size);

    int listenFd = listenOnEphemeralPort();
    int port = portOf(listenFd);
    std::thread server(rawStreamingServer, listenFd, body, std::chrono::milliseconds(20), chunked);
    auto listenSock = listenOnEphemeralPort();
    int port = rawtcp::boundPort(*listenSock);
    std::thread server(rawStreamingServer, listenSock.get(), body, std::chrono::milliseconds(20), chunked);

    std::vector<char> collected;
    bool ok = true;
@@ -269,7 +271,7 @@ void testBlockingStreaming(bool chunked) {
    }

    if (server.joinable()) server.join();
    ::close(listenFd);
    listenSock->close();

    check(ok, label + ": no exception");
    check(collected == body, label + ": collected bytes match the original body exactly");
+20 −28
Changes for test/http2_checksum_test.cpp: 20 added lines, 28 removed lines.
Original line number Diff line number Diff line
@@ -56,11 +56,6 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include <thread>
#include <utility>

#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

#include <netplus/exception.h>
#include <netplus/socket.h>
#include <netplus/crypto/tls.h>
@@ -70,6 +65,7 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include "hpack.h"
#include "https_certs.h"
#include "https_ca_cert.h"
#include "raw_tcp_client.h"

using namespace checksumtest;

@@ -172,10 +168,7 @@ std::pair<int, int> startServer(std::unique_ptr<netplus::tcp> &sockOwner,
    sockOwner = std::make_unique<netplus::tcp>("127.0.0.1", 0, 16, -1);
    sockOwner->bind();

    sockaddr_in bound{};
    socklen_t blen = sizeof(bound);
    ::getsockname(sockOwner->fd(), reinterpret_cast<sockaddr *>(&bound), &blen);
    int port = ntohs(bound.sin_port);
    int port = rawtcp::boundPort(*sockOwner);

    sockOwner->listen();

@@ -207,20 +200,16 @@ public:
    };

    void connectTo(int port) {
        _fd = ::socket(AF_INET, SOCK_STREAM, 0);
        sockaddr_in addr{};
        addr.sin_family = AF_INET;
        addr.sin_port = htons(static_cast<uint16_t>(port));
        inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr);
        if (::connect(_fd, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) < 0) {
            ::close(_fd);
            throw std::runtime_error("H2RawClient: connect failed");
        try {
            rawtcp::connectLocal(_sock, port);
        } catch (netplus::NetException &e) {
            throw std::runtime_error(std::string("H2RawClient: connect failed: ") + e.what());
        }
        sendAll(std::string(CLIENT_PREFACE, sizeof(CLIENT_PREFACE) - 1));
        sendFrame(FRAME_SETTINGS, 0, 0, ""); // empty client SETTINGS
    }

    ~H2RawClient() { if (_fd >= 0) ::close(_fd); }
    ~H2RawClient() { _sock.close(); }

    Response request(const std::string &method, const std::string &path,
                      const std::vector<char> &body) {
@@ -321,11 +310,10 @@ private:
    }

    void sendAll(const std::string &data) {
        size_t sent = 0;
        while (sent < data.size()) {
            ssize_t n = ::send(_fd, data.data() + sent, data.size() - sent, 0);
            if (n <= 0) throw std::runtime_error("H2RawClient: send failed");
            sent += static_cast<size_t>(n);
        try {
            rawtcp::sendAllOrThrow(_sock, data);
        } catch (std::runtime_error &) {
            throw std::runtime_error("H2RawClient: send failed");
        }
    }

@@ -334,10 +322,14 @@ private:
        while (_inbuf.size() < n) {
            if (std::chrono::steady_clock::now() >= deadline)
                throw std::runtime_error("H2RawClient: timeout waiting for frame data");
            char tmp[65536];
            ssize_t r = ::recv(_fd, tmp, sizeof(tmp), 0);
            if (r <= 0) throw std::runtime_error("H2RawClient: connection closed unexpectedly");
            _inbuf.insert(_inbuf.end(), tmp, tmp + r);
            netplus::buffer buf(65536);
            size_t r;
            try {
                r = _sock.recvData(buf);
            } catch (netplus::NetException &) {
                throw std::runtime_error("H2RawClient: connection closed unexpectedly");
            }
            _inbuf.insert(_inbuf.end(), buf.data.buf, buf.data.buf + r);
        }
    }

@@ -360,7 +352,7 @@ private:
        return f;
    }

    int _fd = -1;
    netplus::tcp _sock;
    uint32_t _nextStreamId = 1;
    libhttppp::hpack::Decoder _decoder;
    std::vector<char> _inbuf;
+20 −27
Changes for test/httpd_h2_concurrency_test.cpp: 20 added lines, 27 removed lines.
Original line number Diff line number Diff line
@@ -40,16 +40,12 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include <thread>
#include <vector>

#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

#include <netplus/socket.h>

#include "hpack.h"
#include "http.h"
#include "httpd.h"
#include "raw_tcp_client.h"

namespace {

@@ -125,14 +121,11 @@ std::vector<char> buildSlowRequest(uint32_t streamId, int delayMs) {
std::vector<std::chrono::steady_clock::time_point> runConcurrentRequest(
    int port, const std::vector<uint32_t> &streamIds, int delayMs) {

    int fd = ::socket(AF_INET, SOCK_STREAM, 0);
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(static_cast<uint16_t>(port));
    inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr);
    if (::connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) < 0) {
        ::close(fd);
        throw std::runtime_error("connect failed");
    netplus::tcp sock;
    try {
        rawtcp::connectLocal(sock, port);
    } catch (netplus::NetException &e) {
        throw std::runtime_error(std::string("connect failed: ") + e.what());
    }

    std::vector<char> req(CLIENT_PREFACE, CLIENT_PREFACE + sizeof(CLIENT_PREFACE) - 1);
@@ -141,22 +134,25 @@ std::vector<std::chrono::steady_clock::time_point> runConcurrentRequest(
        auto hdrs = buildSlowRequest(sid, delayMs);
        req.insert(req.end(), hdrs.begin(), hdrs.end());
    }
    size_t sent = 0;
    while (sent < req.size()) {
        ssize_t n = ::send(fd, req.data() + sent, req.size() - sent, 0);
        if (n <= 0) { ::close(fd); throw std::runtime_error("send failed"); }
        sent += static_cast<size_t>(n);
    try {
        rawtcp::sendAllOrThrow(sock, req.data(), req.size());
    } catch (std::runtime_error &) {
        throw std::runtime_error("send failed");
    }

    std::map<uint32_t, std::chrono::steady_clock::time_point> completed;
    std::vector<char> buf;
    char tmp[4096];
    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
    while (completed.size() < streamIds.size() &&
           std::chrono::steady_clock::now() < deadline) {
        ssize_t n = ::recv(fd, tmp, sizeof(tmp), 0);
        if (n <= 0) break;
        buf.insert(buf.end(), tmp, tmp + n);
        netplus::buffer rbuf(4096);
        size_t n;
        try {
            n = sock.recvData(rbuf);
        } catch (netplus::NetException &) {
            break;
        }
        buf.insert(buf.end(), rbuf.data.buf, rbuf.data.buf + n);

        size_t off = 0;
        while (off + FRAME_HEADER_LEN <= buf.size()) {
@@ -179,7 +175,7 @@ std::vector<std::chrono::steady_clock::time_point> runConcurrentRequest(
        }
        buf.erase(buf.begin(), buf.begin() + off);
    }
    ::close(fd);
    sock.close();

    if (completed.size() != streamIds.size())
        throw std::runtime_error("not all streams completed in time");
@@ -210,10 +206,7 @@ RunningServer startServer(int &port, size_t h2OffloadThreads) {
    rs.sock = std::make_unique<netplus::tcp>("127.0.0.1", 0, 16, -1);
    rs.sock->bind();

    sockaddr_in bound{};
    socklen_t blen = sizeof(bound);
    ::getsockname(rs.sock->fd(), reinterpret_cast<sockaddr*>(&bound), &blen);
    port = ntohs(bound.sin_port);
    port = rawtcp::boundPort(*rs.sock);

    rs.sock->listen();

+22 −31
Changes for test/httpd_h2_window_credit_race_test.cpp: 22 added lines, 31 removed lines.
Original line number Diff line number Diff line
@@ -44,7 +44,6 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
// and having _finishH2Dispatch() only fill in the default if no entry
// exists yet, instead of unconditionally overwriting it.

#include <cerrno>
#include <chrono>
#include <cstring>
#include <iostream>
@@ -52,16 +51,12 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include <thread>
#include <vector>

#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

#include <netplus/socket.h>

#include "hpack.h"
#include "http.h"
#include "httpd.h"
#include "raw_tcp_client.h"

namespace {

@@ -135,10 +130,7 @@ RunningServer startServer(int &port) {
    rs.sock = std::make_unique<netplus::tcp>("127.0.0.1", 0, 16, -1);
    rs.sock->bind();

    sockaddr_in bound{};
    socklen_t blen = sizeof(bound);
    ::getsockname(rs.sock->fd(), reinterpret_cast<sockaddr*>(&bound), &blen);
    port = ntohs(bound.sin_port);
    port = rawtcp::boundPort(*rs.sock);

    rs.sock->listen();

@@ -163,14 +155,11 @@ void stopServer(RunningServer &rs) {
// proactively grants extra per-stream credit right after opening the
// stream. Returns the total DATA-frame bytes received for stream 1.
size_t runRaceRequest(int port) {
    int fd = ::socket(AF_INET, SOCK_STREAM, 0);
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(static_cast<uint16_t>(port));
    inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr);
    if (::connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) < 0) {
        ::close(fd);
        throw std::runtime_error("connect failed");
    netplus::tcp sock;
    try {
        rawtcp::connectLocal(sock, port);
    } catch (netplus::NetException &e) {
        throw std::runtime_error(std::string("connect failed: ") + e.what());
    }

    std::string hpack_out = libhttppp::hpack::Encoder::encodeRequestHeaders(
@@ -193,26 +182,28 @@ size_t runRaceRequest(int port) {
    appendWindowUpdate(req, 0, 1'000'000);
    appendWindowUpdate(req, 1, 1'000'000);

    size_t sent = 0;
    while (sent < req.size()) {
        ssize_t n = ::send(fd, req.data() + sent, req.size() - sent, 0);
        if (n <= 0) { ::close(fd); throw std::runtime_error("send failed"); }
        sent += static_cast<size_t>(n);
    try {
        rawtcp::sendAllOrThrow(sock, req.data(), req.size());
    } catch (std::runtime_error &) {
        throw std::runtime_error("send failed");
    }

    timeval rcvTimeout{2, 0};
    ::setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &rcvTimeout, sizeof(rcvTimeout));
    sock.setTimeout(2000); // 2s SO_RCVTIMEO

    size_t dataBytes = 0;
    bool sawEndStream = false;
    std::vector<char> buf;
    char tmp[8192];
    auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
    while (!sawEndStream && std::chrono::steady_clock::now() < deadline) {
        ssize_t n = ::recv(fd, tmp, sizeof(tmp), 0);
        if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) continue;
        if (n <= 0) break;
        buf.insert(buf.end(), tmp, tmp + n);
        netplus::buffer rbuf(8192);
        size_t n;
        try {
            n = sock.recvData(rbuf);
        } catch (netplus::NetException &e) {
            if (e.getErrorType() == netplus::NetException::Note) continue; // would-block/timeout
            break; // EOF or hard error
        }
        buf.insert(buf.end(), rbuf.data.buf, rbuf.data.buf + n);

        size_t off = 0;
        while (off + FRAME_HEADER_LEN <= buf.size()) {
@@ -238,7 +229,7 @@ size_t runRaceRequest(int port) {
        }
        buf.erase(buf.begin(), buf.begin() + off);
    }
    ::close(fd);
    sock.close();
    return dataBytes;
}

Loading