1import {
2 TerminalRpcGroup,
3 WS_METHODS,
4 type TerminalAttachStreamEvent,
5 type TerminalEvent,
6 type TerminalError,
7} from "@t3tools/contracts";
8import * as Effect from "effect/Effect";
9import * as Queue from "effect/Queue";
10import * as Stream from "effect/Stream";
11import * as TerminalManager from "./Manager.ts";
12import { batchTerminalOutput } from "./OutputProtocol.ts";
13
14export const makeTerminalRpcHandlers = (
15 terminalManager: TerminalManager.TerminalManager["Service"],
16) =>
17 TerminalRpcGroup.of({
18 [WS_METHODS.terminalOpen]: (input) => terminalManager.open(input),
19 [WS_METHODS.terminalAttach]: (input) =>
20 Stream.callback<TerminalAttachStreamEvent, TerminalError>((queue) =>
21 Effect.acquireRelease(
22 terminalManager.attachStream(input, (event) => Queue.offer(queue, event)),
23 (unsubscribe) => Effect.sync(unsubscribe),
24 ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))),
25 ).pipe(batchTerminalOutput),
26 [WS_METHODS.terminalWrite]: (input) => terminalManager.write(input),
27 [WS_METHODS.terminalResize]: (input) => terminalManager.resize(input),
28 [WS_METHODS.terminalClear]: (input) => terminalManager.clear(input),
29 [WS_METHODS.terminalRestart]: (input) => terminalManager.restart(input),
30 [WS_METHODS.terminalClose]: (input) => terminalManager.close(input),
31 [WS_METHODS.terminalObserve]: (input) =>
32 Stream.callback<TerminalAttachStreamEvent, TerminalError>((queue) =>
33 Effect.acquireRelease(
34 terminalManager.observeStream(input, (event) => Queue.offer(queue, event)),
35 (unsubscribe) => Effect.sync(unsubscribe),
36 ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))),
37 ).pipe(batchTerminalOutput),
38 [WS_METHODS.subscribeTerminalEvents]: (_input) =>
39 Stream.callback<TerminalEvent>((queue) =>
40 Effect.acquireRelease(
41 terminalManager.subscribe((event) => Queue.offer(queue, event)),
42 (unsubscribe) => Effect.sync(unsubscribe),
43 ),
44 ).pipe(batchTerminalOutput),
45 });