21283815940

Committed 23 Jan 2026 10:59AM UTC coverage: 92.838% (+0.01%) from 92.825%

Build # 21283815940

Build Type

Pull #8423

github

Committed by

web-flow

Commit Message

Merge 74416f189 into 405411fa8

Pull Request Pull Request #8423: fix: Add NotifyEmitter

Run Details

8874 of 10620 branches covered (83.56%)

62 of 64 new or added lines in 3 files covered. (96.88%)

4 existing lines in 2 files now uncovered.

17734 of 19102 relevant lines covered (92.84%)

31510.24 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.22

/packages/cspell-lib/src/rpc/server.ts

import { assert } from './assert.js';
import { MalformedRPCRequestError, UnknownMethodRPCRequestError } from './errors.js';
import type { MessagePortLike } from './messagePort.js';
import type { RequestID, ResponseCode, RPCMessage, RPCRequestMessage } from './models.js';
import {
    createRPCCanceledResponse,
    createRPCError,
    createRPCOkResponse,
    createRPCReadyResponse,
    createRPCResponse,
    isRPCBaseMessage,
    isRPCCancelRequest,
    isRPCOkRequest,
    isRPCReadyRequest,
    isRPCRequest,
} from './modelsHelpers.js';
import type { RPCProtocol, RPCProtocolMethodNames } from './protocol.js';

const RESPONSE_CODES = {
    OK: 200 as const,
    BadRequest: 400 as const,
    RequestTimeout: 408 as const,
    InternalServerError: 500 as const,
    ServiceUnavailable: 503 as const,
} as const;

export interface RPCServerOptions {
    /**
     * If true, the server will close the message port when disposed.
     * @default false
     */
    closePortOnDispose?: boolean;
    /**
     * If true, the server will respond with an error message for unknown or malformed requests.
     * @default false
     */
    returnMalformedRPCRequestError?: boolean;
}

export interface RPCServerConfiguration extends RPCServerOptions {
    /**
     * The message port to use for communication.
     */
    port: MessagePortLike;
}

interface PendingRequest {
    requestMessage: RPCRequestMessage;
    promise: Promise<void>;
}

class RPCServerImpl<
    ServerApi,
    PApi extends RPCProtocol<ServerApi> = RPCProtocol<ServerApi>,
    MethodsNames extends RPCProtocolMethodNames<PApi> = RPCProtocolMethodNames<PApi>,
