stream: add readableEnded

Adds a readableEnded property and improved finished compat with possible
stream-like objects.

PR-URL: https://github.com/nodejs/node/pull/28814
Refs: https://github.com/nodejs/node/issues/28813
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

3 files changed, 13 insertions, 2 deletions

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index 85c365c791..ac37930d12 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -205,6 +205,16 @@ Object.defineProperty(Readable.prototype, 'destroyed', {
   }
 });
 
+Object.defineProperty(Readable.prototype, 'readableEnded', {
+  // Making it explicit this property is not enumerable
+  // because otherwise some prototype manipulation in
+  // userland will fail
+  enumerable: false,
+  get() {
+    return this._readableState ? this._readableState.endEmitted : false;
+  }
+});
+
 Readable.prototype.destroy = destroyImpl.destroy;
 Readable.prototype._undestroy = destroyImpl.undestroy;
 Readable.prototype._destroy = function(err, cb) {
diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
index 89a1dae7fd..0161ca98ef 100644
--- a/lib/internal/streams/async_iterator.js
+++ b/lib/internal/streams/async_iterator.js
@@ -132,7 +132,7 @@ const createReadableStreamAsyncIterator = (stream) => {
     [kLastReject]: { value: null, writable: true },
     [kError]: { value: null, writable: true },
     [kEnded]: {
-      value: stream._readableState.endEmitted,
+      value: stream.readableEnded || stream._readableState.endEmitted,
       writable: true
     },
     // The function passed to new Promise is cached so we avoid allocating a new
diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index 1753996a4f..850e5d3796 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -42,7 +42,8 @@ function eos(stream, opts, callback) {
     if (!readable) callback.call(stream);
   };
 
-  var readableEnded = stream._readableState && stream._readableState.endEmitted;
+  var readableEnded = stream.readableEnded ||
+    (stream._readableState && stream._readableState.endEmitted);
   const onend = () => {
     readable = false;
     readableEnded = true;