chore(release): 4.4.1

Diff: https://github.com/socketio/socket.io/compare/4.4.0...4.4.1

CHANGELOG.md

@@ -1,3 +1,29 @@
+## [4.4.1](https://github.com/socketio/socket.io/compare/4.4.0...4.4.1) (2022-01-06)
+
+
+### Bug Fixes
+
+* **types:** make `RemoteSocket.data` type safe ([#4234](https://github.com/socketio/socket.io/issues/4234)) ([770ee59](https://github.com/socketio/socket.io/commit/770ee5949fb47c2556876c622f06c862573657d6))
+* **types:** pass `SocketData` type to custom namespaces ([#4233](https://github.com/socketio/socket.io/issues/4233)) ([f2b8de7](https://github.com/socketio/socket.io/commit/f2b8de71919e1b4d3e57f15a459972c1d1064787))
+
+
+
+# [4.4.0](https://github.com/socketio/socket.io/compare/4.3.2...4.4.0) (2021-11-18)
+
+
+### Bug Fixes
+
+* only set 'connected' to true after middleware execution ([02b0f73](https://github.com/socketio/socket.io/commit/02b0f73e2c64b09c72c5fbf7dc5f059557bdbe50))
+
+
+### Features
+
+* add an implementation based on uWebSockets.js ([c0d8c5a](https://github.com/socketio/socket.io/commit/c0d8c5ab234d0d2bef0d0dec472973cc9662f647))
+* add timeout feature ([f0ed42f](https://github.com/socketio/socket.io/commit/f0ed42f18cabef20ad976aeec37077b6bf3837a5))
+* add type information to `socket.data` ([#4159](https://github.com/socketio/socket.io/issues/4159)) ([fe8730c](https://github.com/socketio/socket.io/commit/fe8730ca0f15bc92d5de81cf934c89c76d6af329))
+
+
+
 ## [4.3.2](https://github.com/socketio/socket.io/compare/4.3.1...4.3.2) (2021-11-08)
 
 

Readme.md

@@ -115,6 +115,14 @@ io.on('connection', client => { ... });
 io.listen(3000);
 ```
 
+### Module syntax 
+
+```js
+import { Server } from "socket.io";
+const io = new Server(server);
+io.listen(3000);
+```
+
 ### In conjunction with Express
 
 Starting with **3.0**, express applications have become request handler

client-dist/socket.io.js

@@ -1,6 +1,6 @@
 /*!
- * Socket.IO v4.3.2
- * (c) 2014-2021 Guillermo Rauch
+ * Socket.IO v4.4.1
+ * (c) 2014-2022 Guillermo Rauch
  * Released under the MIT License.
  */
 (function (global, factory) {
@@ -3148,8 +3148,12 @@
         packet.options.compress = this.flags.compress !== false; // event ack callback
 
         if ("function" === typeof args[args.length - 1]) {
-          this.acks[this.ids] = args.pop();
-          packet.id = this.ids++;
+          var id = this.ids++;
+          var ack = args.pop();
+
+          this._registerAckCallback(id, ack);
+
+          packet.id = id;
         }
 
         var isTransportWritable = this.io.engine && this.io.engine.transport && this.io.engine.transport.writable;
@@ -3164,6 +3168,46 @@
         this.flags = {};
         return this;
       }
+      /**
+       * @private
+       */
+
+    }, {
+      key: "_registerAckCallback",
+      value: function _registerAckCallback(id, ack) {
+        var _this2 = this;
+
+        var timeout = this.flags.timeout;
+
+        if (timeout === undefined) {
+          this.acks[id] = ack;
+          return;
+        } // @ts-ignore
+
+
+        var timer = this.io.setTimeoutFn(function () {
+          delete _this2.acks[id];
+
+          for (var i = 0; i < _this2.sendBuffer.length; i++) {
+            if (_this2.sendBuffer[i].id === id) {
+              _this2.sendBuffer.splice(i, 1);
+            }
+          }
+
+          ack.call(_this2, new Error("operation has timed out"));
+        }, timeout);
+
+        this.acks[id] = function () {
+          // @ts-ignore
+          _this2.io.clearTimeoutFn(timer);
+
+          for (var _len3 = arguments.length, args = new Array(_len3), _key3 = 0; _key3 < _len3; _key3++) {
+            args[_key3] = arguments[_key3];
+          }
+
+          ack.apply(_this2, [null].concat(args));
+        };
+      }
       /**
        * Sends a packet.
        *
@@ -3187,11 +3231,11 @@
     }, {
       key: "onopen",
       value: function onopen() {
-        var _this2 = this;
+        var _this3 = this;
 
         if (typeof this.auth == "function") {
           this.auth(function (data) {
-            _this2.packet({
+            _this3.packet({
               type: PacketType.CONNECT,
               data: data
             });
@@ -3277,6 +3321,7 @@
             break;
 
           case PacketType.CONNECT_ERROR:
+            this.destroy();
             var err = new Error(packet.data.message); // @ts-ignore
 
             err.data = packet.data.data;
@@ -3345,8 +3390,8 @@
           if (sent) return;
           sent = true;
 
-          for (var _len3 = arguments.length, args = new Array(_len3), _key3 = 0; _key3 < _len3; _key3++) {
-            args[_key3] = arguments[_key3];
+          for (var _len4 = arguments.length, args = new Array(_len4), _key4 = 0; _key4 < _len4; _key4++) {
+            args[_key4] = arguments[_key4];
           }
 
           self.packet({
@@ -3397,14 +3442,14 @@
     }, {
       key: "emitBuffered",
       value: function emitBuffered() {
-        var _this3 = this;
+        var _this4 = this;
 
         this.receiveBuffer.forEach(function (args) {
-          return _this3.emitEvent(args);
+          return _this4.emitEvent(args);
         });
         this.receiveBuffer = [];
         this.sendBuffer.forEach(function (packet) {
-          return _this3.packet(packet);
+          return _this4.packet(packet);
         });
         this.sendBuffer = [];
       }
@@ -3507,6 +3552,28 @@
         this.flags["volatile"] = true;
         return this;
       }
+      /**
+       * Sets a modifier for a subsequent event emission that the callback will be called with an error when the
+       * given number of milliseconds have elapsed without an acknowledgement from the server:
+       *
+       * ```
+       * socket.timeout(5000).emit("my-event", (err) => {
+       *   if (err) {
+       *     // the server did not acknowledge the event in the given delay
+       *   }
+       * });
+       * ```
+       *
+       * @returns self
+       * @public
+       */
+
+    }, {
+      key: "timeout",
+      value: function timeout(_timeout) {
+        this.flags.timeout = _timeout;
+        return this;
+      }
       /**
        * Adds a listener that will be fired when any event is emitted. The event name is passed as the first argument to the
        * callback.
@@ -4020,15 +4087,7 @@
       value: function _close() {
         this.skipReconnect = true;
         this._reconnecting = false;
-
-        if ("opening" === this._readyState) {
-          // `onclose` will not fire because
-          // an open event never happened
-          this.cleanup();
-        }
-
-        this.backoff.reset();
-        this._readyState = "closed";
+        this.onclose("forced close");
         if (this.engine) this.engine.close();
       }
       /**

examples/basic-crud-application/server/lib/util.ts

@@ -7,8 +7,8 @@ export enum Errors {
 
 const errorValues: string[] = Object.values(Errors);
 
-export function sanitizeErrorMessage(message: string) {
-  if (errorValues.includes(message)) {
+export function sanitizeErrorMessage(message: any) {
+  if (typeof message === "string" && errorValues.includes(message)) {
     return message;
   } else {
     return "an unknown error has occurred";

examples/chat/README.md

@@ -6,7 +6,7 @@ A simple chat demo for Socket.IO
 ## How to use
 
 ```
-$ npm ci
+$ npm i
 $ npm start
 ```
 

examples/chat/public/main.js

@@ -264,14 +264,14 @@ $(function() {
     log('you have been disconnected');
   });
 
-  socket.on('reconnect', () => {
+  socket.io.on('reconnect', () => {
     log('you have been reconnected');
     if (username) {
       socket.emit('add user', username);
     }
   });
 
-  socket.on('reconnect_error', () => {
+  socket.io.on('reconnect_error', () => {
     log('attempt to reconnect has failed');
   });
 

examples/cluster-nginx/docker-compose.yml

@@ -1,6 +1,8 @@
 
 nginx:
-  build: ./nginx
+  image: nginx:alpine
+  volumes:
+    - ./nginx.conf:/etc/nginx/nginx.conf:ro
   links:
     - server-john
     - server-paul

examples/cluster-nginx/nginx/Dockerfile

@@ -1,3 +0,0 @@
-
-FROM nginx:alpine
-COPY nginx.conf /etc/nginx/nginx.conf

examples/cluster-nginx/server/index.js

@@ -1,15 +1,18 @@
-// Setup basic express server
-var express = require('express');
-var app = express();
-var server = require('http').createServer(app);
-var io = require('socket.io')(server);
-var redis = require('socket.io-redis');
-var port = process.env.PORT || 3000;
-var serverName = process.env.NAME || 'Unknown';
+const express = require('express');
+const app = express();
+const server = require('http').createServer(app);
+const io = require('socket.io')(server);
+const { createAdapter } = require('@socket.io/redis-adapter');
+const { createClient } = require('redis');
+const port = process.env.PORT || 3000;
+const serverName = process.env.NAME || 'Unknown';
 
-io.adapter(redis({ host: 'redis', port: 6379 }));
+const pubClient = createClient({ host: 'redis', port: 6379 });
+const subClient = pubClient.duplicate();
 
-server.listen(port, function () {
+io.adapter(createAdapter(pubClient, subClient));
+
+server.listen(port, () => {
   console.log('Server listening at port %d', port);
   console.log('Hello, I\'m %s, how can I help?', serverName);
 });
@@ -19,15 +22,15 @@ app.use(express.static(__dirname + '/public'));
 
 // Chatroom
 
-var numUsers = 0;
+let numUsers = 0;
 
-io.on('connection', function (socket) {
+io.on('connection', socket => {
   socket.emit('my-name-is', serverName);
 
-  var addedUser = false;
+  let addedUser = false;
 
   // when the client emits 'new message', this listens and executes
-  socket.on('new message', function (data) {
+  socket.on('new message', data => {
     // we tell the client to execute 'new message'
     socket.broadcast.emit('new message', {
       username: socket.username,
@@ -36,7 +39,7 @@ io.on('connection', function (socket) {
   });
 
   // when the client emits 'add user', this listens and executes
-  socket.on('add user', function (username) {
+  socket.on('add user', username => {
     if (addedUser) return;
 
     // we store the username in the socket session for this client
@@ -54,21 +57,21 @@ io.on('connection', function (socket) {
   });
 
   // when the client emits 'typing', we broadcast it to others
-  socket.on('typing', function () {
+  socket.on('typing', () => {
     socket.broadcast.emit('typing', {
       username: socket.username
     });
   });
 
   // when the client emits 'stop typing', we broadcast it to others
-  socket.on('stop typing', function () {
+  socket.on('stop typing', () => {
     socket.broadcast.emit('stop typing', {
       username: socket.username
     });
   });
 
   // when the user disconnects.. perform this
-  socket.on('disconnect', function () {
+  socket.on('disconnect', () => {
     if (addedUser) {
       --numUsers;
 

examples/cluster-nginx/server/package.json

@@ -7,9 +7,10 @@
   "private": true,
   "license": "MIT",
   "dependencies": {
+    "@socket.io/redis-adapter": "^7.0.1",
     "express": "4.13.4",
-    "socket.io": "^4.0.0",
-    "socket.io-redis": "^6.0.1"
+    "redis": "^3.1.2",
+    "socket.io": "^4.0.0"
   },
   "scripts": {
     "start": "node index.js"

lib/broadcast-operator.ts

@@ -9,7 +9,7 @@ import type {
   TypedEventBroadcaster,
 } from "./typed-events";
 
-export class BroadcastOperator<EmitEvents extends EventsMap>
+export class BroadcastOperator<EmitEvents extends EventsMap, SocketData>
   implements TypedEventBroadcaster<EmitEvents>
 {
   constructor(
@@ -26,7 +26,7 @@ export class BroadcastOperator<EmitEvents extends EventsMap>
    * @return a new BroadcastOperator instance
    * @public
    */
-  public to(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public to(room: Room | Room[]): BroadcastOperator<EmitEvents, SocketData> {
     const rooms = new Set(this.rooms);
     if (Array.isArray(room)) {
       room.forEach((r) => rooms.add(r));
@@ -48,7 +48,7 @@ export class BroadcastOperator<EmitEvents extends EventsMap>
    * @return a new BroadcastOperator instance
    * @public
    */
-  public in(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public in(room: Room | Room[]): BroadcastOperator<EmitEvents, SocketData> {
     return this.to(room);
   }
 
@@ -59,7 +59,9 @@ export class BroadcastOperator<EmitEvents extends EventsMap>
    * @return a new BroadcastOperator instance
    * @public
    */
-  public except(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public except(
+    room: Room | Room[]
+  ): BroadcastOperator<EmitEvents, SocketData> {
     const exceptRooms = new Set(this.exceptRooms);
     if (Array.isArray(room)) {
       room.forEach((r) => exceptRooms.add(r));
@@ -81,7 +83,9 @@ export class BroadcastOperator<EmitEvents extends EventsMap>
    * @return a new BroadcastOperator instance
    * @public
    */
-  public compress(compress: boolean): BroadcastOperator<EmitEvents> {
+  public compress(
+    compress: boolean
+  ): BroadcastOperator<EmitEvents, SocketData> {
     const flags = Object.assign({}, this.flags, { compress });
     return new BroadcastOperator(
       this.adapter,
@@ -99,7 +103,7 @@ export class BroadcastOperator<EmitEvents extends EventsMap>
    * @return a new BroadcastOperator instance
    * @public
    */
-  public get volatile(): BroadcastOperator<EmitEvents> {
+  public get volatile(): BroadcastOperator<EmitEvents, SocketData> {
     const flags = Object.assign({}, this.flags, { volatile: true });
     return new BroadcastOperator(
       this.adapter,
@@ -115,7 +119,7 @@ export class BroadcastOperator<EmitEvents extends EventsMap>
    * @return a new BroadcastOperator instance
    * @public
    */
-  public get local(): BroadcastOperator<EmitEvents> {
+  public get local(): BroadcastOperator<EmitEvents, SocketData> {
     const flags = Object.assign({}, this.flags, { local: true });
     return new BroadcastOperator(
       this.adapter,
@@ -177,7 +181,9 @@ export class BroadcastOperator<EmitEvents extends EventsMap>
    *
    * @public
    */
-  public fetchSockets(): Promise<RemoteSocket<EmitEvents>[]> {
+  public fetchSockets<SocketData = any>(): Promise<
+    RemoteSocket<EmitEvents, SocketData>[]
+  > {
     return this.adapter
       .fetchSockets({
         rooms: this.rooms,
@@ -187,9 +193,12 @@ export class BroadcastOperator<EmitEvents extends EventsMap>
         return sockets.map((socket) => {
           if (socket instanceof Socket) {
             // FIXME the TypeScript compiler complains about missing private properties
-            return socket as unknown as RemoteSocket<EmitEvents>;
+            return socket as unknown as RemoteSocket<EmitEvents, SocketData>;
           } else {
-            return new RemoteSocket(this.adapter, socket as SocketDetails);
+            return new RemoteSocket(
+              this.adapter,
+              socket as SocketDetails<SocketData>
+            );
           }
         });
       });
@@ -247,27 +256,27 @@ export class BroadcastOperator<EmitEvents extends EventsMap>
 /**
  * Format of the data when the Socket instance exists on another Socket.IO server
  */
-interface SocketDetails {
+interface SocketDetails<SocketData> {
   id: SocketId;
   handshake: Handshake;
   rooms: Room[];
-  data: any;
+  data: SocketData;
 }
 
 /**
  * Expose of subset of the attributes and methods of the Socket class
  */
-export class RemoteSocket<EmitEvents extends EventsMap>
+export class RemoteSocket<EmitEvents extends EventsMap, SocketData>
   implements TypedEventBroadcaster<EmitEvents>
 {
   public readonly id: SocketId;
   public readonly handshake: Handshake;
   public readonly rooms: Set<Room>;
-  public readonly data: any;
+  public readonly data: SocketData;
 
-  private readonly operator: BroadcastOperator<EmitEvents>;
+  private readonly operator: BroadcastOperator<EmitEvents, SocketData>;
 
-  constructor(adapter: Adapter, details: SocketDetails) {
+  constructor(adapter: Adapter, details: SocketDetails<SocketData>) {
     this.id = details.id;
     this.handshake = details.handshake;
     this.rooms = new Set(details.rooms);

lib/client.ts

@@ -21,21 +21,27 @@ interface WriteOptions {
 export class Client<
   ListenEvents extends EventsMap,
   EmitEvents extends EventsMap,
-  ServerSideEvents extends EventsMap
+  ServerSideEvents extends EventsMap,
+  SocketData = any
 > {
   public readonly conn: RawSocket;
 
   private readonly id: string;
-  private readonly server: Server<ListenEvents, EmitEvents, ServerSideEvents>;
+  private readonly server: Server<
+    ListenEvents,
+    EmitEvents,
+    ServerSideEvents,
+    SocketData
+  >;
   private readonly encoder: Encoder;
   private readonly decoder: Decoder;
   private sockets: Map<
     SocketId,
-    Socket<ListenEvents, EmitEvents, ServerSideEvents>
+    Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
   > = new Map();
   private nsps: Map<
     string,
-    Socket<ListenEvents, EmitEvents, ServerSideEvents>
+    Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
   > = new Map();
   private connectTimeout?: NodeJS.Timeout;
 
@@ -47,7 +53,7 @@ export class Client<
    * @package
    */
   constructor(
-    server: Server<ListenEvents, EmitEvents, ServerSideEvents>,
+    server: Server<ListenEvents, EmitEvents, ServerSideEvents, SocketData>,
     conn: any
   ) {
     this.server = server;
@@ -112,7 +118,7 @@ export class Client<
       auth,
       (
         dynamicNspName:
-          | Namespace<ListenEvents, EmitEvents, ServerSideEvents>
+          | Namespace<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
           | false
       ) => {
         if (dynamicNspName) {
@@ -171,7 +177,9 @@ export class Client<
    *
    * @private
    */
-  _remove(socket: Socket<ListenEvents, EmitEvents, ServerSideEvents>): void {
+  _remove(
+    socket: Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
+  ): void {
     if (this.sockets.has(socket.id)) {
       const nsp = this.sockets.get(socket.id)!.nsp.name;
       this.sockets.delete(socket.id);

lib/index.ts

@@ -9,6 +9,7 @@ import {
   Server as Engine,
   ServerOptions as EngineOptions,
   AttachOptions,
+  uServer,
 } from "engine.io";
 import { Client } from "./client";
 import { EventEmitter } from "events";
@@ -27,6 +28,7 @@ import {
   StrictEventEmitter,
   EventNames,
 } from "./typed-events";
+import { patchAdapter, restoreAdapter, serveFile } from "./uws.js";
 
 const debug = debugModule("socket.io:server");
 
@@ -72,16 +74,23 @@ interface ServerOptions extends EngineOptions, AttachOptions {
 export class Server<
   ListenEvents extends EventsMap = DefaultEventsMap,
   EmitEvents extends EventsMap = ListenEvents,
-  ServerSideEvents extends EventsMap = DefaultEventsMap
+  ServerSideEvents extends EventsMap = DefaultEventsMap,
+  SocketData = any
 > extends StrictEventEmitter<
   ServerSideEvents,
   EmitEvents,
-  ServerReservedEventsMap<ListenEvents, EmitEvents, ServerSideEvents>
+  ServerReservedEventsMap<
+    ListenEvents,
+    EmitEvents,
+    ServerSideEvents,
+    SocketData
+  >
 > {
   public readonly sockets: Namespace<
     ListenEvents,
     EmitEvents,
-    ServerSideEvents
+    ServerSideEvents,
+    SocketData
   >;
   /**
    * A reference to the underlying Engine.IO server.
@@ -103,11 +112,13 @@ export class Server<
   /**
    * @private
    */
-  _nsps: Map<string, Namespace<ListenEvents, EmitEvents, ServerSideEvents>> =
-    new Map();
+  _nsps: Map<
+    string,
+    Namespace<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
+  > = new Map();
   private parentNsps: Map<
     ParentNspNameMatchFn,
-    ParentNamespace<ListenEvents, EmitEvents, ServerSideEvents>
+    ParentNamespace<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
   > = new Map();
   private _adapter?: AdapterConstructor;
   private _serveClient: boolean;
@@ -188,7 +199,9 @@ export class Server<
     name: string,
     auth: { [key: string]: any },
     fn: (
-      nsp: Namespace<ListenEvents, EmitEvents, ServerSideEvents> | false
+      nsp:
+        | Namespace<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
+        | false
     ) => void
   ): void {
     if (this.parentNsps.size === 0) return fn(false);
@@ -337,6 +350,69 @@ export class Server<
     return this;
   }
 
+  public attachApp(app /*: TemplatedApp */, opts: Partial<ServerOptions> = {}) {
+    // merge the options passed to the Socket.IO server
+    Object.assign(opts, this.opts);
+    // set engine.io path to `/socket.io`
+    opts.path = opts.path || this._path;
+
+    // initialize engine
+    debug("creating uWebSockets.js-based engine with opts %j", opts);
+    const engine = new uServer(opts);
+
+    engine.attach(app, opts);
+
+    // bind to engine events
+    this.bind(engine);
+
+    if (this._serveClient) {
+      // attach static file serving
+      app.get(`${this._path}/*`, (res, req) => {
+        if (!this.clientPathRegex.test(req.getUrl())) {
+          req.setYield(true);
+          return;
+        }
+
+        const filename = req
+          .getUrl()
+          .replace(this._path, "")
+          .replace(/\?.*$/, "")
+          .replace(/^\//, "");
+        const isMap = dotMapRegex.test(filename);
+        const type = isMap ? "map" : "source";
+
+        // Per the standard, ETags must be quoted:
+        // https://tools.ietf.org/html/rfc7232#section-2.3
+        const expectedEtag = '"' + clientVersion + '"';
+        const weakEtag = "W/" + expectedEtag;
+
+        const etag = req.getHeader("if-none-match");
+        if (etag) {
+          if (expectedEtag === etag || weakEtag === etag) {
+            debug("serve client %s 304", type);
+            res.writeStatus("304 Not Modified");
+            res.end();
+            return;
+          }
+        }
+
+        debug("serve client %s", type);
+
+        res.writeHeader("cache-control", "public, max-age=0");
+        res.writeHeader(
+          "content-type",
+          "application/" + (isMap ? "json" : "javascript")
+        );
+        res.writeHeader("etag", expectedEtag);
+
+        const filepath = path.join(__dirname, "../client-dist/", filename);
+        serveFile(res, filepath);
+      });
+    }
+
+    patchAdapter(app);
+  }
+
   /**
    * Initialize engine
    *
@@ -504,8 +580,10 @@ export class Server<
    */
   public of(
     name: string | RegExp | ParentNspNameMatchFn,
-    fn?: (socket: Socket<ListenEvents, EmitEvents, ServerSideEvents>) => void
-  ): Namespace<ListenEvents, EmitEvents, ServerSideEvents> {
+    fn?: (
+      socket: Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
+    ) => void
+  ): Namespace<ListenEvents, EmitEvents, ServerSideEvents, SocketData> {
     if (typeof name === "function" || name instanceof RegExp) {
       const parentNsp = new ParentNamespace(this);
       debug("initializing parent namespace %s", parentNsp.name);
@@ -553,6 +631,9 @@ export class Server<
 
     this.engine.close();
 
+    // restore the Adapter prototype
+    restoreAdapter();
+
     if (this.httpServer) {
       this.httpServer.close(fn);
     } else {
@@ -568,7 +649,7 @@ export class Server<
    */
   public use(
     fn: (
-      socket: Socket<ListenEvents, EmitEvents, ServerSideEvents>,
+      socket: Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>,
       next: (err?: ExtendedError) => void
     ) => void
   ): this {
@@ -583,7 +664,7 @@ export class Server<
    * @return self
    * @public
    */
-  public to(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public to(room: Room | Room[]): BroadcastOperator<EmitEvents, SocketData> {
     return this.sockets.to(room);
   }
 
@@ -594,7 +675,7 @@ export class Server<
    * @return self
    * @public
    */
-  public in(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public in(room: Room | Room[]): BroadcastOperator<EmitEvents, SocketData> {
     return this.sockets.in(room);
   }
 
@@ -605,7 +686,9 @@ export class Server<
    * @return self
    * @public
    */
-  public except(name: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public except(
+    name: Room | Room[]
+  ): BroadcastOperator<EmitEvents, SocketData> {
     return this.sockets.except(name);
   }
 
@@ -661,7 +744,9 @@ export class Server<
    * @return self
    * @public
    */
-  public compress(compress: boolean): BroadcastOperator<EmitEvents> {
+  public compress(
+    compress: boolean
+  ): BroadcastOperator<EmitEvents, SocketData> {
     return this.sockets.compress(compress);
   }
 
@@ -673,7 +758,7 @@ export class Server<
    * @return self
    * @public
    */
-  public get volatile(): BroadcastOperator<EmitEvents> {
+  public get volatile(): BroadcastOperator<EmitEvents, SocketData> {
     return this.sockets.volatile;
   }
 
@@ -683,7 +768,7 @@ export class Server<
    * @return self
    * @public
    */
-  public get local(): BroadcastOperator<EmitEvents> {
+  public get local(): BroadcastOperator<EmitEvents, SocketData> {
     return this.sockets.local;
   }
 
@@ -692,7 +777,7 @@ export class Server<
    *
    * @public
    */
-  public fetchSockets(): Promise<RemoteSocket<EmitEvents>[]> {
+  public fetchSockets(): Promise<RemoteSocket<EmitEvents, SocketData>[]> {
     return this.sockets.fetchSockets();
   }
 
@@ -749,3 +834,4 @@ module.exports.Namespace = Namespace;
 module.exports.Socket = Socket;
 
 export { Socket, ServerOptions, Namespace, BroadcastOperator, RemoteSocket };
+export { Event } from "./socket";

lib/namespace.ts

@@ -21,56 +21,72 @@ export interface ExtendedError extends Error {
 export interface NamespaceReservedEventsMap<
   ListenEvents extends EventsMap,
   EmitEvents extends EventsMap,
-  ServerSideEvents extends EventsMap
+  ServerSideEvents extends EventsMap,
+  SocketData
 > {
-  connect: (socket: Socket<ListenEvents, EmitEvents, ServerSideEvents>) => void;
+  connect: (
+    socket: Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
+  ) => void;
   connection: (
-    socket: Socket<ListenEvents, EmitEvents, ServerSideEvents>
+    socket: Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
   ) => void;
 }
 
 export interface ServerReservedEventsMap<
   ListenEvents,
   EmitEvents,
-  ServerSideEvents
+  ServerSideEvents,
+  SocketData
 > extends NamespaceReservedEventsMap<
     ListenEvents,
     EmitEvents,
-    ServerSideEvents
+    ServerSideEvents,
+    SocketData
   > {
   new_namespace: (
-    namespace: Namespace<ListenEvents, EmitEvents, ServerSideEvents>
+    namespace: Namespace<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
   ) => void;
 }
 
 export const RESERVED_EVENTS: ReadonlySet<string | Symbol> = new Set<
-  keyof ServerReservedEventsMap<never, never, never>
+  keyof ServerReservedEventsMap<never, never, never, never>
 >(<const>["connect", "connection", "new_namespace"]);
 
 export class Namespace<
   ListenEvents extends EventsMap = DefaultEventsMap,
   EmitEvents extends EventsMap = ListenEvents,
-  ServerSideEvents extends EventsMap = DefaultEventsMap
+  ServerSideEvents extends EventsMap = DefaultEventsMap,
+  SocketData = any
 > extends StrictEventEmitter<
   ServerSideEvents,
   EmitEvents,
-  NamespaceReservedEventsMap<ListenEvents, EmitEvents, ServerSideEvents>
+  NamespaceReservedEventsMap<
+    ListenEvents,
+    EmitEvents,
+    ServerSideEvents,
+    SocketData
+  >
 > {
   public readonly name: string;
   public readonly sockets: Map<
     SocketId,
-    Socket<ListenEvents, EmitEvents, ServerSideEvents>
+    Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
   > = new Map();
 
   public adapter: Adapter;
 
   /** @private */
-  readonly server: Server<ListenEvents, EmitEvents, ServerSideEvents>;
+  readonly server: Server<
+    ListenEvents,
+    EmitEvents,
+    ServerSideEvents,
+    SocketData
+  >;
 
   /** @private */
   _fns: Array<
     (
-      socket: Socket<ListenEvents, EmitEvents, ServerSideEvents>,
+      socket: Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>,
       next: (err?: ExtendedError) => void
     ) => void
   > = [];
@@ -85,7 +101,7 @@ export class Namespace<
    * @param name
    */
   constructor(
-    server: Server<ListenEvents, EmitEvents, ServerSideEvents>,
+    server: Server<ListenEvents, EmitEvents, ServerSideEvents, SocketData>,
     name: string
   ) {
     super();
@@ -114,7 +130,7 @@ export class Namespace<
    */
   public use(
     fn: (
-      socket: Socket<ListenEvents, EmitEvents, ServerSideEvents>,
+      socket: Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>,
       next: (err?: ExtendedError) => void
     ) => void
   ): this {
@@ -130,7 +146,7 @@ export class Namespace<
    * @private
    */
   private run(
-    socket: Socket<ListenEvents, EmitEvents, ServerSideEvents>,
+    socket: Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>,
     fn: (err: ExtendedError | null) => void
   ) {
     const fns = this._fns.slice(0);
@@ -159,7 +175,7 @@ export class Namespace<
    * @return self
    * @public
    */
-  public to(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public to(room: Room | Room[]): BroadcastOperator<EmitEvents, SocketData> {
     return new BroadcastOperator(this.adapter).to(room);
   }
 
@@ -170,7 +186,7 @@ export class Namespace<
    * @return self
    * @public
    */
-  public in(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public in(room: Room | Room[]): BroadcastOperator<EmitEvents, SocketData> {
     return new BroadcastOperator(this.adapter).in(room);
   }
 
@@ -181,7 +197,9 @@ export class Namespace<
    * @return self
    * @public
    */
-  public except(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public except(
+    room: Room | Room[]
+  ): BroadcastOperator<EmitEvents, SocketData> {
     return new BroadcastOperator(this.adapter).except(room);
   }
 
@@ -195,7 +213,7 @@ export class Namespace<
     client: Client<ListenEvents, EmitEvents, ServerSideEvents>,
     query,
     fn?: () => void
-  ): Socket<ListenEvents, EmitEvents, ServerSideEvents> {
+  ): Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData> {
     debug("adding socket to nsp %s", this.name);
     const socket = new Socket(this, client, query);
     this.run(socket, (err) => {
@@ -238,7 +256,9 @@ export class Namespace<
    *
    * @private
    */
-  _remove(socket: Socket<ListenEvents, EmitEvents, ServerSideEvents>): void {
+  _remove(
+    socket: Socket<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
+  ): void {
     if (this.sockets.has(socket.id)) {
       this.sockets.delete(socket.id);
     } else {
@@ -256,7 +276,10 @@ export class Namespace<
     ev: Ev,
     ...args: EventParams<EmitEvents, Ev>
   ): boolean {
-    return new BroadcastOperator<EmitEvents>(this.adapter).emit(ev, ...args);
+    return new BroadcastOperator<EmitEvents, SocketData>(this.adapter).emit(
+      ev,
+      ...args
+    );
   }
 
   /**
@@ -328,7 +351,9 @@ export class Namespace<
    * @return self
    * @public
    */
-  public compress(compress: boolean): BroadcastOperator<EmitEvents> {
+  public compress(
+    compress: boolean
+  ): BroadcastOperator<EmitEvents, SocketData> {
     return new BroadcastOperator(this.adapter).compress(compress);
   }
 
@@ -340,7 +365,7 @@ export class Namespace<
    * @return self
    * @public
    */
-  public get volatile(): BroadcastOperator<EmitEvents> {
+  public get volatile(): BroadcastOperator<EmitEvents, SocketData> {
     return new BroadcastOperator(this.adapter).volatile;
   }
 
@@ -350,7 +375,7 @@ export class Namespace<
    * @return self
    * @public
    */
-  public get local(): BroadcastOperator<EmitEvents> {
+  public get local(): BroadcastOperator<EmitEvents, SocketData> {
     return new BroadcastOperator(this.adapter).local;
   }
 
@@ -359,7 +384,7 @@ export class Namespace<
    *
    * @public
    */
-  public fetchSockets(): Promise<RemoteSocket<EmitEvents>[]> {
+  public fetchSockets(): Promise<RemoteSocket<EmitEvents, SocketData>[]> {
     return new BroadcastOperator(this.adapter).fetchSockets();
   }
 

lib/parent-namespace.ts

@@ -1,5 +1,5 @@
 import { Namespace } from "./namespace";
-import type { Server } from "./index";
+import type { Server, RemoteSocket } from "./index";
 import type {
   EventParams,
   EventNames,
@@ -11,13 +11,17 @@ import type { BroadcastOptions } from "socket.io-adapter";
 export class ParentNamespace<
   ListenEvents extends EventsMap = DefaultEventsMap,
   EmitEvents extends EventsMap = ListenEvents,
-  ServerSideEvents extends EventsMap = DefaultEventsMap
-> extends Namespace<ListenEvents, EmitEvents, ServerSideEvents> {
+  ServerSideEvents extends EventsMap = DefaultEventsMap,
+  SocketData = any
+> extends Namespace<ListenEvents, EmitEvents, ServerSideEvents, SocketData> {
   private static count: number = 0;
-  private children: Set<Namespace<ListenEvents, EmitEvents, ServerSideEvents>> =
-    new Set();
+  private children: Set<
+    Namespace<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
+  > = new Set();
 
-  constructor(server: Server<ListenEvents, EmitEvents, ServerSideEvents>) {
+  constructor(
+    server: Server<ListenEvents, EmitEvents, ServerSideEvents, SocketData>
+  ) {
     super(server, "/_" + ParentNamespace.count++);
   }
 
@@ -47,7 +51,7 @@ export class ParentNamespace<
 
   createChild(
     name: string
-  ): Namespace<ListenEvents, EmitEvents, ServerSideEvents> {
+  ): Namespace<ListenEvents, EmitEvents, ServerSideEvents, SocketData> {
     const namespace = new Namespace(this.server, name);
     namespace._fns = this._fns.slice(0);
     this.listeners("connect").forEach((listener) =>
@@ -60,4 +64,13 @@ export class ParentNamespace<
     this.server._nsps.set(name, namespace);
     return namespace;
   }
+
+  fetchSockets(): Promise<RemoteSocket<EmitEvents, SocketData>[]> {
+    // note: we could make the fetchSockets() method work for dynamic namespaces created with a regex (by sending the
+    // regex to the other Socket.IO servers, and returning the sockets of each matching namespace for example), but
+    // the behavior for namespaces created with a function is less clear
+    // note²: we cannot loop over each children namespace, because with multiple Socket.IO servers, a given namespace
+    // may exist on one node but not exist on another (since it is created upon client connection)
+    throw new Error("fetchSockets() is not supported on parent namespaces");
+  }
 }

lib/socket.ts

@@ -1,5 +1,4 @@
 import { Packet, PacketType } from "socket.io-parser";
-import url = require("url");
 import debugModule from "debug";
 import type { Server } from "./index";
 import {
@@ -46,7 +45,7 @@ export interface EventEmitterReservedEventsMap {
 
 export const RESERVED_EVENTS: ReadonlySet<string | Symbol> = new Set<
   | ClientReservedEvents
-  | keyof NamespaceReservedEventsMap<never, never, never>
+  | keyof NamespaceReservedEventsMap<never, never, never, never>
   | keyof SocketReservedEventsMap
   | keyof EventEmitterReservedEventsMap
 >(<const>[
@@ -108,12 +107,13 @@ export interface Handshake {
   auth: { [key: string]: any };
 }
 
-type Event = [eventName: string, ...args: any[]];
+export type Event = [eventName: string, ...args: any[]];
 
 export class Socket<
   ListenEvents extends EventsMap = DefaultEventsMap,
   EmitEvents extends EventsMap = ListenEvents,
-  ServerSideEvents extends EventsMap = DefaultEventsMap
+  ServerSideEvents extends EventsMap = DefaultEventsMap,
+  SocketData = any
 > extends StrictEventEmitter<
   ListenEvents,
   EmitEvents,
@@ -124,16 +124,20 @@ export class Socket<
   /**
    * Additional information that can be attached to the Socket instance and which will be used in the fetchSockets method
    */
-  public data: any = {};
+  public data: Partial<SocketData> = {};
 
-  public connected: boolean;
-  public disconnected: boolean;
+  public connected: boolean = false;
 
-  private readonly server: Server<ListenEvents, EmitEvents, ServerSideEvents>;
+  private readonly server: Server<
+    ListenEvents,
+    EmitEvents,
+    ServerSideEvents,
+    SocketData
+  >;
   private readonly adapter: Adapter;
   private acks: Map<number, () => void> = new Map();
   private fns: Array<(event: Event, next: (err?: Error) => void) => void> = [];
-  private flags: BroadcastFlags = {};
+  private flags: BroadcastFlags & { timeout?: number } = {};
   private _anyListeners?: Array<(...args: any[]) => void>;
 
   /**
@@ -158,8 +162,6 @@ export class Socket<
     } else {
       this.id = base64id.generateId(); // don't reuse the Engine.IO id because it's sensitive information
     }
-    this.connected = true;
-    this.disconnected = false;
     this.handshake = this.buildHandshake(auth);
   }
 
@@ -178,7 +180,8 @@ export class Socket<
       secure: !!this.request.connection.encrypted,
       issued: +new Date(),
       url: this.request.url!,
-      query: url.parse(this.request.url!, true).query,
+      // @ts-ignore
+      query: this.request._query,
       auth,
     };
   }
@@ -204,9 +207,11 @@ export class Socket<
 
     // access last argument to see if it's an ACK callback
     if (typeof data[data.length - 1] === "function") {
-      debug("emitting packet with ack id %d", this.nsp._ids);
-      this.acks.set(this.nsp._ids, data.pop());
-      packet.id = this.nsp._ids++;
+      const id = this.nsp._ids++;
+      debug("emitting packet with ack id %d", id);
+
+      this.registerAckCallback(id, data.pop());
+      packet.id = id;
     }
 
     const flags = Object.assign({}, this.flags);
@@ -217,6 +222,28 @@ export class Socket<
     return true;
   }
 
+  /**
+   * @private
+   */
+  private registerAckCallback(id: number, ack: (...args: any[]) => void): void {
+    const timeout = this.flags.timeout;
+    if (timeout === undefined) {
+      this.acks.set(id, ack);
+      return;
+    }
+
+    const timer = setTimeout(() => {
+      debug("event with ack id %d has timed out after %d ms", id, timeout);
+      this.acks.delete(id);
+      ack.call(this, new Error("operation has timed out"));
+    }, timeout);
+
+    this.acks.set(id, (...args) => {
+      clearTimeout(timer);
+      ack.apply(this, [null, ...args]);
+    });
+  }
+
   /**
    * Targets a room when broadcasting.
    *
@@ -224,7 +251,7 @@ export class Socket<
    * @return self
    * @public
    */
-  public to(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public to(room: Room | Room[]): BroadcastOperator<EmitEvents, SocketData> {
     return this.newBroadcastOperator().to(room);
   }
 
@@ -235,7 +262,7 @@ export class Socket<
    * @return self
    * @public
    */
-  public in(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public in(room: Room | Room[]): BroadcastOperator<EmitEvents, SocketData> {
     return this.newBroadcastOperator().in(room);
   }
 
@@ -246,7 +273,9 @@ export class Socket<
    * @return self
    * @public
    */
-  public except(room: Room | Room[]): BroadcastOperator<EmitEvents> {
+  public except(
+    room: Room | Room[]
+  ): BroadcastOperator<EmitEvents, SocketData> {
     return this.newBroadcastOperator().except(room);
   }
 
@@ -336,6 +365,7 @@ export class Socket<
    */
   _onconnect(): void {
     debug("socket connected - writing packet");
+    this.connected = true;
     this.join(this.id);
     if (this.conn.protocol === 3) {
       this.packet({ type: PacketType.CONNECT });
@@ -483,7 +513,6 @@ export class Socket<
     this.nsp._remove(this);
     this.client._remove(this);
     this.connected = false;
-    this.disconnected = true;
     this.emitReserved("disconnect", reason);
     return;
   }
@@ -550,7 +579,7 @@ export class Socket<
    * @return {Socket} self
    * @public
    */
-  public get broadcast(): BroadcastOperator<EmitEvents> {
+  public get broadcast(): BroadcastOperator<EmitEvents, SocketData> {
     return this.newBroadcastOperator();
   }
 
@@ -560,10 +589,30 @@ export class Socket<
    * @return {Socket} self
    * @public
    */
-  public get local(): BroadcastOperator<EmitEvents> {
+  public get local(): BroadcastOperator<EmitEvents, SocketData> {
     return this.newBroadcastOperator().local;
   }
 
+  /**
+   * Sets a modifier for a subsequent event emission that the callback will be called with an error when the
+   * given number of milliseconds have elapsed without an acknowledgement from the client:
+   *
+   * ```
+   * socket.timeout(5000).emit("my-event", (err) => {
+   *   if (err) {
+   *     // the client did not acknowledge the event in the given delay
+   *   }
+   * });
+   * ```
+   *
+   * @returns self
+   * @public
+   */
+  public timeout(timeout: number): this {
+    this.flags.timeout = timeout;
+    return this;
+  }
+
   /**
    * Dispatch incoming event to socket listeners.
    *
@@ -625,6 +674,13 @@ export class Socket<
     run(0);
   }
 
+  /**
+   * Whether the socket is currently disconnected
+   */
+  public get disconnected() {
+    return !this.connected;
+  }
+
   /**
    * A reference to the request that originated the underlying Engine.IO Socket.
    *
@@ -710,7 +766,7 @@ export class Socket<
     return this._anyListeners || [];
   }
 
-  private newBroadcastOperator(): BroadcastOperator<EmitEvents> {
+  private newBroadcastOperator(): BroadcastOperator<EmitEvents, SocketData> {
     const flags = Object.assign({}, this.flags);
     this.flags = {};
     return new BroadcastOperator(

lib/uws.ts

@@ -0,0 +1,162 @@
+import { Adapter, Room } from "socket.io-adapter";
+import type { WebSocket } from "uWebSockets.js";
+import type { Socket } from "./socket.js";
+import { createReadStream, statSync } from "fs";
+import debugModule from "debug";
+
+const debug = debugModule("socket.io:adapter-uws");
+
+const SEPARATOR = "\x1f"; // see https://en.wikipedia.org/wiki/Delimiter#ASCII_delimited_text
+
+const { addAll, del, broadcast } = Adapter.prototype;
+
+export function patchAdapter(app /* : TemplatedApp */) {
+  Adapter.prototype.addAll = function (id, rooms) {
+    const isNew = !this.sids.has(id);
+    addAll.call(this, id, rooms);
+    const socket: Socket = this.nsp.sockets.get(id);
+    if (!socket) {
+      return;
+    }
+    if (socket.conn.transport.name === "websocket") {
+      subscribe(this.nsp.name, socket, isNew, rooms);
+      return;
+    }
+    if (isNew) {
+      socket.conn.on("upgrade", () => {
+        const rooms = this.sids.get(id);
+        subscribe(this.nsp.name, socket, isNew, rooms);
+      });
+    }
+  };
+
+  Adapter.prototype.del = function (id, room) {
+    del.call(this, id, room);
+    const socket: Socket = this.nsp.sockets.get(id);
+    if (socket && socket.conn.transport.name === "websocket") {
+      // @ts-ignore
+      const sessionId = socket.conn.id;
+      // @ts-ignore
+      const websocket: WebSocket = socket.conn.transport.socket;
+      const topic = `${this.nsp.name}${SEPARATOR}${room}`;
+      debug("unsubscribe connection %s from topic %s", sessionId, topic);
+      websocket.unsubscribe(topic);
+    }
+  };
+
+  Adapter.prototype.broadcast = function (packet, opts) {
+    const useFastPublish = opts.rooms.size <= 1 && opts.except!.size === 0;
+    if (!useFastPublish) {
+      broadcast.call(this, packet, opts);
+      return;
+    }
+
+    const flags = opts.flags || {};
+    const basePacketOpts = {
+      preEncoded: true,
+      volatile: flags.volatile,
+      compress: flags.compress,
+    };
+
+    packet.nsp = this.nsp.name;
+    const encodedPackets = this.encoder.encode(packet);
+
+    const topic =
+      opts.rooms.size === 0
+        ? this.nsp.name
+        : `${this.nsp.name}${SEPARATOR}${opts.rooms.keys().next().value}`;
+    debug("fast publish to %s", topic);
+
+    // fast publish for clients connected with WebSocket
+    encodedPackets.forEach((encodedPacket) => {
+      const isBinary = typeof encodedPacket !== "string";
+      // "4" being the message type in the Engine.IO protocol, see https://github.com/socketio/engine.io-protocol
+      app.publish(
+        topic,
+        isBinary ? encodedPacket : "4" + encodedPacket,
+        isBinary
+      );
+    });
+
+    this.apply(opts, (socket) => {
+      if (socket.conn.transport.name !== "websocket") {
+        // classic publish for clients connected with HTTP long-polling
+        socket.client.writeToEngine(encodedPackets, basePacketOpts);
+      }
+    });
+  };
+}
+
+function subscribe(
+  namespaceName: string,
+  socket: Socket,
+  isNew: boolean,
+  rooms: Set<Room>
+) {
+  // @ts-ignore
+  const sessionId = socket.conn.id;
+  // @ts-ignore
+  const websocket: WebSocket = socket.conn.transport.socket;
+  if (isNew) {
+    debug("subscribe connection %s to topic %s", sessionId, namespaceName);
+    websocket.subscribe(namespaceName);
+  }
+  rooms.forEach((room) => {
+    const topic = `${namespaceName}${SEPARATOR}${room}`; // '#' can be used as wildcard
+    debug("subscribe connection %s to topic %s", sessionId, topic);
+    websocket.subscribe(topic);
+  });
+}
+
+export function restoreAdapter() {
+  Adapter.prototype.addAll = addAll;
+  Adapter.prototype.del = del;
+  Adapter.prototype.broadcast = broadcast;
+}
+
+const toArrayBuffer = (buffer: Buffer) => {
+  const { buffer: arrayBuffer, byteOffset, byteLength } = buffer;
+  return arrayBuffer.slice(byteOffset, byteOffset + byteLength);
+};
+
+// imported from https://github.com/kolodziejczak-sz/uwebsocket-serve
+export function serveFile(res /* : HttpResponse */, filepath: string) {
+  const { size } = statSync(filepath);
+  const readStream = createReadStream(filepath);
+  const destroyReadStream = () => !readStream.destroyed && readStream.destroy();
+
+  const onError = (error: Error) => {
+    destroyReadStream();
+    throw error;
+  };
+
+  const onDataChunk = (chunk: Buffer) => {
+    const arrayBufferChunk = toArrayBuffer(chunk);
+
+    const lastOffset = res.getWriteOffset();
+    const [ok, done] = res.tryEnd(arrayBufferChunk, size);
+
+    if (!done && !ok) {
+      readStream.pause();
+
+      res.onWritable((offset) => {
+        const [ok, done] = res.tryEnd(
+          arrayBufferChunk.slice(offset - lastOffset),
+          size
+        );
+
+        if (!done && ok) {
+          readStream.resume();
+        }
+
+        return ok;
+      });
+    }
+  };
+
+  res.onAborted(destroyReadStream);
+  readStream
+    .on("data", onDataChunk)
+    .on("error", onError)
+    .on("end", destroyReadStream);
+}

package.json

@@ -1,6 +1,6 @@
 {
   "name": "socket.io",
-  "version": "4.3.2",
+  "version": "4.4.1",
   "description": "node.js realtime framework server",
   "keywords": [
     "realtime",
@@ -27,7 +27,8 @@
   "main": "./dist/index.js",
   "exports": {
     "import": "./wrapper.mjs",
-    "require": "./dist/index.js"
+    "require": "./dist/index.js",
+    "types": "./dist/index.d.ts"
   },
   "types": "./dist/index.d.ts",
   "license": "MIT",
@@ -48,8 +49,8 @@
     "accepts": "~1.3.4",
     "base64id": "~2.0.0",
     "debug": "~4.3.2",
-    "engine.io": "~6.0.0",
-    "socket.io-adapter": "~2.3.2",
+    "engine.io": "~6.1.0",
+    "socket.io-adapter": "~2.3.3",
     "socket.io-parser": "~4.0.4"
   },
   "devDependencies": {
@@ -59,13 +60,14 @@
     "nyc": "^15.1.0",
     "prettier": "^2.3.2",
     "rimraf": "^3.0.2",
-    "socket.io-client": "4.3.2",
+    "socket.io-client": "4.4.1",
     "socket.io-client-v2": "npm:socket.io-client@^2.4.0",
     "superagent": "^6.1.0",
     "supertest": "^6.1.6",
     "ts-node": "^10.2.1",
     "tsd": "^0.17.0",
-    "typescript": "^4.4.2"
+    "typescript": "^4.4.2",
+    "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.0.0"
   },
   "contributors": [
     {

test/socket-timeout.ts

@@ -0,0 +1,57 @@
+import { Server } from "..";
+import { createClient, success } from "./support/util";
+import expect from "expect.js";
+
+describe("timeout", () => {
+  it("should timeout if the client does not acknowledge the event", (done) => {
+    const io = new Server(0);
+    const client = createClient(io, "/");
+
+    io.on("connection", (socket) => {
+      socket.timeout(50).emit("unknown", (err) => {
+        expect(err).to.be.an(Error);
+        success(done, io, client);
+      });
+    });
+  });
+
+  it("should timeout if the client does not acknowledge the event in time", (done) => {
+    const io = new Server(0);
+    const client = createClient(io, "/");
+
+    client.on("echo", (arg, cb) => {
+      cb(arg);
+    });
+
+    let count = 0;
+
+    io.on("connection", (socket) => {
+      socket.timeout(0).emit("echo", 42, (err) => {
+        expect(err).to.be.an(Error);
+        count++;
+      });
+    });
+
+    setTimeout(() => {
+      expect(count).to.eql(1);
+      success(done, io, client);
+    }, 200);
+  });
+
+  it("should not timeout if the client does acknowledge the event", (done) => {
+    const io = new Server(0);
+    const client = createClient(io, "/");
+
+    client.on("echo", (arg, cb) => {
+      cb(arg);
+    });
+
+    io.on("connection", (socket) => {
+      socket.timeout(50).emit("echo", 42, (err, value) => {
+        expect(err).to.be(null);
+        expect(value).to.be(42);
+        success(done, io, client);
+      });
+    });
+  });
+});

test/socket.io.ts

@@ -14,6 +14,7 @@ import { io as ioc, Socket as ClientSocket } from "socket.io-client";
 
 import "./support/util";
 import "./utility-methods";
+import "./uws";
 
 type callback = (err: Error | null, success: boolean) => void;
 
@@ -820,29 +821,6 @@ describe("socket.io", () => {
       });
     });
 
-    it("should close a client without namespace (2)", (done) => {
-      const srv = createServer();
-      const sio = new Server(srv, {
-        connectTimeout: 100,
-      });
-
-      sio.use((_, next) => {
-        next(new Error("nope"));
-      });
-
-      srv.listen(() => {
-        const socket = client(srv);
-
-        const success = () => {
-          socket.close();
-          sio.close();
-          done();
-        };
-
-        socket.on("disconnect", success);
-      });
-    });
-
     it("should exclude a specific socket when emitting", (done) => {
       const srv = createServer();
       const io = new Server(srv);
@@ -1072,7 +1050,7 @@ describe("socket.io", () => {
           reconnectionDelay: 100,
         });
         clientSocket.on("connect", () => {
-          srv.close();
+          sio.close();
         });
 
         clientSocket.io.on("reconnect_failed", () => {
@@ -1452,6 +1430,32 @@ describe("socket.io", () => {
       }, 200);
     });
 
+    it("should broadcast only one consecutive volatile event with binary (ws)", (done) => {
+      const srv = createServer();
+      const sio = new Server(srv, { transports: ["websocket"] });
+
+      let counter = 0;
+      srv.listen(() => {
+        sio.on("connection", (s) => {
+          // Wait to make sure there are no packets being sent for opening the connection
+          setTimeout(() => {
+            sio.volatile.emit("ev", Buffer.from([1, 2, 3]));
+            sio.volatile.emit("ev", Buffer.from([4, 5, 6]));
+          }, 20);
+        });
+
+        const socket = client(srv, { transports: ["websocket"] });
+        socket.on("ev", () => {
+          counter++;
+        });
+      });
+
+      setTimeout(() => {
+        expect(counter).to.be(1);
+        done();
+      }, 200);
+    });
+
     it("should emit regular events after trying a failed volatile event (polling)", (done) => {
       const srv = createServer();
       const sio = new Server(srv, { transports: ["polling"] });
@@ -1826,7 +1830,7 @@ describe("socket.io", () => {
           reconnectionDelay: 100,
         });
         clientSocket.once("connect", () => {
-          srv.close(() => {
+          sio.close(() => {
             clientSocket.io.on("reconnect", () => {
               clientSocket.emit("ev", "payload");
             });
@@ -2515,28 +2519,6 @@ describe("socket.io", () => {
         });
       });
     });
-
-    it("should pre encode a broadcast packet", (done) => {
-      const srv = createServer();
-      const sio = new Server(srv);
-
-      srv.listen(() => {
-        const clientSocket = client(srv, { multiplex: false });
-
-        sio.on("connection", (socket) => {
-          socket.conn.on("packetCreate", (packet) => {
-            expect(packet.data).to.eql('2["hello","world"]');
-            expect(packet.options.wsPreEncoded).to.eql('42["hello","world"]');
-
-            clientSocket.close();
-            sio.close();
-            done();
-          });
-
-          sio.emit("hello", "world");
-        });
-      });
-    });
   });
 
   describe("middleware", () => {
@@ -2725,6 +2707,25 @@ describe("socket.io", () => {
         if (++count === 2) done();
       });
     });
+
+    it("should only set `connected` to true after the middleware execution", (done) => {
+      const httpServer = createServer();
+      const io = new Server(httpServer);
+
+      const clientSocket = client(httpServer, "/");
+
+      io.use((socket, next) => {
+        expect(socket.connected).to.be(false);
+        expect(socket.disconnected).to.be(true);
+        next();
+      });
+
+      io.on("connection", (socket) => {
+        expect(socket.connected).to.be(true);
+        expect(socket.disconnected).to.be(false);
+        success(io, clientSocket, done);
+      });
+    });
   });
 
   describe("socket middleware", () => {
@@ -2864,4 +2865,6 @@ describe("socket.io", () => {
       });
     });
   });
+
+  require("./socket-timeout");
 });

test/support/util.ts

@@ -1,3 +1,11 @@
+import type { Server } from "../..";
+import {
+  io as ioc,
+  ManagerOptions,
+  Socket as ClientSocket,
+  SocketOptions,
+} from "socket.io-client";
+
 const expect = require("expect.js");
 const i = expect.stringify;
 
@@ -20,3 +28,19 @@ expect.Assertion.prototype.contain = function (...args) {
   }
   return contain.apply(this, args);
 };
+
+export function createClient(
+  io: Server,
+  nsp: string,
+  opts?: ManagerOptions & SocketOptions
+): ClientSocket {
+  // @ts-ignore
+  const port = io.httpServer.address().port;
+  return ioc(`http://localhost:${port}${nsp}`, opts);
+}
+
+export function success(done: Function, io: Server, client: ClientSocket) {
+  io.close();
+  client.disconnect();
+  done();
+}

test/uws.ts

@@ -0,0 +1,197 @@
+import { App, us_socket_local_port } from "uWebSockets.js";
+import { Server } from "..";
+import { io as ioc, Socket as ClientSocket } from "socket.io-client";
+import request from "supertest";
+import expect from "expect.js";
+
+const createPartialDone = (done: (err?: Error) => void, count: number) => {
+  let i = 0;
+  return () => {
+    if (++i === count) {
+      done();
+    } else if (i > count) {
+      done(new Error(`partialDone() called too many times: ${i} > ${count}`));
+    }
+  };
+};
+
+const shouldNotHappen = (done) => () => done(new Error("should not happen"));
+
+describe("socket.io with uWebSocket.js-based engine", () => {
+  let io: Server,
+    port: number,
+    client: ClientSocket,
+    clientWSOnly: ClientSocket,
+    clientPollingOnly: ClientSocket,
+    clientCustomNamespace: ClientSocket;
+
+  beforeEach((done) => {
+    const app = App();
+    io = new Server();
+    io.attachApp(app);
+
+    io.of("/custom");
+
+    app.listen(0, (listenSocket) => {
+      port = us_socket_local_port(listenSocket);
+
+      client = ioc(`http://localhost:${port}`);
+      clientWSOnly = ioc(`http://localhost:${port}`, {
+        transports: ["websocket"],
+      });
+      clientPollingOnly = ioc(`http://localhost:${port}`, {
+        transports: ["polling"],
+      });
+      clientCustomNamespace = ioc(`http://localhost:${port}/custom`);
+    });
+
+    const partialDone = createPartialDone(done, 4);
+    client.on("connect", partialDone);
+    clientWSOnly.on("connect", partialDone);
+    clientPollingOnly.on("connect", partialDone);
+    clientCustomNamespace.on("connect", partialDone);
+  });
+
+  afterEach(() => {
+    io.close();
+    client.disconnect();
+    clientWSOnly.disconnect();
+    clientPollingOnly.disconnect();
+    clientCustomNamespace.disconnect();
+  });
+
+  it("should broadcast", (done) => {
+    const partialDone = createPartialDone(done, 3);
+
+    client.on("hello", partialDone);
+    clientWSOnly.on("hello", partialDone);
+    clientPollingOnly.on("hello", partialDone);
+    clientCustomNamespace.on("hello", shouldNotHappen(done));
+
+    io.emit("hello");
+  });
+
+  it("should broadcast in a namespace", (done) => {
+    client.on("hello", shouldNotHappen(done));
+    clientWSOnly.on("hello", shouldNotHappen(done));
+    clientPollingOnly.on("hello", shouldNotHappen(done));
+    clientCustomNamespace.on("hello", done);
+
+    io.of("/custom").emit("hello");
+  });
+
+  it("should broadcast in a dynamic namespace", (done) => {
+    const dynamicNamespace = io.of(/\/dynamic-\d+/);
+    const dynamicClient = clientWSOnly.io.socket("/dynamic-101");
+
+    dynamicClient.on("connect", () => {
+      dynamicNamespace.emit("hello");
+    });
+
+    dynamicClient.on("hello", () => {
+      dynamicClient.disconnect();
+      done();
+    });
+  });
+
+  it("should broadcast binary content", (done) => {
+    const partialDone = createPartialDone(done, 3);
+
+    client.on("hello", partialDone);
+    clientWSOnly.on("hello", partialDone);
+    clientPollingOnly.on("hello", partialDone);
+    clientCustomNamespace.on("hello", shouldNotHappen(done));
+
+    io.emit("hello", Buffer.from([1, 2, 3]));
+  });
+
+  it("should broadcast volatile packet with binary content", (done) => {
+    const partialDone = createPartialDone(done, 3);
+
+    client.on("hello", partialDone);
+    clientWSOnly.on("hello", partialDone);
+    clientPollingOnly.on("hello", partialDone);
+    clientCustomNamespace.on("hello", shouldNotHappen(done));
+
+    // wait to make sure there are no packets being sent for opening the connection
+    setTimeout(() => {
+      io.volatile.emit("hello", Buffer.from([1, 2, 3]));
+    }, 20);
+  });
+
+  it("should broadcast in a room", (done) => {
+    const partialDone = createPartialDone(done, 2);
+
+    client.on("hello", shouldNotHappen(done));
+    clientWSOnly.on("hello", partialDone);
+    clientPollingOnly.on("hello", partialDone);
+    clientCustomNamespace.on("hello", shouldNotHappen(done));
+
+    io.of("/").sockets.get(clientWSOnly.id)!.join("room1");
+    io.of("/").sockets.get(clientPollingOnly.id)!.join("room1");
+
+    io.to("room1").emit("hello");
+  });
+
+  it("should broadcast in multiple rooms", (done) => {
+    const partialDone = createPartialDone(done, 2);
+
+    client.on("hello", shouldNotHappen(done));
+    clientWSOnly.on("hello", partialDone);
+    clientPollingOnly.on("hello", partialDone);
+    clientCustomNamespace.on("hello", shouldNotHappen(done));
+
+    io.of("/").sockets.get(clientWSOnly.id)!.join("room1");
+    io.of("/").sockets.get(clientPollingOnly.id)!.join("room2");
+
+    io.to(["room1", "room2"]).emit("hello");
+  });
+
+  it("should broadcast in all but a given room", (done) => {
+    const partialDone = createPartialDone(done, 2);
+
+    client.on("hello", partialDone);
+    clientWSOnly.on("hello", partialDone);
+    clientPollingOnly.on("hello", shouldNotHappen(done));
+    clientCustomNamespace.on("hello", shouldNotHappen(done));
+
+    io.of("/").sockets.get(clientWSOnly.id)!.join("room1");
+    io.of("/").sockets.get(clientPollingOnly.id)!.join("room2");
+
+    io.except("room2").emit("hello");
+  });
+
+  it("should work even after leaving room", (done) => {
+    const partialDone = createPartialDone(done, 2);
+
+    client.on("hello", partialDone);
+    clientWSOnly.on("hello", shouldNotHappen(done));
+    clientPollingOnly.on("hello", partialDone);
+    clientCustomNamespace.on("hello", shouldNotHappen(done));
+
+    io.of("/").sockets.get(client.id)!.join("room1");
+    io.of("/").sockets.get(clientPollingOnly.id)!.join("room1");
+
+    io.of("/").sockets.get(clientWSOnly.id)!.join("room1");
+    io.of("/").sockets.get(clientWSOnly.id)!.leave("room1");
+
+    io.to("room1").emit("hello");
+  });
+
+  it("should serve static files", (done) => {
+    const clientVersion = require("socket.io-client/package.json").version;
+
+    request(`http://localhost:${port}`)
+      .get("/socket.io/socket.io.js")
+      .buffer(true)
+      .end((err, res) => {
+        if (err) return done(err);
+        expect(res.headers["content-type"]).to.be("application/javascript");
+        expect(res.headers.etag).to.be('"' + clientVersion + '"');
+        expect(res.headers["x-sourcemap"]).to.be(undefined);
+        expect(res.text).to.match(/engine\.io/);
+        expect(res.status).to.be(200);
+        done();
+      });
+  });
+});