> {
    #port: MessagePortLike;
    #options: RPCServerOptions;
    #onMessage: (msg: unknown) => void;
    #onClose: () => void;
    #isClosed: boolean;
    #allowedMethods: Set<MethodsNames>;
    #methods: PApi;
    #pendingRequests: Map<number | string, PendingRequest>;

    constructor(config: RPCServerConfiguration, methods: ServerApi) {
        const port = config.port;
        this.#port = port;
        this.#isClosed = false;
        this.#options = config;
        this.#methods = methods as unknown as PApi;

        this.#allowedMethods = new Set(
            Object.keys(this.#methods).filter(
                (k) => typeof this.#methods[k as MethodsNames] === 'function',
            ) as MethodsNames[],
        );

        this.#pendingRequests = new Map();
        this.#onMessage = (msg: unknown) => this.#handleMessage(msg);
        this.#onClose = () => this.#cancelAllRequests(new Error('RPC Server port closed'));
        port.addListener('close', this.#onClose);
        port.addListener('message', this.#onMessage);
        port.start?.();
        this.#sendReadyResponse();
    }

    #sendResponse(response: RPCMessage): void {
        if (this.#isClosed) return;
        this.#port.postMessage(response);
    }

    #handleMessage(msg: unknown): void {
        let id: RequestID = 0;
        try {
            if (!isRPCBaseMessage(msg)) {
                if (this.#options.returnMalformedRPCRequestError) {
                    throw new MalformedRPCRequestError('Malformed RPC request', msg);
                }
                // Not a valid RPC message; ignore.
                return;
            }
            id = msg.id;
            if (isRPCCancelRequest(msg)) {
                // For now just remove it from pending requests.
                // later, implement aborting the request if possible.
                this.#pendingRequests.delete(msg.id);
                this.#sendCancelResponse(msg.id, RESPONSE_CODES.OK);
                return;
            }
            if (isRPCReadyRequest(msg)) {
                this.#sendReadyResponse(msg.id);
                return;
            }
            if (isRPCOkRequest(msg)) {
                this.#sendResponse(createRPCOkResponse(msg.id, RESPONSE_CODES.OK));
                return;
            }
            if (!isRPCRequest(msg)) {
                if (this.#options.returnMalformedRPCRequestError) {
                    throw new MalformedRPCRequestError('Malformed RPC request', msg);
                }
                // Not a request; ignore.
                return;
            }
            this.#handleRequest(msg);
        } catch (err) {
            this.#sendErrorResponse(id, err);
        }
    }

    #sendReadyResponse(id?: RequestID): void {
        this.#sendResponse(createRPCReadyResponse(id || 0, RESPONSE_CODES.OK));
    }

    #isMethod(method: string): method is MethodsNames {
        return this.#allowedMethods.has(method as MethodsNames);
    }

    #handleRequest(msg: RPCRequestMessage): void {
        const handleAsync = async () => {
            if (!this.#isMethod(msg.method)) {
                this.#sendErrorResponse(msg.id, new UnknownMethodRPCRequestError(msg.method));
                return;
            }
            const method = msg.method;
            const params = msg.params as Parameters<PApi[typeof method]>;
            assert(Array.isArray(params), 'RPC method parameters must be an array');
            assert(typeof this.#methods[method] === 'function', `RPC method ${method} is not a function`);

            const result = await this.#methods[method](...params);

            if (this.#pendingRequests.has(msg.id)) {
                const response = createRPCResponse(msg.id, result, RESPONSE_CODES.OK);
                this.#sendResponse(response);
            }
        };

        this.#pendingRequests.set(msg.id, {
            requestMessage: msg,
            promise: handleAsync().catch((err) =>
                this.#sendErrorResponse(msg.id, err, RESPONSE_CODES.InternalServerError),
            ),
        });

        return;
    }

    #sendCancelResponse(id: RequestID, code: ResponseCode = RESPONSE_CODES.ServiceUnavailable): void {
        this.#sendResponse(createRPCCanceledResponse(id, code));
    }

    #sendErrorResponse(id: RequestID, error: unknown, code: ResponseCode = RESPONSE_CODES.BadRequest): void {
        try {
            const err = error instanceof Error ? error : new Error(String(error));
            this.#sendResponse(createRPCError(id, err, code));
        } catch {
            // Nothing to do if the port is closed.
        }
    }

    #cancelAllRequests(reason?: unknown): void {
        if (!this.#pendingRequests.size) return;
        reason ??= new Error('RPC Server is shutting down');
        for (const id of this.#pendingRequests.keys()) {
            this.#sendErrorResponse(id, reason);
        }
        this.#pendingRequests.clear();
    }

    [Symbol.dispose](): void {
        this.#cancelAllRequests();
        this.#port.removeListener('message', this.#onMessage);
        this.#port.removeListener('close', this.#onClose);

        if (this.#options.closePortOnDispose) {
            this.#port.close?.();
        }
    }
}

/**
 * RPC Server implementation.
 * @param ServerApi - The API methods of the server.
 */
export class RPCServer<ServerApi> extends RPCServerImpl<ServerApi> {
    /**
     *
     * @param config - The server configuration, including the message port and options.
     * @param methods - The methods to implement the API.
     */
    constructor(config: RPCServerConfiguration, methods: ServerApi) {
        super(config, methods);
    }
}

