enhance(server): use native AbortController and AbortSignal (#2031)

Author: ardatan

.changeset/metal-moles-care.md

@@ -0,0 +1,7 @@
+---
+'@whatwg-node/server': minor
+'@whatwg-node/node-fetch': patch
+---
+
+- Use native AbortSignal and AbortController for Request.signal
+- Remove custom AbortSignal implementation (ServerAdapterAbortSignal)

packages/node-fetch/src/Request.ts

@@ -73,7 +73,7 @@ export class PonyfillRequest<TJSON = any> extends PonyfillBody<TJSON> implements
     this.redirect = requestInit?.redirect || 'follow';
     this.referrer = requestInit?.referrer || 'about:client';
     this.referrerPolicy = requestInit?.referrerPolicy || 'no-referrer';
-    this._signal = requestInit?.signal;
+    this.signal = requestInit?.signal || new AbortController().signal;
     this.headersSerializer = requestInit?.headersSerializer;
     this.duplex = requestInit?.duplex || 'half';
 
@@ -137,16 +137,7 @@ export class PonyfillRequest<TJSON = any> extends PonyfillBody<TJSON> implements
 
   agent: HTTPAgent | HTTPSAgent | false | undefined;
 
-  private _signal: AbortSignal | undefined | null;
-
-  get signal() {
-    // Create a new signal only if needed
-    // Because the creation of signal is expensive
-    if (!this._signal) {
-      this._signal = new AbortController().signal;
-    }
-    return this._signal!;
-  }
+  signal: AbortSignal;
 
   clone(): PonyfillRequest<TJSON> {
     return this;

packages/node-fetch/src/WritableStream.ts

@@ -44,55 +44,15 @@ export class PonyfillWritableStream<W = any> implements WritableStream<W> {
         },
       });
       this.writable = writable;
-      let onabort: EventListener | null;
-      let reason: any;
+      const abortCtrl = new AbortController();
       const controller: WritableStreamDefaultController = {
-        signal: {
-          any(signals) {
-            return AbortSignal.any([...signals]);
-          },
-          get reason() {
-            return reason;
-          },
-          get aborted() {
-            return writable.destroyed;
-          },
-          addEventListener: (_event: string, eventListener: EventListener) => {
-            writable.once('error', eventListener);
-            writable.once('close', eventListener);
-          },
-          removeEventListener: (_event: string, eventListener: EventListener) => {
-            writable.off('error', eventListener);
-            writable.off('close', eventListener);
-          },
-          dispatchEvent: (_event: Event) => {
-            return false;
-          },
-          get onabort() {
-            return onabort;
-          },
-          set onabort(value) {
-            if (onabort) {
-              this.removeEventListener('abort', onabort);
-            }
-            onabort = value;
-            if (onabort) {
-              this.addEventListener('abort', onabort);
-            }
-          },
-          throwIfAborted() {
-            if (writable.destroyed) {
-              throw reason;
-            }
-          },
-        },
-        error: e => {
-          this.writable.destroy(e);
+        signal: abortCtrl.signal,
+        error(e) {
+          writable.destroy(e);
         },
       };
-      this.writable.once('error', err => {
-        reason = err;
-      });
+      writable.once('error', err => abortCtrl.abort(err));
+      writable.once('close', () => abortCtrl.abort());
     } else {
       this.writable = new Writable();
     }

packages/node-fetch/src/fetchCurl.ts

@@ -30,11 +30,7 @@ export function fetchCurl<TResponseJSON = any, TRequestJSON = any>(
   curlHandle.enable(CurlFeature.StreamResponse);
 
   curlHandle.setStreamProgressCallback(function () {
-    return fetchRequest['_signal']?.aborted
-      ? process.env.DEBUG
-        ? CurlProgressFunc.Continue
-        : 1
-      : 0;
+    return fetchRequest.signal.aborted ? (process.env.DEBUG ? CurlProgressFunc.Continue : 1) : 0;
   });
 
   if (fetchRequest['bodyType'] === 'String') {
@@ -92,17 +88,17 @@ export function fetchCurl<TResponseJSON = any, TRequestJSON = any>(
       }
     }
   }
