Revert "stream: add abort signal for ReadableStream and WritableStream"

doc/api/stream.md

@@ -3228,24 +3228,17 @@ readable.getReader().read().then((result) => {
 
 <!-- YAML
 added: v15.4.0
-changes:
-  - version: v19.7.0
-    pr-url: https://github.com/nodejs/node/pull/46273
-    description: Added support for `ReadableStream` and
-                 `WritableStream`.
 -->
 
 * `signal` {AbortSignal} A signal representing possible cancellation
-* `stream` {Stream|ReadableStream|WritableStream}
-
-A stream to attach a signal to.
+* `stream` {Stream} a stream to attach a signal to
 
 Attaches an AbortSignal to a readable or writeable stream. This lets code
 control stream destruction using an `AbortController`.
 
 Calling `abort` on the `AbortController` corresponding to the passed
 `AbortSignal` will behave the same way as calling `.destroy(new AbortError())`
-on the stream, and `controller.error(new AbortError())` for webstreams.
+on the stream.
 
 ```js
 const fs = require('node:fs');
@@ -3259,61 +3252,6 @@ const read = addAbortSignal(
 controller.abort();
 ```
 
-Or using an `AbortSignal` with a readable stream as an async iterable:
-
-```js
-const controller = new AbortController();
-setTimeout(() => controller.abort(), 10_000); // set a timeout
-const stream = addAbortSignal(
-  controller.signal,
-  fs.createReadStream(('object.json')),
-);
-(async () => {
-  try {
-    for await (const chunk of stream) {
-      await process(chunk);
-    }
-  } catch (e) {
-    if (e.name === 'AbortError') {
-      // The operation was cancelled
-    } else {
-      throw e;
-    }
-  }
-})();
-```
-
-Or using an `AbortSignal` with a ReadableStream:
-
-```js
-const controller = new AbortController();
-const rs = new ReadableStream({
-  start(controller) {
-    controller.enqueue('hello');
-    controller.enqueue('world');
-    controller.close();
-  },
-});
-
-addAbortSignal(controller.signal, rs);
-
-finished(rs, (err) => {
-  if (err) {
-    if (err.name === 'AbortError') {
-      // The operation was cancelled
-    }
-  }
-});
-
-const reader = rs.getReader();
-
-reader.read().then(({ value, done }) => {
-  console.log(value); // hello
-  console.log(done); // false
-  controller.abort();
-});
-```
-
 ## API for stream implementers
 
 <!--type=misc-->

lib/internal/streams/add-abort-signal.js

@@ -5,12 +5,6 @@ const {
   codes,
 } = require('internal/errors');
 
-const {
-  isNodeStream,
-  isWebStream,
-  kControllerErrorFunction,
-} = require('internal/streams/utils');
-
 const eos = require('internal/streams/end-of-stream');
 const { ERR_INVALID_ARG_TYPE } = codes;
 
@@ -24,25 +18,24 @@ const validateAbortSignal = (signal, name) => {
   }
 };
 
+function isNodeStream(obj) {
+  return !!(obj && typeof obj.pipe === 'function');
+}
+
 module.exports.addAbortSignal = function addAbortSignal(signal, stream) {
   validateAbortSignal(signal, 'signal');
-  if (!isNodeStream(stream) && !isWebStream(stream)) {
-    throw new ERR_INVALID_ARG_TYPE('stream', ['ReadableStream', 'WritableStream', 'Stream'], stream);
+  if (!isNodeStream(stream)) {
+    throw new ERR_INVALID_ARG_TYPE('stream', 'stream.Stream', stream);
   }
   return module.exports.addAbortSignalNoValidate(signal, stream);
 };
-
 module.exports.addAbortSignalNoValidate = function(signal, stream) {
   if (typeof signal !== 'object' || !('aborted' in signal)) {
     return stream;
   }
-  const onAbort = isNodeStream(stream) ?
-    () => {
-      stream.destroy(new AbortError(undefined, { cause: signal.reason }));
-    } :
-    () => {
-      stream[kControllerErrorFunction](new AbortError(undefined, { cause: signal.reason }));
-    };
+  const onAbort = () => {
+    stream.destroy(new AbortError(undefined, { cause: signal.reason }));
+  };
   if (signal.aborted) {
     onAbort();
   } else {

lib/internal/streams/utils.js

@@ -13,7 +13,6 @@ const kIsReadable = Symbol('kIsReadable');
 const kIsDisturbed = Symbol('kIsDisturbed');
 
 const kIsClosedPromise = SymbolFor('nodejs.webstream.isClosedPromise');
-const kControllerErrorFunction = SymbolFor('nodejs.webstream.controllerErrorFunction');
 
 function isReadableNodeStream(obj, strict = false) {
   return !!(
@@ -306,7 +305,6 @@ module.exports = {
   isReadable,
   kIsReadable,
   kIsClosedPromise,
-  kControllerErrorFunction,
   isClosed,
   isDestroyed,
   isDuplexNodeStream,

lib/internal/webstreams/readablestream.js

@@ -83,7 +83,6 @@ const {
   kIsErrored,
   kIsReadable,
   kIsClosedPromise,
-  kControllerErrorFunction,
 } = require('internal/streams/utils');
 
 const {
@@ -261,7 +260,6 @@ class ReadableStream {
     };
 
     this[kIsClosedPromise] = createDeferredPromise();
-    this[kControllerErrorFunction] = () => {};
 
     // The spec requires handling of the strategy first
     // here. Specifically, if getting the size and
@@ -1893,6 +1891,7 @@ function readableStreamClose(stream) {
   assert(stream[kState].state === 'readable');
   stream[kState].state = 'closed';
   stream[kIsClosedPromise].resolve();
+
   const {
     reader,
   } = stream[kState];
@@ -2331,7 +2330,6 @@ function setupReadableStreamDefaultController(
     stream,
   };
   stream[kState].controller = controller;
-  stream[kControllerErrorFunction] = FunctionPrototypeBind(controller.error, controller);
 
   const startResult = startAlgorithm();
 

lib/internal/webstreams/writablestream.js

@@ -71,7 +71,6 @@ const {
 
 const {
   kIsClosedPromise,
-  kControllerErrorFunction,
 } = require('internal/streams/utils');
 
 const {
@@ -200,7 +199,6 @@ class WritableStream {
     };
 
     this[kIsClosedPromise] = createDeferredPromise();
-    this[kControllerErrorFunction] = () => {};
 
     const size = extractSizeAlgorithm(strategy?.size);
     const highWaterMark = extractHighWaterMark(strategy?.highWaterMark, 1);
@@ -372,7 +370,6 @@ function TransferredWritableStream() {
         },
       };
       this[kIsClosedPromise] = createDeferredPromise();
-      this[kControllerErrorFunction] = () => {};
     },
     [], WritableStream));
 }
@@ -1285,7 +1282,6 @@ function setupWritableStreamDefaultController(
     writeAlgorithm,
   };
   stream[kState].controller = controller;
-  stream[kControllerErrorFunction] = FunctionPrototypeBind(controller.error, controller);
 
   writableStreamUpdateBackpressure(
     stream,