#13265

Committed 01 Nov 2024 05:58PM UTC coverage: 89.076% (+0.9%) from 88.214%

Build Type

push

Committed by

Commit Message

Merge 30be955de into 7cfaaa047

Run Details

7248 of 8577 branches covered (84.51%)

Branch coverage included in aggregate %.

1095 of 1214 new or added lines in 28 files covered. (90.2%)

24 existing lines in 5 files now uncovered.

9640 of 10382 relevant lines covered (92.85%)

1782.98 hits per line

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

69.17

/src/lsp/worker/MessageHandler.ts

import type { MessagePort, parentPort } from 'worker_threads';
import * as EventEmitter from 'eventemitter3';
import type { DisposableLike } from '../../interfaces';
import util from '../../util';

interface PseudoMessagePort {
    on: (name: 'message', cb: (message: any) => any) => any;
    postMessage: typeof parentPort['postMessage'];
}

export class MessageHandler<T, TRequestName = MethodNames<T>> {
    constructor(
        options: {
            name?: string;
            port: PseudoMessagePort;
            onRequest?: (message: WorkerRequest) => any;
            onResponse?: (message: WorkerResponse) => any;
            onUpdate?: (message: WorkerUpdate) => any;
        }
    ) {
        this.name = options?.name;
        this.port = options?.port;
        const listener = (message: WorkerMessage) => {
            switch (message.type) {
                case 'request':
                    options?.onRequest?.(message);
                    break;
                case 'response':
                    options?.onResponse?.(message);
                    this.emitter.emit(`${message.type}-${message.id}`, message);
                    break;
                case 'update':
                    options?.onUpdate?.(message);
                    break;
            }
        };
        options?.port.on('message', listener);

        this.disposables.push(
            this.emitter.removeAllListeners.bind(this.emitter),
            () => (options?.port as MessagePort).off('message', listener)
        );
    }

    /**
     * An optional name to help with debugging this handler
     */
    public readonly name: string;

    private port: PseudoMessagePort;

    private disposables: DisposableLike[] = [];

    private emitter = new EventEmitter();

    /**
     * Get the response with this ID
     * @param id the ID of the response
     * @returns the message
     */
    private onResponse<T>(id: number) {
        return new Promise<WorkerResponse<T>>((resolve) => {
            this.emitter.once(`response-${id}`, (response) => {
                resolve(response);
            });
        });
    }

    /**
     * A unique sequence for identifying messages
     */
    private idSequence = 0;

    /**
     * Send a request to the worker, and wait for a response.
     * @param name the name of the request
     * @param options the request options
     * @param options.data an array of data that will be passed in as params to the target function
     * @param options.id an id for this request
     */
    public async sendRequest<R>(name: TRequestName, options?: { data: any[]; id?: number }) {
        const request: WorkerMessage = {
            type: 'request',
            name: name as any,
            data: options?.data ?? [],
            id: options?.id ?? this.idSequence++
        };
        const responsePromise = this.onResponse<R>(request.id);
        this.port.postMessage(request);
        const response = await responsePromise;
        if ('error' in response) {
            //throw the error so it causes a rejected promise (like we'd expect)
            throw new Error(`Worker thread encountered an error: ${JSON.stringify(response.error.stack)}`);
        }
        return response;
    }

    /**
     * Send a request to the worker, and wait for a response.
     * @param request the request we are responding to
     * @param options options for this request
     */
    public sendResponse(request: WorkerMessage, options?: { data: any } | { error: Error } | undefined) {
        const response: WorkerResponse = {
            type: 'response',
            name: request.name,
            id: request.id
        };
        if ('error' in options) {
            //hack: turn the error into a plain json object
            response.error = this.errorToObject(options.error);
        } else if ('data' in options) {
            response.data = options.data;
        }
        this.port.postMessage(response);
    }

    /**
     * Send a request to the worker, and wait for a response.
     * @param name the name of the request
     * @param options options for the update
     * @param options.data an array of data that will be passed in as params to the target function
     * @param options.id an id for this update
     */
    public sendUpdate<T>(name: string, options?: { data?: any[]; id?: number }) {
        let update: WorkerMessage = {
            type: 'update',
            name: name,
            data: options?.data ?? [],
            id: options?.id ?? this.idSequence++
        };
        this.port.postMessage(update);
    }

    /**
     * Convert an Error object into a plain object so it can be serialized
     * @param error the error to object-ify
     * @returns an object version of an error
     */
    private errorToObject(error: Error) {
        return {
            name: error.name,
            message: error.message,
            stack: error.stack,
            cause: (error.cause as any)?.message && (error.cause as any)?.stack ? this.errorToObject(error.cause as any) : error.cause
        };
    }

    public dispose() {
        util.applyDispose(this.disposables);
    }
}

export interface WorkerRequest<TData = any> {
    id: number;
    type: 'request';
    name: string;
    data?: TData;
}

export interface WorkerResponse<TData = any> {
    id: number;
    type: 'response';
    name: string;
    data?: TData;
    /**
     * An error occurred on the remote side. There will be no `.data` value
     */
    error?: Error;
}

export interface WorkerUpdate<TData = any> {
    id: number;
    type: 'update';
    name: string;
    data?: TData;
}

export type WorkerMessage<T = any> = WorkerRequest<T> | WorkerResponse<T> | WorkerUpdate<T>;

export type MethodNames<T> = {
    [K in keyof T]: T[K] extends (...args: any[]) => any ? K : never;
}[keyof T];

