test (57cc954b) · Commits · jan.koester / libhttppp

src/http.cpp

+263 −144

Changes for src/http.cpp: 263 added lines, 144 removed lines.
Original line number	Diff line number	Diff line
		@@ -630,9 +630,14 @@ std::vector<char> libhttppp::HttpClient::_h1ReadResponse(const std::string &labe
		}
		} catch (...) {}

		if (!chunked) {
		// getContentLength() returns 0 both when Content-Length is genuinely
		// absent (relying on EOF/connection-close to terminate the body) and
		// when the header is literally "Content-Length: 0" -- indistinguishable
		// from that call alone. Check presence directly so an EOF-terminated
		// response isn't misdetected as a zero-length body.
		if (!chunked && res.getHeaderData("content-length")) {
		try {
		content_len = res.getContentLength();
		content_len = static_cast<ptrdiff_t>(res.getContentLength());
		if (content_len < 0) content_len = -1;
		} catch (...) {
		content_len = -1;
		@@ -1268,9 +1273,11 @@ libhttppp::HttpResponse libhttppp::HttpClient::GetStream(libhttppp::HttpRequest
		}
		} catch (...) {}

		if (!chunked) {
		// See _h1ReadResponse() for why presence is checked explicitly
		// rather than trusting getContentLength()'s ambiguous 0 return.
		if (!chunked && res.getHeaderData("content-length")) {
		try {
		content_len = res.getContentLength();
		content_len = static_cast<ptrdiff_t>(res.getContentLength());
		if (content_len < 0) content_len = -1;
		} catch (...) {
		content_len = -1;
		@@ -1441,9 +1448,11 @@ libhttppp::HttpResponse libhttppp::HttpClient::PostStream(libhttppp::HttpRequest
		}
		} catch (...) {}

		if (!chunked) {
		// See _h1ReadResponse() for why presence is checked explicitly
		// rather than trusting getContentLength()'s ambiguous 0 return.
		if (!chunked && res.getHeaderData("content-length")) {
		try {
		content_len = res.getContentLength();
		content_len = static_cast<ptrdiff_t>(res.getContentLength());
		if (content_len < 0) content_len = -1;
		} catch (...) {
		content_len = -1;
		@@ -1474,18 +1483,61 @@ libhttppp::HttpResponse libhttppp::HttpClient::PostStream(libhttppp::HttpRequest
		}
		}

		size_t libhttppp::HttpClient::readBodyChunk(char *buf, size_t bufsize) {
		if (_streamMode == STREAM_NONE \|\| bufsize == 0)
		return 0;
		// Tries to pull more bytes off the wire into _streamBuf. `blocking` selects
		// _recvBlocking() (waits up to the recv timeout) vs _recvNonBlocking()
		// (returns immediately if nothing is available). Returns true iff bytes
		// were appended. Returns false otherwise; `eof` distinguishes why:
		// blocking=true, false return => eof is always true (a clean EOF, e.g.
		// TLS close_notify -- _recvBlocking() never returns 0
		// without one, and any other failure throws)
		// blocking=false, false return => eof is always false (EAGAIN, i.e. no
		// data is available yet); a real connection-ending
		// condition in non-blocking mode throws instead, exactly
		// like it would in blocking mode
		// Does NOT itself catch exceptions: a chunked body ending mid-frame without
		// its terminator is a real protocol error and should propagate, whereas an
		// EOF-terminated body's caller (_streamReadEof) treats connection-ending
		// exceptions as a normal, successful end of body -- this distinction
		// already existed between the two modes before this method existed, and is
		// preserved by leaving the choice to each caller instead of swallowing
		// exceptions here.
		bool libhttppp::HttpClient::_streamRecvMore(bool blocking, bool &eof) {
		eof = false;
		size_t n = blocking ? _recvBlocking(_recvScratch) : _recvNonBlocking(_recvScratch);
		if (n == 0) {
		if (blocking) eof = true;
		return false;
		}
		// Compact only when consumed portion exceeds half the buffer, to
		// amortize the O(n) memmove cost of dropping the consumed prefix.
		if (_streamBufPos > 0 && _streamBufPos >= _streamBuf.size() / 2) {
		_streamBuf.erase(_streamBuf.begin(), _streamBuf.begin() + (ptrdiff_t)_streamBufPos);
		_streamBufPos = 0;
		}
		_streamBuf.insert(_streamBuf.end(), _recvScratch.data.buf, _recvScratch.data.buf + n);
		return true;
		}

		// Single implementation of chunked-transfer-encoding body reading shared by
		// readBodyChunk() (blocking=true) and readBodyChunkNonBlocking()
		// (blocking=false) -- previously the non-blocking variant didn't implement
		// this mode at all and silently fell back to the blocking one, meaning a
		// caller using the non-blocking API to avoid stalling its thread would
		// still stall for up to the recv timeout on a chunked response.
		// Returns: bytes read (>0), 0 (blocking=false only: no data available yet),
		// or (size_t)-1 (body complete).
		size_t libhttppp::HttpClient::_streamReadChunked(char *buf, size_t bufsize, bool blocking) {
		if (_streamChunkDone) {
		_streamMode = STREAM_NONE;
		return static_cast<size_t>(-1);
		}

		// Helper: drain from internal leftover buffer
		auto drainBuf = [&](size_t maxBytes) -> size_t {
		size_t have = _streamBuf.size() - _streamBufPos;
		if (have == 0) return 0;
		size_t take = (std::min)(have, maxBytes);
		std::memcpy(buf, _streamBuf.data() + _streamBufPos, take);
		_streamBufPos += take;
		// Compact buffer when fully consumed
		if (_streamBufPos == _streamBuf.size()) {
		_streamBuf.clear();
		_streamBufPos = 0;
		@@ -1493,192 +1545,237 @@ size_t libhttppp::HttpClient::readBodyChunk(char *buf, size_t bufsize) {
		return take;
		};

		// Helper: receive more data into internal buffer
		auto recvMore = [&]() -> size_t {
		size_t n = _recvBlocking(_recvScratch);
		if (n > 0) {
		// Compact only when consumed portion exceeds half the buffer
		// to amortize the O(n) memmove cost
		if (_streamBufPos > 0 && _streamBufPos >= _streamBuf.size() / 2) {
		_streamBuf.erase(_streamBuf.begin(),
		_streamBuf.begin() + (ptrdiff_t)_streamBufPos);
		_streamBufPos = 0;
		enum class LineResult { Found, WouldBlock, Eof };
		auto readLine = [&](std::string &out) -> LineResult {
		for (;;) {
		size_t i = _streamBufPos;
		for (; i + 1 < _streamBuf.size(); ++i) {
		if (_streamBuf[i] == '\r' && _streamBuf[i + 1] == '\n') break;
		}
		_streamBuf.insert(_streamBuf.end(), _recvScratch.data.buf, _recvScratch.data.buf + n);
		if (i + 1 < _streamBuf.size()) {
		out.assign(_streamBuf.begin() + (ptrdiff_t)_streamBufPos,
		_streamBuf.begin() + (ptrdiff_t)i);
		_streamBufPos = i + 2;
		return LineResult::Found;
		}
		return n;
		};

		// ── Content-Length mode ──
		if (_streamMode == STREAM_CONTENT_LENGTH) {
		if (_streamRemaining == 0) {
		_streamMode = STREAM_NONE;
		return 0;
		bool eof = false;
		if (!_streamRecvMore(blocking, eof))
		return (!blocking) ? LineResult::WouldBlock : LineResult::Eof;
		}
		};

		size_t want = (std::min)(bufsize, _streamRemaining);

		// First try leftover buffer
		enum class DataResult { Got, WouldBlock, Eof };
		auto readChunkData = [&](size_t &outGot) -> DataResult {
		size_t want = (std::min)(bufsize, _streamChunkRemaining);
		size_t got = drainBuf(want);
		if (got > 0) {
		_streamRemaining -= got;
		if (_streamRemaining == 0)
		_streamMode = STREAM_NONE;
		return got;
		}
		while (got == 0) {
		bool eof = false;
		if (!_streamRecvMore(blocking, eof))
		return (!blocking) ? DataResult::WouldBlock : DataResult::Eof;
		got = drainBuf(want);
		}
		outGot = got;
		return DataResult::Got;
		};

		// Read from socket
		size_t n = _recvBlocking(_recvScratch);
		if (n == 0) {
		_streamMode = STREAM_NONE;
		if (_streamChunkRemaining > 0) {
		size_t got = 0;
		switch (readChunkData(got)) {
		case DataResult::WouldBlock:
		return 0;
		case DataResult::Eof:
		// Clean EOF mid-chunk: matches the pre-existing (lenient)
		// behavior of treating it as an early-but-clean body end
		// rather than a hard error.
		_streamChunkDone = true;
		_streamMode = STREAM_NONE;
		return static_cast<size_t>(-1);
		case DataResult::Got:
		_streamChunkRemaining -= got;
		return got;
		}
		size_t take = (std::min)({n, want, bufsize});
		std::memcpy(buf, _recvScratch.data.buf, take);
		_streamRemaining -= take;

		// Save excess data for next call
		if (take < n) {
		_streamBuf.assign(_recvScratch.data.buf + take, _recvScratch.data.buf + n);
		_streamBufPos = 0;
		}

		if (_streamRemaining == 0)
		for (;;) {
		std::string line;
		LineResult lr = readLine(line);
		if (lr == LineResult::WouldBlock) return 0;
		if (lr == LineResult::Eof) {
		_streamChunkDone = true;
		_streamMode = STREAM_NONE;
		return take;
		return static_cast<size_t>(-1);
		}
		// An empty line here is the previous chunk's trailing CRLF
		// terminator (deferred from the last call if it hadn't arrived yet
		// when that chunk's data was fully delivered) -- skip it and read
		// the actual next line.
		if (line.empty()) continue;

		// ── Chunked mode ──
		if (_streamMode == STREAM_CHUNKED) {
		if (_streamChunkDone) {
		auto sem = line.find(';');
		if (sem != std::string::npos) line.resize(sem);

		size_t chunk_size = 0;
		try {
		chunk_size = static_cast<size_t>(std::stoul(line, nullptr, 16));
		} catch (...) {
		_streamChunkDone = true;
		_streamMode = STREAM_NONE;
		return 0;
		return static_cast<size_t>(-1);
		}

		// Helper: read a line from the stream buffer (refilling as needed)
		auto readLine = [&]() -> std::string {
		for (;;) {
		for (size_t i = _streamBufPos; i + 1 < _streamBuf.size(); ++i) {
		if (_streamBuf[i] == '\r' && _streamBuf[i+1] == '\n') {
		std::string line(_streamBuf.begin() + (ptrdiff_t)_streamBufPos,
		_streamBuf.begin() + (ptrdiff_t)i);
		_streamBufPos = i + 2;
		return line;
		}
		if (chunk_size == 0) {
		// Final chunk -- consume trailers (bounded to prevent unbounded
		// memory growth from a malicious server).
		for (int trailer_i = 0; trailer_i < 128; ++trailer_i) {
		std::string tline;
		LineResult tr = readLine(tline);
		if (tr == LineResult::WouldBlock) return 0;
		if (tr == LineResult::Eof \|\| tline.empty()) break;
		}
		// Need more data
		if (recvMore() == 0) {
		_streamChunkDone = true;
		return {};
		}
		_streamMode = STREAM_NONE;
		return static_cast<size_t>(-1);
		}
		};

		// Iterative loop: avoids tail-recursion when a new chunk is parsed
		// but no data is returned yet (chunk header consumed, data follows).
		for (;;) {

		// If we're in the middle of a chunk, continue reading chunk data
		if (_streamChunkRemaining > 0) {
		size_t want = (std::min)(bufsize, _streamChunkRemaining);
		size_t got = drainBuf(want);
		if (got > 0) {
		_streamChunkRemaining = chunk_size;
		size_t got = 0;
		switch (readChunkData(got)) {
		case DataResult::WouldBlock:
		return 0;
		case DataResult::Eof:
		_streamChunkDone = true;
		_streamMode = STREAM_NONE;
		return static_cast<size_t>(-1);
		case DataResult::Got:
		_streamChunkRemaining -= got;
		if (_streamChunkRemaining == 0) {
		// Skip chunk terminator \r\n
		while (_streamBuf.size() - _streamBufPos < 2)
		recvMore();
		_streamBufPos += 2;
		}
		return got;
		}
		// Read from socket directly
		size_t n = _recvBlocking(_recvScratch);
		if (n == 0) {
		_streamChunkDone = true;
		_streamMode = STREAM_NONE;
		return 0;
		}
		size_t take = (std::min)({n, want, bufsize});
		std::memcpy(buf, _recvScratch.data.buf, take);
		_streamChunkRemaining -= take;
		if (take < n) {
		_streamBuf.assign(_recvScratch.data.buf + take, _recvScratch.data.buf + n);
		_streamBufPos = 0;
		}
		if (_streamChunkRemaining == 0) {
		// Skip chunk terminator \r\n
		while (_streamBuf.size() - _streamBufPos < 2)
		recvMore();
		_streamBufPos += 2;

		// Single implementation of EOF-terminated (no Content-Length, not chunked)
		// body reading shared by readBodyChunk()/readBodyChunkNonBlocking(); see
		// _streamReadChunked() for why this collapses what used to be a
		// blocking-only implementation (silently reused via fallback for the
		// non-blocking caller) into one implementation parameterized by `blocking`.
		// Returns: bytes read (>0), 0 (blocking=false only: no data available yet),
		// or (size_t)-1 (body complete).
		size_t libhttppp::HttpClient::_streamReadEof(char *buf, size_t bufsize, bool blocking) {
		size_t have = _streamBuf.size() - _streamBufPos;
		if (have > 0) {
		size_t take = (std::min)(have, bufsize);
		std::memcpy(buf, _streamBuf.data() + _streamBufPos, take);
		_streamBufPos += take;
		if (_streamBufPos == _streamBuf.size()) {
		_streamBuf.clear();
		_streamBufPos = 0;
		}
		return take;
		}

		// Read next chunk size line
		std::string szline = readLine();
		if (szline.empty() && _streamChunkDone) {
		_streamMode = STREAM_NONE;
		return 0;
		}
		auto sem = szline.find(';');
		if (sem != std::string::npos) szline.resize(sem);

		size_t chunk_size = 0;
		bool eof = false;
		bool progressed;
		try {
		chunk_size = (size_t)std::stoul(szline, nullptr, 16);
		progressed = _streamRecvMore(blocking, eof);
		} catch (...) {
		_streamChunkDone = true;
		// An EOF-terminated body has no explicit terminator to wait for, so
		// any connection-ending condition (including a plain-TCP "peer
		// closed" exception, unlike the clean 0-return TLS gives) means the
		// body is done, not that something went wrong.
		_streamMode = STREAM_NONE;
		return 0;
		return static_cast<size_t>(-1);
		}
		if (!progressed) {
		if (eof) {
		_streamMode = STREAM_NONE;
		return static_cast<size_t>(-1);
		}
		return 0; // non-blocking: no data yet
		}

		if (chunk_size == 0) {
		// Final chunk — consume trailers (limit to 128 to prevent
		// unbounded memory growth from a malicious server).
		for (int trailer_i = 0; trailer_i < 128; ++trailer_i) {
		std::string tline = readLine();
		if (tline.empty()) break;
		have = _streamBuf.size() - _streamBufPos;
		size_t take = (std::min)(have, bufsize);
		std::memcpy(buf, _streamBuf.data() + _streamBufPos, take);
		_streamBufPos += take;
		if (_streamBufPos == _streamBuf.size()) {
		_streamBuf.clear();
		_streamBufPos = 0;
		}
		_streamChunkDone = true;
		_streamMode = STREAM_NONE;
		return take;
		}

		size_t libhttppp::HttpClient::readBodyChunk(char *buf, size_t bufsize) {
		if (_streamMode == STREAM_NONE \|\| bufsize == 0)
		return 0;

		// Helper: drain from internal leftover buffer
		auto drainBuf = [&](size_t maxBytes) -> size_t {
		size_t have = _streamBuf.size() - _streamBufPos;
		if (have == 0) return 0;
		size_t take = (std::min)(have, maxBytes);
		std::memcpy(buf, _streamBuf.data() + _streamBufPos, take);
		_streamBufPos += take;
		// Compact buffer when fully consumed
		if (_streamBufPos == _streamBuf.size()) {
		_streamBuf.clear();
		_streamBufPos = 0;
		}
		return take;
		};

		_streamChunkRemaining = chunk_size;
		// Loop back to read data from this new chunk (iterative instead of
		// recursive to avoid stack overflow on many consecutive chunks).
		continue;
		} // end for(;;)
		// ── Content-Length mode ──
		if (_streamMode == STREAM_CONTENT_LENGTH) {
		if (_streamRemaining == 0) {
		_streamMode = STREAM_NONE;
		return 0;
		}

		// ── EOF mode (no content-length, not chunked) ──
		if (_streamMode == STREAM_EOF) {
		// First drain leftover buffer
		size_t got = drainBuf(bufsize);
		if (got > 0) return got;
		size_t want = (std::min)(bufsize, _streamRemaining);

		// Read from socket; EOF signals end of body
		size_t n = 0;
		try {
		n = _recvBlocking(_recvScratch);
		} catch (...) {
		// First try leftover buffer
		size_t got = drainBuf(want);
		if (got > 0) {
		_streamRemaining -= got;
		if (_streamRemaining == 0)
		_streamMode = STREAM_NONE;
		return 0;
		return got;
		}

		// Read from socket
		size_t n = _recvBlocking(_recvScratch);
		if (n == 0) {
		_streamMode = STREAM_NONE;
		return 0;
		}
		size_t take = (std::min)(n, bufsize);
		size_t take = (std::min)({n, want, bufsize});
		std::memcpy(buf, _recvScratch.data.buf, take);
		_streamRemaining -= take;

		// Save excess data for next call
		if (take < n) {
		_streamBuf.assign(_recvScratch.data.buf + take, _recvScratch.data.buf + n);
		_streamBufPos = 0;
		}

		if (_streamRemaining == 0)
		_streamMode = STREAM_NONE;
		return take;
		}

		// ── Chunked / EOF modes ──
		// Shared with readBodyChunkNonBlocking() via one implementation
		// parameterized by `blocking` (see _streamReadChunked/_streamReadEof).
		// (size_t)-1 (their "body complete" signal) is translated to 0 here to
		// match this function's own contract.
		if (_streamMode == STREAM_CHUNKED) {
		size_t n = _streamReadChunked(buf, bufsize, /blocking=/true);
		return (n == static_cast<size_t>(-1)) ? 0 : n;
		}
		if (_streamMode == STREAM_EOF) {
		size_t n = _streamReadEof(buf, bufsize, /blocking=/true);
		return (n == static_cast<size_t>(-1)) ? 0 : n;
		}

		// ── HTTP/2 streaming mode ──
		if (_streamMode == STREAM_H2) {
		// First drain any body data buffered during GetStream
		@@ -2263,8 +2360,17 @@ size_t libhttppp::HttpClient::readBodyChunkNonBlocking(char *buf, size_t bufsize
		return 0;
		}

		// For chunked/EOF modes, fall back to blocking readBodyChunk
		return readBodyChunk(buf, bufsize);
		// Chunked / EOF modes -- shared with readBodyChunk() via one
		// implementation parameterized by `blocking` (see
		// _streamReadChunked/_streamReadEof). Unlike readBodyChunk(), this
		// function's own contract already uses (size_t)-1 for "body complete",
		// so no translation is needed here.
		if (_streamMode == STREAM_CHUNKED)
		return _streamReadChunked(buf, bufsize, /blocking=/false);
		if (_streamMode == STREAM_EOF)
		return _streamReadEof(buf, bufsize, /blocking=/false);

		return static_cast<size_t>(-1);
		}

		static bool is_redirect_status(int code) {
		@@ -3771,6 +3877,19 @@ size_t libhttppp::HttpResponse::parse(const char *data, size_t inlen) {
		throw excep;
		}

		// Discard the default constructor's placeholder headers (content-length:
		// 0, connection: keep-alive) before parsing wire data. Without this, a
		// genuinely EOF-terminated response (no Content-Length, no
		// Transfer-Encoding on the wire) would keep reporting
		// getContentLength()==0 from the leftover default instead of reflecting
		// that no such header was actually present -- misdetected as a
		// zero-length body instead of "read until connection close".
		clear();
		_ContentLength = nullptr;
		_Connection = nullptr;
		_ContentType = nullptr;
		_TransferEncoding = nullptr;

		// --- helpers: trim ---
		auto ltrim_inplace = [](std::string& s) {
		size_t i = 0;

src/http.h

+12 −0

Changes for src/http.h: 12 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -173,6 +173,18 @@ namespace libhttppp {
		// Shared HTTP/1.x response reader (avoids code duplication)
		std::vector<char> _h1ReadResponse(const std::string &label);

		// Shared blocking/non-blocking body readers for STREAM_CHUNKED and
		// STREAM_EOF (used by both readBodyChunk() and
		// readBodyChunkNonBlocking() -- a single implementation parameterized
		// by `blocking` instead of two independently-maintained copies, the
		// second of which used to not exist at all and silently fell back to
		// the blocking one). Return value matches readBodyChunkNonBlocking's
		// contract: 0 = no data yet (blocking=false only) or a benign empty
		// read, (size_t)-1 = body complete, otherwise bytes written to buf.
		bool _streamRecvMore(bool blocking, bool &eof);
		size_t _streamReadChunked(char *buf, size_t bufsize, bool blocking);
		size_t _streamReadEof(char *buf, size_t bufsize, bool blocking);

		// Shared Get/Post/Put/Delete implementation: builds+sends the request
		// over whichever transport is active (H1/H2/H3), follows redirects
		// per RFC 7231 §6.4 (303 always converts to a bodyless GET; 307/308

test/http1_nonblocking_stream_test.cpp

+68 −12

File changed.

Preview size limit exceeded, changes collapsed.