Real-Time Web Overview

// common.mjs - a tiny in-memory "room" shared by the HTTP examples
import { createServer } from "node:http";

export function createRoom() {
  const messages = [];
  const listeners = new Set();
  return {
    messages,
    add(text) {
      const message = { seq: messages.length + 1, text, ts: Date.now() };
      messages.push(message);
      for (const listener of listeners) listener(message);
      return message;
    },
    after(seq) {
      return messages.filter((m) => m.seq > seq);
    },
    onMessage(listener) {
      listeners.add(listener);
      return () => listeners.delete(listener);
    },
  };
}

export function readBody(req) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    req.on("data", (c) => chunks.push(c));
    req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8")));
    req.on("error", reject);
  });
}

export function json(res, status, body) {
  res.writeHead(status, { "Content-Type": "application/json", "Cache-Control": "no-store" });
  res.end(JSON.stringify(body));
}

// Handles POST /messages for every example; returns true when the request was handled.
export async function handlePost(room, req, res) {
  if (req.method === "POST" && req.url === "/messages") {
    const { text } = JSON.parse(await readBody(req));
    json(res, 201, room.add(String(text).slice(0, 4000)));
    return true;
  }
  return false;
}

export { createServer };

// Posts one message to the HTTP examples: node send.mjs "Hello, Chatterbox"
const res = await fetch("http://localhost:3000/messages", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ text: process.argv[2] ?? "Hello, Chatterbox" }),
});
console.log("sent", res.status, await res.json());

import { createServer, createRoom, handlePost, json } from "./common.mjs";

const room = createRoom();

createServer(async (req, res) => {
  if (await handlePost(room, req, res)) return;
  const url = new URL(req.url, "http://localhost");
  if (req.method === "GET" && url.pathname === "/poll") {
    // Answer immediately, even when there is nothing new.
    json(res, 200, room.after(Number(url.searchParams.get("after") ?? 0)));
    return;
  }
  json(res, 404, { error: "not_found" });
}).listen(3000, () => console.log("short polling server on :3000"));

let after = 0;
let requests = 0;

async function poll() {
  requests += 1;
  const res = await fetch(`http://localhost:3000/poll?after=${after}`);
  for (const m of await res.json()) {
    after = m.seq;
    console.log(`received "${m.text}" after ${Date.now() - m.ts} ms (${requests} requests so far)`);
  }
}

setInterval(() => poll().catch(console.error), 1000); // the polling interval is the latency floor

received "Hello, Chatterbox" after 769 ms (2 requests so far)

import { createServer, createRoom, handlePost, json } from "./common.mjs";

const room = createRoom();
const HOLD_MS = 25_000; // stay below the idle timeout of proxies and load balancers

createServer(async (req, res) => {
  if (await handlePost(room, req, res)) return;
  const url = new URL(req.url, "http://localhost");
  if (req.method === "GET" && url.pathname === "/poll") {
    const after = Number(url.searchParams.get("after") ?? 0);
    const pending = room.after(after);
    if (pending.length > 0) return json(res, 200, pending); // something is already waiting

    // Park the request until a message arrives or the hold time expires.
    const finish = (body) => {
      clearTimeout(timer);
      stop();
      json(res, 200, body);
    };
    const stop = room.onMessage(() => finish(room.after(after)));
    const timer = setTimeout(() => finish([]), HOLD_MS);
    req.on("close", () => { clearTimeout(timer); stop(); }); // the client went away
    return;
  }
  json(res, 404, { error: "not_found" });
}).listen(3000, () => console.log("long polling server on :3000"));

let after = 0;
let requests = 0;

async function loop() {
  for (;;) {
    try {
      requests += 1;
      const res = await fetch(`http://localhost:3000/poll?after=${after}`);
      for (const m of await res.json()) {
        after = m.seq;
        console.log(`received "${m.text}" after ${Date.now() - m.ts} ms (${requests} requests so far)`);
      }
    } catch (err) {
      console.error("poll failed, retrying in 2 s", err.message);
      await new Promise((r) => setTimeout(r, 2000)); // back off; never spin on errors
    }
  }
}

loop(); // the next request is issued immediately after each response

received "Hello, Chatterbox" after 4 ms (1 requests so far)

import { createServer, createRoom, handlePost, json } from "./common.mjs";

const room = createRoom();

createServer(async (req, res) => {
  if (await handlePost(room, req, res)) return;
  if (req.method === "GET" && req.url === "/events") {
    res.writeHead(200, {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
    });
    res.write("retry: 3000\n\n"); // reconnection delay hint in milliseconds

    const send = (m) => res.write(`id: ${m.seq}\nevent: message\ndata: ${JSON.stringify(m)}\n\n`);

    // The browser sends Last-Event-ID automatically when it reconnects.
    const last = Number(req.headers["last-event-id"] ?? 0);
    room.after(last).forEach(send);

    const stop = room.onMessage(send);
    const keepAlive = setInterval(() => res.write(": keep-alive\n\n"), 15_000); // comment line
    req.on("close", () => { stop(); clearInterval(keepAlive); });
    return;
  }
  json(res, 404, { error: "not_found" });
}).listen(3000, () => console.log("SSE server on :3000"));

// Run with: node --experimental-eventsource sse-client.mjs
// Browsers expose the same EventSource API without any flag.
const source = new EventSource("http://localhost:3000/events");