1	import type { MessagePort, parentPort } from 'worker_threads';
2	import * as EventEmitter from 'eventemitter3';	1✔
3	import type { DisposableLike } from '../../interfaces';
4	import util from '../../util';	1✔
5
6	interface PseudoMessagePort {
7	on: (name: 'message', cb: (message: any) => any) => any;
8	postMessage: typeof parentPort['postMessage'];
9	}
10
11	export class MessageHandler<T, TRequestName = MethodNames<T>> {	1✔
12	constructor(
13	options: {
14	name?: string;
15	port: PseudoMessagePort;
16	onRequest?: (message: WorkerRequest) => any;
17	onResponse?: (message: WorkerResponse) => any;
18	onUpdate?: (message: WorkerUpdate) => any;
19	}
20	) {
21	this.name = options?.name;	5!
22	this.port = options?.port;	5!
23	const listener = (message: WorkerMessage) => {	5✔
24	switch (message.type) {	15✔
25	case 'request':	15✔
26	options?.onRequest?.(message);	1!
27	break;	1✔
28	case 'response':
29	options?.onResponse?.(message);	5!
30	this.emitter.emit(`${message.type}-${message.id}`, message);	5✔
31	break;	5✔
32	case 'update':
33	options?.onUpdate?.(message);	9!
34	break;	9✔
35	}
36	};
37	options?.port.on('message', listener);	5!
38
39	this.disposables.push(	5✔
40	this.emitter.removeAllListeners.bind(this.emitter),
41	() => (options?.port as MessagePort).off('message', listener)	3!
42	);
43	}
44
45	/**
46	* An optional name to help with debugging this handler
47	*/
48	public readonly name: string;
49
50	private port: PseudoMessagePort;
51
52	private disposables: DisposableLike[] = [];	5✔
53
54	private emitter = new EventEmitter();	5✔
55
56	/**
57	* Get the response with this ID
58	* @param id the ID of the response
59	* @returns the message
60	*/
61	private onResponse<T>(id: number) {
62	return new Promise<WorkerResponse<T>>((resolve) => {	5✔
63	this.emitter.once(`response-${id}`, (response) => {	5✔
64	resolve(response);	5✔
65	});
66	});
67	}
68
69	/**
70	* A unique sequence for identifying messages
71	*/
72	private idSequence = 0;	5✔
73
74	/**
75	* Send a request to the worker, and wait for a response.
76	* @param name the name of the request
77	* @param options the request options
78	* @param options.data an array of data that will be passed in as params to the target function
79	* @param options.id an id for this request
80	*/
81	public async sendRequest<R>(name: TRequestName, options?: { data: any[]; id?: number }) {
82	const request: WorkerMessage = {	5✔
83	type: 'request',
84	name: name as any,
85	data: options?.data ?? [],	30✔
86	id: options?.id ?? this.idSequence++	30!
87	};
88	const responsePromise = this.onResponse<R>(request.id);	5✔
89	this.port.postMessage(request);	5✔
90	const response = await responsePromise;	5✔
91	if ('error' in response) {	5✔
92	//throw the error so it causes a rejected promise (like we'd expect)
93	throw new Error(`Worker thread encountered an error: ${JSON.stringify(response.error.stack)}`);	1✔
94	}
95	return response;	4✔
96	}
97
98	/**
99	* Send a request to the worker, and wait for a response.
100	* @param request the request we are responding to
101	* @param options options for this request
102	*/
103	public sendResponse(request: WorkerMessage, options?: { data: any } \| { error: Error } \| undefined) {
104	const response: WorkerResponse = {	1✔
105	type: 'response',
106	name: request.name,
107	id: request.id
108	};
109	if ('error' in options) {	1!
110	//hack: turn the error into a plain json object
111	response.error = this.errorToObject(options.error);	1✔
NEW 112	} else if ('data' in options) {	×
NEW 113	response.data = options.data;	×
114	}
115	this.port.postMessage(response);	1✔
116	}
117
118	/**
119	* Send a request to the worker, and wait for a response.
120	* @param name the name of the request
121	* @param options options for the update
122	* @param options.data an array of data that will be passed in as params to the target function
123	* @param options.id an id for this update
124	*/
125	public sendUpdate<T>(name: string, options?: { data?: any[]; id?: number }) {
NEW 126	let update: WorkerMessage = {	×
127	type: 'update',
128	name: name,
129	data: options?.data ?? [],	×
130	id: options?.id ?? this.idSequence++	×
131	};
NEW 132	this.port.postMessage(update);	×
133	}
134
135	/**
136	* Convert an Error object into a plain object so it can be serialized
137	* @param error the error to object-ify
138	* @returns an object version of an error
139	*/
140	private errorToObject(error: Error) {
141	return {	1✔
142	name: error.name,
143	message: error.message,
144	stack: error.stack,
145	cause: (error.cause as any)?.message && (error.cause as any)?.stack ? this.errorToObject(error.cause as any) : error.cause	5!
146	};
147	}
148
149	public dispose() {
150	util.applyDispose(this.disposables);	3✔
151	}
152	}
153
154	export interface WorkerRequest<TData = any> {
155	id: number;
156	type: 'request';
157	name: string;
158	data?: TData;
159	}
160
161	export interface WorkerResponse<TData = any> {
162	id: number;
163	type: 'response';
164	name: string;
165	data?: TData;
166	/**
167	* An error occurred on the remote side. There will be no `.data` value
168	*/
169	error?: Error;
170	}
171
172	export interface WorkerUpdate<TData = any> {
173	id: number;
174	type: 'update';
175	name: string;
176	data?: TData;
177	}
178
179	export type WorkerMessage<T = any> = WorkerRequest<T> \| WorkerResponse<T> \| WorkerUpdate<T>;
180
181	export type MethodNames<T> = {
182	[K in keyof T]: T[K] extends (...args: any[]) => any ? K : never;
183	}[keyof T];

rokucommunity / brighterscript / #13265

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