feat: implement Request.signal

.changeset/spotty-cougars-yawn.md

@@ -0,0 +1,6 @@
+---
+'@whatwg-node/node-fetch': patch
+'@whatwg-node/server': patch
+---
+
+Handle AbortSignal

packages/node-fetch/src/fetchCurl.ts

@@ -1,4 +1,5 @@
 import { Readable } from 'node:stream';
+import { PonyfillAbortError } from './AbortError.js';
 import { PonyfillRequest } from './Request.js';
 import { PonyfillResponse } from './Response.js';
 import { defaultHeadersSerializer } from './utils.js';
@@ -10,12 +11,6 @@ export function fetchCurl<TResponseJSON = any, TRequestJSON = any>(
 
   const curlHandle = new Curl();
 
-  if (fetchRequest['_signal']) {
-    fetchRequest['_signal'].onabort = () => {
-      curlHandle.pause(CurlPause.Recv);
-    };
-  }
-
   curlHandle.enable(CurlFeature.NoDataParsing);
 
   curlHandle.setOpt('URL', fetchRequest.url);
@@ -77,6 +72,17 @@ export function fetchCurl<TResponseJSON = any, TRequestJSON = any>(
   curlHandle.enable(CurlFeature.NoHeaderParsing);
 
   return new Promise(function promiseResolver(resolve, reject) {
+    let streamResolved = false;
+    if (fetchRequest['_signal']) {
+      fetchRequest['_signal'].onabort = () => {
+        if (streamResolved) {
+          curlHandle.pause(CurlPause.Recv);
+        } else {
+          reject(new PonyfillAbortError());
+          curlHandle.close();
+        }
+      };
+    }
     curlHandle.once('end', function endListener() {
       curlHandle.close();
     });
@@ -115,6 +121,7 @@ export function fetchCurl<TResponseJSON = any, TRequestJSON = any>(
           url: fetchRequest.url,
         }),
       );
+      streamResolved = true;
     });
     curlHandle.perform();
   });

packages/server/src/createServerAdapter.ts

@@ -27,6 +27,7 @@ import {
   NodeResponse,
   normalizeNodeRequest,
   sendNodeResponse,
+  ServerAdapterRequestAbortSignal,
 } from './utils.js';
 import {
   getRequestFromUWSRequest,
@@ -244,6 +245,7 @@ function createServerAdapter<
     let resAborted = false;
     res.onAborted(() => {
       resAborted = true;
+      (request.signal as ServerAdapterRequestAbortSignal).sendAbort();
     });
     let response$: Response | Promise<Response> | undefined;
     try {

packages/server/src/utils.ts

@@ -23,6 +23,7 @@ export interface NodeRequest {
   raw?: IncomingMessage | Http2ServerRequest;
   socket?: Socket;
   query?: any;
+  once?(event: string, listener: (...args: any[]) => void): void;
 }
 
 export type NodeResponse = ServerResponse | Http2ServerResponse;
@@ -80,6 +81,36 @@ function isRequestBody(body: any): body is BodyInit {
   return false;
 }
 
+export class ServerAdapterRequestAbortSignal extends EventTarget implements AbortSignal {
+  aborted = false;
+  _onabort: ((this: AbortSignal, ev: Event) => any) | null = null;
+  reason: any;
+
+  throwIfAborted(): void {
+    if (this.aborted) {
+      throw new DOMException('Aborted', 'AbortError');
+    }
+  }
+
+  sendAbort() {
+    this.aborted = true;
+    this.dispatchEvent(new Event('abort'));
+  }
+
+  get onabort() {
+    return this._onabort;
+  }
+
+  set onabort(value) {
+    this._onabort = value;
+    if (value) {
+      this.addEventListener('abort', value);
+    } else {
+      this.removeEventListener('abort', value);
+    }
+  }
+}
+
 export function normalizeNodeRequest(
   nodeRequest: NodeRequest,
   RequestCtor: typeof Request,
@@ -94,10 +125,18 @@ export function normalizeNodeRequest(
     fullUrl = url.toString();
   }
 
+  const signal = new ServerAdapterRequestAbortSignal();
+
+  if (rawRequest.once) {
+    rawRequest.once('end', () => signal.sendAbort());
+    rawRequest.once('close', () => signal.sendAbort());
+  }
+
   if (nodeRequest.method === 'GET' || nodeRequest.method === 'HEAD') {
     return new RequestCtor(fullUrl, {
       method: nodeRequest.method,
       headers: nodeRequest.headers,
+      signal,
     });
   }
 
@@ -114,11 +153,13 @@ export function normalizeNodeRequest(
         method: nodeRequest.method,
         headers: nodeRequest.headers,
         body: maybeParsedBody,
+        signal,
       });
     }
     const request = new RequestCtor(fullUrl, {
       method: nodeRequest.method,
       headers: nodeRequest.headers,
+      signal,
     });
     if (!request.headers.get('content-type')?.includes('json')) {
       request.headers.set('content-type', 'application/json; charset=utf-8');
@@ -142,6 +183,7 @@ export function normalizeNodeRequest(
     method: nodeRequest.method,
     headers: nodeRequest.headers,
     body: rawRequest as any,
+    signal,
   });
 }
 

