26120672946

Committed 19 May 2026 07:37PM UTC coverage: 70.727% (+0.7%) from 70.049%

Build # 26120672946

Build Type

Pull #351

github

Committed by

web-flow

Commit Message

Merge eb0b2e542 into 5bbd82240

Pull Request Pull Request #351: 0.23.8

Coverage Stats

3328 of 5046 branches covered (65.95%)

Branch coverage included in aggregate %.

5834 of 7908 relevant lines covered (73.77%)

35.01 hits per line

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

64.54

/src/bsc/threading/ThreadMessageHandler.ts

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

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

export class ThreadMessageHandler<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(),
            () => (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 as R);
            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 unknown as Error) : 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';	2✔
3	import type { DisposableLike } from '../../interfaces';	2✔
4	import { Deferred } from 'brighterscript';	2✔
5	import { util } from '../../util';	2✔
6		2✔
7	interface PseudoMessagePort {
8	on: (name: 'message', cb: (message: any) => any) => any;
9	postMessage: typeof parentPort['postMessage'];	5✔
10	}	5✔
11		5✔
12	export class ThreadMessageHandler<T, TRequestName = MethodNames<T>> {
13	constructor(
14	options: {
15	name?: string;	5✔
16	port: PseudoMessagePort;	5!
17	onRequest?: (message: WorkerRequest) => any;	5!
18	onResponse?: (message: WorkerResponse) => any;	5✔
19	onUpdate?: (message: WorkerUpdate) => any;
20	}	4!
21	) {
22	this.name = options?.name;	2!
23	this.port = options?.port;	2✔
24	const listener = (message: WorkerMessage) => {
25	switch (message.type) {	2!
26	case 'request':	2✔
27	options?.onRequest?.(message);	2✔
28	break;
29	case 'response':	×
30	options?.onResponse?.(message);	×
31	this.emitter.emit(`${message.type}-${message.id}`, message);
32	break;
33	case 'update':	5!
34	options?.onUpdate?.(message);	5!
35	break;
36	}
37	};
38	options?.port.on('message', listener);
39
40	this.disposables.push(
41	() => this.emitter.removeAllListeners(),
42	() => (options?.port as MessagePort).off('message', listener)	3✔
43	);
44	}	3✔
45
46	/**
47	* An optional name to help with debugging this handler
48	*/	3✔
49	public readonly name: string;	2✔
50		2✔
51	private port: PseudoMessagePort;
52		3✔
53	private disposables: DisposableLike[] = [];
54
55	private emitter = new EventEmitter();
56
57	private activeRequests = new Map<number, {
58	id: number;
59	deferred: Deferred<any>;
60	}>();
61
62	/**
63	* Get the response with this ID	3✔
64	* @param id the ID of the response
65	* @returns the message
66	*/	18✔
67	private onResponse<T, R = WorkerResponse<T>>(id: number): Promise<R> {	18!
68	const deferred = new Deferred<R>();
69		3✔
70	//store this request so we can resolve it later, or reject if this class is disposed	3✔
71	this.activeRequests.set(id, {	3✔
72	id: id,	2✔
73	deferred: deferred
74	});	1✔
75
76	this.emitter.once(`response-${id}`, (response) => {	1✔
77	deferred.resolve(response as R);
78	this.activeRequests.delete(id);
79	});
80
81	return deferred.promise;
82	}
83
84	/**	1✔
85	* A unique sequence for identifying messages
86	*/
87	private idSequence = 0;
88
89	/**	1!
90	* Send a request to the worker, and wait for a response.
91	* @param name the name of the request	1✔
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 }) {	1✔
97	const request: WorkerMessage = {
98	type: 'request',
99	name: name as any,
100	data: options?.data ?? [],
101	id: options?.id ?? this.idSequence++
102	};
103	const responsePromise = this.onResponse<R>(request.id);
104	this.port.postMessage(request);
105	const response = await responsePromise;
106	if ('error' in response) {
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)}`);
109	}
110	return response;	×
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 = {
120	type: 'response',
121	name: request.name,
122	id: request.id	1✔
123	};
124	if ('error' in options) {
125	//hack: turn the error into a plain json object
126	response.error = this.errorToObject(options.error);	5!
127	} else if ('data' in options) {
128	response.data = options.data;
129	}
130	this.port.postMessage(response);	2✔
131	}
132		2✔
133	/**	1✔
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	2✔
138	* @param options.id an id for this update
139	*/
140	public sendUpdate<T>(name: string, options?: { data?: any[]; id?: number }) {
141	let update: WorkerMessage = {
142	type: 'update',
143	name: name,
144	data: options?.data ?? [],
145	id: options?.id ?? this.idSequence++
146	};
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 {
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 unknown as Error) : error.cause
161	};
162	}
163
164	public dispose() {
165	util.applyDispose(this.disposables);
166	//reject all active requests
167	for (const request of this.activeRequests.values()) {
168	request.deferred.reject(new Error(`Request ${request.id} has been rejected because MessageHandler is now disposed`));
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 / roku-debug / 26120672946

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