stream: move webstream abort hooks onto prototypes

Author: ikeyan

lib/internal/webstreams/readablestream.js

@@ -254,7 +254,6 @@ class ReadableStream {
     this[kState] = createReadableStreamState();
 
     this[kIsClosedPromise] = PromiseWithResolvers();
-    this[kControllerAbortFunction] = () => {};
 
     // The spec requires handling of the strategy first
     // here. Specifically, if getting the size and
@@ -284,6 +283,20 @@ class ReadableStream {
     }
   }
 
+  [kControllerAbortFunction](error) {
+    if (!isReadableStream(this))
+      throw new ERR_INVALID_THIS('ReadableStream');
+
+    const controller = this[kState].controller;
+    if (isReadableStreamDefaultController(controller)) {
+      readableStreamDefaultControllerError(controller, error);
+      return;
+    }
+
+    assert(isReadableByteStreamController(controller));
+    readableByteStreamControllerError(controller, error);
+  }
+
   get [kIsDisturbed]() {
     return this[kState].disturbed;
   }
@@ -2540,8 +2553,6 @@ function setupReadableStreamDefaultController(
     stream,
   };
   stream[kState].controller = controller;
-  stream[kControllerAbortFunction] =
-    (error) => readableStreamDefaultControllerError(controller, error);
 
   const startResult = startAlgorithm();
 
@@ -3365,8 +3376,6 @@ function setupReadableByteStreamController(
     pendingPullIntos: [],
   };
   stream[kState].controller = controller;
-  stream[kControllerAbortFunction] =
-    (error) => readableByteStreamControllerError(controller, error);
 
   const startResult = startAlgorithm();
 

lib/internal/webstreams/writablestream.js

@@ -173,7 +173,6 @@ class WritableStream {
     this[kState] = createWritableStreamState();
 
     this[kIsClosedPromise] = PromiseWithResolvers();
-    this[kControllerAbortFunction] = () => {};
 
     const size = extractSizeAlgorithm(strategy?.size);
     const highWaterMark = extractHighWaterMark(strategy?.highWaterMark, 1);
@@ -185,6 +184,12 @@ class WritableStream {
       size);
   }
 
+  [kControllerAbortFunction](reason) {
+    if (!isWritableStream(this))
+      throw new ERR_INVALID_THIS('WritableStream');
+    setPromiseHandled(writableStreamAbort(this, reason));
+  }
+
   get [kIsErrored]() {
     return this[kState].state === 'errored';
   }
@@ -1322,9 +1327,6 @@ function setupWritableStreamDefaultController(
     writeAlgorithm,
   };
   stream[kState].controller = controller;
-  stream[kControllerAbortFunction] = (reason) => {
-    setPromiseHandled(writableStreamAbort(stream, reason));
-  };
 
   writableStreamUpdateBackpressure(
     stream,

test/parallel/test-webstreams-abort-controller.js

@@ -171,7 +171,7 @@ function assertWritableStreamEventuallyAborted(ws, writer, {
 {
   /** @member {import('internal/webstreams/readablestream').ReadableStreamDefaultController} */
   let controller;
-  let pullPromiseWithResolvers = Promise.withResolvers();
+  const pullPromiseWithResolvers = Promise.withResolvers();
   const rs = new ReadableStream({
     start(c) { controller = c; },
     pull() { return pullPromiseWithResolvers.promise; },