fix: make promise throttle level consistent

src/throttledPromiseAll.ts

@@ -18,6 +18,15 @@ export type PromiseItem<T, O = T | undefined> = {
   producer: (source: T, throttledPromise: ThrottledPromiseAll<T, O | undefined>) => Promise<O | undefined>;
 };
 
+type IndexedProducer<T, O = T> = PromiseItem<T, O> & {
+  index: number;
+};
+
+type IndexedResult<O> = {
+  index: number;
+  result: O | undefined;
+};
+
 /**
  * A promise that throttles the number of promises running at a time.
  *
@@ -39,7 +48,7 @@ export class ThrottledPromiseAll<T, O = T> {
   private readonly concurrency: number;
   private wait: Duration;
   private timeout: NodeJS.Timeout | undefined;
-  readonly #results: Array<O | undefined> = [];
+  readonly #results: Array<IndexedResult<O> | undefined> = [];
 
   /**
    * Construct a new ThrottledPromiseAll.
@@ -56,7 +65,7 @@ export class ThrottledPromiseAll<T, O = T> {
    * Returns the results of the promises that have been resolved.
    */
   public get results(): Array<O | undefined> {
-    return this.#results;
+    return this.#results.sort((a, b) => (a?.index ?? 0) - (b?.index ?? 0)).map((r) => r?.result);
   }
 
   /**
@@ -109,7 +118,7 @@ export class ThrottledPromiseAll<T, O = T> {
         await this.dequeue();
       }
       this.stop();
-      return this.#results;
+      return this.results;
     } catch (e) {
       this.stop();
       throw e;
@@ -124,14 +133,35 @@ export class ThrottledPromiseAll<T, O = T> {
   }
 
   private async dequeue(): Promise<void> {
-    while (this.queue.length > 0) {
-      const next = this.queue.slice(0, this.concurrency);
-      this.queue.splice(0, this.concurrency);
-      // eslint-disable-next-line no-await-in-loop
-      const results = await Promise.all(
-        next.map((item) => item.producer(item.source, this).catch((e) => Promise.reject(e)))
+    const generator = function* (
+      data: Array<PromiseItem<T, O | undefined>>
+    ): Generator<PromiseItem<T, O | undefined> | undefined> {
+      while (data.length > 0) {
+        yield data.shift();
+      }
+    };
+    const concurrencyPool: Array<Promise<IndexedResult<O> | undefined>> = [];
+    const get = generator(this.queue);
+    let index = 0;
+    while (this.queue.length > 0 || concurrencyPool.length > 0) {
+      while (concurrencyPool.length < this.concurrency) {
+        const item = get.next().value as PromiseItem<T, O | undefined>;
+        if (!item) {
+          break;
+        }
+
+        const p: IndexedProducer<T, O> = { ...item, index: index++ };
+        concurrencyPool.push(
+          p
+            .producer(item.source, this)
+            .then((result) => ({ index: p.index, result }))
+            .catch((e) => Promise.reject(e))
+        );
+      }
+      this.#results.push(
+        // eslint-disable-next-line no-await-in-loop
+        await Promise.race([concurrencyPool.shift(), ...concurrencyPool])
       );
-      this.#results.push(...results);
     }
   }
 }

test/promiseQueue.test.ts

@@ -5,8 +5,8 @@
  * For full license text, see LICENSE.txt file in the repo root or https://opensource.org/licenses/BSD-3-Clause
  */
 import { expect } from 'chai';
-import { ThrottledPromiseAll } from '../src/throttledPromiseAll';
-import { Duration } from '../src/duration';
+import { ThrottledPromiseAll } from '../src';
+import { Duration } from '../src';
 
 describe('throttledPromiseAll', () => {
   const numberProducer = (
@@ -19,7 +19,10 @@ describe('throttledPromiseAll', () => {
     const throttledPromiseAll: ThrottledPromiseAll<number, number> = new ThrottledPromiseAll({ concurrency: 1 });
     for (const i of [1, 2, 3, 4, 5]) {
       // eslint-disable-next-line no-await-in-loop
-      throttledPromiseAll.add(i, numberProducer);
+      throttledPromiseAll.add(
+        i,
+        (source) => new Promise((resolve) => setTimeout(() => resolve(source + 1), (5 - i) * 100))
+      );
     }
     await throttledPromiseAll.all();
     const results = throttledPromiseAll.results as number[];
@@ -65,7 +68,7 @@ describe('throttledPromiseAll', () => {
       });
       throttledPromiseAll.add(
         [1, 2, 3, 4, 5],
-        (source) => new Promise((resolve) => setTimeout(() => resolve(source + 1), 10000))
+        (source) => new Promise((resolve) => setTimeout(() => resolve(source + 1), 200))
       );
       await throttledPromiseAll.all();
     } catch (e) {