@setoelkahfi / sigit-vscode
Code Commits Pulls 1 CI/CD
feature/sigit-vscode-v1
development feature/add-registry feature/sigit-vscode-v1 main
ts 216 lines 6.2 KB
Raw
1 import { EventEmitter } from "events";
2 import { Readable, Writable } from "stream";
3
4 /**
5 * A generic JSON-RPC 2.0 peer over newline-delimited JSON (ndjson).
6 *
7 * Reads from a readable stream (the agent's stdout) and writes to a writable
8 * stream (the agent's stdin). Both incoming requests/notifications and our own
9 * outbound requests/notifications are supported, so this is a symmetric peer
10 * rather than a one-directional client.
11 */
12
13 type JsonValue =
14 | null
15 | boolean
16 | number
17 | string
18 | JsonValue[]
19 | { [key: string]: JsonValue };
20
21 interface JsonRpcRequest {
22 jsonrpc: "2.0";
23 id: number | string;
24 method: string;
25 params?: unknown;
26 }
27
28 interface JsonRpcNotification {
29 jsonrpc: "2.0";
30 method: string;
31 params?: unknown;
32 }
33
34 interface JsonRpcResponse {
35 jsonrpc: "2.0";
36 id: number | string;
37 result?: unknown;
38 error?: JsonRpcError;
39 }
40
41 interface JsonRpcError {
42 code: number;
43 message: string;
44 data?: unknown;
45 }
46
47 type RequestHandler = (params: unknown) => unknown | Promise<unknown>;
48 type NotificationHandler = (params: unknown) => void;
49
50 interface PendingRequest {
51 resolve: (value: unknown) => void;
52 reject: (reason: unknown) => void;
53 }
54
55 export class Connection extends EventEmitter {
56 private readonly writable: Writable;
57 private readonly pending = new Map<number | string, PendingRequest>();
58 private readonly requestHandlers = new Map<string, RequestHandler>();
59 private readonly notificationHandlers = new Map<string, NotificationHandler>();
60 private buffer = "";
61 private nextId = 1;
62 private disposed = false;
63
64 constructor(readable: Readable, writable: Writable) {
65 super();
66 this.writable = writable;
67 readable.on("data", (chunk: Buffer | string) => this.onData(chunk.toString()));
68 readable.on("error", (err) => this.emit("error", err));
69 readable.on("close", () => this.emit("close"));
70 }
71
72 /** Send a request and resolve with its result (or reject on error). */
73 request<T = unknown>(method: string, params?: unknown): Promise<T> {
74 if (this.disposed) {
75 return Promise.reject(new Error("Connection disposed"));
76 }
77 const id = this.nextId++;
78 const payload: JsonRpcRequest = { jsonrpc: "2.0", id, method, params };
79 return new Promise<T>((resolve, reject) => {
80 this.pending.set(id, {
81 resolve: resolve as (value: unknown) => void,
82 reject
83 });
84 this.send(payload);
85 });
86 }
87
88 /** Send a fire-and-forget notification. */
89 notify(method: string, params?: unknown): void {
90 if (this.disposed) {
91 return;
92 }
93 const payload: JsonRpcNotification = { jsonrpc: "2.0", method, params };
94 this.send(payload);
95 }
96
97 /** Register a handler for inbound requests of a given method. */
98 onRequest(method: string, handler: RequestHandler): void {
99 this.requestHandlers.set(method, handler);
100 }
101
102 /** Register a handler for inbound notifications of a given method. */
103 onNotification(method: string, handler: NotificationHandler): void {
104 this.notificationHandlers.set(method, handler);
105 }
106
107 dispose(): void {
108 if (this.disposed) {
109 return;
110 }
111 this.disposed = true;
112 const err = new Error("Connection disposed");
113 for (const { reject } of this.pending.values()) {
114 reject(err);
115 }
116 this.pending.clear();
117 this.requestHandlers.clear();
118 this.notificationHandlers.clear();
119 this.removeAllListeners();
120 }
121
122 private send(message: JsonValue | JsonRpcRequest | JsonRpcNotification | JsonRpcResponse): void {
123 try {
124 this.writable.write(JSON.stringify(message) + "\n");
125 } catch (err) {
126 this.emit("error", err);
127 }
128 }
129
130 private onData(text: string): void {
131 this.buffer += text;
132 let newlineIndex: number;
133 while ((newlineIndex = this.buffer.indexOf("\n")) !== -1) {
134 const line = this.buffer.slice(0, newlineIndex).trim();
135 this.buffer = this.buffer.slice(newlineIndex + 1);
136 if (line.length === 0) {
137 continue;
138 }
139 this.handleLine(line);
140 }
141 }
142
143 private handleLine(line: string): void {
144 let message: JsonRpcRequest | JsonRpcResponse | JsonRpcNotification;
145 try {
146 message = JSON.parse(line);
147 } catch (err) {
148 this.emit("error", new Error(`Failed to parse message: ${line}`));
149 return;
150 }
151
152 if ("id" in message && ("result" in message || "error" in message)) {
153 this.handleResponse(message as JsonRpcResponse);
154 } else if ("method" in message && "id" in message) {
155 void this.handleRequest(message as JsonRpcRequest);
156 } else if ("method" in message) {
157 this.handleNotification(message as JsonRpcNotification);
158 } else {
159 this.emit("error", new Error(`Unrecognized message: ${line}`));
160 }
161 }
162
163 private handleResponse(message: JsonRpcResponse): void {
164 const pending = this.pending.get(message.id);
165 if (!pending) {
166 return;
167 }
168 this.pending.delete(message.id);
169 if (message.error) {
170 const err = new Error(message.error.message);
171 (err as Error & { code?: number; data?: unknown }).code = message.error.code;
172 (err as Error & { code?: number; data?: unknown }).data = message.error.data;
173 pending.reject(err);
174 } else {
175 pending.resolve(message.result);
176 }
177 }
178
179 private async handleRequest(message: JsonRpcRequest): Promise<void> {
180 const handler = this.requestHandlers.get(message.method);
181 if (!handler) {
182 this.send({
183 jsonrpc: "2.0",
184 id: message.id,
185 error: { code: -32601, message: `Method not found: ${message.method}` }
186 });
187 return;
188 }
189 try {
190 const result = await handler(message.params);
191 this.send({ jsonrpc: "2.0", id: message.id, result: (result ?? null) as JsonValue });
192 } catch (err) {
193 const error = err as Error & { code?: number; data?: unknown };
194 this.send({
195 jsonrpc: "2.0",
196 id: message.id,
197 error: {
198 code: typeof error.code === "number" ? error.code : -32603,
199 message: error.message ?? "Internal error",
200 data: error.data
201 }
202 });
203 }
204 }
205
206 private handleNotification(message: JsonRpcNotification): void {
207 const handler = this.notificationHandlers.get(message.method);
208 if (handler) {
209 try {
210 handler(message.params);
211 } catch (err) {
212 this.emit("error", err);
213 }
214 }
215 }
216 }