History

nits

Julius Marminge committed 8 months ago

Commit 5a755ad1 parent b211d053 files · +33 −29Browse files

Showing the changes to apps/server/src (1 of 3 files). Show all

Unified · Split

apps/server/src/wsServer.ts+30−27
@@ -183,23 +183,22 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 182 unmodified lines
183 });183 });
184 }184 }
185 185
186 const onTerminalEvent = (event: TerminalEvent) =>186 const onTerminalEvent = Effect.fnUntraced(function* (event: TerminalEvent) {
187 Effect.gen(function* () {187 const push: WsPush = {
188 const push: WsPush = {188 type: "push",
189 type: "push",189 channel: WS_CHANNELS.terminalEvent,
190 channel: WS_CHANNELS.terminalEvent,190 data: event,
191 data: event,191 };
192 };192 const message = yield* Schema.encodeEffect(Schema.fromJsonString(WsPush))(push);
193 const message = yield* Schema.encodeEffect(Schema.fromJsonString(WsPush))(push);193 let recipients = 0;
194 let recipients = 0;194 for (const client of yield* Ref.get(clients)) {
195 for (const client of yield* Ref.get(clients)) {195 if (client.readyState === client.OPEN) {
196 if (client.readyState === client.OPEN) {196 client.send(message);
197 client.send(message);197 recipients += 1;
198 recipients += 1;
199 }
200 }198 }
201 logOutgoingPush(push, recipients);199 }
202 }).pipe(Effect.runPromise);200 logOutgoingPush(push, recipients);
201 });
203 202
204 // HTTP server — serves static files or redirects to Vite dev server203 // HTTP server — serves static files or redirects to Vite dev server
205 const httpServer = http.createServer((req, res) => {204 const httpServer = http.createServer((req, res) => {
@@ -289,9 +288,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 83 unmodified lines
289 const { openInEditor } = yield* Open;288 const { openInEditor } = yield* Open;
290 289
291 const subscriptionsScope = yield* Scope.make("sequential");290 const subscriptionsScope = yield* Scope.make("sequential");
292 yield* Effect.addFinalizer(() =>291 yield* Effect.addFinalizer(() => Scope.close(subscriptionsScope, Exit.void));
293 Scope.close(subscriptionsScope, Exit.void).pipe(Effect.orElseSucceed(() => undefined)),
294 );
295 292
296 yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) =>293 yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) =>
297 Effect.gen(function* () {294 Effect.gen(function* () {
@@ -312,7 +309,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 14 unmodified lines
312 }),309 }),
313 ).pipe(Effect.forkIn(subscriptionsScope));310 ).pipe(Effect.forkIn(subscriptionsScope));
314 311
315 yield* orchestrationReactor.start.pipe(Scope.provide(subscriptionsScope));312 yield* Scope.provide(orchestrationReactor.start, subscriptionsScope);
316 313
317 if (autoBootstrapProjectFromCwd) {314 if (autoBootstrapProjectFromCwd) {
318 const snapshot = yield* projectionReadModelQuery.getSnapshot();315 const snapshot = yield* projectionReadModelQuery.getSnapshot();
@@ -332,7 +329,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 13 unmodified lines
332 }329 }
333 }330 }
334 331
335 const routeRequest = Effect.fn(function* (request: WebSocketRequest) {332 const routeRequest = Effect.fnUntraced(function* (request: WebSocketRequest) {
336 switch (request.body._tag) {333 switch (request.body._tag) {
337 case ORCHESTRATION_WS_METHODS.getSnapshot:334 case ORCHESTRATION_WS_METHODS.getSnapshot:
338 return yield* projectionReadModelQuery.getSnapshot();335 return yield* projectionReadModelQuery.getSnapshot();
@@ -573,18 +570,24 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E · 234 unmodified lines
573 ),570 ),
574 );571 );
575 572
576 const unsubscribeTerminalEvents = yield* terminalManager.subscribe(onTerminalEvent);573 const unsubscribeTerminalEvents = yield* terminalManager.subscribe(
574 (event) => void Effect.runPromise(onTerminalEvent(event)),
575 );
577 yield* Effect.addFinalizer(() =>576 yield* Effect.addFinalizer(() =>
578 Effect.sync(() => {577 Effect.all([Effect.sync(() => unsubscribeTerminalEvents()), terminalManager.dispose()]),
579 unsubscribeTerminalEvents();
580 }),
581 );578 );
582 yield* Effect.addFinalizer(terminalManager.dispose);
583 579
584 yield* NodeHttpServer.make(() => httpServer, listenOptions);580 yield* NodeHttpServer.make(() => httpServer, listenOptions);
585 581
586 yield* Effect.addFinalizer(() =>582 yield* Effect.addFinalizer(() =>
587 Effect.all([closeAllClients, closeWebSocketServer.pipe(Effect.orElseSucceed(() => undefined))]),583 Effect.all([
584 closeAllClients,
585 closeWebSocketServer.pipe(
586 Effect.catch((error) =>
587 Effect.logWarning("failed to close web socket server", { cause: error }),
588 ),
589 ),
590 ]),
588 );591 );
589 592
590 return httpServer;593 return httpServer;