feat(utils): add process-queue

src/index.d.ts

@@ -10,3 +10,4 @@
  * governing permissions and limitations under the License.
  */
 export * from './wrap';
+export * from './process-queue';

src/index.js

@@ -22,6 +22,7 @@ const Async = require('./async.js');
 const MarkupConfig = require('./MarkupConfig');
 const Condition = require('./Condition.js');
 const wrap = require('./wrap.js');
+const processQueue = require('./process-queue.js');
 
 module.exports = {
   GitUrl,
@@ -37,4 +38,5 @@ module.exports = {
   async: Async,
   Condition,
   wrap,
+  processQueue,
 };

src/process-queue.d.ts

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 Adobe. All rights reserved.
+ * This file is licensed to you under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License. You may obtain a copy
+ * of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under
+ * the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR REPRESENTATIONS
+ * OF ANY KIND, either express or implied. See the License for the specific language
+ * governing permissions and limitations under the License.
+ */
+
+
+/**
+ * Queue entry.
+ */
+export declare interface QueueEntry {}
+
+/**
+ * A (asynchronous) handler function that is invoked for every queue entry.
+ * Values added to the `results` array will be returned by `processQueue` function.
+ * The handler can modify the `queue` if needed.
+ *
+ * @param {QueueEntry} entry The queue entry.
+ * @param {QueueEntry[]} queue the queue.
+ * @param {[]} results the process queue results
+ */
+export declare interface ProcessQueueHandler {
+  (entry: QueueEntry, queue:[QueueEntry], results:[any]): void;
+}
+
+/**
+ * Processes the given queue concurrently.
+ *
+ * @param {QueueEntry[]} queue A list of entries to be processed
+ * @param {ProcessQueueHandler} fn A handler function
+ * @param {number} [maxConcurrent = 8] Concurrency level
+ * @returns {[]} the results
+ */
+export declare function processQueue(queue:[QueueEntry], fn:ProcessQueueHandler, maxConcurrent?:number): [any];

src/process-queue.js

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2021 Adobe. All rights reserved.
+ * This file is licensed to you under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License. You may obtain a copy
+ * of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under
+ * the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR REPRESENTATIONS
+ * OF ANY KIND, either express or implied. See the License for the specific language
+ * governing permissions and limitations under the License.
+ */
+
+/**
+ * Processes the given queue concurrently. The handler functions can add more items to the queue
+ * if needed.
+ *
+ * @param {Array<*>} queue A list of tasks
+ * @param {function} fn A handler function `fn(task:any, queue:array, results:array)`
+ * @param {number} [maxConcurrent = 8] Concurrency level
+ * @returns the results
+ */
+async function processQueue(queue, fn, maxConcurrent = 8) {
+  const running = [];
+  const results = [];
+  while (queue.length || running.length) {
+    if (running.length < maxConcurrent && queue.length) {
+      const task = fn(queue.shift(), queue, results);
+      if (task.then) {
+        running.push(task);
+        task.catch(() => {}).finally(() => {
+          running.splice(running.indexOf(task), 1);
+        });
+      }
+    } else {
+      // eslint-disable-next-line no-await-in-loop
+      await Promise.race(running);
+    }
+  }
+  return results;
+}
+
+module.exports = processQueue;

test/process-queue.test.js

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2021 Adobe. All rights reserved.
+ * This file is licensed to you under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License. You may obtain a copy
+ * of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under
+ * the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR REPRESENTATIONS
+ * OF ANY KIND, either express or implied. See the License for the specific language
+ * governing permissions and limitations under the License.
+ */
+
+/* eslint-env mocha */
+const assert = require('assert');
+const processQueue = require('../src/process-queue.js');
+
+const nop = () => {};
+
+let concurrency = 0;
+let maxConcurrency = 0;
+
+async function testFunction({ time, number, error }, queue, results) {
+  concurrency += 1;
+  maxConcurrency = Math.max(maxConcurrency, concurrency);
+  await new Promise((resolve) => setTimeout(resolve, time));
+  concurrency -= 1;
+  if (error) {
+    throw error;
+  }
+  results.push(number * number);
+}
+
+describe('Process Queue', () => {
+  beforeEach(() => {
+    maxConcurrency = 0;
+  });
+
+  it('works with empty queue', async () => {
+    const result = await processQueue([], nop);
+    assert.deepEqual(result, []);
+  });
+
+  it('works with non async function', async () => {
+    const result = await processQueue([5], (n, queue, results) => results.push(n * n));
+    assert.deepEqual(result, [25]);
+  });
+
+  it('processes queue', async () => {
+    const result = await processQueue([{
+      time: 100,
+      number: 4,
+    }, {
+      time: 50,
+      number: 3,
+    }], testFunction);
+    assert.deepEqual(result, [9, 16]);
+  });
+
+  it('respects concurrency', async () => {
+    const tasks = [];
+    const expected = [];
+    for (let i = 0; i < 32; i += 1) {
+      tasks.push({
+        time: 5,
+        number: i,
+      });
+      expected.push(i * i);
+    }
+    const result = await processQueue(tasks, testFunction);
+    assert.deepEqual(result.sort((a, b) => a - b), expected);
+    assert.equal(maxConcurrency, 8);
+  });
+
+  it('aborts queue on early error', async () => {
+    await assert.rejects(processQueue([{
+      time: 100,
+      number: 4,
+    }, {
+      time: 50,
+      number: 3,
+      error: new Error('aborted'),
+    }], testFunction));
+  });
+});