feat: support `uWebSockets.js

.gitignore

@@ -8,5 +8,5 @@ dist
 *.env*
 .wrangler
 
-adapters
+/adapters
 websocket.d.ts

README.md

@@ -5,7 +5,7 @@
 
 👉 Elegant, typed, and simple interface to implement platform-agnostic WebSocket servers
 
-🧩 Seamlessly integrates with, [Node.js](https://nodejs.org/en), [Bun](https://bun.sh/), [Deno](https://deno.com/) and [Cloudflare Workers](https://workers.cloudflare.com/)!
+🧩 Seamlessly integrates with, [Bun](https://bun.sh/), [Deno](https://deno.com/), [Cloudflare Workers](https://workers.cloudflare.com/) and [Node.js](https://nodejs.org/en) ([ws](https://github.com/websockets/ws) || [uWebSockets](https://github.com/uNetworking/uWebSockets.js)).
 
 🚀 High-performance server hooks, avoiding heavy per-connection events API ([why](https://bun.sh/docs/api/websockets#lcYFjkFYJC-summary))
 
@@ -154,6 +154,45 @@ server.on("upgrade", handleUpgrade);
 
 See [playground/node.ts](./playground/node.ts) for demo and [src/adapters/node.ts](./src/adapters/node.ts) for implementation.
 
+### Integration with **Node.js** (uWebSockets)
+
+You can alternatively use [uWebSockets.js](https://github.com/uNetworking/uWebSockets.js) server for Node.js WebSockets.
+
+```ts
+import { App } from "uWebSockets.js";
+
+import nodeUwsAdapter from "crossws/adapters/node-uws";
+import { createDemo, getIndexHTMLURL } from "./_common";
+
+const { websocket } = nodeWSAdapter({ message: console.log });
+
+const server = App().ws("/*", websocket);
+
+server.get("/*", (res, req) => {
+  res.writeStatus("200 OK").writeHeader("Content-Type", "text/html");
+  res.end(
+    `<script>new WebSocket("ws://localhost:3000").addEventListener('open', (e) => e.target.send("Hello from client!"));</script>`,
+  );
+});
+
+server.listen(3001, () => {
+  console.log("Listening to port 3001");
+});
+```
+
+**Adapter specific hooks:**
+
+- `uws:open(ws)`
+- `uws:message(ws, message, isBinary)`
+- `uws:close(ws, code, message)`
+- `uws:ping(ws, message)`
+- `uws:pong(ws, message)`
+- `uws:drain(ws)`
+- `uws:upgrade (res, req, context)`
+- `subscription(ws, topic, newCount, oldCount)`
+
+See [playground/node-uws.ts](./playground/node-uws.ts) for demo and [src/adapters/node-uws.ts](./src/adapters/node-uws.ts) for implementation.
+
 ### Integration with **Bun**
 
 To integrate CrossWS with your Bun server, you need to check for `server.upgrade` and also pass the `websocket` object returned from the adapter to server options. CrossWS leverages native Bun WebSocket API.

package.json

@@ -62,6 +62,7 @@
   "scripts": {
     "build": "unbuild",
     "play:node": "jiti playground/node.ts",
+    "play:uws": "jiti playground/node-uws.ts",
     "play:bun": "bun playground/bun.ts",
     "play:deno": "deno run -A playground/deno.ts",
     "play:cf": "wrangler dev --port 3001",
@@ -91,4 +92,4 @@
     "ws": "^8.16.0"
   },
   "packageManager": "pnpm@8.15.0"
-}
+}

playground/node-uws.ts

@@ -0,0 +1,23 @@
+// You can run this demo using `npm run play:node-uws` in repo
+
+import { readFileSync } from "node:fs";
+
+import { App } from "uWebSockets.js";
+
+import nodeAdapter from "../src/adapters/node-uws.ts";
+import { createDemo, getIndexHTMLURL } from "./_common";
+
+const adapter = createDemo(nodeAdapter);
+
+const app = App().ws("/*", adapter.websocket);
+
+app.get("/*", (res, req) => {
+  res.writeStatus("200 OK");
+  res.writeHeader("Content-Type", "text/html");
+  const indexHTML = readFileSync(getIndexHTMLURL(), "utf8");
+  res.end(indexHTML);
+});
+
+app.listen(3001, () => {
+  console.log("Listening to port 3001");
+});

src/adapters/node-uws.ts

@@ -0,0 +1,106 @@
+// https://github.com/websockets/ws
+// https://github.com/websockets/ws/blob/master/doc/ws.md
+
+import { WebSocketBehavior, WebSocket } from "uWebSockets.js";
+import { WebSocketPeerBase } from "../peer";
+import { WebSocketMessage } from "../message";
+import { defineWebSocketAdapter } from "../adapter";
+
+type UserData = { _peer?: any };
+type WebSocketHandler = WebSocketBehavior<UserData>;
+
+export interface AdapterOptions
+  extends Exclude<
+    WebSocketBehavior<any>,
+    | "close"
+    | "drain"
+    | "message"
+    | "open"
+    | "ping"
+    | "pong"
+    | "subscription"
+    | "upgrade"
+  > {}
+
+export interface Adapter {
+  websocket: WebSocketHandler;
+}
+
+export default defineWebSocketAdapter<Adapter, AdapterOptions>(
+  (hooks, opts = {}) => {
+    const getPeer = (ws: WebSocket<UserData>) => {
+      const userData = ws.getUserData();
+      if (userData._peer) {
+        return userData._peer as WebSocketPeer;
+      }
+      const peer = new WebSocketPeer({ uws: { ws } });
+      userData._peer = peer;
+      return peer;
+    };
+
+    const websocket: WebSocketHandler = {
+      ...opts,
+      close(ws, code, message) {
+        const peer = getPeer(ws);
+        hooks["uws:close"]?.(peer, ws, code, message);
+        hooks.close?.(peer, { code, reason: message?.toString() });
+      },
+      drain(ws) {
+        const peer = getPeer(ws);
+        hooks["uws:drain"]?.(peer, ws);
+      },
+      message(ws, message, isBinary) {
+        const peer = getPeer(ws);
+        hooks["uws:message"]?.(peer, ws, message, isBinary);
+        const msg = new WebSocketMessage(message, isBinary);
+        hooks.message?.(peer, msg);
+      },
+      open(ws) {
+        const peer = getPeer(ws);
+        hooks["uws:open"]?.(peer, ws);
+        hooks.open?.(peer);
+      },
+      ping(ws, message) {
+        const peer = getPeer(ws);
+        hooks["uws:ping"]?.(peer, ws, message);
+      },
+      pong(ws, message) {
+        const peer = getPeer(ws);
+        hooks["uws:pong"]?.(peer, ws, message);
+      },
+      subscription(ws, topic, newCount, oldCount) {
+        const peer = getPeer(ws);
+        hooks["uws:subscription"]?.(peer, ws, topic, newCount, oldCount);
+      },
+      // error ? TODO
+      // upgrade(res, req, context) {}
+    };
+
+    return {
+      websocket,
+    };
+  },
+);
+
+class WebSocketPeer extends WebSocketPeerBase<{
+  uws: {
+    ws: WebSocket<UserData>;
+  };
+}> {
+  get id() {
+    try {
+      const addr = this.ctx.uws.ws?.getRemoteAddressAsText();
+      return new TextDecoder().decode(addr);
+    } catch {
+      // Error: Invalid access of closed uWS.WebSocket/SSLWebSocket.
+    }
+  }
+
+  // TODO
+  // get readyState() {}
+
+  send(message: string, compress?: boolean) {
+    this.ctx.uws.ws.send(message, false, compress);
+    return 0;
+  }
+}

src/hooks.ts

@@ -47,7 +47,7 @@ export interface WebSocketHooks {
   "deno:close": WSHook<[]>;
   "deno:error": WSHook<[error: any]>;
 
-  // Node
+  // ws (Node)
   "node:open": WSHook<[]>;
   "node:message": WSHook<[data: any, isBinary: boolean]>;
   "node:close": WSHook<[code: number, reason: Buffer]>;
@@ -56,4 +56,16 @@ export interface WebSocketHooks {
   "node:pong": WSHook<[data: Buffer]>;
   "node:unexpected-response": WSHook<[req: any, res: any]>;
   "node:upgrade": WSHook<[req: any]>;
+
+  // uws (Node)
+  "uws:open": WSHook<[ws: any]>;
+  "uws:message": WSHook<[ws: any, message: any, isBinary: boolean]>;
+  "uws:close": WSHook<[ws: any, code: number, message: any]>;
+  "uws:ping": WSHook<[ws: any, message: any]>;
+  "uws:pong": WSHook<[ws: any, message: any]>;
+  "uws:drain": WSHook<[ws: any]>;
+  "uws:upgrade": WSHook<[res: any, req: any, context: any]>;
+  "uws:subscription": WSHook<
+    [ws: any, topic: any, newCount: number, oldCount: number]
+  >;
 }

src/message.ts

@@ -5,7 +5,10 @@ export class WebSocketMessage {
   ) {}
 
   text(): string {
-    return this.rawData.toString();
+    if (typeof this.rawData === "string") {
+      return this.rawData;
+    }
+    return new TextDecoder().decode(this.rawData);
   }
 
   toString() {