Something went wrong. Try again.
Demo comparing html-over-the-wire w/ WS and SSE.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243// Uniform client facade over both architectures, so bench code is identical.//// The SSE client deliberately mirrors what a browser does per origin: ONE// connection (h2 session / h1 keep-alive agent) carrying both the event stream// and every command POST.
import http from 'node:http'import https from 'node:https'import http2 from 'node:http2'import zlib from 'node:zlib'import WebSocket from 'ws'
function makeDecompressor(kind, onText) { if (kind === 'none') { return { write: (b) => onText(b.toString('utf8')), end: () => {} } } const z = kind === 'br' ? zlib.createBrotliDecompress() : kind === 'deflate' ? zlib.createInflate() : zlib.createGunzip() z.on('data', (b) => onText(b.toString('utf8'))) z.on('error', () => {}) return { write: (b) => z.write(b), end: () => z.end() }}
// Incremental SSE framing. Returns complete events as raw blocks.function makeSseParser(onEvent) { let buf = '' return (text) => { buf += text let i while ((i = buf.indexOf('\n\n')) !== -1) { const block = buf.slice(0, i) buf = buf.slice(i + 2) onEvent(block) } }}
// Matches both raw SSE payloads and the JSON-escaped form the WS envelope uses.const NONCE_RE = /data-nonce=\\?"(\d+)\\?"/g
function noncesIn(s) { const out = [] let m NONCE_RE.lastIndex = 0 while ((m = NONCE_RE.exec(s))) out.push(Number(m[1])) return out}
// ------------------------------------------------------------------ WS client
function wsClient({ host, port, tls, room, compress }) { const url = `${tls ? 'wss' : 'ws'}://${host}:${port}/?room=${room}` let sock const handlers = [] let firstMsg return { connect() { return new Promise((resolve, reject) => { firstMsg = resolve sock = new WebSocket(url, { rejectUnauthorized: false, perMessageDeflate: compress !== 'none', }) sock.on('error', reject) sock.on('message', (data) => { const s = data.toString('utf8') if (firstMsg) { const f = firstMsg firstMsg = null f() return } const ns = noncesIn(s) for (const h of handlers) h(ns, s) }) }) }, onPatch(cb) { handlers.push(cb) }, dispatch({ id, nonce }) { sock.send(JSON.stringify({ t: 'cmd', id, nonce, room })) return Promise.resolve() }, close() { try { sock.close() } catch {} }, }}
// ------------------------------------------------------------- SSE/h2 client
function h2Client({ host, port, tls, room, compress }) { const origin = `${tls ? 'https' : 'http'}://${host}:${port}` let session let stream const handlers = [] return { connect() { return new Promise((resolve, reject) => { session = http2.connect(origin, { rejectUnauthorized: false }) session.on('error', reject) session.setMaxListeners(0) stream = session.request({ ':path': `/events?room=${room}`, ':method': 'GET', 'accept-encoding': compress === 'none' ? 'identity' : compress, }) let first = true const parse = makeSseParser((block) => { if (first) { first = false resolve() return } const ns = noncesIn(block) for (const h of handlers) h(ns, block) }) const dec = makeDecompressor(compress, parse) stream.on('data', (c) => dec.write(c)) stream.on('error', reject) stream.on('response', () => {}) }) }, onPatch(cb) { handlers.push(cb) }, dispatch({ id, nonce }) { return new Promise((resolve, reject) => { const body = JSON.stringify({ id, nonce, room }) // Same h2 session as the SSE stream -> new stream, not new TCP conn. const req = session.request({ ':path': '/command', ':method': 'POST', 'content-type': 'application/json', 'content-length': Buffer.byteLength(body), }) req.on('response', () => {}) req.on('end', resolve) req.on('error', reject) req.resume() req.end(body) }) }, close() { try { stream?.close() session?.close() } catch {} }, }}
// ------------------------------------------------------------- SSE/h1 client
function h1Client({ host, port, tls, room, compress }) { const mod = tls ? https : http // A browser gives an origin 6 sockets on HTTP/1.1 and the SSE stream eats one. const agent = new mod.Agent({ keepAlive: true, maxSockets: 5, rejectUnauthorized: false }) let streamReq const handlers = [] return { connect() { return new Promise((resolve, reject) => { streamReq = mod.request( { host, port, path: `/events?room=${room}`, method: 'GET', rejectUnauthorized: false, headers: { 'accept-encoding': compress === 'none' ? 'identity' : compress }, // The stream gets its own socket, exactly like the browser. agent: new mod.Agent({ keepAlive: true, maxSockets: 1, rejectUnauthorized: false }), }, (res) => { let first = true const parse = makeSseParser((block) => { if (first) { first = false resolve() return } const ns = noncesIn(block) for (const h of handlers) h(ns, block) }) const dec = makeDecompressor(compress, parse) res.on('data', (c) => dec.write(c)) res.on('error', reject) }, ) streamReq.on('error', reject) streamReq.end() }) }, onPatch(cb) { handlers.push(cb) }, dispatch({ id, nonce }) { return new Promise((resolve, reject) => { const body = JSON.stringify({ id, nonce, room }) const req = mod.request( { host, port, path: '/command', method: 'POST', agent, rejectUnauthorized: false, headers: { 'content-type': 'application/json', 'content-length': Buffer.byteLength(body), }, }, (res) => { res.resume() res.on('end', resolve) }, ) req.on('error', reject) req.end(body) }) }, close() { try { streamReq?.destroy() agent.destroy() } catch {} }, }}
export function createClient(opts) { if (opts.arch === 'ws') return wsClient(opts) return opts.h1 ? h1Client(opts) : h2Client(opts)}