Skip to content

Commit

Permalink
stream: add readableDidRead if has been read from
Browse files Browse the repository at this point in the history
Adds did read accessor used to determine whether a readable has been
read from.

PR-URL: #39589
Refs: nodejs/undici#907
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
  • Loading branch information
ronag authored and targos committed Sep 4, 2021
1 parent 989c204 commit a62d4d6
Show file tree
Hide file tree
Showing 3 changed files with 133 additions and 1 deletion.
11 changes: 11 additions & 0 deletions doc/api/stream.md
Expand Up @@ -1211,6 +1211,17 @@ added: v11.4.0
Is `true` if it is safe to call [`readable.read()`][stream-read], which means
the stream has not been destroyed or emitted `'error'` or `'end'`.

##### `readable.readableDidRead`
<!-- YAML
added: REPLACEME
-->

* {boolean}

Allows determining if the stream has been or is about to be read.
Returns true if `'data'`, `'end'`, `'error'` or `'close'` has been
emitted.

##### `readable.readableEncoding`
<!-- YAML
added: v12.7.0
Expand Down
19 changes: 18 additions & 1 deletion lib/internal/streams/readable.js
Expand Up @@ -154,6 +154,8 @@ function ReadableState(options, stream, isDuplex) {
// If true, a maybeReadMore has been scheduled.
this.readingMore = false;

this.dataEmitted = false;

this.decoder = null;
this.encoding = null;
if (options && options.encoding) {
Expand Down Expand Up @@ -287,6 +289,7 @@ function addChunk(stream, state, chunk, addToFront) {
} else {
state.awaitDrainWriters = null;
}
state.dataEmitted = true;
stream.emit('data', chunk);
} else {
// Update the buffer info.
Expand Down Expand Up @@ -496,8 +499,10 @@ Readable.prototype.read = function(n) {
endReadable(this);
}

if (ret !== null)
if (ret !== null) {
state.dataEmitted = true;
this.emit('data', ret);
}

return ret;
};
Expand Down Expand Up @@ -1169,6 +1174,18 @@ ObjectDefineProperties(Readable.prototype, {
}
},

readableDidRead: {
enumerable: false,
get: function() {
return (
this._readableState.dataEmitted ||
this._readableState.endEmitted ||
this._readableState.errorEmitted ||
this._readableState.closeEmitted
);
}
},

readableHighWaterMark: {
enumerable: false,
get: function() {
Expand Down
104 changes: 104 additions & 0 deletions test/parallel/test-stream-readable-didRead.js
@@ -0,0 +1,104 @@
'use strict';
const common = require('../common');
const assert = require('assert');
const Readable = require('stream').Readable;

function noop() {}

function check(readable, data, fn) {
assert.strictEqual(readable.readableDidRead, false);
if (data === -1) {
readable.on('error', common.mustCall());
readable.on('data', common.mustNotCall());
readable.on('end', common.mustNotCall());
} else {
readable.on('error', common.mustNotCall());
if (data === -2) {
readable.on('end', common.mustNotCall());
} else {
readable.on('end', common.mustCall());
}
if (data > 0) {
readable.on('data', common.mustCallAtLeast(data));
} else {
readable.on('data', common.mustNotCall());
}
}
readable.on('close', common.mustCall());
fn();
setImmediate(() => {
assert.strictEqual(readable.readableDidRead, true);
});
}

{
const readable = new Readable({
read() {
this.push(null);
}
});
check(readable, 0, () => {
readable.read();
});
}

{
const readable = new Readable({
read() {
this.push(null);
}
});
check(readable, 0, () => {
readable.resume();
});
}

{
const readable = new Readable({
read() {
this.push(null);
}
});
check(readable, -2, () => {
readable.destroy();
});
}

{
const readable = new Readable({
read() {
this.push(null);
}
});

check(readable, -1, () => {
readable.destroy(new Error());
});
}

{
const readable = new Readable({
read() {
this.push('data');
this.push(null);
}
});

check(readable, 1, () => {
readable.on('data', noop);
});
}

{
const readable = new Readable({
read() {
this.push('data');
this.push(null);
}
});

check(readable, 1, () => {
readable.on('data', noop);
readable.off('data', noop);
});
}

0 comments on commit a62d4d6

Please sign in to comment.