stream: use internal abort hooks for web streams

Author: ikeyan

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

@@ -14,7 +14,7 @@ const {
 const {
   isNodeStream,
   isWebStream,
-  kControllerErrorFunction,
+  kControllerAbortFunction,
 } = require('internal/streams/utils');
 
 const { eos } = require('internal/streams/end-of-stream');
@@ -47,7 +47,7 @@ module.exports.addAbortSignalNoValidate = function(signal, stream) {
       stream.destroy(new AbortError(undefined, { cause: signal.reason }));
     } :
     () => {
-      stream[kControllerErrorFunction](new AbortError(undefined, { cause: signal.reason }));
+      stream[kControllerAbortFunction](new AbortError(undefined, { cause: signal.reason }));
     };
   if (signal.aborted) {
     onAbort();

lib/internal/streams/utils.js

@@ -20,7 +20,7 @@ const kIsDisturbed = SymbolFor('nodejs.stream.disturbed');
 const kOnConstructed = Symbol('kOnConstructed');
 
 const kIsClosedPromise = SymbolFor('nodejs.webstream.isClosedPromise');
-const kControllerErrorFunction = SymbolFor('nodejs.webstream.controllerErrorFunction');
+const kControllerAbortFunction = SymbolFor('nodejs.webstream.controllerAbortFunction');
 
 const kState = Symbol('kState');
 const kObjectMode = 1 << 0;
@@ -326,7 +326,7 @@ module.exports = {
   isReadable,
   kIsReadable,
   kIsClosedPromise,
-  kControllerErrorFunction,
+  kControllerAbortFunction,
   kIsWritable,
   isClosed,
   isDuplexNodeStream,

lib/internal/webstreams/readablestream.js

@@ -86,7 +86,7 @@ const {
   kIsErrored,
   kIsReadable,
   kIsClosedPromise,
-  kControllerErrorFunction,
+  kControllerAbortFunction,
 } = require('internal/streams/utils');
 
 const {
@@ -254,7 +254,7 @@ class ReadableStream {
     this[kState] = createReadableStreamState();
 
     this[kIsClosedPromise] = PromiseWithResolvers();
-    this[kControllerErrorFunction] = () => {};
+    this[kControllerAbortFunction] = () => {};
 
     // The spec requires handling of the strategy first
     // here. Specifically, if getting the size and
@@ -2540,7 +2540,8 @@ function setupReadableStreamDefaultController(
     stream,
   };
   stream[kState].controller = controller;
-  stream[kControllerErrorFunction] = FunctionPrototypeBind(controller.error, controller);
+  stream[kControllerAbortFunction] =
+    (error) => readableStreamDefaultControllerError(controller, error);
 
   const startResult = startAlgorithm();
 
@@ -3364,6 +3365,8 @@ function setupReadableByteStreamController(
     pendingPullIntos: [],
   };
   stream[kState].controller = controller;
+  stream[kControllerAbortFunction] =
+    (error) => readableByteStreamControllerError(controller, error);
 
   const startResult = startAlgorithm();
 

lib/internal/webstreams/writablestream.js

@@ -78,7 +78,7 @@ const {
   kIsClosedPromise,
   kIsErrored,
   kIsWritable,
-  kControllerErrorFunction,
+  kControllerAbortFunction,
 } = require('internal/streams/utils');
 
 const {
@@ -173,7 +173,7 @@ class WritableStream {
     this[kState] = createWritableStreamState();
 
     this[kIsClosedPromise] = PromiseWithResolvers();
-    this[kControllerErrorFunction] = () => {};
+    this[kControllerAbortFunction] = () => {};
 
     const size = extractSizeAlgorithm(strategy?.size);
     const highWaterMark = extractHighWaterMark(strategy?.highWaterMark, 1);
@@ -1322,7 +1322,9 @@ function setupWritableStreamDefaultController(
     writeAlgorithm,
   };
   stream[kState].controller = controller;
-  stream[kControllerErrorFunction] = FunctionPrototypeBind(controller.error, controller);
+  stream[kControllerAbortFunction] = (reason) => {
+    setPromiseHandled(writableStreamAbort(stream, reason));
+  };
 
   writableStreamUpdateBackpressure(
     stream,