✨ Add WebSocket server to @effectionx/websocket

websocket/README.md

@@ -26,7 +26,7 @@ await main(function* () {
   let socket = yield* useWebSocket("ws://websocket.example.org");
 
   // Send messages to the server
-  socket.send("Hello World");
+  yield* socket.send("Hello World");
 
   // Receive messages using a simple iterator
   for (let message of yield* each(socket)) {
@@ -46,6 +46,68 @@ await main(function* () {
 - **Clean Resource Management**: Connections are properly cleaned up when the
   operation completes
 
+## WebSocket Server
+
+`useWebSocketServer()` is the server counterpart of `useWebSocket()`. It yields a
+stream of incoming connections, where **each connection is the same full-duplex
+`WebSocketResource`** produced by the client — you receive messages by iterating
+it and reply with `yield* connection.send()`.
+
+The underlying server is supplied through a factory, so this package never
+imports a concrete server implementation and stays platform-agnostic. On Node
+this is typically the [`ws`](https://github.com/websockets/ws) `WebSocketServer`.
+
+```typescript
+import { each, main, spawn } from "effection";
+import { WebSocketServer } from "ws";
+import {
+  useWebSocketServer,
+  type WebSocketServerLike,
+} from "@effectionx/websocket";
+
+await main(function* () {
+  let server = yield* useWebSocketServer<string>(
+    () => new WebSocketServer({ port: 3000 }) as unknown as WebSocketServerLike,
+  );
+
+  // A stream is consumed sequentially, so spawn a handler per connection to
+  // serve many clients concurrently.
+  for (let connection of yield* each(server)) {
+    yield* spawn(function* () {
+      for (let message of yield* each(connection)) {
+        yield* connection.send(`echo: ${message.data}`);
+        yield* each.next();
+      }
+    });
+    yield* each.next();
+  }
+});
+```
+
+A client — using `useWebSocket()` from the same package — pairs with it directly.
+Because `send` is an `Operation`, invoke it with `yield*` on both sides:
+
+```typescript
+import { each, main } from "effection";
+import { useWebSocket } from "@effectionx/websocket";
+
+await main(function* () {
+  let socket = yield* useWebSocket<string>("ws://localhost:3000");
+
+  yield* socket.send("hello"); // client -> server
+
+  for (let message of yield* each(socket)) {
+    console.log(message.data); // "echo: hello"  (server -> client)
+    yield* each.next();
+  }
+});
+```
+
+Connections are buffered, so none are dropped between the moment the server
+starts listening and the moment you begin iterating. The server — and every live
+connection it produced — is automatically closed when the resource passes out of
+scope.
+
 ## Advanced Usage
 
 ### Custom WebSocket Implementations

websocket/mod.ts

@@ -1 +1,2 @@
 export * from "./websocket.ts";
+export * from "./server.ts";

websocket/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/websocket",
-  "description": "WebSocket client with stream-based message handling and automatic cleanup",
-  "version": "2.3.3",
+  "description": "WebSocket client and server with stream-based message handling and automatic cleanup",
+  "version": "3.0.0",
   "keywords": ["io", "streams"],
   "type": "module",
   "main": "./dist/mod.js",
@@ -14,6 +14,10 @@
       "default": "./dist/mod.js"
     }
   },
+  "dependencies": {
+    "@effectionx/node": "workspace:*",
+    "@effectionx/timebox": "workspace:*"
+  },
   "peerDependencies": {
     "effection": "^3 || ^4"
   },

websocket/server.test.ts

@@ -0,0 +1,254 @@
+import { createServer } from "node:http";
+import type { AddressInfo } from "node:net";
+import { describe, it } from "@effectionx/vitest";
+import {
+  type Operation,
+  type Subscription,
+  createQueue,
+  resource,
+  spawn,
+  suspend,
+  withResolvers,
+} from "effection";
+import { expect } from "expect";
+import { WebSocketServer, type WebSocket as WsWebSocket } from "ws";
+
+import {
+  type WebSocketServerLike,
+  type WebSocketServerResource,
+  useWebSocketServer,
+} from "./server.ts";
+import { type WebSocketResource, useWebSocket } from "./websocket.ts";
+
+describe("WebSocketServer", () => {
+  it("yields a connection and receives a message from the client", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    let client = yield* connect(port);
+    let connection = (yield* incoming.next()).value;
+
+    let messages = yield* connection;
+    yield* client.send("hello from client");
+
+    let { value } = yield* messages.next();
+    expect(value).toMatchObject({ data: "hello from client" });
+  });
+
+  it("sends a message from a server connection to the client", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    let client = yield* connect(port);
+    let connection = (yield* incoming.next()).value;
+
+    let clientMessages = yield* client;
+    yield* connection.send("hello from server");
+
+    let { value } = yield* clientMessages.next();
+    expect(value).toMatchObject({ data: "hello from server" });
+  });
+
+  it("completes a connection stream when its client disconnects", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    let raw = new WebSocket(`ws://localhost:${port}`);
+    yield* useWebSocket<string>(() => raw);
+    let connection = (yield* incoming.next()).value;
+    let messages = yield* connection;
+
+    raw.close(4001, "goodbye");
+
+    let event = yield* drain(messages);
+    expect(event.type).toEqual("close");
+    expect(event.wasClean).toEqual(true);
+    expect(event.code).toEqual(4001);
+    expect(event.reason).toEqual("goodbye");
+  });
+
+  it("closes live client connections when the server is torn down", function* () {
+    let { httpServer, port } = yield* useHttp();
+    let accepted = createQueue<void, never>();
+
+    let serverTask = yield* spawn(function* () {
+      let server = yield* useWebSocketServer<string>(
+        () =>
+          new WebSocketServer({
+            server: httpServer,
+          }) as unknown as WebSocketServerLike,
+      );
+      let incoming = yield* server;
+      yield* incoming.next();
+      accepted.add();
+      yield* suspend();
+    });
+
+    let client = yield* connect(port);
+    let messages = yield* client;
+
+    // wait until the server has accepted the connection before tearing it down
+    yield* accepted.next();
+    yield* serverTask.halt();
+
+    let event = yield* drain(messages);
+    expect(event.type).toEqual("close");
+    expect(event.wasClean).toEqual(true);
+    expect(event.code).toEqual(1001);
+    expect(event.reason).toEqual("server shutting down");
+  });
+
+  it("closes a connection with an explicit code and reason", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    let client = yield* connect(port);
+    let connection = (yield* incoming.next()).value;
+    let clientMessages = yield* client;
+
+    yield* connection.close(4002, "custom");
+
+    let event = yield* drain(clientMessages);
+    expect(event.code).toEqual(4002);
+    expect(event.reason).toEqual("custom");
+  });
+
+  it("buffers a connection that arrives before it is consumed", function* () {
+    let { server, port } = yield* useTestServer();
+
+    // connect the client before subscribing to the server stream
+    let client = yield* connect(port);
+    let incoming = yield* server;
+
+    // the connection was buffered while nobody was subscribed
+    let connection = (yield* incoming.next()).value;
+
+    let messages = yield* connection;
+    yield* client.send("buffered hello");
+
+    let { value } = yield* messages.next();
+    expect(value).toMatchObject({ data: "buffered hello" });
+  });
+
+  it("isolates a connection error so the server keeps serving other clients", function* () {
+    let { httpServer, port } = yield* useHttp();
+
+    // capture the raw accepted sockets so we can force an error on one
+    let wss = new WebSocketServer({ server: httpServer });
+    let rawSockets = createQueue<WsWebSocket, never>();
+    wss.on("connection", (ws) => rawSockets.add(ws));
+
+    let server = yield* useWebSocketServer<string>(
+      () => wss as unknown as WebSocketServerLike,
+    );
+    let incoming = yield* server;
+    let serverErrors = yield* server.errors;
+
+    // accept one client, then make its underlying socket error
+    yield* connect(port);
+    yield* incoming.next();
+    let raw = (yield* rawSockets.next()).value;
+    raw.emit("error", new Error("boom"));
+
+    // the failure is surfaced on the errors stream, not thrown at the server.
+    // A socket failure surfaces as the DOM `error` event, whose message is the
+    // underlying error's message.
+    let { value: error } = yield* serverErrors.next();
+    expect((error as ErrorEvent).message).toContain("boom");
+
+    // and the server survives, serving a fresh client
+    let client = yield* connect(port);
+    let connection = (yield* incoming.next()).value;
+    let messages = yield* connection;
+    yield* client.send("still alive");
+    let { value } = yield* messages.next();
+    expect(value).toMatchObject({ data: "still alive" });
+  });
+
+  it("surfaces each simultaneous client as a distinct connection", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    // connect two clients, then read two buffered connections back out
+    let clientA = yield* connect(port);
+    let clientB = yield* connect(port);
+
+    let first = (yield* incoming.next()).value;
+    let second = (yield* incoming.next()).value;
+
+    expect(first).not.toBe(second);
+
+    // each connection receives only its own client's message, regardless of order
+    let firstMessages = yield* first;
+    let secondMessages = yield* second;
+
+    yield* clientA.send("A");
+    yield* clientB.send("B");
+
+    let received = [
+      ((yield* firstMessages.next()).value as MessageEvent<string>).data,
+      ((yield* secondMessages.next()).value as MessageEvent<string>).data,
+    ].sort();
+
+    expect(received).toEqual(["A", "B"]);
+  });
+});
+
+interface TestServer {
+  server: WebSocketServerResource<string>;
+  port: number;
+}
+
+function useTestServer(): Operation<TestServer> {
+  return resource(function* (provide) {
+    let { httpServer, port } = yield* useHttp();
+
+    let server = yield* useWebSocketServer<string>(
+      () =>
+        new WebSocketServer({
+          server: httpServer,
+        }) as unknown as WebSocketServerLike,
+    );
+
+    yield* provide({ server, port });
+  });
+}
+
+function useHttp(): Operation<{
+  httpServer: ReturnType<typeof createServer>;
+  port: number;
+}> {
+  return resource(function* (provide) {
+    let httpServer = createServer();
+
+    let listening = withResolvers<void>();
+    httpServer.listen(0, listening.resolve);
+    yield* listening.operation;
+
+    let port = (httpServer.address() as AddressInfo).port;
+
+    try {
+      yield* provide({ httpServer, port });
+    } finally {
+      let closed = withResolvers<void>();
+      httpServer.close(() => closed.resolve());
+      yield* closed.operation;
+    }
+  });
+}
+
+function* connect(port: number): Operation<WebSocketResource<string>> {
+  return yield* useWebSocket<string>(
+    () => new WebSocket(`ws://localhost:${port}`),
+  );
+}
+
+function* drain<T, TClose>(
+  subscription: Subscription<T, TClose>,
+): Operation<TClose> {
+  let next = yield* subscription.next();
+  while (!next.done) {
+    next = yield* subscription.next();
+  }
+  return next.value;
+}