Modified everything to use 'res.abortCh'

Author: Guthib-of-Dan

bun.lock

@@ -1,6 +1,6 @@
 {
   "name": "@ublitzjs/payload",
-  "version": "1.0.2",
+  "version": "1.0.3",
   "types": "./dist/types/index.d.ts",
   "files": ["dist"],
   "tsd": {
@@ -31,7 +31,7 @@
     }
   },
   "dependencies": {
-    "@ublitzjs/core": "^1.1.0",
+    "@ublitzjs/core": "^1.2.0",
     "busboy": "^1.6.0",
     "nanoid": "^5.1.6",
     "tseep": "^1.3.1"

package.json

@@ -33,7 +33,7 @@ export type FilesOnDisk<Repeated extends boolean> = FilesType<FileOnDisk, Repeat
 // type of "files" object, returned by parseFormDataBody with memory mode
 export type FilesInMemory<Repeated extends boolean> = FilesType<FileInMemory, Repeated>;
 /**
-* This function asynchronously parses multipart requests until the end, so that any work considering unfinished request wouldn't need to be undone.
+* This function asynchronously parses multipart requests until the end, so that any work considering unfinished request wouldn't need to be undone. It requires "regAbort" from "@ublitzjs/core" package to be used on response.
 * This utility expects "unpaused" response object and doesn't set any additional properties on it.  
 * When limits are exceeded - acts as like because of an error (as described below)
 * @param limits - limits passed to "busboy"
@@ -54,7 +54,7 @@ export type FilesInMemory<Repeated extends boolean> = FilesType<FileInMemory, Re
 *       }
 *     }, 
 *     // mode "memory" | "disk"
-*     "disk", 
+*     "disk",  It requires "regAbort" from "core" package to be used on response.
 *     // whether to accept repeated files/fields
 *     false
 *   );
@@ -270,15 +270,14 @@ export async function parseFormDataBody<T extends "memory" | "disk", Repeated ex
     parserStream.write(copy);
     if (isLast) { parserStream.end();}
   });
-  // calling emitter.emit many times won't hurt, as listener is ONE and emitEmergency might be called IN ONE PLACE
   var endReceivingBody: ()=>void | undefined
   function onAborted() {
     if(!lastError) { lastError = "aborted" }
     emitEmergency()
   }
-  res.emitter.once("abort", onAborted);
+  res.abortCh.sub(onAborted);
   await new Promise<void>((resolve) => { endReceivingBody = resolve });
-  if(!res.aborted) res.emitter.off("abort", onAborted);
+  if(!res.aborted) res.abortCh.unsub(onAborted);
   if(lastError && !emergency) { emitEmergency(); }
   if(emergency && emergency !== true) await (emergency as Promise<void[]>);
   return !lastError 
@@ -287,7 +286,7 @@ export async function parseFormDataBody<T extends "memory" | "disk", Repeated ex
 }
 type AccumulatedBody<T extends boolean> =  Buffer<T extends true ? SharedArrayBuffer : ArrayBuffer>
 /**
-* This utility just accumulates body and verifies if it stays within the given limit.
+* This utility just accumulates body and verifies if it stays within the given limit. It requires "regAbort" from "@ublitzjs/core" package to be used.
 * @param CL this is Content-Length to be compared against + acts as a preallocation amount.
 * If not given by developer OR == 0 - constantly happen memory reallocations with Buffer.concat
 * @param shared whether return Buffer of SharedArrayBuffer or simple ArrayBuffer. SharedArrayBuffer lets you pass data to a worker thread if work there is cpu intensive
@@ -318,11 +317,11 @@ export async function accumulateBody<T extends boolean = false>(
   data = (shared ? Buffer.from(new SharedArrayBuffer(CL)) : Buffer.allocUnsafe(CL)) as AccumulatedBody<T>
   return new Promise((resolve) => {
     function onAborted() { resolve(undefined) }
-    res.emitter.once("abort", onAborted);
+    res.abortCh.sub(onAborted);
     res.onData((ab, isLast) => {
       write(ab);
       writtenBytes += ab.byteLength;
-      if (isLast) { res.emitter.off("abort", onAborted); resolve(data) }
+      if (isLast) { res.abortCh.unsub(onAborted); resolve(data) }
     });
   })
 }

src/index.ts

@@ -1,5 +1,5 @@
 import {App} from "uWebSockets.js"
-import {extendApp, registerAbort} from "@ublitzjs/core"
+import {extendApp, regAbort} from "@ublitzjs/core"
 import {accumulateBody, parseFormDataBody} from "@ublitzjs/payload"
 import {nanoid} from "nanoid"
 import {mkdirSync, rm} from "node:fs"
@@ -8,7 +8,7 @@ var server = extendApp(App())
 var outDir = "tmp/" + nanoid(7)  
 mkdirSync(outDir)
 server.post("/multipart-disk", async (res, req)=>{
-  registerAbort(res)
+  regAbort(res)
   var result = await parseFormDataBody({
     CT: req.getHeader("content-type"), res, outDir,
     parseLimits: {fileSize: 1024*1024+1}
@@ -24,7 +24,7 @@ server.post("/multipart-disk", async (res, req)=>{
   ) + '}'))
 })
 server.post("/multipart-memory", async (res, req)=>{
-  registerAbort(res)
+  regAbort(res)
   var result = await parseFormDataBody({
     CT: req.getHeader("content-type"), res, outDir,
     parseLimits: {fileSize: 1024*1024+1}
@@ -40,7 +40,7 @@ server.post("/multipart-memory", async (res, req)=>{
   ) + '}'))
 })
 server.post("/upload", async (res, req)=>{
-  registerAbort(res)
+  regAbort(res)
   var result = await accumulateBody(res, Number(req.getHeader("content-length")), false)
   res.cork(()=>res.end('{"receivedBytes": ' + result.byteLength + '}'))
 })

tests/k6/server.mjs

@@ -1,7 +1,7 @@
 import type { ClientRequest } from "node:http";
 import type { FileInMemory, FileOnDisk, parseFormDataBody } from "@ublitzjs/payload"
 import type { Limits } from "busboy";
-import { registerAbort, type HttpRequest, type HttpResponse } from "@ublitzjs/core";
+import { regAbort, type HttpRequest, type HttpResponse } from "@ublitzjs/core";
 import { expect } from "vitest";
 import { readFileSync, rmSync } from "node:fs";
 
@@ -61,7 +61,7 @@ export function setupHandler(parseBody: typeof parseFormDataBody) {
     opts: { limits?: Limits, outDir?: string } = {}
   ) {
     return async (res: HttpResponse, req: HttpRequest) => {
-      registerAbort(res)
+      regAbort(res)
       var result = await parseBody({
         res, CT: req.getHeader("content-type"), parseLimits: opts.limits,
         outDir: opts.outDir,

tests/vitest/http/form-data.ts

@@ -1,4 +1,4 @@
-import { registerAbort } from "@ublitzjs/core";
+import { regAbort } from "@ublitzjs/core";
 import { request } from "node:http";
 import {expectType} from "tsd"
 //import { setTimeout as awaitTimeout } from "node:timers/promises"
@@ -37,7 +37,7 @@ export default async function(module: typeof import("@ublitzjs/payload")) {
         it("is type-safe", skipHelper(), ()=>{
           if(runningTsd) {
             server.post("/types", async (res, req)=>{
-              registerAbort(res);
+              regAbort(res);
               var resultMemorySingular = await module.parseFormDataBody({
                 res, CT:req.getHeader("content-type")
               }, "memory", false);
@@ -238,7 +238,7 @@ export default async function(module: typeof import("@ublitzjs/payload")) {
         it("successfully aborts request", skipHelper(), async () => {
           await new Promise<void>((resolve) => {
             server.post("/files/memory/abort", async (res, req) => {
-              registerAbort(res)
+              regAbort(res)
               var result = await module.parseFormDataBody({
                 CT: req.getHeader("content-type"),
                 res
@@ -360,7 +360,7 @@ export default async function(module: typeof import("@ublitzjs/payload")) {
           function doIt(type: "/singular" | "/repeated") {
             return new Promise<void>((resolve) => {
               server.post("/files/disk/abort" + type, async (res, req) => {
-                registerAbort(res)
+                regAbort(res)
                 var result = await module.parseFormDataBody({
                   CT: req.getHeader("content-type"),
                   res, outDir
@@ -399,7 +399,7 @@ export default async function(module: typeof import("@ublitzjs/payload")) {
           var outDir = "tmp/" + nanoid(10);
           mkdirSync(outDir)
           server.post("/files/disk/empty", async (res, req)=>{
-            registerAbort(res)
+            regAbort(res)
             var result = await module.parseFormDataBody({ res, CT: req.getHeader("content-type"), outDir },"disk", false)
             if (result.ok) {
               try {
@@ -437,7 +437,7 @@ export default async function(module: typeof import("@ublitzjs/payload")) {
       it("is type-safe", ()=>{
         if(runningTsd) {
           server.post("/body/types", async (res, req)=>{
-            registerAbort(res)
+            regAbort(res)
             var result = await module.accumulateBody(res, Number(req.getHeader("content-length")), false)
             if(res.aborted) return;
             expectType<Buffer<ArrayBuffer>|undefined>(result)
@@ -451,7 +451,7 @@ export default async function(module: typeof import("@ublitzjs/payload")) {
       it("accumulates larger bodies then 64KiB", async ()=>{
         var body = Buffer.allocUnsafeSlow(65*1024)
         server.post("/body/large", async (res, req)=>{
-          registerAbort(res)
+          regAbort(res)
           var result = await module.accumulateBody(res, Number(req.getHeader("content-length")))
           if(!result) return;
           expect(body.compare(result)).toBe(0)