review

Author: sommeeeer

packages/open-next/src/http/openNextResponse.ts

@@ -87,8 +87,8 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse {
 
     // We want to destroy this response when the original response is closed. (i.e when the client disconnects)
     // This is to support `request.signal.onabort` in route handlers
-    if (streamCreator?.onClose) {
-      streamCreator.onClose(() => {
+    if (streamCreator?.abortSignal) {
+      streamCreator.abortSignal.addEventListener("abort", () => {
         this.destroy();
       });
     }

packages/open-next/src/overrides/wrappers/cloudflare-node.ts

@@ -66,6 +66,9 @@ const handler: WrapperHandler<InternalEvent, InternalResult> =
 
         return Writable.fromWeb(writable);
       },
+      // For some reason the signal from `request` here did not work.
+      // @ts-expect-error This is defined in init from cloudflare
+      abortSignal: globalThis[Symbol.for("__cloudflare-context__")].signal,
     };
 
     ctx.waitUntil(

packages/open-next/src/overrides/wrappers/express-dev.ts

@@ -42,7 +42,8 @@ const wrapper: WrapperHandler = async (handler, converter) => {
     }
     const internalEvent = await converter.convertFrom(req);
 
-    let onCloseCallback: (() => void) | undefined;
+    const abortController = new AbortController();
+
     const streamCreator: StreamCreator = {
       writeHeaders: (prelude) => {
         res.setHeader("Set-Cookie", prelude.cookies);
@@ -51,15 +52,11 @@ const wrapper: WrapperHandler = async (handler, converter) => {
         return res;
       },
       onFinish: () => {},
-      onClose: (callback) => {
-        onCloseCallback = callback;
-      },
+      abortSignal: abortController.signal,
     };
 
     res.on("close", () => {
-      if (onCloseCallback) {
-        onCloseCallback();
-      }
+      abortController.abort();
     });
 
     await handler(internalEvent, { streamCreator });

packages/open-next/src/overrides/wrappers/node.ts

@@ -9,23 +9,20 @@ const wrapper: WrapperHandler = async (handler, converter) => {
   const server = createServer(async (req, res) => {
     const internalEvent = await converter.convertFrom(req);
 
-    let onCloseCallback: (() => void) | undefined;
+    const abortController = new AbortController();
+
     const streamCreator: StreamCreator = {
       writeHeaders: (prelude) => {
         res.setHeader("Set-Cookie", prelude.cookies);
         res.writeHead(prelude.statusCode, prelude.headers);
         res.flushHeaders();
         return res;
       },
-      onClose: (callback) => {
-        onCloseCallback = callback;
-      },
+      abortSignal: abortController.signal,
     };
 
     res.on("close", () => {
-      if (onCloseCallback) {
-        onCloseCallback();
-      }
+      abortController.abort();
     });
 
     if (internalEvent.rawPath === "/__health") {

packages/open-next/src/types/open-next.ts

@@ -49,8 +49,7 @@ export interface StreamCreator {
   // Just to fix an issue with aws lambda streaming with empty body
   onWrite?: () => void;
   onFinish?: (length: number) => void;
-  // This is called when the wrappers response is closed, i.e. when the client disconnects
-  onClose?: (callback: () => void) => void;
+  abortSignal?: AbortSignal;
 }
 
 export type WaitUntil = (promise: Promise<void>) => void;