Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 28 additions & 2 deletions src/js/node/_http_server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,13 @@ const MathFloor = Math.floor;

let cluster;

// diagnostics_channel channels for the HTTP server. Mirrors Node's
// lib/_http_server.js. Inactive channels are no-ops until someone subscribes.
const dc = require("node:diagnostics_channel");
const onRequestStartChannel = dc.channel("http.server.request.start");
const onResponseCreatedChannel = dc.channel("http.server.response.created");
const onResponseFinishChannel = dc.channel("http.server.response.finish");

function emitCloseServer(self: Server) {
callCloseCallback(self);
self.emit("close");
Expand Down Expand Up @@ -661,7 +668,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
if (!requestShouldKeepAlive(http_req)) {
http_res[kMustCloseConnection] = true;
}
http_res.once("finish", endSocketOnFinishIfNeeded.bind(undefined, socket, http_res));
http_res.once("finish", resOnFinish.bind(undefined, http_req, http_res, socket, server));

if (hasObserver("http")) {
startPerf(http_res, kServerResponseStatistics, {
Expand Down Expand Up @@ -758,6 +765,11 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
http_req._dumpAndCloseReadable();
}

// Match Node's parserOnIncoming: publish once, before branching, for
// non-upgrade requests (fires on 503/checkContinue/417/normal paths).
if (!is_upgrade && onRequestStartChannel.hasSubscribers) {
onRequestStartChannel.publish({ request: http_req, response: http_res, socket, server });
}
Comment thread
claude[bot] marked this conversation as resolved.
if (reachedRequestsLimit) {
server.emit("dropRequest", http_req, socket);
http_res.writeHead(503);
Expand Down Expand Up @@ -1519,6 +1531,12 @@ function ServerResponse(req, options): void {
this.statusCode = 200;
this.statusMessage = undefined;
this.chunkedEncoding = false;

// Publish response.created from the constructor (matches Node) so it also
// fires for direct `new ServerResponse(req)` use — light-my-request etc.
if (onResponseCreatedChannel.hasSubscribers) {
onResponseCreatedChannel.publish({ request: req, response: this });
}
Comment thread
robobun marked this conversation as resolved.
}
$toClass(ServerResponse, "ServerResponse", OutgoingMessage);

Expand Down Expand Up @@ -1684,7 +1702,15 @@ function stopServerResponsePerf(this: any) {
}
}

function endSocketOnFinishIfNeeded(socket, res) {
function resOnFinish(req, res, socket, server) {
if (onResponseFinishChannel.hasSubscribers) {
onResponseFinishChannel.publish({
request: req,
response: res,
socket,
server,
});
}
if (res[kMustCloseConnection]) {
socket?.end();
}
Expand Down
37 changes: 30 additions & 7 deletions src/js/node/http2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,13 @@ const onServerStreamStartChannel = dc.channel("http2.server.stream.start");
const onServerStreamErrorChannel = dc.channel("http2.server.stream.error");
const onServerStreamFinishChannel = dc.channel("http2.server.stream.finish");
const onServerStreamCloseChannel = dc.channel("http2.server.stream.close");
// node:_http_server's HTTP server channels, for the allowHTTP1 fallback path.
// response.created is published by the ServerResponse constructor; the other
// two are published in connectionListenerHTTP1 so subscribers see the same
// three events Node fires on this path. Same channel objects as
// node:_http_server (keyed by name in diagnostics_channel's registry).
const onHttp1RequestStartChannel = dc.channel("http.server.request.start");
const onHttp1ResponseFinishChannel = dc.channel("http.server.response.finish");
const { Readable } = Stream;
type Http2ConnectOptions = {
settings?: Settings;
Expand Down Expand Up @@ -5436,6 +5443,7 @@ function createHttp1FallbackResponseHandle(socket, shouldKeepAlive, keepAliveTim
ended: false,
finished: false,
aborted: false,
closeDelimited: false,
bufferedAmount: 0,
shouldKeepAlive,
onfinished: null,
Expand Down Expand Up @@ -5473,15 +5481,15 @@ function createHttp1FallbackResponseHandle(socket, shouldKeepAlive, keepAliveTim
if (chunked && !noBody) socket.write("0\r\n\r\n");
this.ended = true;
this.finished = true;
// A close-delimited body ends at EOF, so the response must end the
// connection; the 'finish' listener in connectionListenerHTTP1 does
// that after the diagnostics publish.
this.closeDelimited = closeDelimited;
const onfinished = this.onfinished;
if (onfinished) {
this.onfinished = null;
onfinished();
}
// A close-delimited body ends at EOF, so the response ends the connection.
if (closeDelimited && !socket.destroyed) {
socket.end();
}
return length;
},
abort() {
Expand Down Expand Up @@ -5562,16 +5570,31 @@ function connectionListenerHTTP1(server, socket, options) {
};

const res = new ServerResponseClass(req);
// Stable reference for the diagnostics closure: the outer `req` is reused
// across pipelined requests on this connection.
const request = req;
const handle = createHttp1FallbackResponseHandle(socket, shouldKeepAlive, keepAliveTimeout);
handle.onfinished = function () {
socket[kHttp1ActiveRequests] = Math.max(0, (socket[kHttp1ActiveRequests] || 1) - 1);
if (!shouldKeepAlive && !socket.destroyed) {
socket.end();
}
};
res[kHttp1ResponseHandle] = handle;
res.assignSocket(socket);

// Like resOnFinish in node:_http_server: publish from the 'finish' event
// (so subscribers observe the finished response) and only then end the
// socket on non-keep-alive / close-delimited responses.
res.once("finish", () => {
if (onHttp1ResponseFinishChannel.hasSubscribers) {
onHttp1ResponseFinishChannel.publish({ request, response: res, socket, server });
}
Comment thread
claude[bot] marked this conversation as resolved.
if ((!shouldKeepAlive || handle.closeDelimited) && !socket.destroyed) {
socket.end();
}
});

if (onHttp1RequestStartChannel.hasSubscribers) {
onHttp1RequestStartChannel.publish({ request, response: res, socket, server });
}
server.emit("request", req, res);
return 0;
};
Expand Down
Loading
Loading