packages/server/src/uwebsockets.ts

@@ -1,5 +1,6 @@
 import type { Readable } from 'node:stream';
 import type { FetchAPI } from './types.js';
+import { ServerAdapterRequestAbortSignal } from './utils.js';
 
 export interface UWSRequest {
   getMethod(): string;
@@ -32,39 +33,6 @@ interface GetRequestFromUWSOpts {
   fetchAPI: FetchAPI;
 }
 
-class UWSAbortSignal extends EventTarget implements AbortSignal {
-  aborted = false;
-  _onabort: ((this: AbortSignal, ev: Event) => any) | null = null;
-  reason: any;
-
-  throwIfAborted(): void {
-    if (this.aborted) {
-      throw new DOMException('Aborted', 'AbortError');
-    }
-  }
-
-  constructor(res: UWSResponse) {
-    super();
-    res.onAborted(() => {
-      this.aborted = true;
-      this.dispatchEvent(new Event('request aborted'));
-    });
-  }
-
-  get onabort() {
-    return this._onabort;
-  }
-
-  set onabort(value) {
-    this._onabort = value;
-    if (value) {
-      this.addEventListener('request aborted', value);
-    } else {
-      this.removeEventListener('request aborted', value);
-    }
-  }
-}
-
 export function getRequestFromUWSRequest({ req, res, fetchAPI }: GetRequestFromUWSOpts) {
   let body: ReadableStream | undefined;
   const method = req.getMethod();
@@ -97,7 +65,7 @@ export function getRequestFromUWSRequest({ req, res, fetchAPI }: GetRequestFromU
     method,
     headers,
     body: body as any,
-    signal: new UWSAbortSignal(res),
+    signal: new ServerAdapterRequestAbortSignal(),
   });
 }
 

packages/server/test/node.spec.ts

@@ -146,6 +146,27 @@ describe('Node Specific Cases', () => {
       const response = await fetch(testServer.url);
       expect(response.status).toBe(418);
     });
+
+    it('handles AbortSignal correctly', async () => {
+      const abortListener = jest.fn();
+      const serverAdapter = createServerAdapter(
+        req =>
+          new Promise(resolve => {
+            req.signal.onabort = () => {
+              abortListener();
+              resolve(new Response('Hello World', { status: 200 }));
+            };
+          }),
+      );
+      testServer.addOnceHandler(serverAdapter);
+      const controller = new AbortController();
+      setTimeout(() => controller.abort(), 1000);
+      await expect(() => fetch(testServer.url, { signal: controller.signal })).rejects.toEqual(
+        new Error('The operation was aborted'),
+      );
+      await new Promise(resolve => setTimeout(resolve, 300));
+      expect(abortListener).toHaveBeenCalledTimes(1);
+    });
   });
 });