-  if (fetchRequest['_signal']) {
-    fetchRequest['_signal'].addEventListener('abort', onAbort, { once: true });
+  if (fetchRequest.signal) {
+    fetchRequest.signal.addEventListener('abort', onAbort, { once: true });
   }
   curlHandle.once('end', function endListener() {
     try {
       curlHandle.close();
     } catch (e) {
       deferredPromise.reject(e);
     }
-    if (fetchRequest['_signal']) {
-      fetchRequest['_signal'].removeEventListener('abort', onAbort);
+    if (fetchRequest.signal) {
+      fetchRequest.signal.removeEventListener('abort', onAbort);
     }
   });
   curlHandle.once('error', function errorListener(error: any) {
@@ -127,7 +123,7 @@ export function fetchCurl<TResponseJSON = any, TRequestJSON = any>(
 
       pipeline(stream, outputStream, {
         end: true,
-        signal: fetchRequest['_signal'] ?? undefined,
+        signal: fetchRequest.signal,
       })
         .then(() => {
           if (!stream.destroyed) {

packages/node-fetch/src/fetchNodeHttp.ts

@@ -47,14 +47,14 @@ export function fetchNodeHttp<TResponseJSON = any, TRequestJSON = any>(
         nodeRequest = requestFn(fetchRequest.parsedUrl, {
           method: fetchRequest.method,
           headers: nodeHeaders,
-          signal: fetchRequest['_signal'] ?? undefined,
+          signal: fetchRequest.signal,
           agent: fetchRequest.agent,
         });
       } else {
         nodeRequest = requestFn(fetchRequest.url, {
           method: fetchRequest.method,
           headers: nodeHeaders,
-          signal: fetchRequest['_signal'] ?? undefined,
+          signal: fetchRequest.signal,
           agent: fetchRequest.agent,
         });
       }
@@ -107,7 +107,7 @@ export function fetchNodeHttp<TResponseJSON = any, TRequestJSON = any>(
           }
         }
         pipeline(nodeResponse, outputStream, {
-          signal: fetchRequest['_signal'] ?? undefined,
+          signal: fetchRequest.signal,
           end: true,
         })
           .then(() => {

packages/server/src/createServerAdapter.ts

@@ -32,7 +32,6 @@ import {
   NodeResponse,
   normalizeNodeRequest,
   sendNodeResponse,
-  ServerAdapterRequestAbortSignal,
 } from './utils.js';
 import {
   fakePromise,
@@ -319,7 +318,7 @@ function createServerAdapter<
         ? completeAssign(defaultServerContext, ...ctx)
         : defaultServerContext;
 
-    const signal = new ServerAdapterRequestAbortSignal();
+    const controller = new AbortController();
     const originalResEnd = res.end.bind(res);
     let resEnded = false;
     res.end = function (data: any) {
@@ -328,16 +327,16 @@ function createServerAdapter<
     };
     const originalOnAborted = res.onAborted.bind(res);
     originalOnAborted(function () {
-      signal.sendAbort();
+      controller.abort();
     });
     res.onAborted = function (cb: () => void) {
-      signal.addEventListener('abort', cb);
+      controller.signal.addEventListener('abort', cb, { once: true });
     };
     const request = getRequestFromUWSRequest({
       req,
       res,
       fetchAPI,
-      signal,
+      controller,
     });
     let response$: Response | Promise<Response> | undefined;
     try {
@@ -349,8 +348,8 @@ function createServerAdapter<
       return response$
         .catch((e: any) => handleErrorFromRequestHandler(e, fetchAPI.Response))
         .then(response => {
-          if (!signal.aborted && !resEnded) {
-            return sendResponseToUwsOpts(res, response, signal, fetchAPI);
+          if (!controller.signal.aborted && !resEnded) {
+            return sendResponseToUwsOpts(res, response, controller, fetchAPI);
           }
         })
         .catch(err => {
@@ -360,8 +359,8 @@ function createServerAdapter<
         });
     }
     try {
-      if (!signal.aborted && !resEnded) {
-        return sendResponseToUwsOpts(res, response$, signal, fetchAPI);
+      if (!controller.signal.aborted && !resEnded) {
+        return sendResponseToUwsOpts(res, response$, controller, fetchAPI);
       }
     } catch (err: any) {
       console.error(
@@ -406,13 +405,17 @@ function createServerAdapter<
       if (isRequestInit(initOrCtx)) {
         const request = new fetchAPI.Request(input, initOrCtx);
         const res$ = handleRequestWithWaitUntil(request, ...restOfCtx);
-        return handleAbortSignalAndPromiseResponse(res$, (initOrCtx as RequestInit)?.signal);
+        const signal = (initOrCtx as RequestInit).signal;
+        if (signal) {
+          return handleAbortSignalAndPromiseResponse(res$, signal);
+        }
+        return res$;
       }
       const request = new fetchAPI.Request(input);
       return handleRequestWithWaitUntil(request, ...maybeCtx);
     }
     const res$ = handleRequestWithWaitUntil(input, ...maybeCtx);
-    return handleAbortSignalAndPromiseResponse(res$, (input as any)._signal);
+    return handleAbortSignalAndPromiseResponse(res$, input.signal);
   };
 
   const genericRequestHandler = (