From 9bd74924ad5781925a244abb88fa846b9dec925f Mon Sep 17 00:00:00 2001 From: Robert Nagy Date: Wed, 6 Jan 2021 21:35:01 +0100 Subject: [PATCH] stream: add readableDidRead Adds readableDidRead to streams and applies usage to http, http2 and quic. PR-URL: https://github.com/nodejs/node/pull/36820 Reviewed-By: James M Snell Reviewed-By: Matteo Collina Reviewed-By: Benjamin Gruenbaum --- lib/_http_incoming.js | 1 + lib/_http_server.js | 2 +- lib/internal/streams/readable.js | 5 +++++ 3 files changed, 7 insertions(+), 1 deletion(-) diff --git a/lib/_http_incoming.js b/lib/_http_incoming.js index e0f1354e6c969c..071721cefd23b7 100644 --- a/lib/_http_incoming.js +++ b/lib/_http_incoming.js @@ -87,6 +87,7 @@ function IncomingMessage(socket) { this.statusMessage = null; this.client = socket; + // TODO: Deprecate and remove. this._consuming = false; // Flag for when we decide that this message cannot possibly be // read by the user, so there's no point continuing to handle it. diff --git a/lib/_http_server.js b/lib/_http_server.js index 11119169d56f2c..02a9ecb6e69efd 100644 --- a/lib/_http_server.js +++ b/lib/_http_server.js @@ -804,7 +804,7 @@ function resOnFinish(req, res, socket, state, server) { // If the user never called req.read(), and didn't pipe() or // .resume() or .on('data'), then we call req._dump() so that the // bytes will be pulled off the wire. - if (!req._consuming && !req._readableState.resumeScheduled) + if (!req.readableDidRead) req._dump(); // Make sure the requestTimeout is cleared before finishing. diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js index 03953c37eac280..82e621826d22af 100644 --- a/lib/internal/streams/readable.js +++ b/lib/internal/streams/readable.js @@ -527,6 +527,8 @@ Readable.prototype.read = function(n) { this.emit('data', ret); } + state.didRead = true; + return ret; }; @@ -830,7 +832,9 @@ function pipeOnDrain(src, dest) { if ((!state.awaitDrainWriters || state.awaitDrainWriters.size === 0) && EE.listenerCount(src, 'data')) { + // TODO(ronag): Call resume() instead? state.flowing = true; + state.didRead = true; flow(src); } }; @@ -978,6 +982,7 @@ Readable.prototype.resume = function() { function resume(stream, state) { if (!state.resumeScheduled) { state.resumeScheduled = true; + state.didRead = true; process.nextTick(resume_, stream, state); } }