feat: add an implementation based on uWebSockets.js

```js
const { App } = require("uWebSockets.js");
const { uServer } = require("engine.io");

const app = new App();
const server = new uServer();

server.attach(app);

app.listen(3000);
```

Reference: https://github.com/uNetworking/uWebSockets.js

Related: #578

Author: darrachequesne

.github/workflows/ci.yml

@@ -12,7 +12,7 @@ jobs:
 
     strategy:
       matrix:
-        node-version: [10.x, 12.x, 14.x]
+        node-version: [12.x, 14.x]
 
     steps:
       - uses: actions/checkout@v2

lib/engine.io.ts

@@ -5,6 +5,7 @@ import * as parser from "engine.io-parser";
 
 export { Server, transports, listen, attach, parser };
 export { AttachOptions, ServerOptions } from "./server";
+export { uServer } from "./userver";
 export { Socket } from "./socket";
 export { Transport } from "./transport";
 export const protocol = parser.protocol;

lib/server.ts

@@ -113,15 +113,12 @@ export interface ServerOptions {
   allowEIO3?: boolean;
 }
 
-export class Server extends EventEmitter {
+export abstract class BaseServer extends EventEmitter {
   public opts: ServerOptions;
-  public httpServer?: HttpServer;
 
-  private clients: any;
+  protected clients: any;
   private clientsCount: number;
-  private ws: any;
-  private corsMiddleware: Function;
-  private perMessageDeflate: any;
+  protected corsMiddleware: Function;
 
   /**
    * Server constructor.
@@ -182,42 +179,7 @@ export class Server extends EventEmitter {
     this.init();
   }
 
-  /**
-   * Initialize websocket server
-   *
-   * @api private
-   */
-  private init() {
-    if (!~this.opts.transports.indexOf("websocket")) return;
-
-    if (this.ws) this.ws.close();
-
-    this.ws = new this.opts.wsEngine({
-      noServer: true,
-      clientTracking: false,
-      perMessageDeflate: this.opts.perMessageDeflate,
-      maxPayload: this.opts.maxHttpBufferSize
-    });
-
-    if (typeof this.ws.on === "function") {
-      this.ws.on("headers", (headersArray, req) => {
-        // note: 'ws' uses an array of headers, while Engine.IO uses an object (response.writeHead() accepts both formats)
-        // we could also try to parse the array and then sync the values, but that will be error-prone
-        const additionalHeaders = {};
-
-        const isInitialRequest = !req._query.sid;
-        if (isInitialRequest) {
-          this.emit("initial_headers", additionalHeaders, req);
-        }
-
-        this.emit("headers", additionalHeaders, req);
-
-        Object.keys(additionalHeaders).forEach(key => {
-          headersArray.push(`${key}: ${additionalHeaders[key]}`);
-        });
-      });
-    }
-  }
+  protected abstract init();
 
   /**
    * Returns a list of available transports for upgrade given a certain transport.
@@ -237,7 +199,7 @@ export class Server extends EventEmitter {
    * @return {Boolean} whether the request is valid
    * @api private
    */
-  private verify(req, upgrade, fn) {
+  protected verify(req, upgrade, fn) {
     // transport check
     const transport = req._query.transport;
     if (!~this.opts.transports.indexOf(transport)) {
@@ -298,18 +260,6 @@ export class Server extends EventEmitter {
     fn();
   }
 
-  /**
-   * Prepares a request by processing the query string.
-   *
-   * @api private
-   */
-  private prepare(req) {
-    // try to leverage pre-existing `req._query` (e.g: from connect)
-    if (!req._query) {
-      req._query = ~req.url.indexOf("?") ? qs.parse(parse(req.url).query) : {};
-    }
-  }
-
   /**
    * Closes all clients.
    *
@@ -322,56 +272,11 @@ export class Server extends EventEmitter {
         this.clients[i].close(true);
       }
     }
-    if (this.ws) {
-      debug("closing webSocketServer");
-      this.ws.close();
-      // don't delete this.ws because it can be used again if the http server starts listening again
-    }
+    this.cleanup();
     return this;
   }
 
-  /**
-   * Handles an Engine.IO HTTP request.
-   *
-   * @param {http.IncomingMessage} request
-   * @param {http.ServerResponse|http.OutgoingMessage} response
-   * @api public
-   */
-  public handleRequest(req, res) {
-    debug('handling "%s" http request "%s"', req.method, req.url);
-    this.prepare(req);
-    req.res = res;
-
-    const callback = (errorCode, errorContext) => {
-      if (errorCode !== undefined) {
-        this.emit("connection_error", {
-          req,
-          code: errorCode,
-          message: Server.errorMessages[errorCode],
-          context: errorContext
-        });
-        abortRequest(res, errorCode, errorContext);
-        return;
-      }
-
-      if (req._query.sid) {
-        debug("setting new request for existing client");
-        this.clients[req._query.sid].transport.onRequest(req);
-      } else {
-        const closeConnection = (errorCode, errorContext) =>
-          abortRequest(res, errorCode, errorContext);
-        this.handshake(req._query.transport, req, closeConnection);
-      }
-    };
-
-    if (this.corsMiddleware) {
-      this.corsMiddleware.call(null, req, res, () => {
-        this.verify(req, false, callback);
-      });
-    } else {
-      this.verify(req, false, callback);
-    }
-  }
+  protected abstract cleanup();
 
   /**
    * generate a socket id.
@@ -391,9 +296,9 @@ export class Server extends EventEmitter {
    * @param {Object} request object
    * @param {Function} closeConnection
    *
-   * @api private
+   * @api protected
    */
-  private async handshake(transportName, req, closeConnection) {
+  protected async handshake(transportName, req, closeConnection) {
     const protocol = req._query.EIO === "4" ? 4 : 3; // 3rd revision by default
     if (protocol === 3 && !this.opts.allowEIO3) {
       debug("unsupported protocol version");
@@ -431,7 +336,7 @@ export class Server extends EventEmitter {
     debug('handshaking client "%s"', id);
 
     try {
-      var transport = new transports[transportName](req);
+      var transport = this.createTransport(transportName, req);
       if ("polling" === transportName) {
         transport.maxHttpBufferSize = this.opts.maxHttpBufferSize;
         transport.httpCompression = this.opts.httpCompression;
@@ -486,6 +391,141 @@ export class Server extends EventEmitter {
     });
 
     this.emit("connection", socket);
+
+    return transport;
+  }
+
+  protected abstract createTransport(transportName, req);
+
+  /**
+   * Protocol errors mappings.
+   */
+
+  static errors = {
+    UNKNOWN_TRANSPORT: 0,
+    UNKNOWN_SID: 1,
+    BAD_HANDSHAKE_METHOD: 2,
+    BAD_REQUEST: 3,
+    FORBIDDEN: 4,
+    UNSUPPORTED_PROTOCOL_VERSION: 5
+  };
+
+  static errorMessages = {
+    0: "Transport unknown",
+    1: "Session ID unknown",
+    2: "Bad handshake method",
+    3: "Bad request",
+    4: "Forbidden",
+    5: "Unsupported protocol version"
+  };
+}
+
+export class Server extends BaseServer {
+  public httpServer?: HttpServer;
+  private ws: any;
+
+  /**
+   * Initialize websocket server
+   *
+   * @api protected
+   */
+  protected init() {
+    if (!~this.opts.transports.indexOf("websocket")) return;
+
+    if (this.ws) this.ws.close();
+
+    this.ws = new this.opts.wsEngine({
+      noServer: true,
+      clientTracking: false,
+      perMessageDeflate: this.opts.perMessageDeflate,
+      maxPayload: this.opts.maxHttpBufferSize
+    });
+
+    if (typeof this.ws.on === "function") {
+      this.ws.on("headers", (headersArray, req) => {
+        // note: 'ws' uses an array of headers, while Engine.IO uses an object (response.writeHead() accepts both formats)
+        // we could also try to parse the array and then sync the values, but that will be error-prone
+        const additionalHeaders = {};
+
+        const isInitialRequest = !req._query.sid;
+        if (isInitialRequest) {
+          this.emit("initial_headers", additionalHeaders, req);
+        }
+
+        this.emit("headers", additionalHeaders, req);
+
+        Object.keys(additionalHeaders).forEach(key => {
+          headersArray.push(`${key}: ${additionalHeaders[key]}`);
+        });
+      });
+    }
+  }
+
+  protected cleanup() {
+    if (this.ws) {
+      debug("closing webSocketServer");
+      this.ws.close();
+      // don't delete this.ws because it can be used again if the http server starts listening again
+    }
+  }
+
+  /**
+   * Prepares a request by processing the query string.
+   *
+   * @api private
+   */
+  private prepare(req) {
+    // try to leverage pre-existing `req._query` (e.g: from connect)
+    if (!req._query) {
+      req._query = ~req.url.indexOf("?") ? qs.parse(parse(req.url).query) : {};
+    }
+  }
+
+  protected createTransport(transportName, req) {
+    return new transports[transportName](req);
+  }
+
+  /**
+   * Handles an Engine.IO HTTP request.
+   *
+   * @param {http.IncomingMessage} request
+   * @param {http.ServerResponse|http.OutgoingMessage} response
+   * @api public
+   */
+  public handleRequest(req, res) {
+    debug('handling "%s" http request "%s"', req.method, req.url);
+    this.prepare(req);
+    req.res = res;
+
+    const callback = (errorCode, errorContext) => {
+      if (errorCode !== undefined) {
+        this.emit("connection_error", {
+          req,
+          code: errorCode,
+          message: Server.errorMessages[errorCode],
+          context: errorContext
+        });
+        abortRequest(res, errorCode, errorContext);
+        return;
+      }
+
+      if (req._query.sid) {
+        debug("setting new request for existing client");
+        this.clients[req._query.sid].transport.onRequest(req);
+      } else {
+        const closeConnection = (errorCode, errorContext) =>
+          abortRequest(res, errorCode, errorContext);
+        this.handshake(req._query.transport, req, closeConnection);
+      }
+    };
+
+    if (this.corsMiddleware) {
+      this.corsMiddleware.call(null, req, res, () => {
+        this.verify(req, false, callback);
+      });
+    } else {
+      this.verify(req, false, callback);
+    }
   }
 
   /**
@@ -559,13 +599,13 @@ export class Server extends EventEmitter {
         // transport error handling takes over
         websocket.removeListener("error", onUpgradeError);
 
-        const transport = new transports[req._query.transport](req);
+        const transport = this.createTransport(req._query.transport, req);
         if (req._query && req._query.b64) {
           transport.supportsBinary = false;
         } else {
           transport.supportsBinary = true;
         }
-        transport.perMessageDeflate = this.perMessageDeflate;
+        transport.perMessageDeflate = this.opts.perMessageDeflate;
         client.maybeUpgrade(transport);
       }
     } else {
@@ -590,7 +630,7 @@ export class Server extends EventEmitter {
    * @param {Object} options
    * @api public
    */
-  public attach(server, options: AttachOptions = {}) {
+  public attach(server: HttpServer, options: AttachOptions = {}) {
     let path = (options.path || "/engine.io").replace(/\/$/, "");
 
     const destroyUpgradeTimeout = options.destroyUpgradeTimeout || 1000;
@@ -632,6 +672,7 @@ export class Server extends EventEmitter {
           // and if no eio thing handles the upgrade
           // then the socket needs to die!
           setTimeout(function() {
+            // @ts-ignore
             if (socket.writable && socket.bytesWritten <= 0) {
               return socket.end();
             }
@@ -640,28 +681,6 @@ export class Server extends EventEmitter {
       });
     }
   }
-
-  /**
-   * Protocol errors mappings.
-   */
-
-  static errors = {
-    UNKNOWN_TRANSPORT: 0,
-    UNKNOWN_SID: 1,
-    BAD_HANDSHAKE_METHOD: 2,
-    BAD_REQUEST: 3,
-    FORBIDDEN: 4,
-    UNSUPPORTED_PROTOCOL_VERSION: 5
-  };
-
-  static errorMessages = {
-    0: "Transport unknown",
-    1: "Session ID unknown",
-    2: "Bad handshake method",
-    3: "Bad request",
-    4: "Forbidden",
-    5: "Unsupported protocol version"
-  };
 }
 
 /**