diff --git a/lib/_http_incoming.js b/lib/_http_incoming.js index 3961b583de9ddc..fa7f314ba80040 100644 --- a/lib/_http_incoming.js +++ b/lib/_http_incoming.js @@ -88,6 +88,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 7f6876599cc7fc..9ed171dfced9dd 100644 --- a/lib/internal/streams/readable.js +++ b/lib/internal/streams/readable.js @@ -167,6 +167,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) { @@ -520,6 +522,8 @@ Readable.prototype.read = function(n) { if (ret !== null) this.emit('data', ret); + state.didRead = true; + return ret; }; @@ -823,7 +827,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); } }; @@ -971,6 +977,7 @@ Readable.prototype.resume = function() { function resume(stream, state) { if (!state.resumeScheduled) { state.resumeScheduled = true; + state.didRead = true; process.nextTick(resume_, stream, state); } } @@ -1181,6 +1188,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); +}