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

1.37

/modules/worker-utils/src/lib/worker-api/create-worker.ts

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

import type {
  WorkerMessageType,
  WorkerMessagePayload,
  WorkerContext,
  WorkerJobContext,
  Process,
  ProcessInBatches
} from '../../types';
import AsyncQueue from '../async-queue/async-queue';
import WorkerBody from '../worker-farm/worker-body';
// import {validateWorkerVersion} from './validate-worker-version';

/** Counter for jobs */
let requestId = 0;
let inputBatches: AsyncQueue<any>;
let options: {[key: string]: any};

export type ProcessOnMainThread = (
  data: any,
  options?: {[key: string]: any},
  jobContext?: WorkerJobContext
) => any;

/**
 * Set up a WebWorkerGlobalScope to talk with the main thread
 */
export async function createWorker(
  process: Process,
  processInBatches?: ProcessInBatches
): Promise<void> {
  if (!(await WorkerBody.inWorkerThread())) {
    return;
  }

  const context: WorkerContext = {
    process: processOnMainThread
  };

  // eslint-disable-next-line complexity
  WorkerBody.onmessage = async (type: WorkerMessageType, payload: WorkerMessagePayload) => {
    try {
      switch (type) {
        case 'preload':
          WorkerBody.postMessage('done', {});
          break;

        case 'process':
          if (!process) {
            throw new Error('Worker does not support atomic processing');
          }
          const result = await process(
            payload.input,
            payload.options || {},
            context,
            payload.context || {}
          );
          WorkerBody.postMessage('done', {result});
          break;

        case 'process-in-batches':
          if (!processInBatches) {
            throw new Error('Worker does not support batched processing');
          }
          inputBatches = new AsyncQueue<any>();
          options = payload.options || {};
          const resultIterator = processInBatches(
            inputBatches,
            options,
            context,
            payload.context || {}
          );
          for await (const batch of resultIterator) {
            WorkerBody.postMessage('output-batch', {result: batch});
          }
          WorkerBody.postMessage('done', {});
          break;

        case 'input-batch':
          inputBatches.push(payload.input);
          break;

        case 'input-done':
          inputBatches.close();
          break;

        default:
      }
    } catch (error) {
      const message = error instanceof Error ? error.message : '';
      WorkerBody.postMessage('error', {error: message});
    }
  };
}

function processOnMainThread(arrayBuffer: ArrayBuffer, options = {}, jobContext = {}) {
  return new Promise((resolve, reject) => {
    const id = requestId++;

    /**
     */
    const onMessage = (type: string, payload: WorkerMessagePayload) => {
      if (payload.id !== id) {
        // not ours
        return;
      }

      switch (type) {
        case 'done':
          WorkerBody.removeEventListener(onMessage);
          resolve(payload.result);
          break;

        case 'error':
          WorkerBody.removeEventListener(onMessage);
          reject(payload.error);
          break;

        default:
        // ignore
      }
    };

    WorkerBody.addEventListener(onMessage);

    // Ask the main thread to decode data
    const payload = {id, input: arrayBuffer, options, context: jobContext};
    WorkerBody.postMessage('process', payload);
  });
}

1	// loaders.gl
2	// SPDX-License-Identifier: MIT
3	// Copyright (c) vis.gl contributors
4
5	import type {
6	WorkerMessageType,
7	WorkerMessagePayload,
8	WorkerContext,
9	WorkerJobContext,
10	Process,
11	ProcessInBatches
12	} from '../../types';
13	import AsyncQueue from '../async-queue/async-queue';
14	import WorkerBody from '../worker-farm/worker-body';
15	// import {validateWorkerVersion} from './validate-worker-version';
16
17	/** Counter for jobs */
18	let requestId = 0;	748✔
19	let inputBatches: AsyncQueue<any>;
20	let options: {[key: string]: any};
21
22	export type ProcessOnMainThread = (
23	data: any,
24	options?: {[key: string]: any},
25	jobContext?: WorkerJobContext
26	) => any;
27
28	/**
29	* Set up a WebWorkerGlobalScope to talk with the main thread
30	*/
31	export async function createWorker(
32	process: Process,
33	processInBatches?: ProcessInBatches
34	): Promise<void> {
35	if (!(await WorkerBody.inWorkerThread())) {	×
36	return;	×
37	}
38
39	const context: WorkerContext = {	×
40	process: processOnMainThread
41	};
42
43	// eslint-disable-next-line complexity
44	WorkerBody.onmessage = async (type: WorkerMessageType, payload: WorkerMessagePayload) => {	×
45	try {	×
46	switch (type) {	×
47	case 'preload':
NEW 48	WorkerBody.postMessage('done', {});	×
NEW 49	break;	×
50
51	case 'process':
52	if (!process) {	×
53	throw new Error('Worker does not support atomic processing');	×
54	}
NEW 55	const result = await process(	×
56	payload.input,
57	payload.options \|\| {},	×
58	context,
59	payload.context \|\| {}	×
60	);
61	WorkerBody.postMessage('done', {result});	×
62	break;	×
63
64	case 'process-in-batches':
65	if (!processInBatches) {	×
66	throw new Error('Worker does not support batched processing');	×
67	}
68	inputBatches = new AsyncQueue<any>();	×
69	options = payload.options \|\| {};	×
NEW 70	const resultIterator = processInBatches(	×
71	inputBatches,
72	options,
73	context,
74	payload.context \|\| {}	×
75	);
76	for await (const batch of resultIterator) {	×
77	WorkerBody.postMessage('output-batch', {result: batch});	×
78	}
79	WorkerBody.postMessage('done', {});	×
80	break;	×
81
82	case 'input-batch':
83	inputBatches.push(payload.input);	×
84	break;	×
85
86	case 'input-done':
87	inputBatches.close();	×
88	break;	×
89
90	default:
91	}
92	} catch (error) {
93	const message = error instanceof Error ? error.message : '';	×
94	WorkerBody.postMessage('error', {error: message});	×
95	}
96	};
97	}
98
99	function processOnMainThread(arrayBuffer: ArrayBuffer, options = {}, jobContext = {}) {	×
100	return new Promise((resolve, reject) => {	×
101	const id = requestId++;	×
102
103	/**
104	*/
105	const onMessage = (type: string, payload: WorkerMessagePayload) => {	×
106	if (payload.id !== id) {	×
107	// not ours
108	return;	×
109	}
110
111	switch (type) {	×
112	case 'done':
113	WorkerBody.removeEventListener(onMessage);	×
114	resolve(payload.result);	×
115	break;	×
116
117	case 'error':
118	WorkerBody.removeEventListener(onMessage);	×
119	reject(payload.error);	×
120	break;	×
121
122	default:
123	// ignore
124	}
125	};
126
127	WorkerBody.addEventListener(onMessage);	×
128
129	// Ask the main thread to decode data
NEW 130	const payload = {id, input: arrayBuffer, options, context: jobContext};	×
131	WorkerBody.postMessage('process', payload);	×
132	});
133	}

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