History

fix

Julius Marminge committed 8 months ago

Commit b211d051 parent 52681631 file · +39 −33Browse files

Showing the changes to apps/server/src/wsServer.ts (1 of 1 file). Show all

Unified · Split

apps/server/src/wsServer.ts+39−33
@@ -16,7 +16,7 @@ import { · 15 unmodified lines
1616 WsResponse,
1717 } from "@t3tools/contracts";
1818 import { NodeHttpServer } from "@effect/platform-node";
19−import { Cause, Effect, Exit, Layer, Schema, Scope, ServiceMap, Stream, Struct } from "effect";
19+import { Cause, Effect, Exit, Layer, Ref, Schema, Scope, ServiceMap, Stream, Struct } from "effect";
2020 import { WebSocketServer, type WebSocket } from "ws";
2121
2222 import { createLogger } from "./logger";
@@ -171,7 +171,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 148 unmodified lines
171171 const keybindingsManager = yield* Keybindings;
172172 const git = yield* GitCore;
173173
174− const clients = new Set<WebSocket>();
174+ const clients = yield* Ref.make(new Set<WebSocket>());
175175 const logger = createLogger("ws");
176176
177177 function logOutgoingPush(push: WsPush, recipients: number) {
@@ -183,22 +183,23 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 5 unmodified lines
183183 });
184184 }
185185
186− const onTerminalEvent = (event: TerminalEvent) => {
187− const push: WsPush = {
188− type: "push",
189− channel: WS_CHANNELS.terminalEvent,
190− data: event,
191− };
192− const message = JSON.stringify(push);
193− let recipients = 0;
194− for (const client of clients) {
195− if (client.readyState === client.OPEN) {
196− client.send(message);
197− recipients += 1;
186+ const onTerminalEvent = (event: TerminalEvent) =>
187+ Effect.gen(function* () {
188+ const push: WsPush = {
189+ type: "push",
190+ channel: WS_CHANNELS.terminalEvent,
191+ data: event,
192+ };
193+ const message = yield* Schema.encodeEffect(Schema.fromJsonString(WsPush))(push);
194+ let recipients = 0;
195+ for (const client of yield* Ref.get(clients)) {
196+ if (client.readyState === client.OPEN) {
197+ client.send(message);
198+ recipients += 1;
199+ }
198200 }
199− }
200− logOutgoingPush(push, recipients);
201− };
201+ logOutgoingPush(push, recipients);
202+ }).pipe(Effect.runPromise);
202203
203204 // HTTP server — serves static files or redirects to Vite dev server
204205 const httpServer = http.createServer((req, res) => {
@@ -273,12 +274,10 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 68 unmodified lines
273274 });
274275 });
275276
276− const closeAllClients = () => {
277− for (const client of clients) {
278− client.close();
279− }
280− clients.clear();
281− };
277+ const closeAllClients = Ref.get(clients).pipe(
278+ Effect.flatMap(Effect.forEach((client) => Effect.sync(() => client.close()))),
279+ Effect.flatMap(() => Ref.set(clients, new Set())),
280+ );
282281
283282 const listenOptions = host ? { host, port } : { port };
284283
@@ -295,15 +294,15 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 10 unmodified lines
295294 );
296295
297296 yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) =>
298− Effect.sync(() => {
297+ Effect.gen(function* () {
299298 const push: WsPush = {
300299 type: "push",
301300 channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
302301 data: event,
303302 };
304− const message = JSON.stringify(push);
303+ const message = yield* Schema.encodeEffect(Schema.fromJsonString(WsPush))(push);
305304 let recipients = 0;
306− for (const client of clients) {
305+ for (const client of yield* Ref.get(clients)) {
307306 if (client.readyState === client.OPEN) {
308307 client.send(message);
309308 recipients += 1;
@@ -532,7 +531,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 222 unmodified lines
532531 });
533532
534533 wss.on("connection", (ws) => {
535− clients.add(ws);
534+ void Effect.runPromise(Ref.update(clients, (clients) => clients.add(ws)));
536535
537536 const segments = cwd.split(/[/\\]/).filter(Boolean);
538537 const projectName = segments[segments.length - 1] ?? "project";
@@ -550,11 +549,21 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 11 unmodified lines
550549 });
551550
552551 ws.on("close", () => {
553− clients.delete(ws);
552+ void Effect.runPromise(
553+ Ref.update(clients, (clients) => {
554+ clients.delete(ws);
555+ return clients;
556+ }),
557+ );
554558 });
555559
556560 ws.on("error", () => {
557− clients.delete(ws);
561+ void Effect.runPromise(
562+ Ref.update(clients, (clients) => {
563+ clients.delete(ws);
564+ return clients;
565+ }),
566+ );
558567 });
559568 });
560569
@@ -575,10 +584,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 14 unmodified lines
575584 yield* NodeHttpServer.make(() => httpServer, listenOptions);
576585
577586 yield* Effect.addFinalizer(() =>
578− Effect.all([
579− Effect.sync(() => closeAllClients()),
580− closeWebSocketServer.pipe(Effect.orElseSucceed(() => undefined)),
581− ]),
587+ Effect.all([closeAllClients, closeWebSocketServer.pipe(Effect.orElseSucceed(() => undefined))]),
582588 );
583589
584590 return httpServer;