From 3bd9a531169bbf49a316a4e8373aca0d52228110 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 2 Aug 2021 13:08:32 +0200
Subject: [PATCH] stream: add isDisturbed helper

Adds a helper util used to determine whether a stream has been
disturbed (read or cancelled).

Refs: https://github.com/nodejs/node/issues/39627
---
 lib/internal/streams/utils.js             | 13 +++++++++++++
 lib/internal/webstreams/readablestream.js |  9 +++++++++
 lib/stream.js                             |  1 +
 3 files changed, 23 insertions(+)

diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js
index 01396b5113340f6..0b7f80ea1b93b6e 100644
--- a/lib/internal/streams/utils.js
+++ b/lib/internal/streams/utils.js
@@ -7,6 +7,7 @@ const {
 } = primordials;
 
 const kDestroyed = Symbol('kDestroyed');
+const kIsDisturbed = Symbol('kIsDisturbed');
 
 function isReadableNodeStream(obj) {
   return !!(
@@ -195,8 +196,20 @@ function willEmitClose(stream) {
   );
 }
 
+function isDisturbed(stream) {
+  const state = stream && stream._readableState;
+  return stream && (
+    stream.readableDidRead ||
+    isDestroyed(stream) ||
+    stream[kIsDisturbed] ||
+    (state && state.endEmitted)
+  );
+}
+
 module.exports = {
   kDestroyed,
+  isDisturbed,
+  kIsDisturbed,
   isClosed,
   isDestroyed,
   isFinished,
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
index 62da68ff28086f9..efe890b7ebedaf6 100644
--- a/lib/internal/webstreams/readablestream.js
+++ b/lib/internal/webstreams/readablestream.js
@@ -80,6 +80,10 @@ const {
   queueMicrotask,
 } = require('internal/process/task_queues');
 
+const {
+  kIsDisturbed,
+} = require('internal/stream/utils');
+
 const {
   ArrayBufferViewGetBuffer,
   ArrayBufferViewGetByteLength,
@@ -200,6 +204,7 @@ class ReadableStream {
         promise: undefined,
       }
     };
+
     // The spec requires handling of the strategy first
     // here. Specifically, if getting the size and
     // highWaterMark from the strategy fail, that has
@@ -232,6 +237,10 @@ class ReadableStream {
     return makeTransferable(this);
   }
 
+  get [kIsDisturbed]() {
+    return this[kState].disturbed;
+  }
+
   /**
    * @readonly
    * @type {boolean}
diff --git a/lib/stream.js b/lib/stream.js
index b84efb0fd8862d5..43f59788f62bc8a 100644
--- a/lib/stream.js
+++ b/lib/stream.js
@@ -38,6 +38,7 @@ const internalBuffer = require('internal/buffer');
 const promises = require('stream/promises');
 
 const Stream = module.exports = require('internal/streams/legacy').Stream;
+Stream.isDisturbed = require('internal/streams/utils').isDisturbed;
 Stream.Readable = require('internal/streams/readable');
 Stream.Writable = require('internal/streams/writable');
 Stream.Duplex = require('internal/streams/duplex');