Open History to see the last commit
1import * as Effect from "effect/Effect";
2import { WebSocketServer, type Server } from "ws";
3import * as RpcServer from "./RpcServer.ts";
4import { SESSION_ENV_PARAM } from "./RpcServerEnvironment.ts";
5
6export const RpcServerNode = RpcServer.layerServer(
7 Effect.fn(function* ({ parentConnected, parentDisconnected, createRpcSession }) {
8 const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
9 const url = yield* Effect.callback<string>((resume) => {
10 server.on("connection", (ws, req) => {
11 if (req.url?.startsWith("/parent")) {
12 parentConnected();
13 ws.on("close", () => {
14 parentDisconnected();
15 });
16 return;
17 }
18 const sessionEnv =
19 new URL(req.url ?? "/", "http://localhost").searchParams.get(SESSION_ENV_PARAM) ??
20 undefined;
21 const session = createRpcSession(ws, sessionEnv);
22 ws.on("message", (data) => {
23 session.dispatch.message(
24 (Array.isArray(data)
25 ? Buffer.concat(data)
26 : Buffer.isBuffer(data)
27 ? data
28 : Buffer.from(data)
29 ).toString(),
30 );
31 });
32 ws.on("close", (code, reason) => {
33 session.dispatch.close(code, reason.toString());
34 });
35 });
36 server.on("error", (error) => {
37 resume(Effect.die(error));
38 });
39 server.on("listening", () => {
40 resume(getServerAddress(server));
41 });
42 });
43 yield* Effect.addFinalizer(() => Effect.sync(() => server.close()));
44 return { url };
45 }),
46);
47
48function getServerAddress(server: Server): Effect.Effect<string> {
49 const address = server.address();
50 if (
51 typeof address === "object" &&
52 address !== null &&
53 "address" in address &&
54 "port" in address
55 ) {
56 return Effect.succeed(
57 `ws://${address.address === "::" ? "localhost" : address.address}:${address.port}`,
58 );
59 }
60 return Effect.die(
61 new Error(
62 `Server address is not an object with address and port properties: ${JSON.stringify(address)}`,
63 ),
64 );
65}