source.addEventListener("open", () => console.log("stream open"));
source.addEventListener("message", (event) => {
  const m = JSON.parse(event.data);
  console.log(`received "${m.text}" after ${Date.now() - m.ts} ms (id ${event.lastEventId})`);
});
source.addEventListener("error", () => {
  // readyState CONNECTING means the browser is retrying by itself; CLOSED means it gave up.
  console.log("stream error, readyState =", source.readyState);
});

stream open
received "Hello, Chatterbox" after 1 ms (id 1)

import { createServer } from "node:http";
import { WebSocketServer } from "ws";

const ALLOWED_ORIGINS = new Set(["https://chat.example.test"]);
const wss = new WebSocketServer({ noServer: true, maxPayload: 16 * 1024 });
const server = createServer((req, res) => {
  res.writeHead(426, { "Content-Type": "text/plain" }); // plain HTTP is not served here
  res.end("Upgrade Required");
});

// Reject cross-site handshakes before the upgrade completes (Origin allow-list).
// Non-browser clients send no Origin header; allow that only because this is a local demo.
server.on("upgrade", (req, socket, head) => {
  const { origin } = req.headers;
  if (origin !== undefined && !ALLOWED_ORIGINS.has(origin)) {
    socket.write("HTTP/1.1 403 Forbidden\r\nConnection: close\r\n\r\n");
    socket.destroy();
    return;
  }
  wss.handleUpgrade(req, socket, head, (ws) => wss.emit("connection", ws, req));
});

let seq = 0;

wss.on("connection", (socket) => {
  socket.on("message", (data, isBinary) => {
    if (isBinary) return socket.close(1003, "text only");
    const { text } = JSON.parse(data.toString("utf8"));
    const message = JSON.stringify({ seq: ++seq, text: String(text).slice(0, 4000), ts: Date.now() });
    for (const client of wss.clients) {
      if (client.readyState === client.OPEN) client.send(message); // fan out, including the sender
    }
  });
});

server.listen(3000, () => console.log("WebSocket server on :3000"));

// Node 22 or later has a global WebSocket client with the browser API.
// Usage: node websocket-client.mjs            (listen)
//        node websocket-client.mjs "Hello"    (send one message after connecting)
const socket = new WebSocket("ws://localhost:3000");
const toSend = process.argv[2];

socket.addEventListener("open", () => {
  console.log("socket open");
  if (toSend) {
    setTimeout(() => socket.send(JSON.stringify({ text: toSend })), 2500);
    setTimeout(() => socket.close(1000), 3500); // the sender leaves after the broadcast
  }
});
socket.addEventListener("message", (event) => {
  const m = JSON.parse(event.data);
  console.log(`received "${m.text}" after ${Date.now() - m.ts} ms`);
});
socket.addEventListener("close", (event) => console.log("closed", event.code));

// Companion for latency.mjs: a WebSocket echo on :3001 and an HTTP ping on :3002.
import { createServer } from "node:http";
import { WebSocketServer } from "ws";

new WebSocketServer({ port: 3001 }).on("connection", (s) => s.on("message", (d) => s.send(d)));
createServer((req, res) => res.end("pong")).listen(3002);
console.log("echo on ws://localhost:3001 and http://localhost:3002/ping");

// Measures round-trip time of a WebSocket echo and an HTTP request on the same host.
// Usage: node latency.mjs ws://localhost:3001 http://localhost:3002/ping
import { WebSocket } from "ws";

const [wsUrl, httpUrl] = process.argv.slice(2);
const SAMPLES = 50;

function percentile(sorted, p) {
  return sorted[Math.min(sorted.length - 1, Math.ceil((p / 100) * sorted.length) - 1)];
}
function report(label, values) {
  const sorted = [...values].sort((a, b) => a - b);
  const mean = sorted.reduce((a, b) => a + b, 0) / sorted.length;
  console.log(`${label.padEnd(10)} p50 ${percentile(sorted, 50).toFixed(2)} ms  p95 ${percentile(sorted, 95).toFixed(2)} ms  max ${sorted.at(-1).toFixed(2)} ms  mean ${mean.toFixed(2)} ms`);
}

async function measureWebSocket(url) {
  const socket = new WebSocket(url);
  await new Promise((resolve, reject) => { socket.once("open", resolve); socket.once("error", reject); });
  const samples = [];
  for (let i = 0; i < SAMPLES; i += 1) {
    const start = performance.now();
    const reply = new Promise((resolve) => socket.once("message", resolve));
    socket.send(String(i));
    await reply;
    samples.push(performance.now() - start);
  }
  socket.close();
  return samples;
}

async function measureHttp(url) {
  const samples = [];
  for (let i = 0; i < SAMPLES; i += 1) {
    const start = performance.now();
    const res = await fetch(url, { cache: "no-store" });
    await res.text();
    samples.push(performance.now() - start);
  }
  return samples;
}

if (wsUrl) report("WebSocket", await measureWebSocket(wsUrl));
if (httpUrl) report("HTTP GET", await measureHttp(httpUrl));

WebSocket  p50 0.05 ms  p95 0.13 ms  max 0.94 ms  mean 0.08 ms
HTTP GET   p50 0.17 ms  p95 0.58 ms  max 11.89 ms  mean 0.47 ms