feat(concurrent): support max concurrent installs

Support max concurrent installs in the programmatic API

Author: nicojs

src/LocalInstaller.ts

@@ -1,5 +1,6 @@
 import { EventEmitter } from 'events';
 import { promises as fs } from 'fs';
+import os from 'os';
 import path from 'path';
 import { helpers } from './helpers.ts';
 import type { InstallTarget, PackageJson } from './index.ts';
@@ -17,7 +18,8 @@ export interface Env {
 
 export interface Options {
   npmEnv?: Env;
-  packageManager: PackageManager | undefined;
+  packageManager?: PackageManager;
+  concurrent?: number;
 }
 
 export interface ListByPackage {
@@ -82,7 +84,13 @@ export class LocalInstaller extends EventEmitter {
 
   private async installAll(installTargets: InstallTarget[]): Promise<void> {
     this.emit('install_start', this.sourcesByTarget);
-    await Promise.all(installTargets.map((target) => this.installOne(target)));
+    // Install targets with max concurrent operations
+    const batchSize = this.options.concurrent ?? os.cpus().length;
+    for (let i = 0; i < installTargets.length; i += batchSize) {
+      const batch = installTargets.slice(i, i + batchSize);
+      await Promise.all(batch.map((target) => this.installOne(target)));
+    }
+
     this.emit('install_end');
   }
 

test/unit/LocalInstaller.spec.ts

@@ -1,5 +1,5 @@
 import { expect } from 'chai';
-import type { ResultPromise } from 'execa';
+import type { Options, Result, ResultPromise } from 'execa';
 import { promises as fs } from 'fs';
 import os from 'os';
 import { resolve } from 'path';
@@ -267,6 +267,44 @@ describe('LocalInstaller install', () => {
     });
   });
 
+  describe('with concurrent', () => {
+    beforeEach(() => {
+      sut = new LocalInstaller(
+        { '/a': ['c'], '/b': ['c'] },
+        { packageManager: 'npm', concurrent: 1 },
+      );
+      stubPackageJson({
+        '/a': 'a',
+        '/b': 'b',
+        c: 'c',
+      });
+      helper.rmStub.resolves();
+    });
+
+    it('should install with the specified concurrency', async () => {
+      const calls: ((res: Result<Options>) => void)[] = [];
+      helper.execStub.callsFake((file, args) => {
+        if(file === 'npm' && args?.includes('pack')) {
+          return Promise.resolve(createExecaResult()) as ResultPromise;
+        }
+        return new Promise(res => {
+          calls.push(res);
+        }) as ResultPromise;
+      });
+
+      const onGoingInstall = sut.install();
+      await tick();
+      await tick();
+      expect(calls).to.have.lengthOf(1);
+      calls[0](createExecaResult());
+      await tick();
+      await tick();
+      expect(calls).to.have.lengthOf(2);
+      calls[1](createExecaResult());
+      await onGoingInstall
+    });
+  });
+
   describe('when readFile errors', () => {
     it('should propagate the error', () => {
       helper.readFileStub.rejects(new Error('file error'));
@@ -320,3 +358,7 @@ describe('LocalInstaller install', () => {
     });
   };
 });
+
+function tick(): Promise<void> {
+  return new Promise((res) => process.nextTick(res));
+}