Loading src/http/HttpProxyServer.cpp +128 −0 Changes for src/http/HttpProxyServer.cpp: 128 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -603,6 +603,75 @@ void HttpProxyController::RequestEvent(libhttppp::HttpRequest &curreq, const int body = client.Head(outreq); } else { outreq.setRequestType(GETREQUEST); // A GET carrying a Range header is almost always a browser <video>/<audio> // element probing for, or seeking into, a large media file -- both blogi's // and bolli's own media-serving routes implement real Range/206 support (see // media.cpp/bolli.cpp), so this is a reliable, cheap-to-check signal without // needing any route-specific configuration. client.Get() below fully buffers // the entire response into `body` before a single byte reaches the browser -- // fine for ordinary HTML/JSON/CSS/JS traffic (small, and often compressible), // but exactly what made video playback through the proxy slow to start and // unable to play progressively. Stream this response through instead: fetch // just the headers via GetStream() (fast), send those to the client // immediately, and hand the still-open upstream connection off to // ResponseEvent to drain in chunks as the event loop reports this connection // writable -- mirrors blogi/bolli's own identical GetStream()+ResponseEvent // pattern one hop further out. Skips applyResponseCompression entirely: a // byte-range response is inherently uncompressible-as-a-whole (the range // refers to the uncompressed representation), and neither app's media routes // send a Content-Encoding on these responses anyway. if (curreq.getHeaderData("range")) { libhttppp::HttpResponse streamResp = client.GetStream(outreq); int rangeStatus = streamResp.getStatusCode(); libhttppp::HttpResponse curres; std::string reason = streamResp.getState(); curres.setState(forwardedResponseState(rangeStatus, reason)); curres.setContentType(streamResp.getContentType()); // Same idea as kResponseSkip below, minus content-encoding/vary -- // nothing else claims those here since compression is skipped outright, // so they forward through this generic loop like any other header // instead of needing their own special-cased copy. static const std::set<std::string> kRangeResponseSkip = { "content-type", "content-length", "connection", "transfer-encoding", "keep-alive", "upgrade", "trailer", }; for (auto *hd = streamResp.getfirstHeaderData(); hd; hd = hd->nextHeaderData()) { const std::string &key = hd->getkey(); if (key.empty() || key[0] == ':' || kRangeResponseSkip.count(key)) continue; auto *out = curres.setHeaderData(key); for (auto *v = hd->getfirstValue(); v; v = v->nextvalue()) { out->push_back(v->getvalue()); } } size_t rangeContentLength = streamResp.getContentLength(); curres.setContentLength(rangeContentLength); // Headers only -- ResponseEvent below appends the body over however many // further event-loop passes it takes (see HttpResponse::send's datalen=-1 // contract: skips both setContentLength and the body-append branch). curres.send(curreq, "", -1); // Move the checked-out upstream connection into this request's own // extension instead of leaving it in poolGuard -- that RAII release fires // when RequestEvent returns, far too early for a body that ResponseEvent // hasn't finished draining yet. auto *ext = ensureExt(curreq); ext->streaming.client = std::move(poolGuard.client); ext->streaming.pool = poolGuard.pool; ext->streaming.origin = poolGuard.origin; ext->streaming.send_size = rangeContentLength; ext->streaming.produced = 0; poolGuard.pool = nullptr; // detach -- its destructor is now a no-op upstreamOk = rangeStatus < 500; lbGuard.ok = upstreamOk; poolGuard.ok = false; return; } body = client.Get(outreq); } Loading Loading @@ -663,6 +732,65 @@ void HttpProxyController::RequestEvent(libhttppp::HttpRequest &curreq, const int } } void HttpProxyController::ResponseEvent(libhttppp::HttpRequest &curreq, const int /*tid*/, ULONG_PTR /*args*/) { auto *ext = getExt(curreq); if (!ext || !ext->streaming.client) return; // every non-streamed response has no extension at all ProxyStreamingState &streaming = ext->streaming; if (streaming.produced < streaming.send_size) { // 256KB per event and a 64KB read chunk -- same budget blogi/bolli's own identical // ResponseEvent streaming loops use, for the same reason: bound how much of one // event-loop pass a single connection's body-drain can consume. static constexpr size_t MAX_FILL = 262144; size_t filled = 0; while (filled < MAX_FILL && streaming.produced < streaming.send_size) { const size_t remaining = streaming.send_size - streaming.produced; static thread_local char chunk[65536]; const size_t toRead = remaining < sizeof(chunk) ? remaining : sizeof(chunk); size_t n; try { n = streaming.client->readBodyChunkNonBlocking(chunk, toRead); } catch (const std::exception &e) { std::cerr << "[http] streaming range response failed after " << streaming.produced << "/" << streaming.send_size << " bytes: " << e.what() << std::endl; n = static_cast<size_t>(-1); } catch (...) { n = static_cast<size_t>(-1); } if (n == static_cast<size_t>(-1)) { // Upstream connection broke mid-stream -- its state is now unknown, so let it // close (don't hand it back to _connectionPool for a future request to trip // over) rather than trying to salvage it. streaming.client.reset(); streaming.send_size = streaming.produced; return; } if (n > 0) { if (n > remaining) n = remaining; curreq.SendData.append(chunk, n); streaming.produced += n; filled += n; } else { if (filled == 0) curreq.slots[0].ResponsePending.store(true, std::memory_order_release); break; } } } if (streaming.produced >= streaming.send_size) { // Clean finish -- the connection is exactly as reusable as any other completed // request, so give it back to the pool instead of closing it (mirrors // ScopedPoolRelease's ok=true path in RequestEvent above). if (streaming.pool && streaming.client) streaming.pool->release(std::move(streaming.client), streaming.origin); streaming.client.reset(); } } HttpProxyServer::HttpProxyServer(config::ListenerConfig listener, std::string certPath, std::string keyPath, store::ConfigStore &configStore, auth::IAuthService &authService, Loading Loading
src/http/HttpProxyServer.cpp +128 −0 Changes for src/http/HttpProxyServer.cpp: 128 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -603,6 +603,75 @@ void HttpProxyController::RequestEvent(libhttppp::HttpRequest &curreq, const int body = client.Head(outreq); } else { outreq.setRequestType(GETREQUEST); // A GET carrying a Range header is almost always a browser <video>/<audio> // element probing for, or seeking into, a large media file -- both blogi's // and bolli's own media-serving routes implement real Range/206 support (see // media.cpp/bolli.cpp), so this is a reliable, cheap-to-check signal without // needing any route-specific configuration. client.Get() below fully buffers // the entire response into `body` before a single byte reaches the browser -- // fine for ordinary HTML/JSON/CSS/JS traffic (small, and often compressible), // but exactly what made video playback through the proxy slow to start and // unable to play progressively. Stream this response through instead: fetch // just the headers via GetStream() (fast), send those to the client // immediately, and hand the still-open upstream connection off to // ResponseEvent to drain in chunks as the event loop reports this connection // writable -- mirrors blogi/bolli's own identical GetStream()+ResponseEvent // pattern one hop further out. Skips applyResponseCompression entirely: a // byte-range response is inherently uncompressible-as-a-whole (the range // refers to the uncompressed representation), and neither app's media routes // send a Content-Encoding on these responses anyway. if (curreq.getHeaderData("range")) { libhttppp::HttpResponse streamResp = client.GetStream(outreq); int rangeStatus = streamResp.getStatusCode(); libhttppp::HttpResponse curres; std::string reason = streamResp.getState(); curres.setState(forwardedResponseState(rangeStatus, reason)); curres.setContentType(streamResp.getContentType()); // Same idea as kResponseSkip below, minus content-encoding/vary -- // nothing else claims those here since compression is skipped outright, // so they forward through this generic loop like any other header // instead of needing their own special-cased copy. static const std::set<std::string> kRangeResponseSkip = { "content-type", "content-length", "connection", "transfer-encoding", "keep-alive", "upgrade", "trailer", }; for (auto *hd = streamResp.getfirstHeaderData(); hd; hd = hd->nextHeaderData()) { const std::string &key = hd->getkey(); if (key.empty() || key[0] == ':' || kRangeResponseSkip.count(key)) continue; auto *out = curres.setHeaderData(key); for (auto *v = hd->getfirstValue(); v; v = v->nextvalue()) { out->push_back(v->getvalue()); } } size_t rangeContentLength = streamResp.getContentLength(); curres.setContentLength(rangeContentLength); // Headers only -- ResponseEvent below appends the body over however many // further event-loop passes it takes (see HttpResponse::send's datalen=-1 // contract: skips both setContentLength and the body-append branch). curres.send(curreq, "", -1); // Move the checked-out upstream connection into this request's own // extension instead of leaving it in poolGuard -- that RAII release fires // when RequestEvent returns, far too early for a body that ResponseEvent // hasn't finished draining yet. auto *ext = ensureExt(curreq); ext->streaming.client = std::move(poolGuard.client); ext->streaming.pool = poolGuard.pool; ext->streaming.origin = poolGuard.origin; ext->streaming.send_size = rangeContentLength; ext->streaming.produced = 0; poolGuard.pool = nullptr; // detach -- its destructor is now a no-op upstreamOk = rangeStatus < 500; lbGuard.ok = upstreamOk; poolGuard.ok = false; return; } body = client.Get(outreq); } Loading Loading @@ -663,6 +732,65 @@ void HttpProxyController::RequestEvent(libhttppp::HttpRequest &curreq, const int } } void HttpProxyController::ResponseEvent(libhttppp::HttpRequest &curreq, const int /*tid*/, ULONG_PTR /*args*/) { auto *ext = getExt(curreq); if (!ext || !ext->streaming.client) return; // every non-streamed response has no extension at all ProxyStreamingState &streaming = ext->streaming; if (streaming.produced < streaming.send_size) { // 256KB per event and a 64KB read chunk -- same budget blogi/bolli's own identical // ResponseEvent streaming loops use, for the same reason: bound how much of one // event-loop pass a single connection's body-drain can consume. static constexpr size_t MAX_FILL = 262144; size_t filled = 0; while (filled < MAX_FILL && streaming.produced < streaming.send_size) { const size_t remaining = streaming.send_size - streaming.produced; static thread_local char chunk[65536]; const size_t toRead = remaining < sizeof(chunk) ? remaining : sizeof(chunk); size_t n; try { n = streaming.client->readBodyChunkNonBlocking(chunk, toRead); } catch (const std::exception &e) { std::cerr << "[http] streaming range response failed after " << streaming.produced << "/" << streaming.send_size << " bytes: " << e.what() << std::endl; n = static_cast<size_t>(-1); } catch (...) { n = static_cast<size_t>(-1); } if (n == static_cast<size_t>(-1)) { // Upstream connection broke mid-stream -- its state is now unknown, so let it // close (don't hand it back to _connectionPool for a future request to trip // over) rather than trying to salvage it. streaming.client.reset(); streaming.send_size = streaming.produced; return; } if (n > 0) { if (n > remaining) n = remaining; curreq.SendData.append(chunk, n); streaming.produced += n; filled += n; } else { if (filled == 0) curreq.slots[0].ResponsePending.store(true, std::memory_order_release); break; } } } if (streaming.produced >= streaming.send_size) { // Clean finish -- the connection is exactly as reusable as any other completed // request, so give it back to the pool instead of closing it (mirrors // ScopedPoolRelease's ok=true path in RequestEvent above). if (streaming.pool && streaming.client) streaming.pool->release(std::move(streaming.client), streaming.origin); streaming.client.reset(); } } HttpProxyServer::HttpProxyServer(config::ListenerConfig listener, std::string certPath, std::string keyPath, store::ConfigStore &configStore, auth::IAuthService &authService, Loading