stream: refactor to use validateAbortSignal

PR-URL: #46520
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: James M Snell <[email protected]>

Author: aduh95

lib/internal/abort_controller.js

@@ -394,7 +394,6 @@ ObjectDefineProperty(AbortController.prototype, SymbolToStringTag, {
 });
 
 module.exports = {
-  kAborted,
   AbortController,
   AbortSignal,
   ClonedAbortSignal,

lib/internal/webstreams/readablestream.js

@@ -58,14 +58,11 @@ const {
 } = require('internal/util');
 
 const {
+  validateAbortSignal,
   validateBuffer,
   validateObject,
 } = require('internal/validators');
 
-const {
-  kAborted,
-} = require('internal/abort_controller');
-
 const {
   MessageChannel,
 } = require('internal/worker/io');
@@ -381,8 +378,9 @@ class ReadableStream {
     const preventClose = options?.preventClose;
     const signal = options?.signal;
 
-    if (signal !== undefined && signal?.[kAborted] === undefined)
-      throw new ERR_INVALID_ARG_TYPE('options.signal', 'AbortSignal', signal);
+    if (signal !== undefined) {
+      validateAbortSignal(signal, 'options.signal');
+    }
 
     if (isReadableStreamLocked(this))
       throw new ERR_INVALID_STATE.TypeError('The ReadableStream is locked');
@@ -422,8 +420,9 @@ class ReadableStream {
       const preventClose = options?.preventClose;
       const signal = options?.signal;
 
-      if (signal !== undefined && signal?.[kAborted] === undefined)
-        throw new ERR_INVALID_ARG_TYPE('options.signal', 'AbortSignal', signal);
+      if (signal !== undefined) {
+        validateAbortSignal(signal, 'options.signal');
+      }
 
       if (isReadableStreamLocked(this))
         throw new ERR_INVALID_STATE.TypeError('The ReadableStream is locked');
@@ -1269,12 +1268,12 @@ function readableStreamPipeTo(
 
   let shuttingDown = false;
 
-  if (signal !== undefined && signal?.[kAborted] === undefined) {
-    return PromiseReject(
-      new ERR_INVALID_ARG_TYPE(
-        'options.signal',
-        'AbortSignal',
-        signal));
+  if (signal !== undefined) {
+    try {
+      validateAbortSignal(signal, 'options.signal');
+    } catch (error) {
+      return PromiseReject(error);
+    }
   }
 
   const promise = createDeferredPromise();