diff --git a/src/__tests__/__snapshots__/index.test.ts.snap b/src/__tests__/__snapshots__/index.test.ts.snap index 7b0f8eb18..c44dcf1ec 100644 --- a/src/__tests__/__snapshots__/index.test.ts.snap +++ b/src/__tests__/__snapshots__/index.test.ts.snap @@ -198,6 +198,7 @@ exports[`existence of exported functions 1`] = ` "isPowerOfTwo", "nextPowerOfTwo", "recursiveResolve", + "recursiveResolveTasks", "recursiveUntypeArrays", "stringify", "calculateAdaptiveWeights", diff --git a/src/utils/__tests__/recursiveResolve.test.ts b/src/utils/__tests__/recursiveResolve.test.ts index 39590876c..992d3d117 100644 --- a/src/utils/__tests__/recursiveResolve.test.ts +++ b/src/utils/__tests__/recursiveResolve.test.ts @@ -43,3 +43,17 @@ test('with array', async () => { }, }); }); + +test('null and undefined values', async () => { + const object = { + a: null, + b: undefined, + c: { d: null, e: Promise.resolve(1) }, + }; + + await expect(recursiveResolve(object)).resolves.toStrictEqual({ + a: null, + b: undefined, + c: { d: null, e: 1 }, + }); +}); diff --git a/src/utils/__tests__/recursiveResolveTasks.test.ts b/src/utils/__tests__/recursiveResolveTasks.test.ts new file mode 100644 index 000000000..8d5c27a99 --- /dev/null +++ b/src/utils/__tests__/recursiveResolveTasks.test.ts @@ -0,0 +1,119 @@ +import { expect, test } from 'vitest'; + +import { recursiveResolveTasks } from '../recursiveResolveTasks.ts'; + +function delay(ms: number) { + return new Promise((resolve) => { + setTimeout(resolve, ms); + }); +} + +test('primitive', async () => { + await expect(recursiveResolveTasks(1)).resolves.toBe(1); + await expect(recursiveResolveTasks({})).resolves.toStrictEqual({}); + await expect(recursiveResolveTasks(null)).resolves.toBeNull(); +}); + +test('simple object', async () => { + const object = { + a: { + b: { + c: () => Promise.resolve(1), + d: () => 2, + e: null, + }, + }, + }; + + await expect(recursiveResolveTasks(object)).resolves.toStrictEqual({ + a: { + b: { + c: 1, + d: 2, + e: null, + }, + }, + }); +}); + +test('with array', async () => { + const object = { + a: { + b: { + c: [() => Promise.resolve(1), () => Promise.resolve(2)], + }, + }, + }; + + await expect(recursiveResolveTasks(object)).resolves.toStrictEqual({ + a: { + b: { + c: [1, 2], + }, + }, + }); +}); + +test('bounds the number of tasks running at the same time', async () => { + let running = 0; + let peak = 0; + const object = { + values: Array.from({ length: 10 }, (_, index) => async () => { + running++; + if (running > peak) peak = running; + await delay(5); + running--; + return index; + }), + }; + + await recursiveResolveTasks(object, { concurrency: 3 }); + + expect(peak).toBe(3); + expect(object.values).toStrictEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]); +}); + +test('runs everything at once by default', async () => { + let running = 0; + let peak = 0; + const object = { + values: Array.from({ length: 10 }, (_, index) => async () => { + running++; + if (running > peak) peak = running; + await delay(5); + running--; + return index; + }), + }; + + await recursiveResolveTasks(object); + + expect(peak).toBe(10); + expect(object.values).toStrictEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]); +}); + +test('a failing task stops the remaining ones from starting', async () => { + let started = 0; + const object = { + values: Array.from({ length: 10 }, (_, index) => async () => { + started++; + if (index === 0) throw new Error('task 0 failed'); + await delay(10); + return index; + }), + }; + + await expect( + recursiveResolveTasks(object, { concurrency: 2 }), + ).rejects.toThrow('task 0 failed'); + + await delay(50); + + expect(started).toBe(2); +}); + +test('concurrency must be at least 1', async () => { + await expect(recursiveResolveTasks({}, { concurrency: 0 })).rejects.toThrow( + 'concurrency must be at least 1', + ); +}); diff --git a/src/utils/index.ts b/src/utils/index.ts index 58b8388cb..d7b069db1 100644 --- a/src/utils/index.ts +++ b/src/utils/index.ts @@ -9,6 +9,7 @@ export * from './getRescaler.ts'; export * from './isPowerOfTwo.ts'; export * from './nextPowerOfTwo.ts'; export * from './recursiveResolve.ts'; +export * from './recursiveResolveTasks.ts'; export * from './recursiveUntypeArrays.ts'; export * from './stringify.ts'; export * from './calculateAdaptiveWeights.ts'; diff --git a/src/utils/recursiveResolve.ts b/src/utils/recursiveResolve.ts index fe3faad72..154cfc020 100644 --- a/src/utils/recursiveResolve.ts +++ b/src/utils/recursiveResolve.ts @@ -15,12 +15,14 @@ export async function recursiveResolve(object: unknown) { function appendPromises(object: any, promises: Array>) { if (typeof object !== 'object') return object; for (const key in object) { - if (typeof object[key].then === 'function') { + const value = object[key]; + if (value === null || value === undefined) continue; + if (typeof value.then === 'function') { promises.push( - object[key].then((value: unknown) => (object[key] = value)), + value.then((resolved: unknown) => (object[key] = resolved)), ); - } else if (typeof object[key] === 'object') { - appendPromises(object[key], promises); + } else if (typeof value === 'object') { + appendPromises(value, promises); } } return object; diff --git a/src/utils/recursiveResolveTasks.ts b/src/utils/recursiveResolveTasks.ts new file mode 100644 index 000000000..caa86d3f5 --- /dev/null +++ b/src/utils/recursiveResolveTasks.ts @@ -0,0 +1,71 @@ +export interface RecursiveResolveTasksOptions { + /** + * Maximum number of tasks running at the same time. + * @default Number.MAX_SAFE_INTEGER + */ + concurrency?: number; +} + +/** + * Runs all the functions in an object recursively, at most `concurrency` of them + * at a time. Each function is replaced by the value it returns, awaited if it is + * a promise. Because the tasks are only called here, the concurrency applies to + * the work itself, unlike `recursiveResolve`, which receives promises that are + * already running. + * The changes are done in-place ! If a task fails, the returned promise rejects + * with that error and no further task is started. + * @param object - object whose functions should be run. + * @param options - options. + * @returns the resolved object. + */ +export async function recursiveResolveTasks( + object: unknown, + options: RecursiveResolveTasksOptions = {}, +) { + const { concurrency = Number.MAX_SAFE_INTEGER } = options; + if (concurrency < 1) { + throw new RangeError('concurrency must be at least 1'); + } + if (typeof object !== 'object' || object === null) return object; + const tasks: Array<() => Promise> = []; + appendTasks(object, tasks); + + const workerCount = Math.min(concurrency, tasks.length); + const cursor = { next: 0, failed: false }; + const workers = new Array>(workerCount); + for (let i = 0; i < workerCount; i++) { + workers[i] = runWorker(tasks, cursor); + } + await Promise.all(workers); + return object; +} + +function appendTasks(object: any, tasks: Array<() => Promise>) { + for (const key in object) { + const value = object[key]; + if (typeof value === 'function') { + tasks.push(async () => { + object[key] = await value(); + }); + } else if (typeof value === 'object' && value !== null) { + appendTasks(value, tasks); + } + } +} + +async function runWorker( + tasks: Array<() => Promise>, + cursor: { next: number; failed: boolean }, +) { + while (!cursor.failed && cursor.next < tasks.length) { + const task = tasks[cursor.next++]; + if (task === undefined) return; + try { + // eslint-disable-next-line no-await-in-loop -- sequential on purpose: this is what bounds the concurrency + await task(); + } catch (error) { + cursor.failed = true; + throw error; + } + } +}