Skip to content

Commit 317fa3a

Browse files
ronagTrott
authored andcommitted
stream: add readableEnded
Adds a readableEnded property and improved finished compat with possible stream-like objects. PR-URL: #28814 Refs: #28813 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Rich Trott <rtrott@gmail.com>
1 parent 0d7acfa commit 317fa3a

File tree

6 files changed

+64
-2
lines changed

6 files changed

+64
-2
lines changed

doc/api/stream.md

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1129,6 +1129,15 @@ added: v12.7.0
11291129
Getter for the property `encoding` of a given `Readable` stream. The `encoding`
11301130
property can be set using the [`readable.setEncoding()`][] method.
11311131

1132+
##### readable.readableEnded
1133+
<!-- YAML
1134+
added: REPLACEME
1135+
-->
1136+
1137+
* {boolean}
1138+
1139+
Becomes `true` when [`'end'`][] event is emitted.
1140+
11321141
##### readable.readableHighWaterMark
11331142
<!-- YAML
11341143
added: v9.3.0

lib/_stream_readable.js

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -205,6 +205,16 @@ Object.defineProperty(Readable.prototype, 'destroyed', {
205205
}
206206
});
207207

208+
Object.defineProperty(Readable.prototype, 'readableEnded', {
209+
// Making it explicit this property is not enumerable
210+
// because otherwise some prototype manipulation in
211+
// userland will fail
212+
enumerable: false,
213+
get() {
214+
return this._readableState ? this._readableState.endEmitted : false;
215+
}
216+
});
217+
208218
Readable.prototype.destroy = destroyImpl.destroy;
209219
Readable.prototype._undestroy = destroyImpl.undestroy;
210220
Readable.prototype._destroy = function(err, cb) {

lib/internal/streams/async_iterator.js

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,7 @@ const createReadableStreamAsyncIterator = (stream) => {
132132
[kLastReject]: { value: null, writable: true },
133133
[kError]: { value: null, writable: true },
134134
[kEnded]: {
135-
value: stream._readableState.endEmitted,
135+
value: stream.readableEnded || stream._readableState.endEmitted,
136136
writable: true
137137
},
138138
// The function passed to new Promise is cached so we avoid allocating a new

lib/internal/streams/end-of-stream.js

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,8 @@ function eos(stream, opts, callback) {
4242
if (!readable) callback.call(stream);
4343
};
4444

45-
var readableEnded = stream._readableState && stream._readableState.endEmitted;
45+
var readableEnded = stream.readableEnded ||
46+
(stream._readableState && stream._readableState.endEmitted);
4647
const onend = () => {
4748
readable = false;
4849
readableEnded = true;

test/parallel/test-stream-finished.js

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
const common = require('../common');
44
const { Writable, Readable, Transform, finished } = require('stream');
55
const assert = require('assert');
6+
const EE = require('events');
67
const fs = require('fs');
78
const { promisify } = require('util');
89

@@ -175,3 +176,11 @@ const { promisify } = require('util');
175176
rs.push(null);
176177
rs.resume();
177178
}
179+
180+
{
181+
const streamLike = new EE();
182+
streamLike.readableEnded = true;
183+
streamLike.readable = true;
184+
finished(streamLike, common.mustCall);
185+
streamLike.emit('close');
186+
}
Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
'use strict';
2+
3+
const common = require('../common');
4+
const { Readable } = require('stream');
5+
const assert = require('assert');
6+
7+
// basic
8+
{
9+
// Find it on Readable.prototype
10+
assert(Readable.prototype.hasOwnProperty('readableEnded'));
11+
}
12+
13+
// event
14+
{
15+
const readable = new Readable();
16+
17+
readable._read = () => {
18+
// The state ended should start in false.
19+
assert.strictEqual(readable.readableEnded, false);
20+
readable.push('asd');
21+
assert.strictEqual(readable.readableEnded, false);
22+
readable.push(null);
23+
assert.strictEqual(readable.readableEnded, false);
24+
};
25+
26+
readable.on('end', common.mustCall(() => {
27+
assert.strictEqual(readable.readableEnded, true);
28+
}));
29+
30+
readable.on('data', common.mustCall(() => {
31+
assert.strictEqual(readable.readableEnded, false);
32+
}));
33+
}

0 commit comments

Comments
 (0)