24839896359

Committed 23 Apr 2026 02:06PM UTC coverage: 59.334% (-0.3%) from 59.627%

Build # 24839896359

Build Type

push

github

Committed by

web-flow

Commit Message

fix(json) Only emit batches when we have complete elements (#3400)

Coverage Stats

11234 of 20699 branches covered (54.27%)

Branch coverage included in aggregate %.

24 of 25 new or added lines in 1 file covered. (96.0%)

123 existing lines in 8 files now uncovered.

23043 of 37071 relevant lines covered (62.16%)

16510.97 hits per line

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

78.57

/modules/loader-utils/src/lib/worker-loader-utils/parse-with-worker.ts

import {
  WorkerJob,
  WorkerMessageType,
  WorkerMessagePayload,
  isBrowser,
  WorkerFarm,
  getWorkerURL
} from '@loaders.gl/worker-utils';
import type {Loader, LoaderOptions, LoaderContext} from '../../loader-types';

/**
 * Determines if a loader can parse with worker
 * @param loader
 * @param options
 */
export function canParseWithWorker(loader: Loader, options?: LoaderOptions) {
  if (!WorkerFarm.isSupported()) {
    return false;
  }

  // Node workers are still experimental
  const nodeWorkers = options?._nodeWorkers ?? options?.core?._nodeWorkers;
  if (!isBrowser && !nodeWorkers) {
    return false;
  }

  // Excel Arrow output currently relies on main-thread conversion helpers that
  // should stay aligned with the primary loader path.
  if (
    loader.id === 'excel' &&
    (options as {excel?: {shape?: string}} | undefined)?.excel?.shape === 'arrow-table'
  ) {
    return false;
  }

  const useWorkers = options?.worker ?? options?.core?.worker;
  return Boolean(loader.worker && useWorkers);
}

/**
 * this function expects that the worker function sends certain messages,
 * this can be automated if the worker is wrapper by a call to createLoaderWorker in @loaders.gl/loader-utils.
 */
export async function parseWithWorker(
  loader: Loader,
  data: any,
  options?: LoaderOptions,
  context?: LoaderContext,
  parseOnMainThread?: (arrayBuffer: ArrayBuffer, options: {[key: string]: any}) => Promise<unknown>
) {
  const name = loader.id; // TODO
  const url = getWorkerURL(loader, options);

  const workerFarm = WorkerFarm.getWorkerFarm(options?.core);
  const workerPool = workerFarm.getWorkerPool({name, url});

  // options.log object contains functions which cannot be transferred
  // context.fetch & context.parse functions cannot be transferred
  // TODO - decide how to handle logging on workers
  options = JSON.parse(JSON.stringify(options));
  context = JSON.parse(JSON.stringify(context || {}));

  const job = await workerPool.startJob(
    'process-on-worker',
    // @ts-expect-error
    onMessage.bind(null, parseOnMainThread) // eslint-disable-line @typescript-eslint/no-misused-promises
  );

  job.postMessage('process', {
    // @ts-ignore
    input: data,
    options,
    context
  });

  const result = await job.result;
  // TODO - what is going on here?
  return await result.result;
}

/**
 * Handle worker's responses to the main thread
 * @param job
 * @param type
 * @param payload
 */
async function onMessage(
  parseOnMainThread: (arrayBuffer: ArrayBuffer, options?: {[key: string]: any}) => Promise<void>,
  job: WorkerJob,
  type: WorkerMessageType,
  payload: WorkerMessagePayload
) {
  switch (type) {
    case 'done':
      job.done(payload);
      break;

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

    case 'process':
      // Worker is asking for main thread to parseO
      const {id, input, options} = payload;
      try {
        const result = await parseOnMainThread(input, options);
        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(`parse-with-worker unknown message ${type}`);
  }
}

1	import {
2	WorkerJob,
3	WorkerMessageType,
4	WorkerMessagePayload,
5	isBrowser,
6	WorkerFarm,
7	getWorkerURL
8	} from '@loaders.gl/worker-utils';
9	import type {Loader, LoaderOptions, LoaderContext} from '../../loader-types';
10
11	/**
12	* Determines if a loader can parse with worker
13	* @param loader
14	* @param options
15	*/
16	export function canParseWithWorker(loader: Loader, options?: LoaderOptions) {
17	if (!WorkerFarm.isSupported()) {	1,908!
18	return false;	×
19	}
20
21	// Node workers are still experimental
22	const nodeWorkers = options?._nodeWorkers ?? options?.core?._nodeWorkers;	1,908✔
23	if (!isBrowser && !nodeWorkers) {	1,908✔
24	return false;	901✔
25	}
26
27	// Excel Arrow output currently relies on main-thread conversion helpers that
28	// should stay aligned with the primary loader path.
29	if (	1,007✔
30	loader.id === 'excel' &&	1,018✔
31	(options as {excel?: {shape?: string}} \| undefined)?.excel?.shape === 'arrow-table'
32	) {
33	return false;	4✔
34	}
35
36	const useWorkers = options?.worker ?? options?.core?.worker;	1,003✔
37	return Boolean(loader.worker && useWorkers);	1,908✔
38	}
39
40	/**
41	* this function expects that the worker function sends certain messages,
42	* this can be automated if the worker is wrapper by a call to createLoaderWorker in @loaders.gl/loader-utils.
43	*/
44	export async function parseWithWorker(
45	loader: Loader,
46	data: any,
47	options?: LoaderOptions,
48	context?: LoaderContext,
49	parseOnMainThread?: (arrayBuffer: ArrayBuffer, options: {[key: string]: any}) => Promise<unknown>
50	) {
51	const name = loader.id; // TODO	102✔
52	const url = getWorkerURL(loader, options);	102✔
53
54	const workerFarm = WorkerFarm.getWorkerFarm(options?.core);	102✔
55	const workerPool = workerFarm.getWorkerPool({name, url});	102✔
56
57	// options.log object contains functions which cannot be transferred
58	// context.fetch & context.parse functions cannot be transferred
59	// TODO - decide how to handle logging on workers
60	options = JSON.parse(JSON.stringify(options));	102✔
61	context = JSON.parse(JSON.stringify(context \|\| {}));	102!
62
63	const job = await workerPool.startJob(	102✔
64	'process-on-worker',
65	// @ts-expect-error
66	onMessage.bind(null, parseOnMainThread) // eslint-disable-line @typescript-eslint/no-misused-promises
67	);
68
69	job.postMessage('process', {	102✔
70	// @ts-ignore
71	input: data,
72	options,
73	context
74	});
75
76	const result = await job.result;	102✔
77	// TODO - what is going on here?
78	return await result.result;	102✔
79	}
80
81	/**
82	* Handle worker's responses to the main thread
83	* @param job
84	* @param type
85	* @param payload
86	*/
87	async function onMessage(
88	parseOnMainThread: (arrayBuffer: ArrayBuffer, options?: {[key: string]: any}) => Promise<void>,
89	job: WorkerJob,
90	type: WorkerMessageType,
91	payload: WorkerMessagePayload
92	) {
93	switch (type) {	109!
94	case 'done':
95	job.done(payload);	102✔
96	break;	102✔
97
98	case 'error':
UNCOV 99	job.error(new Error(payload.error));	×
100	break;	×
101
102	case 'process':
103	// Worker is asking for main thread to parseO
104	const {id, input, options} = payload;	7✔
105	try {	7✔
106	const result = await parseOnMainThread(input, options);	7✔
107	job.postMessage('done', {id, result});	7✔
108	} catch (error) {
UNCOV 109	const message = error instanceof Error ? error.message : 'unknown error';	×
UNCOV 110	job.postMessage('error', {id, error: message});	×
111	}
112	break;	7✔
113
114	default:
115	// eslint-disable-next-line
UNCOV 116	console.warn(`parse-with-worker unknown message ${type}`);	×
117	}
118	}

visgl / loaders.gl / 24839896359

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