fix multiform

.github/workflows/test.yml

@@ -19,3 +19,5 @@ jobs:
         run: npm install
       - name: Run tests
         run: npm test
+      - name: Run benchmark
+        run: npm run benchmark

benchmarks/fastify-node.js

@@ -0,0 +1,13 @@
+import fastify from 'fastify'
+
+const app = fastify({
+  logger: false,
+})
+
+app.get('/', (req, reply) => {
+  reply.send({ hello: 'world' })
+})
+
+app.listen({ port: 3000 }, async (err, address) => {
+  if (err) throw err
+})

benchmarks/fastify-uws.js

@@ -0,0 +1,16 @@
+import fastify from 'fastify'
+
+import { serverFactory } from '../src/server.js'
+
+const app = fastify({
+  logger: false,
+  serverFactory,
+})
+
+app.get('/', (req, reply) => {
+  reply.send({ hello: 'world' })
+})
+
+app.listen({ port: 3000 }, async (err, address) => {
+  if (err) throw err
+})

benchmarks/index.js

@@ -0,0 +1,53 @@
+import { fork } from 'node:child_process'
+
+import autocannon from 'autocannon'
+import autocannonCompare from 'autocannon-compare'
+
+const benchs = [
+  {
+    name: 'fastify-node',
+    file: 'fastify-node.js',
+  },
+  {
+    name: 'fastify-uws',
+    file: 'fastify-uws.js',
+  },
+]
+
+const results = new Map()
+
+for (const bench of benchs) {
+  const child = fork(`./benchmarks/${bench.file}`)
+
+  await new Promise(resolve => setTimeout(resolve, 1000))
+
+  const result = await autocannon({
+    url: `http://localhost:3000/${bench.file}`,
+    connections: 100,
+    pipelining: 10,
+    duration: 3,
+  })
+
+  results.set(bench.name, {
+    name: bench.name,
+    ...result,
+  })
+
+  child.kill('SIGINT')
+}
+
+const a = results.get('fastify-node')
+const b = results.get('fastify-uws')
+
+const comp = autocannonCompare(a, b)
+
+console.log(`a: ${a.requests.average}`)
+console.log(`b: ${b.requests.average}`)
+
+if (comp.equal) {
+  console.log('Same performance!')
+} else if (comp.aWins) {
+  console.log(`${a.name} is faster than ${b.name} by ${comp.requests.difference} of difference`)
+} else {
+  console.log(`${b.name} is faster than ${a.name} by ${autocannonCompare(b, a).requests.difference} of difference`)
+}

package.json

@@ -53,24 +53,28 @@
     "lint": "eslint .",
     "lint:fix": "npm run lint -- --fix",
     "prepublishOnly": "npm test && npm run build && npm run types",
