fixup

Author: ronag

doc/api/stream.md

@@ -2046,6 +2046,18 @@ added: REPLACEME
   * `signal` {AbortSignal}
 * Returns: {stream.Readable}
 
+### `stream.Readable.isDisturbed(stream)`
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `stream` {stream.Readable|ReadableStream}
+* Returns: `boolean`
+
+Returns whether the stream has been read from.
+
 ### `stream.Readable.toWeb(streamReadable)`
 <!-- YAML
 added: REPLACEME

lib/internal/streams/readable.js

@@ -1185,7 +1185,7 @@ ObjectDefineProperties(Readable.prototype, {
   readableDidRead: {
     enumerable: false,
     get: function() {
-      return (
+      return !!(
         this._readableState.dataEmitted ||
         this._readableState.endEmitted ||
         this._readableState.errorEmitted ||

lib/internal/streams/utils.js

@@ -198,12 +198,12 @@ function willEmitClose(stream) {
 
 function isDisturbed(stream) {
   const state = stream && stream._readableState;
-  return stream && (
+  return !!(stream && (
     stream.readableDidRead ||
     isDestroyed(stream) ||
     stream[kIsDisturbed] ||
     (state && state.endEmitted)
-  );
+  ));
 }
 
 module.exports = {

lib/internal/webstreams/readablestream.js

@@ -82,7 +82,7 @@ const {
 
 const {
   kIsDisturbed,
-} = require('internal/stream/utils');
+} = require('internal/streams/utils');
 
 const {
   ArrayBufferViewGetBuffer,

test/parallel/test-stream-readable-didRead.js

@@ -1,12 +1,13 @@
 'use strict';
 const common = require('../common');
 const assert = require('assert');
-const Readable = require('stream').Readable;
+const { isDisturbed, Readable } = require('stream');
 
 function noop() {}
 
 function check(readable, data, fn) {
   assert.strictEqual(readable.readableDidRead, false);
+  assert.strictEqual(isDisturbed(readable), false);
   if (data === -1) {
     readable.on('error', common.mustCall());
     readable.on('data', common.mustNotCall());
@@ -28,6 +29,7 @@ function check(readable, data, fn) {
   fn();
   setImmediate(() => {
     assert.strictEqual(readable.readableDidRead, true);
+    assert.strictEqual(isDisturbed(readable), true);
   });
 }
 

test/parallel/test-whatwg-readablestream.js

@@ -2,6 +2,7 @@
 'use strict';
 
 const common = require('../common');
+const { isDisturbed } = require('stream');
 const assert = require('assert');
 const {
   isPromise,
@@ -1520,3 +1521,52 @@ class Source {
   readableByteStreamControllerClose(controller);
   readableByteStreamControllerEnqueue(controller);
 }
+
+{
+  const stream = new ReadableStream({
+    start(controller) {
+      controller.enqueue('a');
+      controller.close();
+    },
+    pull: common.mustNotCall(),
+  });
+
+  const reader = stream.getReader();
+  (async () => {
+    isDisturbed(stream, false);
+    await reader.read();
+    isDisturbed(stream, true);
+  })().then(common.mustCall());
+}
+
+{
+  const stream = new ReadableStream({
+    start(controller) {
+      controller.close();
+    },
+    pull: common.mustNotCall(),
+  });
+
+  const reader = stream.getReader();
+  (async () => {
+    isDisturbed(stream, false);
+    await reader.read();
+    isDisturbed(stream, true);
+  })().then(common.mustCall());
+}
+
+{
+  const stream = new ReadableStream({
+    start(controller) {
+    },
+    pull: common.mustNotCall(),
+  });
+  stream.cancel();
+
+  const reader = stream.getReader();
+  (async () => {
+    isDisturbed(stream, false);
+    await reader.read();
+    isDisturbed(stream, true);
+  })().then(common.mustCall());
+}