#13272

Committed 04 Nov 2024 02:16PM UTC coverage: 89.102%. Remained the same

Build Type

push

Committed by

Commit Message

Merge f33c894f1 into 30be955de

Run Details

7249 of 8577 branches covered (84.52%)

Branch coverage included in aggregate %.

10 of 10 new or added lines in 1 file covered. (100.0%)

4 existing lines in 1 file now uncovered.

9650 of 10389 relevant lines covered (92.89%)

1782.12 hits per line

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

70.71

/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';
import { Deferred } from '../../deferred';

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();

    private activeRequests = new Map<number, {
        id: number;
        deferred: Deferred<any>;
    }>();

    /**
     * Get the response with this ID
     * @param id the ID of the response
     * @returns the message
     */
    private onResponse<T, R = WorkerResponse<T>>(id: number): Promise<R> {
        const deferred = new Deferred<R>();

        //store this request so we can resolve it later, or reject if this class is disposed
        this.activeRequests.set(id, {
            id: id,
            deferred: deferred
        });

        this.emitter.once(`response-${id}`, (response) => {
            deferred.resolve(response);
            this.activeRequests.delete(id);
        });

        return deferred.promise;
    }

    /**
     * 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);
        //reject all active requests
        for (const request of this.activeRequests.values()) {
            request.deferred.reject(new Error(`Request ${request.id} has been rejected because MessageHandler is now disposed`));
        }
    }
}

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

rokucommunity / brighterscript / #13272

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