25256585712

Committed 02 May 2026 04:35PM UTC coverage: 59.717% (-0.06%) from 59.776%

Build # 25256585712

Build Type

push

github

Committed by

web-flow

Commit Message

chore(loader-utils): Consolidate `parseWithWorker` with `processOnWorker` (#1564)

Coverage Stats

12514 of 23182 branches covered (53.98%)

Branch coverage included in aggregate %.

497 of 804 new or added lines in 22 files covered. (61.82%)

25 existing lines in 4 files now uncovered.

25948 of 41225 relevant lines covered (62.94%)

14803.35 hits per line

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

76.19

/modules/worker-utils/src/lib/worker-api/process-on-worker.ts

// loaders.gl
// SPDX-License-Identifier: MIT
// Copyright (c) vis.gl contributors

import type {
  WorkerObject,
  WorkerOptions,
  WorkerContext,
  WorkerJobContext,
  WorkerMessageType,
  WorkerMessagePayload
} from '../../types';
import type WorkerJob from '../worker-farm/worker-job';
import WorkerFarm from '../worker-farm/worker-farm';
import {getWorkerURL, getWorkerName} from './get-worker-url';
import {getTransferListForWriter} from '../worker-utils/get-transfer-list';

/** Options for worker processing */
export type ProcessOnWorkerOptions = WorkerOptions & {
  jobName?: string;
  [key: string]: any;
};

/** Options for preloading workers. */
export type PreloadWorkerOptions = {
  /** Number of workers to warm in the worker pool. */
  count?: number;
};

/**
 * Determines if we can parse with worker
 * @param loader
 * @param data
 * @param options
 */
export function canProcessOnWorker(worker: WorkerObject, options?: WorkerOptions) {
  if (!WorkerFarm.isSupported()) {
    return false;
  }

  return worker.worker && options?.worker;
}

/**
 * This function expects that the worker thread sends certain messages,
 * Creating such a worker can be automated if the worker is wrapper by a call to
 * createWorker in @loaders.gl/worker-utils.
 */
export async function processOnWorker(
  worker: WorkerObject,
  data: any,
  options: ProcessOnWorkerOptions = {},
  context: WorkerContext = {},
  jobContext: WorkerJobContext = {}
): Promise<any> {
  const name = getWorkerName(worker);

  const workerFarm = WorkerFarm.getWorkerFarm(options);
  const {source} = options;
  const workerPoolProps: {name: string; source?: string; url?: string} = {name, source};
  if (!source) {
    workerPoolProps.url = getWorkerURL(worker, options);
  }
  const workerPool = workerFarm.getWorkerPool(workerPoolProps);

  const jobName = options.jobName || worker.name;
  const job = await workerPool.startJob(
    jobName,
    // eslint-disable-next-line
    onMessage.bind(null, context)
  );

  // Kick off the processing in the worker
  const transferableOptions = getTransferListForWriter(options);
  const transferableContext = getTransferListForWriter(jobContext);
  job.postMessage('process', {
    input: data,
    options: transferableOptions,
    context: transferableContext
  });

  const result = await job.result;
  return result.result;
}

/**
 * Warm-start one or more workers in the same pool used by processOnWorker.
 * @param worker Worker object to preload.
 * @param options Worker options used to resolve the worker pool.
 * @param preloadOptions Preload options.
 */
export async function preloadWorker(
  worker: WorkerObject,
  options: ProcessOnWorkerOptions = {},
  preloadOptions: PreloadWorkerOptions = {}
): Promise<void> {
  const name = getWorkerName(worker);
  const workerFarm = WorkerFarm.getWorkerFarm(options);
  const {source} = options;
  const workerPoolProps: {name: string; source?: string; url?: string} = {name, source};
  if (!source) {
    workerPoolProps.url = getWorkerURL(worker, options);
  }
  const workerPool = workerFarm.getWorkerPool(workerPoolProps);
  const count = preloadOptions.count ?? options.maxConcurrency ?? options.core?.maxConcurrency ?? 1;

  const preloadJobs = Array.from({length: count}, async () => {
    const job = await workerPool.startJob(`${worker.name} preload`, onPreloadMessage);
    job.postMessage('preload', {});
    return await job.result;
  });

  await Promise.all(preloadJobs);
}

/**
 * Job completes when we receive the result
 * @param job
 * @param message
 */
async function onMessage(
  context: WorkerContext,
  job: WorkerJob,
  type: WorkerMessageType,
  payload: WorkerMessagePayload
) {
  switch (type) {
    case 'done':
      // Worker is done
      job.done(payload);
      break;

    case 'error':
      // Worker encountered an error
      job.error(new Error(payload.error));
      break;

    case 'process':
      // Worker is asking for us (main thread) to process something
      const {id, input, options} = payload;
      try {
        if (!context.process) {
          job.postMessage('error', {id, error: 'Worker not set up to process on main thread'});
          return;
        }
        const result = await context.process(input, options, undefined, payload.context || {});
        job.postMessage('done', {id, result});
      } catch (error) {
        const message = error instanceof Error ? error.message : 'unknown error';
        job.postMessage('error', {id, error: message});
      }
      break;

    default:
      // eslint-disable-next-line
      console.warn(`process-on-worker: unknown message ${type}`);
  }
}

/**
 * Completes a preload job when the worker acknowledges the preload message.
 * @param job Worker job.
 * @param type Worker message type.
 * @param payload Worker message payload.
 */
function onPreloadMessage(job: WorkerJob, type: WorkerMessageType, payload: WorkerMessagePayload) {
  switch (type) {
    case 'done':
      job.done(payload);
      break;

    case 'error':
      job.error(new Error(payload.error));
      break;

    default:
      // eslint-disable-next-line
      console.warn(`process-on-worker: unknown preload message ${type}`);
  }
}

1	// loaders.gl
2	// SPDX-License-Identifier: MIT
3	// Copyright (c) vis.gl contributors
4
5	import type {
6	WorkerObject,
7	WorkerOptions,
8	WorkerContext,
9	WorkerJobContext,
10	WorkerMessageType,
11	WorkerMessagePayload
12	} from '../../types';
13	import type WorkerJob from '../worker-farm/worker-job';
14	import WorkerFarm from '../worker-farm/worker-farm';
15	import {getWorkerURL, getWorkerName} from './get-worker-url';
16	import {getTransferListForWriter} from '../worker-utils/get-transfer-list';
17
18	/** Options for worker processing */
19	export type ProcessOnWorkerOptions = WorkerOptions & {
20	jobName?: string;
21	[key: string]: any;
22	};
23
24	/** Options for preloading workers. */
25	export type PreloadWorkerOptions = {
26	/** Number of workers to warm in the worker pool. */
27	count?: number;
28	};
29
30	/**
31	* Determines if we can parse with worker
32	* @param loader
33	* @param data
34	* @param options
35	*/
36	export function canProcessOnWorker(worker: WorkerObject, options?: WorkerOptions) {
37	if (!WorkerFarm.isSupported()) {	1,074!
38	return false;	×
39	}
40
41	return worker.worker && options?.worker;	1,074✔
42	}
43
44	/**
45	* This function expects that the worker thread sends certain messages,
46	* Creating such a worker can be automated if the worker is wrapper by a call to
47	* createWorker in @loaders.gl/worker-utils.
48	*/
49	export async function processOnWorker(
50	worker: WorkerObject,
51	data: any,
52	options: ProcessOnWorkerOptions = {},	133✔
53	context: WorkerContext = {},	133✔
54	jobContext: WorkerJobContext = {}	133✔
55	): Promise<any> {
56	const name = getWorkerName(worker);	133✔
57
58	const workerFarm = WorkerFarm.getWorkerFarm(options);	133✔
59	const {source} = options;	133✔
60	const workerPoolProps: {name: string; source?: string; url?: string} = {name, source};	133✔
61	if (!source) {	133✔
62	workerPoolProps.url = getWorkerURL(worker, options);	129✔
63	}
64	const workerPool = workerFarm.getWorkerPool(workerPoolProps);	133✔
65
66	const jobName = options.jobName \|\| worker.name;	133✔
67	const job = await workerPool.startJob(	133✔
68	jobName,
69	// eslint-disable-next-line
70	onMessage.bind(null, context)
71	);
72
73	// Kick off the processing in the worker
74	const transferableOptions = getTransferListForWriter(options);	133✔
75	const transferableContext = getTransferListForWriter(jobContext);	133✔
76	job.postMessage('process', {	133✔
77	input: data,
78	options: transferableOptions,
79	context: transferableContext
80	});
81
82	const result = await job.result;	133✔
83	return result.result;	131✔
84	}
85
86	/**
87	* Warm-start one or more workers in the same pool used by processOnWorker.
88	* @param worker Worker object to preload.
89	* @param options Worker options used to resolve the worker pool.
90	* @param preloadOptions Preload options.
91	*/
92	export async function preloadWorker(
93	worker: WorkerObject,
94	options: ProcessOnWorkerOptions = {},	2✔
95	preloadOptions: PreloadWorkerOptions = {}	2✔
96	): Promise<void> {
97	const name = getWorkerName(worker);	2✔
98	const workerFarm = WorkerFarm.getWorkerFarm(options);	2✔
99	const {source} = options;	2✔
100	const workerPoolProps: {name: string; source?: string; url?: string} = {name, source};	2✔
101	if (!source) {	2!
102	workerPoolProps.url = getWorkerURL(worker, options);	2✔
103	}
104	const workerPool = workerFarm.getWorkerPool(workerPoolProps);	2✔
105	const count = preloadOptions.count ?? options.maxConcurrency ?? options.core?.maxConcurrency ?? 1;	2!
106
107	const preloadJobs = Array.from({length: count}, async () => {	2✔
108	const job = await workerPool.startJob(`${worker.name} preload`, onPreloadMessage);	8✔
109	job.postMessage('preload', {});	8✔
110	return await job.result;	8✔
111	});
112
113	await Promise.all(preloadJobs);	2✔
114	}
115
116	/**
117	* Job completes when we receive the result
118	* @param job
119	* @param message
120	*/
121	async function onMessage(
122	context: WorkerContext,
123	job: WorkerJob,
124	type: WorkerMessageType,
125	payload: WorkerMessagePayload
126	) {
127	switch (type) {	144!
128	case 'done':
129	// Worker is done
130	job.done(payload);	131✔
131	break;	131✔
132
133	case 'error':
134	// Worker encountered an error
135	job.error(new Error(payload.error));	2✔
136	break;	2✔
137
138	case 'process':
139	// Worker is asking for us (main thread) to process something
140	const {id, input, options} = payload;	11✔
141	try {	11✔
142	if (!context.process) {	11!
143	job.postMessage('error', {id, error: 'Worker not set up to process on main thread'});	×
144	return;	×
145	}
146	const result = await context.process(input, options, undefined, payload.context \|\| {});	11✔
147	job.postMessage('done', {id, result});	11✔
148	} catch (error) {
149	const message = error instanceof Error ? error.message : 'unknown error';	×
150	job.postMessage('error', {id, error: message});	×
151	}
152	break;	11✔
153
154	default:
155	// eslint-disable-next-line
156	console.warn(`process-on-worker: unknown message ${type}`);	×
157	}
158	}
159
160	/**
161	* Completes a preload job when the worker acknowledges the preload message.
162	* @param job Worker job.
163	* @param type Worker message type.
164	* @param payload Worker message payload.
165	*/
166	function onPreloadMessage(job: WorkerJob, type: WorkerMessageType, payload: WorkerMessagePayload) {
167	switch (type) {	8!
168	case 'done':
169	job.done(payload);	8✔
170	break;	8✔
171
172	case 'error':
NEW 173	job.error(new Error(payload.error));	×
NEW 174	break;	×
175
176	default:
177	// eslint-disable-next-line
NEW 178	console.warn(`process-on-worker: unknown preload message ${type}`);	×
179	}
180	}

visgl / loaders.gl / 25256585712

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