packages/shared/concurrency.ts


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

export class AsyncSemaphore {
  private permits: number;
  private queue: (() => void)[] = [];

  constructor(permits: number) {
    this.permits = permits;
  }

  acquire(): Promise<void> {
    if (this.permits > 0) {
      this.permits--;
      return Promise.resolve();
    } else {
      return new Promise<void>((resolve) => {
        this.queue.push(resolve);
      });
    }
  }

  release(): void {
    if (this.queue.length > 0) {
      const resolve = this.queue.shift();
      if (resolve) {
        resolve();
      }
    } else {
      this.permits++;
    }
  }

  get available(): number {
    return this.permits;
  }
}

export function limitConcurrency<T>(
  promises: (() => Promise<T>)[],
  concurrencyLimit: number,
): Promise<T>[] {
  const semaphore = new AsyncSemaphore(concurrencyLimit);
  const results: Promise<T>[] = [];

  for (const promiseFunction of promises) {
    results.push(
      semaphore
        .acquire()
        .then(() => {
          return promiseFunction();
        })
        .finally(() => {
          semaphore.release();
        }),
    );
  }
  return results;
}