http2 streaming (77c28034) · Commits · jan.koester / libhttppp

src/httpd.cpp

+61 −0

Original line number	Diff line number	Diff line
		@@ -455,6 +455,33 @@ void libhttppp::HttpEvent::_dispatchH2Stream(HttpRequest &cureq,
		cureq.SendData.append(out.data(), out.size());
		out.clear();

		// Immediately flush HEADERS to the socket so the browser
		// can start preparing its video player while we stream DATA.
		if (cureq.csock) {
		while (cureq.SendData.pos < cureq.SendData.size()) {
		size_t remaining = cureq.SendData.size() - cureq.SendData.pos;
		size_t sendlen = std::min(remaining, (size_t)BLOCKSIZE);
		netplus::buffer buf(cureq.SendData.data() + cureq.SendData.pos, sendlen);
		try {
		size_t consumed = cureq.csock->sendData(buf, 0);
		if (cureq.csock->hasPendingWrite())
		cureq.csock->flush_out();
		if (consumed > 0)
		cureq.SendData.pos += consumed;
		else
		break;
		} catch (netplus::NetException &e) {
		if (e.getErrorType() == netplus::NetException::Note)
		break;
		throw;
		}
		}
		if (cureq.SendData.pos >= cureq.SendData.size()) {
		cureq.SendData.clear();
		cureq.SendData.pos = 0;
		}
		}

		size_t total_sent = body.size();
		body.clear();
		body.shrink_to_fit();
		@@ -490,6 +517,40 @@ void libhttppp::HttpEvent::_dispatchH2Stream(HttpRequest &cureq,
		total_sent += sa;
		empty_count = 0;
		backoff_ms = 1;

		// Flush SendData to the socket so the browser receives
		// data while we continue reading from upstream.
		// Without this, the entire video buffers in memory
		// before the event loop can write any bytes.
		if (cureq.csock) {
		while (cureq.SendData.pos < cureq.SendData.size()) {
		size_t remaining = cureq.SendData.size() - cureq.SendData.pos;
		size_t sendlen = std::min(remaining, (size_t)BLOCKSIZE);
		netplus::buffer buf(cureq.SendData.data() + cureq.SendData.pos, sendlen);
		try {
		size_t consumed = cureq.csock->sendData(buf, 0);
		if (cureq.csock->hasPendingWrite())
		cureq.csock->flush_out();
		if (consumed > 0) {
		cureq.SendData.pos += consumed;
		} else {
		break;
		}
		} catch (netplus::NetException &e) {
		if (e.getErrorType() == netplus::NetException::Note)
		break; // Socket would block — continue loop
		throw;
		}
		}
		if (cureq.SendData.pos >= cureq.SendData.size()) {
		cureq.SendData.clear();
		cureq.SendData.pos = 0;
		} else if (cureq.SendData.pos > BLOCKSIZE * 4) {
		cureq.SendData.erase(cureq.SendData.begin(),
		cureq.SendData.begin() + cureq.SendData.pos);
		cureq.SendData.pos = 0;
		}
		}
		} else {
		++empty_count;
		if (empty_count > 500) {