From 83060510015c2ad05b156af54938d05e9b2df5ae 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 | 14 +++++++++++ test/parallel/test-stream-readable-didRead.js | 24 +++++++++++++++++++ 4 files changed, 40 insertions(+), 1 deletion(-) create mode 100644 test/parallel/test-stream-readable-didRead.js diff --git a/lib/_http_incoming.js b/lib/_http_incoming.js index 31b7db6f6c7c99..bb76d6eaefc34d 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 64cd44c066cf8a..8d9321d28b9919 100644 --- a/lib/_http_server.js +++ b/lib/_http_server.js @@ -802,7 +802,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 69f5b52496ad29..c789a51a4b3cdb 100644 --- a/lib/internal/streams/readable.js +++ b/lib/internal/streams/readable.js @@ -171,6 +171,8 @@ function ReadableState(options, stream, isDuplex) { // If true, a maybeReadMore has been scheduled. this.readingMore = false; + this.didRead = false; + this.decoder = null; this.encoding = null; if (options && options.encoding) { @@ -542,6 +544,8 @@ Readable.prototype.read = function(n) { if (ret !== null) this.emit('data', ret); + state.didRead = true; + return ret; }; @@ -845,7 +849,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); } }; @@ -993,6 +999,7 @@ Readable.prototype.resume = function() { function resume(stream, state) { if (!state.resumeScheduled) { state.resumeScheduled = true; + state.didRead = true; process.nextTick(resume_, stream, state); } } @@ -1177,6 +1184,13 @@ ObjectDefineProperties(Readable.prototype, { } }, + readableDidRead: { + enumerable: false, + get: function() { + return this._readableState.didRead; + } + }, + readableHighWaterMark: { enumerable: false, get: function() { diff --git a/test/parallel/test-stream-readable-didRead.js b/test/parallel/test-stream-readable-didRead.js new file mode 100644 index 00000000000000..18e2da97e88e94 --- /dev/null +++ b/test/parallel/test-stream-readable-didRead.js @@ -0,0 +1,24 @@ +'use strict'; +require('../common'); +const assert = require('assert'); +const Readable = require('stream').Readable; + +{ + const readable = new Readable({ + read: () => {} + }); + + assert.strictEqual(readable.readableDidRead, false); + readable.read(); + assert.strictEqual(readable.readableDidRead, true); +} + +{ + const readable = new Readable({ + read: () => {} + }); + + assert.strictEqual(readable.readableDidRead, false); + readable.resume(); + assert.strictEqual(readable.readableDidRead, true); +}