-    "types": "node scripts/generate-dts.js"
+    "types": "node scripts/generate-dts.js",
+    "benchmark": "node benchmarks/index.js"
   },
   "dependencies": {
+    "eventemitter3": "^5.0.1",
     "fastify-plugin": "^4.5.1",
-    "fastq": "^1.17.1",
     "ipaddr.js": "^2.2.0",
     "nanoerror": "^2.0.0",
-    "streamx": "^2.18.0",
+    "streamx": "npm:@geut/streamx@^2.20.1",
     "tempy": "^3.1.0",
-    "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.44.0"
+    "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.48.0"
   },
   "devDependencies": {
+    "@fastify/multipart": "^8.3.0",
     "@fastify/static": "^7.0.4",
     "@types/events": "^3.0.2",
     "@types/node": "^20.8.10",
     "@types/streamx": "^2.9.3",
+    "autocannon": "^7.15.0",
+    "autocannon-compare": "^0.4.0",
     "eslint": "^8.57.0",
-    "eslint-config-standard-ext": "^0.0.27",
+    "eslint-config-standard-ext": "^2.0.0",
     "execa": "^8.0.1",
     "fastify": "^4.28.1",
     "require-inject": "^1.4.4",

src/http-socket.js

@@ -1,14 +1,11 @@
-import EventEmitter from 'node:events'
-
-import fastq from 'fastq'
+import { EventEmitter } from 'eventemitter3'
 
 import { ERR_STREAM_DESTROYED } from './errors.js'
 import {
   kAddress,
   kEncoding,
   kHead,
   kHttps,
-  kQueue,
   kReadyState,
   kRemoteAdress,
   kRes,
@@ -64,23 +61,6 @@ function onTimeout() {
   }
 }
 
-function end(socket, data) {
-  socket._clearTimeout()
-
-  const res = socket[kRes]
-
-  res.cork(() => {
-    if (socket[kHead]) {
-      writeHead(res, socket[kHead])
-      socket[kHead] = null
-    }
-    res.end(getChunk(data))
-    socket.bytesWritten += byteLength(data)
-    socket.emit('close')
-    socket.emit('finish')
-  })
-}
-
 function drain(socket, cb) {
   socket.writableNeedDrain = true
   let done = false
@@ -229,7 +209,6 @@ export class HTTPSocket extends EventEmitter {
   abort() {
     if (this.aborted) return
     this.aborted = true
-    this[kQueue] && this[kQueue].kill()
     if (!this[kWs] && !this.writableEnded) {
       this[kRes].close()
     }
@@ -256,12 +235,12 @@ export class HTTPSocket extends EventEmitter {
       this[kRes].onData((chunk, isLast) => {
         if (done) return
 
-        chunk = Buffer.from(chunk)
-
-        this.bytesRead += Buffer.byteLength(chunk)
+        this.bytesRead += chunk.byteLength
 
         if (encoding) {
-          chunk = chunk.toString(encoding)
+          chunk = Buffer.from(chunk).toString(encoding)
+        } else {
+          chunk = Buffer.copyBytesFrom(new Uint8Array(chunk))
         }
 
         this.emit('data', chunk)
@@ -285,34 +264,27 @@ export class HTTPSocket extends EventEmitter {
     if (!data) return this.abort()
 
     this.writableEnded = true
-    const queue = this[kQueue]
 
-    // fast end
-    if (!queue || queue.idle()) {
-      end(this, data)
-      cb()
-      return
-    }
+    this._clearTimeout()
+
+    const res = this[kRes]
 
-    queue.push(data, cb)
+    res.cork(() => {
+      if (this[kHead]) {
+        writeHead(res, this[kHead])
+        this[kHead] = null
+      }
+      res.end(getChunk(data))
+      this.bytesWritten += byteLength(data)
+      this.emit('close')
+      this.emit('finish')
+      cb()
+    })
   }
 
   write(data, _, cb = noop) {
     if (this.destroyed) throw new ERR_STREAM_DESTROYED()
 
-    if (!this[kQueue]) {
-      this[kQueue] = fastq(this, this._onWrite, 1)
-    }
-
-    this[kQueue].push(data, cb)
-    return !this.writableNeedDrain
-  }
-
-  _clearTimeout() {
-    this[kTimeoutRef] && clearTimeout(this[kTimeoutRef])
-  }
-
-  _onWrite(data, cb) {
     const res = this[kRes]
 
     this[kReadyState].write = true
@@ -329,5 +301,11 @@ export class HTTPSocket extends EventEmitter {
       if (drained) return cb()
       drain(this, cb)
     })
+
+    return !this.writableNeedDrain
+  }
+
+  _clearTimeout() {
+    this[kTimeoutRef] && clearTimeout(this[kTimeoutRef])
   }
 }

src/request.js

@@ -79,9 +79,8 @@ export class Request extends Readable {
 
       if (!data) {
         this.readableEnded = true
+        cb()
       }
-
-      cb()
     })
   }
 }

src/server.js

@@ -19,9 +19,9 @@
 
 import assert from 'node:assert'
 import dns from 'node:dns/promises'
-import EventEmitter from 'node:events'
 import { writeFileSync } from 'node:fs'
 
+import { EventEmitter } from 'eventemitter3'
 import ipaddr from 'ipaddr.js'
 import { temporaryFile } from 'tempy'
 import uws from 'uWebSockets.js'
@@ -290,22 +290,22 @@ export const getUws = (fastify) => {
 export { WebSocketStream } from './websocket-server.js'
 
 export {
-  DEDICATED_COMPRESSOR_128KB,
-  DEDICATED_COMPRESSOR_16KB,
-  DEDICATED_COMPRESSOR_256KB,
-  DEDICATED_COMPRESSOR_32KB,
   DEDICATED_COMPRESSOR_3KB,
   DEDICATED_COMPRESSOR_4KB,
-  DEDICATED_COMPRESSOR_64KB,
   DEDICATED_COMPRESSOR_8KB,
+  DEDICATED_COMPRESSOR_16KB,
+  DEDICATED_COMPRESSOR_32KB,
+  DEDICATED_COMPRESSOR_64KB,
+  DEDICATED_COMPRESSOR_128KB,
+  DEDICATED_COMPRESSOR_256KB,
   DEDICATED_DECOMPRESSOR,
-  DEDICATED_DECOMPRESSOR_16KB,
   DEDICATED_DECOMPRESSOR_1KB,
   DEDICATED_DECOMPRESSOR_2KB,
-  DEDICATED_DECOMPRESSOR_32KB,
   DEDICATED_DECOMPRESSOR_4KB,
-  DEDICATED_DECOMPRESSOR_512B,
   DEDICATED_DECOMPRESSOR_8KB,
+  DEDICATED_DECOMPRESSOR_16KB,
+  DEDICATED_DECOMPRESSOR_32KB,
+  DEDICATED_DECOMPRESSOR_512B,
   DISABLED,
   SHARED_COMPRESSOR,
   SHARED_DECOMPRESSOR,

src/symbols.js

@@ -20,7 +20,6 @@ export const kWs = Symbol('uws.ws')
 export const kTopic = Symbol('uws.topic')
 export const kDestroyError = Symbol('uws.destroyError')
 export const kUwsRemoteAddress = Symbol('uws.uwsRemoteAddress')
-export const kQueue = Symbol('uws.queue')
 export const kHead = Symbol('uws.head')
 export const kWebSocketOptions = Symbol('uws.webSocketOptions')
 export const kListenAll = Symbol('uws.listenAll')

src/websocket-server.js

@@ -52,8 +52,7 @@
  * }} WebsocketServerEvent
  */
 
-import EventEmitter from 'node:events'
-
+import { EventEmitter } from 'eventemitter3'
 import { Duplex } from 'streamx'
 import uws from 'uWebSockets.js'
 
@@ -71,6 +70,9 @@ const defaultWebSocketConfig = {
 const SEP = '!'
 const SEP_BUFFER = Buffer.from(SEP)
 
+/**
+ * @extends {EventEmitter<keyof WebsocketEvent, WebsocketEvent[keyof WebsocketEvent]>}
+ */
 export class WebSocket extends EventEmitter {
   /**
    * @param {Buffer} namespace
@@ -208,24 +210,6 @@ export class WebSocket extends EventEmitter {
     if (this[kEnded]) return
     return this.connection.ping(message)
   }
-
-  /**
-   * @template {keyof WebsocketEvent} T
-   * @param {T} eventName
-   * @param {WebsocketEvent[T]} listener
-   */
-  on(eventName, listener) {
-    return super.on(eventName, listener)
-  }
-
-  /**
-   * @template {keyof WebsocketEvent} T
-   * @param {T} eventName
-   * @param {WebsocketEvent[T]} listener
-   */
-  once(eventName, listener) {
-    return super.once(eventName, listener)
-  }
 }
 
 export class WebSocketStream extends Duplex {
@@ -290,6 +274,9 @@ export class WebSocketStream extends Duplex {
   }
 }
 
+/**
+ * @extends {EventEmitter<keyof WebsocketServerEvent, WebsocketServerEvent[keyof WebsocketServerEvent]>}
+ */
 export class WebSocketServer extends EventEmitter {
   /**
    * @param {WSOptions} options
@@ -380,22 +367,4 @@ export class WebSocketServer extends EventEmitter {
       ...options,
     })
   }
-
-  /**
-   * @template {keyof WebsocketServerEvent} T
-   * @param {T} eventName
-   * @param {WebsocketServerEvent[T]} listener
-   */
-  on(eventName, listener) {
-    return super.on(eventName, listener)
-  }
-
-  /**
-   * @template {keyof WebsocketServerEvent} T
-   * @param {T} eventName
-   * @param {WebsocketServerEvent[T]} listener
-   */
-  once(eventName, listener) {
-    return super.once(eventName, listener)
-  }
 }

tests/fastify/index.cjs

@@ -1,6 +1,6 @@
-const { Readable } = require('node:stream')
-const path = require('node:path')
 const { once } = require('node:events')
+const path = require('node:path')
+const { Readable } = require('node:stream')
 const requireInject = require('require-inject')
 const _sget = require('simple-get').concat