feat: AbortSignal for SSE (#2966)

Copied from #2965 

Related discussion
#991 (reply in thread)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- New Features
- Added AbortSignal support to event stream subscriptions; emitters now
expose a signal that aborts when the client disconnects to enable proper
cancellation and cleanup.

- Documentation
  - Updated changelog for version 25.5.0 to note AbortSignal support.
  - Added contributor acknowledgements to the README.

- Tests
- Added tests ensuring the signal is present on emitters and aborts on
connection close.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: jakub-msqt <[email protected]>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: 3 people

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Version 25
 
+### v25.5.0
+
+- Feature: `AbortSignal` for subscription handlers:
+  - The `handler` of `EventStreamFactory::build()` is now equipped with `signal` option;
+  - That signal is aborted when the client disconnects;
+  - The feature was suggested and implemented by [@jakub-msqt](https://github.com/jakub-msqt).
+
 ### v25.4.1
 
 - This patch fixes the issue for users facing `TypeError: .example is not a function`, but not using that method:

README.md

@@ -85,6 +85,7 @@ Therefore, many basic tasks can be accomplished faster and easier, in particular
 
 These people contributed to the improvement of the framework by reporting bugs, making changes and suggesting ideas:
 
+[<img src="https://github.com/jakub-msqt.png" alt="@jakub-msqt" width="50px" />](https://github.com/jakub-msqt)
 [<img src="https://github.com/misha-z1nchuk.png" alt="@misha-z1nchuk" width="50px" />](https://github.com/misha-z1nchuk)
 [<img src="https://github.com/GreaterTamarack.png" alt="@GreaterTamarack" width="50px" />](https://github.com/GreaterTamarack)
 [<img src="https://github.com/pepegc.png" alt="@pepegc" width="50px" />](https://github.com/pepegc)

express-zod-api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "express-zod-api",
-  "version": "25.4.1",
+  "version": "25.5.0-beta.1",
   "description": "A Typescript framework to help you get an API server up and running with I/O schema validation and custom middlewares in minutes.",
   "license": "MIT",
   "repository": {

express-zod-api/src/sse.ts

@@ -16,6 +16,8 @@ type EventsMap = Record<string, z.ZodType>;
 export interface Emitter<E extends EventsMap> extends FlatObject {
   /** @desc Returns true when the connection was closed or terminated */
   isClosed: () => boolean;
+  /** @desc Abort signal bound to the client connection lifecycle */
+  signal: AbortSignal;
   /** @desc Sends an event to the stream according to the declared schema */
   emit: <K extends keyof E>(event: K, data: z.input<E[K]>) => void;
 }
@@ -55,9 +57,18 @@ export const ensureStream = (response: Response) =>
 
 export const makeMiddleware = <E extends EventsMap>(events: E) =>
   new Middleware({
-    handler: async ({ response }): Promise<Emitter<E>> =>
-      setTimeout(() => ensureStream(response), headersTimeout) && {
+    handler: async ({ request, response }): Promise<Emitter<E>> => {
+      const controller = new AbortController();
+
+      request.once("close", () => {
+        controller.abort();
+      });
+
+      setTimeout(() => ensureStream(response), headersTimeout);
+
+      return {
         isClosed: () => response.writableEnded || response.closed,
+        signal: controller.signal,
         emit: (event, data) => {
           ensureStream(response);
           response.write(formatEvent(events, event, data), "utf-8");
@@ -67,7 +78,8 @@ export const makeMiddleware = <E extends EventsMap>(events: E) =>
            * */
           response.flush?.();
         },
-      },
+      };
+    },
   });
 
 export const makeResultHandler = <E extends EventsMap>(events: E) =>

express-zod-api/tests/sse.spec.ts

@@ -87,6 +87,7 @@ describe("SSE", () => {
         if (flushMock) responseMock.flush = flushMock;
         expect(output).toEqual({
           isClosed: expect.any(Function),
+          signal: expect.any(AbortSignal),
           emit: expect.any(Function),
         });
         const { isClosed, emit } = output as Emitter<{ test: z.ZodString }>;
@@ -100,6 +101,15 @@ describe("SSE", () => {
         expect(isClosed()).toBeTruthy();
       },
     );
+
+    test("should abort signal on connection close", async () => {
+      const middleware = makeMiddleware({ test: z.string() });
+      const { requestMock, output } = await testMiddleware({ middleware });
+      const { signal } = output as Emitter<{ test: z.ZodString }>;
+      expect(signal.aborted).toBeFalsy();
+      requestMock.emit("close");
+      expect(signal.aborted).toBeTruthy();
+    });
   });
 
   describe("makeResultHandler()", () => {