Choosing a WebSocket Stack

mkdir chatterbox-ws && cd chatterbox-ws
npm init -y
npm install ws@8.22.0

import { WebSocketServer } from 'ws';
import { randomBytes } from 'node:crypto';

const ALLOWED_ORIGINS = new Set(['https://chat.example.test']);
const wss = new WebSocketServer({
  port: Number(process.env.PORT ?? 8080),
  path: '/ws',
  maxPayload: 16384,
  handleProtocols: (protocols) => (protocols.has('chatterbox.v1') ? 'chatterbox.v1' : false),
  verifyClient: ({ origin }, done) => {
    if (origin === undefined || ALLOWED_ORIGINS.has(origin)) return done(true);
    done(false, 403, 'Forbidden');
  },
});

let seq = 0;
const sid = () => 's_' + randomBytes(10).toString('hex').toUpperCase();
const send = (ws, type, room, payload) =>
  ws.send(JSON.stringify({ v: 1, type, id: sid(), room, ts: Date.now(), payload }));

wss.on('connection', (ws) => {
  let ready = false;
  const timer = setTimeout(() => ws.close(4408, 'hello timeout'), 5000);
  ws.on('close', () => clearTimeout(timer));
  ws.on('message', (data, isBinary) => {
    if (isBinary) return ws.close(4400, 'text only');
    let env;
    try { env = JSON.parse(data.toString()); } catch { return ws.close(4400, 'bad json'); }
    if (!ready) {
      if (env.type !== 'hello') {
        send(ws, 'error', null, { ref: env.id ?? null, code: 'unauthorized', message: 'hello first', retryable: false });
        return ws.close(4401, 'hello first');
      }
      ready = true;
      clearTimeout(timer);
      return send(ws, 'hello', null, {
        ref: env.id, userId: 'u1', name: 'Ada', sessionId: sid(),
        heartbeatMs: 20000, maxFrameBytes: 16384, resumed: false, rooms: [],
      });
    }
    if (env.type === 'message') {
      const assigned = ++seq;
      send(ws, 'ack', env.room, { ref: env.id, ok: true, seq: assigned, dup: false, result: {} });
    } else {
      send(ws, 'error', env.room ?? null, { ref: env.id ?? null, code: 'bad_request', message: 'unsupported in smoke test', retryable: false });
    }
  });
});
console.log('listening on ws://localhost:8080/ws');

const ws = new WebSocket(`ws://localhost:${process.env.PORT ?? 8080}/ws`, 'chatterbox.v1');
const frame = (type, id, room, payload) => JSON.stringify({ v: 1, type, id, room, ts: 0, payload });
ws.onopen = () => ws.send(frame('hello', '01JAB3W7Q8Z4N5R6T7V8X9Y0Z1', null, { client: 'node/24' }));
ws.onmessage = (e) => {
  const m = JSON.parse(e.data);
  console.log(m.type, JSON.stringify(m.payload));
  if (m.type === 'hello') ws.send(frame('message', '01JAB3W7Q8Z4N5R6T7V8X9Y0Z2', 'general', { text: 'Hello, Chatterbox', replyTo: null, attachments: [] }));
  if (m.type === 'ack') ws.close(1000);
};

hello {"ref":"01JAB3W7Q8Z4N5R6T7V8X9Y0Z1","userId":"u1","name":"Ada","sessionId":"s_103F4477D7B13427B425","heartbeatMs":20000,"maxFrameBytes":16384,"resumed":false,"rooms":[]}
ack {"ref":"01JAB3W7Q8Z4N5R6T7V8X9Y0Z2","ok":true,"seq":1,"dup":false,"result":{}}

docker run --rm -v "$PWD":/app -w /app -p 8081:8081 python:3.13-slim sh -c \
  "pip install -q websockets==17.3 && python server.py"

import asyncio
import json
import os
import time

from websockets.asyncio.server import serve