1	import { assert } from './assert.js';
2	import { MalformedRPCRequestError, UnknownMethodRPCRequestError } from './errors.js';
3	import type { MessagePortLike } from './messagePort.js';
4	import type { RequestID, ResponseCode, RPCMessage, RPCRequestMessage } from './models.js';
5	import {
6	createRPCCanceledResponse,
7	createRPCError,
8	createRPCOkResponse,
9	createRPCReadyResponse,
10	createRPCResponse,
11	isRPCBaseMessage,
12	isRPCCancelRequest,
13	isRPCOkRequest,
14	isRPCReadyRequest,
15	isRPCRequest,
16	} from './modelsHelpers.js';
17	import type { RPCProtocol, RPCProtocolMethodNames } from './protocol.js';
18
19	const RESPONSE_CODES = {	3✔
20	OK: 200 as const,
21	BadRequest: 400 as const,
22	RequestTimeout: 408 as const,
23	InternalServerError: 500 as const,
24	ServiceUnavailable: 503 as const,
25	} as const;
26
27	export interface RPCServerOptions {
28	/**
29	* If true, the server will close the message port when disposed.
30	* @default false
31	*/
32	closePortOnDispose?: boolean;
33	/**
34	* If true, the server will respond with an error message for unknown or malformed requests.
35	* @default false
36	*/
37	returnMalformedRPCRequestError?: boolean;
38	}
39
40	export interface RPCServerConfiguration extends RPCServerOptions {
41	/**
42	* The message port to use for communication.
43	*/
44	port: MessagePortLike;
45	}
46
47	interface PendingRequest {
48	requestMessage: RPCRequestMessage;
49	promise: Promise<void>;
50	}
51
52	class RPCServerImpl<
53	ServerApi,
54	PApi extends RPCProtocol<ServerApi> = RPCProtocol<ServerApi>,
55	MethodsNames extends RPCProtocolMethodNames<PApi> = RPCProtocolMethodNames<PApi>,
56	> {
57	#port: MessagePortLike;
58	#options: RPCServerOptions;
59	#onMessage: (msg: unknown) => void;
60	#onClose: () => void;
61	#isClosed: boolean;
62	#allowedMethods: Set<MethodsNames>;
63	#methods: PApi;
64	#pendingRequests: Map<number \| string, PendingRequest>;
65
66	constructor(config: RPCServerConfiguration, methods: ServerApi) {
67	const port = config.port;	17✔
68	this.#port = port;	17✔
69	this.#isClosed = false;	17✔
70	this.#options = config;	17✔
71	this.#methods = methods as unknown as PApi;	17✔
72
73	this.#allowedMethods = new Set(	17✔
74	Object.keys(this.#methods).filter(
75	(k) => typeof this.#methods[k as MethodsNames] === 'function',	82✔
76	) as MethodsNames[],
77	);
78
79	this.#pendingRequests = new Map();	17✔
80	this.#onMessage = (msg: unknown) => this.#handleMessage(msg);	45✔
81	this.#onClose = () => this.#cancelAllRequests(new Error('RPC Server port closed'));	17✔
82	port.addListener('close', this.#onClose);	17✔
83	port.addListener('message', this.#onMessage);	17✔
84	port.start?.();	17✔
85	this.#sendReadyResponse();	17✔
86	}
87
88	#sendResponse(response: RPCMessage): void {
89	if (this.#isClosed) return;	56!
90	this.#port.postMessage(response);	56✔
91	}
92
93	#handleMessage(msg: unknown): void {
94	let id: RequestID = 0;	45✔
95	try {	45✔
96	if (!isRPCBaseMessage(msg)) {	45✔
97	if (this.#options.returnMalformedRPCRequestError) {	3✔
98	throw new MalformedRPCRequestError('Malformed RPC request', msg);	2✔
99	}
100	// Not a valid RPC message; ignore.
101	return;	1✔
102	}
103	id = msg.id;	42✔
104	if (isRPCCancelRequest(msg)) {	42✔
105	// For now just remove it from pending requests.
106	// later, implement aborting the request if possible.
107	this.#pendingRequests.delete(msg.id);	6✔
108	this.#sendCancelResponse(msg.id, RESPONSE_CODES.OK);	6✔
109	return;	6✔
110	}
111	if (isRPCReadyRequest(msg)) {	36✔
112	this.#sendReadyResponse(msg.id);	6✔
113	return;	6✔
114	}
115	if (isRPCOkRequest(msg)) {	30✔
116	this.#sendResponse(createRPCOkResponse(msg.id, RESPONSE_CODES.OK));	2✔
117	return;	2✔
118	}
119	if (!isRPCRequest(msg)) {	28✔
120	if (this.#options.returnMalformedRPCRequestError) {	3!
UNCOV 121	throw new MalformedRPCRequestError('Malformed RPC request', msg);	×
122	}
123	// Not a request; ignore.
124	return;	3✔
125	}
126	this.#handleRequest(msg);	25✔
127	} catch (err) {
128	this.#sendErrorResponse(id, err);	2✔
129	}
130	}
131
132	#sendReadyResponse(id?: RequestID): void {
133	this.#sendResponse(createRPCReadyResponse(id \|\| 0, RESPONSE_CODES.OK));	23✔
134	}
135
136	#isMethod(method: string): method is MethodsNames {
137	return this.#allowedMethods.has(method as MethodsNames);	25✔
138	}
139
140	#handleRequest(msg: RPCRequestMessage): void {
141	const handleAsync = async () => {	25✔
142	if (!this.#isMethod(msg.method)) {	25✔
143	this.#sendErrorResponse(msg.id, new UnknownMethodRPCRequestError(msg.method));	1✔
144	return;	1✔
145	}
146	const method = msg.method;	24✔
147	const params = msg.params as Parameters<PApi[typeof method]>;	24✔
148	assert(Array.isArray(params), 'RPC method parameters must be an array');	24✔
149	assert(typeof this.#methods[method] === 'function', `RPC method ${method} is not a function`);	24✔
150
151	const result = await this.#methods[method](...params);	24✔
152
153	if (this.#pendingRequests.has(msg.id)) {	17!
154	const response = createRPCResponse(msg.id, result, RESPONSE_CODES.OK);	17✔
155	this.#sendResponse(response);	17✔
156	}
157	};
158
159	this.#pendingRequests.set(msg.id, {	25✔
160	requestMessage: msg,
161	promise: handleAsync().catch((err) =>
UNCOV 162	this.#sendErrorResponse(msg.id, err, RESPONSE_CODES.InternalServerError),	×
163	),
164	});
165
166	return;	25✔
167	}
168
169	#sendCancelResponse(id: RequestID, code: ResponseCode = RESPONSE_CODES.ServiceUnavailable): void {	×
170	this.#sendResponse(createRPCCanceledResponse(id, code));	6✔
171	}
172
173	#sendErrorResponse(id: RequestID, error: unknown, code: ResponseCode = RESPONSE_CODES.BadRequest): void {	8✔
174	try {	8✔
175	const err = error instanceof Error ? error : new Error(String(error));	8!
176	this.#sendResponse(createRPCError(id, err, code));	8✔
177	} catch {
178	// Nothing to do if the port is closed.
179	}
180	}
181
182	#cancelAllRequests(reason?: unknown): void {
183	if (!this.#pendingRequests.size) return;	5✔
184	reason ??= new Error('RPC Server is shutting down');	3✔
185	for (const id of this.#pendingRequests.keys()) {	3✔
186	this.#sendErrorResponse(id, reason);	5✔
187	}
188	this.#pendingRequests.clear();	3✔
189	}
190
191	[Symbol.dispose](): void {
192	this.#cancelAllRequests();	5✔
193	this.#port.removeListener('message', this.#onMessage);	5✔
194	this.#port.removeListener('close', this.#onClose);	5✔
195
196	if (this.#options.closePortOnDispose) {	5✔
197	this.#port.close?.();	1✔
198	}
199	}
200	}
201
202	/**
203	* RPC Server implementation.
204	* @param ServerApi - The API methods of the server.
205	*/
206	export class RPCServer<ServerApi> extends RPCServerImpl<ServerApi> {
207	/**
208	*
209	* @param config - The server configuration, including the message port and options.
210	* @param methods - The methods to implement the API.
211	*/
212	constructor(config: RPCServerConfiguration, methods: ServerApi) {
213	super(config, methods);	17✔
214	}
215	}

streetsidesoftware / cspell / 21283815940

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous