Skip to content

Commit

Permalink
stream: add readableDidRead
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.

Refs: nodejs/undici#907
  • Loading branch information
ronag committed Jul 30, 2021
1 parent e605c86 commit b8f5fb9
Show file tree
Hide file tree
Showing 3 changed files with 138 additions and 1 deletion.
11 changes: 11 additions & 0 deletions doc/api/stream.md
Original file line number Diff line number Diff line change
Expand Up @@ -1222,6 +1222,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
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,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 @@ -314,6 +316,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 @@ -539,8 +542,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 @@ -1177,6 +1182,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
109 changes: 109 additions & 0 deletions test/parallel/test-stream-readable-didRead.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
'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);
assert.strictEqual(readable.readableUsed, false);
if (data === -1) {
readable.on('error', common.mustCall());
readable.on('data', common.mustNotCall());
assert.strictEqual(readable.readableUsed, true);
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));
assert.strictEqual(readable.readableUsed, true);
} else {
readable.on('data', common.mustNotCall());
assert.strictEqual(readable.readableUsed, true);
}
}
readable.on('close', common.mustCall());
fn();
setImmediate(() => {
assert.strictEqual(readable.readableDidRead, true);
assert.strictEqual(readable.readableUsed, 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 b8f5fb9

Please sign in to comment.