Something went wrong. Try again.
Demo comparing html-over-the-wire w/ WS and SSE.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351// One server, two architectures, identical domain output.//// --arch ws : single WebSocket carries commands up and HTML patches down// --arch sse : GET /events (SSE) down, POST /command up (CQRS split)//// Flags: --tls --h1 --compress <none|gzip|br|deflate> --mode <narrow|fat>// --port --control-port --rooms//// A control port (plain HTTP, never behind the delay proxy) exposes /stats so// the harness can read RSS/CPU without perturbing the measured path.
import fs from 'node:fs'import http from 'node:http'import https from 'node:https'import http2 from 'node:http2'import net from 'node:net'import tls from 'node:tls'import zlib from 'node:zlib'import { fileURLToPath } from 'node:url'import path from 'node:path'import { WebSocketServer } from 'ws'import { makeBoard, apply, renderPatch, renderBoard, sseEvent, wsMessage } from './domain.js'
const __dirname = path.dirname(fileURLToPath(import.meta.url))
function parseArgs(argv) { const out = {} for (let i = 0; i < argv.length; i++) { const a = argv[i] if (!a.startsWith('--')) continue const key = a.slice(2).replace(/-/g, '_') const next = argv[i + 1] if (next === undefined || next.startsWith('--')) out[key] = true else out[key] = argv[++i] } return out}
const args = parseArgs(process.argv.slice(2))const ARCH = args.arch ?? 'sse'const USE_TLS = !!args.tlsconst USE_H1 = !!args.h1const COMPRESS = args.compress ?? 'none'const MODE = args.mode ?? 'narrow'// Brotli window exponent. Matters a lot: it sets both the ratio ceiling AND// the per-connection memory, and those pull in opposite directions.const LGWIN = Number(args.lgwin ?? 22)const WS_NO_CTX = !!args.ws_no_context_takeoverconst PORT = Number(args.port ?? 0)const CONTROL_PORT = Number(args.control_port ?? 0)
const board = makeBoard()
// ---------------------------------------------------------------- subscribers
// room -> Set<subscriber>. A subscriber is whatever the transport needs to// push bytes to one client. This registry is the thing both architectures// need; the article implies WS needs it and SSE doesn't, which is false.const rooms = new Map()let subCount = 0
function join(room, sub) { let set = rooms.get(room) if (!set) rooms.set(room, (set = new Set())) set.add(sub) subCount++ return () => { set.delete(sub) subCount-- if (set.size === 0) rooms.delete(room) }}
function broadcast(room, framer) { const set = rooms.get(room) if (!set) return 0 let n = 0 for (const sub of set) { try { sub.send(framer) n++ } catch {} } return n}
// ------------------------------------------------------------------ compress
// A per-connection compression context, flushed on every patch. This is where// SSE quietly wins on HTML: the dictionary persists across the whole stream,// so the Nth copy of the same markup costs almost nothing.function makeCompressor(kind, sink) { if (kind === 'none') { return { write: (buf) => sink(buf), end: () => {} } } const z = kind === 'br' ? zlib.createBrotliCompress({ params: { [zlib.constants.BROTLI_PARAM_QUALITY]: 4, [zlib.constants.BROTLI_PARAM_LGWIN]: LGWIN, [zlib.constants.BROTLI_PARAM_MODE]: zlib.constants.BROTLI_MODE_TEXT, }, }) : kind === 'deflate' ? zlib.createDeflate({ level: 6 }) : zlib.createGzip({ level: 6 }) z.on('data', sink) z.on('error', () => {}) const FLUSH = kind === 'br' ? zlib.constants.BROTLI_OPERATION_FLUSH : zlib.constants.Z_SYNC_FLUSH return { write(buf) { z.write(buf) z.flush(FLUSH) }, end() { z.end() }, }}
const contentEncoding = { none: null, gzip: 'gzip', br: 'br', deflate: 'deflate' }[COMPRESS]
// --------------------------------------------------------------- SSE + CQRS
let eventId = 0
function handleSse(req, res, roomOf) { const room = roomOf const headers = { 'content-type': 'text/event-stream', 'cache-control': 'no-cache', } if (contentEncoding) headers['content-encoding'] = contentEncoding res.writeHead(200, headers) if (res.flushHeaders) res.flushHeaders()
const comp = makeCompressor(COMPRESS, (buf) => { if (!res.destroyed && res.writable) res.write(buf) })
const leave = join(room, { send: (framer) => comp.write(framer()), })
// Initial render, exactly as a hypermedia app would. comp.write(sseEvent(renderBoard(board), ++eventId))
const done = () => { leave() comp.end() } res.on('close', done) res.on('error', done)}
function handleCommand(room, id, nonce) { const row = apply(board, { id, nonce }) const html = renderPatch(board, row, MODE) const evId = ++eventId // One render, many sends: framer is called per subscriber but the string is // built once, so we are not measuring template cost. const sseFrame = sseEvent(html, evId) return broadcast(room, () => sseFrame)}
function sseRequestHandler(req, res) { const url = new URL(req.url, 'http://x') if (url.pathname === '/events') { handleSse(req, res, url.searchParams.get('room') ?? '0') return } if (url.pathname === '/command') { // Read the body — a real command has one, and it costs bytes + a turn. let body = '' req.on('data', (c) => (body += c)) req.on('end', () => { let cmd try { cmd = JSON.parse(body || '{}') } catch { cmd = {} } const n = handleCommand(String(cmd.room ?? '0'), Number(cmd.id ?? 0), cmd.nonce ?? 0) res.writeHead(204, { 'x-fanout': String(n) }) res.end() }) return } res.writeHead(404) res.end()}
// ------------------------------------------------------------------- WS side
function attachWs(server) { const wss = new WebSocketServer({ server, perMessageDeflate: COMPRESS === 'none' ? false : { zlibDeflateOptions: { level: 6 }, // Context takeover is what makes WS competitive with SSE on // repeated HTML — and what costs ~256 KB of zlib state per // connection. Proxies routinely force it off to bound that. serverNoContextTakeover: WS_NO_CTX, clientNoContextTakeover: WS_NO_CTX, threshold: 0, }, })
wss.on('connection', (socket, req) => { const url = new URL(req.url, 'http://x') const room = url.searchParams.get('room') ?? '0' const leave = join(room, { send: (framer) => socket.send(framer()) })
socket.send(wsMessage(renderBoard(board), ++eventId))
socket.on('message', (data) => { let cmd try { cmd = JSON.parse(data) } catch { return } if (cmd.t !== 'cmd') return const row = apply(board, { id: Number(cmd.id ?? 0), nonce: cmd.nonce ?? 0 }) const html = renderPatch(board, row, MODE) const msg = wsMessage(html, ++eventId) broadcast(String(cmd.room ?? room), () => msg) }) socket.on('close', leave) socket.on('error', leave) }) return wss}
// ------------------------------------------------------------------- listen
const tlsOpts = () => ({ key: fs.readFileSync(path.join(__dirname, '../certs/key.pem')), cert: fs.readFileSync(path.join(__dirname, '../certs/cert.pem')),})
let server
if (ARCH === 'ws') { server = USE_TLS ? https.createServer({ ...tlsOpts(), ALPNProtocols: ['http/1.1'] }, (_, res) => { res.writeHead(404) res.end() }) : http.createServer((_, res) => { res.writeHead(404) res.end() }) attachWs(server)} else if (USE_H1) { server = USE_TLS ? https.createServer({ ...tlsOpts(), ALPNProtocols: ['http/1.1'] }, sseRequestHandler) : http.createServer(sseRequestHandler) server.keepAliveTimeout = 120000} else { // HTTP/2. allowHTTP1 stays off so we can be certain what we measured. const h2opts = { settings: { maxConcurrentStreams: 1000 } } server = USE_TLS ? http2.createSecureServer({ ...tlsOpts(), ALPNProtocols: ['h2'], ...h2opts }) : http2.createServer(h2opts) server.on('stream', (stream, headers) => { const req = Object.assign(stream, { url: headers[':path'], method: headers[':method'], }) const res = { writeHead(status, h) { stream.respond({ ':status': status, ...h }) }, write: (b) => stream.write(b), end: (b) => stream.end(b), on: (e, f) => stream.on(e, f), get destroyed() { return stream.destroyed }, get writable() { return stream.writable }, } sseRequestHandler(req, res) }) // A long-lived SSE stream must not be reaped, and we need lots of them. server.on('session', (s) => { s.setLocalWindowSize?.(1 << 20) })}
server.on('error', (e) => { console.error('server error', e.message) process.exit(1)})
// ------------------------------------------------------------------ control
let cpuBase = process.cpuUsage()const control = http .createServer((req, res) => { if (req.url === '/reset') { cpuBase = process.cpuUsage() if (global.gc) global.gc() res.writeHead(204) return res.end() } if (req.url === '/shutdown') { res.writeHead(204) res.end() return setTimeout(() => process.exit(0), 50) } if (global.gc && req.url === '/stats') global.gc() const mem = process.memoryUsage() const cpu = process.cpuUsage(cpuBase) res.writeHead(200, { 'content-type': 'application/json' }) res.end( JSON.stringify({ arch: ARCH, subscribers: subCount, rooms: rooms.size, rss: mem.rss, heapUsed: mem.heapUsed, external: mem.external, arrayBuffers: mem.arrayBuffers, cpuUserMs: cpu.user / 1000, cpuSystemMs: cpu.system / 1000, }), ) })
// Both listeners bind port 0 by default and report what the OS handed out, so// concurrent/aborted runs can never collide on a fixed port.control.on('error', (e) => { console.error('control error', e.message) process.exit(1)})control.listen(CONTROL_PORT, '127.0.0.1', () => { server.listen(PORT, '127.0.0.1', () => { console.log( `[server] arch=${ARCH} tls=${USE_TLS} h1=${USE_H1} compress=${COMPRESS} mode=${MODE} ` + `lgwin=${LGWIN} port=${server.address().port} control=${control.address().port}`, ) })})