Socket fun

server/client.test.ts

@@ -1,28 +1,20 @@
 /// <reference lib="deno.ns" />
 import { assertEquals } from "jsr:@std/assert";
+import { startServer } from "./deno.ts";
 
-const url = "ws://0.0.0.0:6969";
+const hostname = "0.0.0.0";
+const port = 8042;
+const href = `ws://${hostname}:${port}`;
 
-function connect(url: string): Promise<WebSocket> {
+async function connect(url: string): Promise<[WebSocket, string[]]> {
-	return new Promise((resolve, reject) => {
+	return new Promise((resolve) => {
+		const messages: string[] = [];
 		const socket = new WebSocket(url);
-		socket.onopen = () => resolve(socket);
-		socket.onerror = (e) => reject(e);
-	});
-}
 
-function nextMessage(socket: WebSocket): Promise<unknown> {
+		socket.onmessage = ({ data }) => {
-	return new Promise((resolve, reject) => {
+			messages.push(data);
-		const onMessage = (e: MessageEvent) => {
-			socket.removeEventListener("message", onMessage);
-			try {
-				resolve(JSON.parse(e.data));
-			} catch {
-				resolve(e.data);
-			}
 		};
-		socket.addEventListener("message", onMessage);
+		socket.onopen = () => resolve([socket, messages]);
-		socket.addEventListener("error", reject, { once: true });
 	});
 }
 
@@ -36,36 +28,51 @@ function close(socket: WebSocket): Promise<void> {
 	});
 }
 
-Deno.test("broadcasts a message on connect to all clients", async () => {
+function setup(
-	const s1 = await connect(url);
+	fn: (ctx: Deno.TestContext) => Promise<void>,
-	const s2 = await connect(url);
+): (ctx: Deno.TestContext) => Promise<void> {
-	assertEquals(await nextMessage(s1), "2 has joined!");
+	return async (ctx) => {
+		const shutdown = startServer(hostname, port);
+		try {
+			await fn(ctx);
+		} finally {
+			await shutdown();
+		}
+	};
+}
 
-	const s3 = connect(url);
+Deno.test(
-	Promise.all([nextMessage(s1), nextMessage(s2)]).then((messages) => {
+	"broadcasts a message on connect to all clients",
-		messages.forEach((message) => {
+	setup(async () => {
-			assertEquals(message, "3 has joined!");
+		const [s1, m1] = await connect(href);
-		});
+		const [s2, m2] = await connect(href);
-	});
+
+		assertEquals(m1, ["1 has joined!", "2 has joined!"]);
+		assertEquals(m2, ["2 has joined!"]);
 
-	await close(s1);
 		await close(s2);
-	await close(await s3);
+		await close(s1);
-});
+	}),
+);
 
-Deno.test("removes a client from broadcast after it disconnects", async () => {
+Deno.test(
-	const s4 = await connect(url);
+	"removes a client from broadcast after it disconnects",
-	const s5 = await connect(url);
+	setup(async () => {
-	const s6 = await connect(url);
+		const [s1, m1] = await connect(href);
+		const [s2, m2] = await connect(href);
+		const [s3, m3] = await connect(href);
 
-	await close(s5);
+		m1.length = 0;
+		m2.length = 0;
+		m3.length = 0;
 
-	Promise.all([nextMessage(s4), nextMessage(s6)]).then((messages) => {
+		await close(s3);
-		messages.forEach((message) => {
-			assertEquals(message, "5 has left!");
-		});
-	});
 
-	await close(s4);
+		assertEquals(m1, ["3 has left!"]);
-	await close(s6);
+		assertEquals(m2, ["3 has left!"]);
-});
+		assertEquals(m3, []);
+
+		await close(s2);
+		await close(s1);
+	}),
+);

server/deno.ts

@@ -1,7 +1,19 @@
 /// <reference lib="deno.ns" />
 
-const clientsMap = new Map<number, WebSocket>();
+export function startServer(
+	hostname: string = "localhost",
+	port: number = 8042,
+): () => Promise<void> {
 	let nextId = 1;
+	const clientsMap = new Map<number, WebSocket>();
+
+	const broadcast = (message: string): void => {
+		for (const socket of clientsMap.values()) {
+			if (socket.readyState === WebSocket.OPEN) {
+				socket.send(message);
+			}
+		}
+	};
 
 	const handleConnection = (socket: WebSocket): void => {
 		const id = nextId++;
@@ -20,15 +32,10 @@ const handleConnection = (socket: WebSocket): void => {
 		socket.onerror = console.error;
 	};
 
-function broadcast(message: string): void {
+	const ac = new AbortController();
-	for (const socket of clientsMap.values()) {
+	const server = Deno.serve(
-		if (socket.readyState === WebSocket.OPEN) {
+		{ hostname, port, signal: ac.signal },
-			socket.send(message);
+		(request) => {
-		}
-	}
-}
-
-Deno.serve({ port: 6969 }, (request) => {
 			if (request.headers.get("upgrade") !== "websocket") {
 				return new Response("Upgrade Required", { status: 426 });
 			}
@@ -37,4 +44,11 @@ Deno.serve({ port: 6969 }, (request) => {
 			handleConnection(socket);
 
 			return response;
-});
+		},
+	);
+
+	return () => {
+		ac.abort();
+		return server.finished;
+	};
+}

server/main.ts

@@ -0,0 +1,3 @@
+import { startServer } from "./deno.ts";
+
+startServer();