stream support (6b9d162e) · Commits · jan.koester / libhttppp

src/http.cpp

+204 −0

Original line number	Diff line number	Diff line
		@@ -1297,6 +1297,173 @@ libhttppp::HttpResponse libhttppp::HttpClient::GetStream(libhttppp::HttpRequest
		}
		}

		libhttppp::HttpResponse libhttppp::HttpClient::PostStream(libhttppp::HttpRequest &nreq,
		const std::vector<char> &postBody) {
		try {
		_ensureConnected();

		// Reset any previous streaming state (mirrors GetStream)
		_streamMode = STREAM_NONE;
		_streamBuf.clear();
		_streamBufPos = 0;
		_streamRemaining = 0;
		_streamChunkDone = false;
		_streamChunkRemaining = 0;
		_streamH2Sid = 0;
		_streamH2EndStream = false;
		_streamH2Raw.clear();
		_streamH3Sid = 0;
		_streamH3EndStream = false;
		_streamH3Raw.clear();
		_streamH3Body.clear();
		_streamH3BodyPos = 0;
		_streamH3InDataFrame = false;
		_streamH3DataRemaining = 0;

		if (dynamic_cast<netplus::quic*>(_cltsock.get()) \|\| _isH2) {
		HTTPException he;
		he[HTTPException::Error] << "PostStream: only HTTP/1.1 upstream connections are supported";
		throw he;
		}

		std::stringstream host;
		host << _url.getHost() << ":" << _url.getPort();

		if (!nreq.getHeaderData("host"))
		nreq.setHeaderData("host")->push_back(host.str());
		nreq.setRequestType(POSTREQUEST);

		if (nreq.getRequestURL().empty())
		nreq.setRequestURL(_url.getPath());

		nreq.setHeaderData("content-length")->push_back(std::to_string(postBody.size()));

		if (nreq.getRequestVersion().empty())
		nreq.setRequestVersion(HTTPVERSION(1.1));

		// Send header + body together for best TCP segment utilization
		{
		std::string header;
		nreq.printHeader(header);
		if (!_cltsock \|\| _cltsock->fd() < 0)
		resetConnection();

		std::vector<char> combined;
		combined.reserve(header.size() + postBody.size());
		combined.insert(combined.end(), header.begin(), header.end());
		combined.insert(combined.end(), postBody.begin(), postBody.end());
		_sendAll(combined.data(), combined.size());
		}

		// Read until we have full response header (\r\n\r\n) -- identical to
		// GetStream's HTTP/1.x header-read + streaming-mode detection below.
		std::vector<char> raw;
		raw.reserve(16384);
		size_t header_end = std::string::npos;
		size_t search_from = 0;

		auto header_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);

		for (;;) {
		if (raw.size() >= 4) {
		size_t start = (search_from >= 3) ? search_from - 3 : 0;
		for (size_t i = start; i + 3 < raw.size(); ++i) {
		if (raw[i] == '\r' && raw[i+1] == '\n' &&
		raw[i+2] == '\r' && raw[i+3] == '\n')
		{
		header_end = i + 4;
		break;
		}
		}
		search_from = raw.size();
		}
		if (header_end != std::string::npos)
		break;

		auto now = std::chrono::steady_clock::now();
		if (now >= header_deadline) {
		HTTPException he;
		he[HTTPException::Error] << "HTTP PostStream: timeout while reading header";
		throw he;
		}
		int remaining_sec = static_cast<int>(std::chrono::duration_cast<std::chrono::seconds>(
		header_deadline - now).count());
		if (remaining_sec <= 0)
		remaining_sec = 1;

		netplus::buffer buf(CHUNKSIZE);
		size_t n = _recvBlocking(buf, remaining_sec);
		if (n == 0) {
		netplus::NetException ne;
		ne[netplus::NetException::Error] << "HTTP PostStream: EOF while reading header";
		throw ne;
		}
		raw.insert(raw.end(), buf.data.buf, buf.data.buf + n);
		}

		libhttppp::HttpResponse res;
		size_t parsed_hsize = res.parse(raw.data(), raw.size());

		if (parsed_hsize == 0 \|\| parsed_hsize > raw.size()) {
		libhttppp::HTTPException he;
		he[libhttppp::HTTPException::Error] << "HTTP PostStream: response header parse failed";
		throw he;
		}

		size_t body_off = parsed_hsize;
		if (raw.size() > body_off) {
		_streamBuf.assign(raw.begin() + (ptrdiff_t)body_off, raw.end());
		}
		_streamBufPos = 0;

		bool chunked = false;
		ptrdiff_t content_len = -1;

		try {
		for (libhttppp::HttpHeader::HeaderData::Values *v = res.getTransferEncoding();
		v; v = v->nextvalue())
		{
		std::string val = tolower_copy(v->getvalue());
		if (val == "chunked") {
		chunked = true;
		break;
		}
		}
		} catch (...) {}

		if (!chunked) {
		try {
		content_len = res.getContentLength();
		if (content_len < 0) content_len = -1;
		} catch (...) {
		content_len = -1;
		}
		}

		if (chunked) {
		_streamMode = STREAM_CHUNKED;
		_streamChunkDone = false;
		_streamChunkRemaining = 0;
		} else if (content_len >= 0) {
		_streamMode = STREAM_CONTENT_LENGTH;
		_streamRemaining = (size_t)content_len;
		size_t have = _streamBuf.size() - _streamBufPos;
		if (have > _streamRemaining) {
		_streamBuf.resize(_streamBufPos + _streamRemaining);
		}
		} else {
		_streamMode = STREAM_EOF;
		}

		return res;

		} catch (netplus::NetException &e) {
		libhttppp::HTTPException ee;
		ee[libhttppp::HTTPException::Error] << e.what();
		throw ee;
		}
		}

		size_t libhttppp::HttpClient::readBodyChunk(char *buf, size_t bufsize) {
		if (_streamMode == STREAM_NONE \|\| bufsize == 0)
		return 0;
		@@ -3816,6 +3983,43 @@ void libhttppp::HttpResponse::send(netplus::con &curconnection,const std::vector
		curconnection.SendData.append(data.data(),datalen);
		}

		void libhttppp::HttpResponse::sendChunkedHeaders(netplus::con &curconnection) {
		// No Content-Length for a chunked body -- and _ContentLength would dangle
		// once its HeaderData node is unlinked, so drop the cached pointer too.
		deldata("content-length");
		_ContentLength = nullptr;
		setTransferEncoding("chunked");

		if (auto req = dynamic_cast<libhttppp::HttpRequest>(&curconnection)) {
		const std::string &reqver = req->getRequestVersion();
		if (!reqver.empty()) {
		_Version = reqver;
		}
		}

		std::vector<char> header;
		printHeader(header);
		curconnection.SendData.append(header.data(), header.size());
		}

		void libhttppp::HttpResponse::sendChunk(netplus::con &curconnection, const char *data, size_t len) {
		if (len == 0 \|\| !data)
		return;

		std::stringstream sizeLine;
		sizeLine << std::hex << len << "\r\n";
		std::string prefix = sizeLine.str();

		curconnection.SendData.append(prefix.data(), prefix.size());
		curconnection.SendData.append(data, len);
		curconnection.SendData.append("\r\n", 2);
		}

		void libhttppp::HttpResponse::endChunked(netplus::con &curconnection) {
		static const char terminator[] = "0\r\n\r\n";
		curconnection.SendData.append(terminator, 5);
		}

		size_t libhttppp::HttpResponse::parse(const char *data, size_t inlen) {
		HTTPException excep;

src/http.h

+17 −0

Original line number	Diff line number	Diff line
		@@ -101,6 +101,14 @@ namespace libhttppp {
		// After this call, use readBodyChunk() to read body data incrementally.
		HttpResponse GetStream(HttpRequest &nreq);

		// Same as GetStream, but POSTs a body first (e.g. an OpenAI-compatible
		// "stream": true chat-completions request) before entering streaming-
		// read mode. HTTP/1.1 upstream connections only -- throws HTTPException
		// for HTTP/2 or HTTP/3 connections (no redirect handling either, unlike
		// Post()); local LLM inference backends are plain HTTP/1.1 services, so
		// this deliberately doesn't replicate Post()'s H2/H3/redirect handling.
		HttpResponse PostStream(HttpRequest &nreq, const std::vector<char> &postBody);

		// Read the next chunk of body data (up to bufsize bytes).
		// Returns number of bytes written to buf, 0 when body is complete.
		size_t readBodyChunk(char *buf, size_t bufsize);
		@@ -315,6 +323,15 @@ namespace libhttppp {
		void send(netplus::con &curconnection,const unsigned char *data,int datalen); //only use as server
		void send(netplus::con &curconnection,const std::vector<char> &data,int datalen=0); //only use as server

		// Outbound chunked-transfer-encoding streaming (HTTP/1.1 only -- caller
		// is responsible for only using these on a connection confirmed to be
		// HTTP/1.1; they don't go through _storeResponseInfo's H2/H3 path).
		// Usage: sendChunkedHeaders() once, then sendChunk() any number of
		// times as data becomes available, then endChunked() exactly once.
		void sendChunkedHeaders(netplus::con &curconnection);
		void sendChunk(netplus::con &curconnection, const char *data, size_t len);
		void endChunked(netplus::con &curconnection);

		/client method/
		size_t parse(const char *in,size_t inlen);