ALLOWED_ORIGINS = {"https://chat.example.test"}
seq = 0


def frame(kind, room, payload):
    return json.dumps({"v": 1, "type": kind, "id": "s_" + os.urandom(10).hex().upper(),
                       "room": room, "ts": int(time.time() * 1000), "payload": payload})


def check_origin(connection, request):
    origin = request.headers.get("Origin")
    if origin is not None and origin not in ALLOWED_ORIGINS:
        return connection.respond(403, "origin not allowed\n")
    return None


async def handler(ws):
    global seq
    try:
        raw = await asyncio.wait_for(ws.recv(), timeout=5)
    except asyncio.TimeoutError:
        return await ws.close(4408, "hello timeout")
    env = json.loads(raw)
    if env.get("type") != "hello":
        await ws.send(frame("error", None, {"ref": env.get("id"), "code": "unauthorized",
                                            "message": "hello first", "retryable": False}))
        return await ws.close(4401, "hello first")
    await ws.send(frame("hello", None, {"ref": env["id"], "userId": "u1", "name": "Ada",
                                        "sessionId": "s_" + os.urandom(10).hex().upper(),
                                        "heartbeatMs": 20000, "maxFrameBytes": 16384,
                                        "resumed": False, "rooms": []}))
    async for raw in ws:
        env = json.loads(raw)
        if env["type"] == "message":
            seq += 1
            await ws.send(frame("ack", env["room"], {"ref": env["id"], "ok": True, "seq": seq,
                                                     "dup": False, "result": {}}))
        else:
            await ws.send(frame("error", env.get("room"), {"ref": env.get("id"), "code": "bad_request",
                                                           "message": "unsupported in smoke test",
                                                           "retryable": False}))


async def main():
    async with serve(handler, "0.0.0.0", 8081, subprotocols=["chatterbox.v1"], max_size=16384,
                     process_request=check_origin):
        await asyncio.Future()

asyncio.run(main())

import asyncio
import json

from websockets.asyncio.client import connect


def frame(kind, id_, room, payload):
    return json.dumps({"v": 1, "type": kind, "id": id_, "room": room, "ts": 0, "payload": payload})


async def main():
    async with connect("ws://localhost:8081/", subprotocols=["chatterbox.v1"]) as ws:
        await ws.send(frame("hello", "01JAB3W7Q8Z4N5R6T7V8X9Y0Z1", None, {"client": "py/1"}))
        print(json.loads(await ws.recv())["type"])
        await ws.send(frame("message", "01JAB3W7Q8Z4N5R6T7V8X9Y0Z2", "general", {"text": "Hello", "replyTo": None, "attachments": []}))
        print(json.loads(await ws.recv())["payload"])

asyncio.run(main())

const n = +process.argv[2]; let open = 0; const socks = [];
for (let i = 0; i < n; i++) {
  const ws = new WebSocket(`ws://localhost:${process.env.PORT ?? 8080}/ws`, 'chatterbox.v1');
  ws.onopen = () => ws.send(JSON.stringify({ v: 1, type: 'hello', id: 'x' + i, room: null, ts: 0, payload: {} }));
  ws.onmessage = () => { open++; if (open === n) console.log('open', open); };
  socks.push(ws);
  if (i % 200 === 199) await new Promise(r => setTimeout(r, 100));
}
setTimeout(() => process.exit(0), 20000);

import asyncio, sys, json
from websockets.asyncio.client import connect
async def main(port, n, hold):
    conns = []
    for i in range(n):
        ws = await connect(f"ws://localhost:{port}/", subprotocols=["chatterbox.v1"])
        await ws.send(json.dumps({"v":1,"type":"hello","id":"x%d"%i,"room":None,"ts":0,"payload":{}}))
        await ws.recv()
        conns.append(ws)
    print("open", len(conns), flush=True)
    await asyncio.sleep(hold)
asyncio.run(main(int(sys.argv[1]), int(sys.argv[2]), float(sys.argv[3])))