From 4e282cc95f5092fa60db4ad6135ec9cf8d9a6268 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jo=C3=A3o=20Silva?= <threequartersjohn@gmail.com>
Date: Sat, 3 Aug 2019 11:51:53 +0100
Subject: [PATCH] Add `concurrency` option (#69)

---
 index.d.ts      |  7 +++++++
 index.js        | 29 ++++++++++++++++++-----------
 index.test-d.ts |  4 ++++
 package.json    |  3 ++-
 readme.md       |  7 +++++++
 test.js         |  5 +++++
 6 files changed, 43 insertions(+), 12 deletions(-)

diff --git a/index.d.ts b/index.d.ts
index e010a37..f4e8c88 100644
--- a/index.d.ts
+++ b/index.d.ts
@@ -32,6 +32,13 @@ declare namespace cpy {
 		```
 		*/
 		readonly rename?: string | ((basename: string) => string);
+
+		/**
+		Number of files being copied concurrently.
+
+		@default (os.cpus().length || 1) * 2
+		*/
+		readonly concurrency?: number;
 	}
 
 	interface ProgressData {
diff --git a/index.js b/index.js
index 39fabae..15ce53c 100644
--- a/index.js
+++ b/index.js
@@ -1,6 +1,8 @@
 'use strict';
 const EventEmitter = require('events');
 const path = require('path');
+const os = require('os');
+const pAll = require('p-all');
 const arrify = require('arrify');
 const globby = require('globby');
 const cpFile = require('cp-file');
@@ -29,7 +31,10 @@ const preprocessDestinationPath = (source, destination, options) => {
 	return path.join(destination, basename);
 };
 
-module.exports = (source, destination, options = {}) => {
+module.exports = (source, destination, {
+	concurrency = (os.cpus().lengh || 1) * 2,
+	...options
+} = {}) => {
 	const progressEmitter = new EventEmitter();
 
 	const promise = (async () => {
@@ -84,18 +89,20 @@ module.exports = (source, destination, options = {}) => {
 			}
 		};
 
-		return Promise.all(files.map(async sourcePath => {
-			const from = preprocessSourcePath(sourcePath, options);
-			const to = preprocessDestinationPath(sourcePath, destination, options);
+		return pAll(files.map(sourcePath => {
+			return async () => {
+				const from = preprocessSourcePath(sourcePath, options);
+				const to = preprocessDestinationPath(sourcePath, destination, options);
 
-			try {
-				await cpFile(from, to, options).on('progress', fileProgressHandler);
-			} catch (error) {
-				throw new CpyError(`Cannot copy from \`${from}\` to \`${to}\`: ${error.message}`, error);
-			}
+				try {
+					await cpFile(from, to, options).on('progress', fileProgressHandler);
+				} catch (error) {
+					throw new CpyError(`Cannot copy from \`${from}\` to \`${to}\`: ${error.message}`, error);
+				}
 
-			return to;
-		}));
+				return to;
+			};
+		}), {concurrency});
 	})();
 
 	promise.on = (...arguments_) => {
diff --git a/index.test-d.ts b/index.test-d.ts
index c6a6500..39fec27 100644
--- a/index.test-d.ts
+++ b/index.test-d.ts
@@ -23,6 +23,10 @@ expectType<Promise<string[]> & ProgressEmitter>(
 expectType<Promise<string[]> & ProgressEmitter>(
 	cpy('foo.js', 'destination', {overwrite: false})
 );
+expectType<Promise<string[]> & ProgressEmitter>(
+	cpy('foo.js', 'destination', {concurrency: 2})
+);
+
 
 expectType<Promise<string[]>>(
 	cpy('foo.js', 'destination').on('progress', progress => {
diff --git a/package.json b/package.json
index 946ffb7..482fe6a 100644
--- a/package.json
+++ b/package.json
@@ -45,7 +45,8 @@
 		"arrify": "^2.0.1",
 		"cp-file": "^7.0.0",
 		"globby": "^9.2.0",
-		"nested-error-stacks": "^2.1.0"
+		"nested-error-stacks": "^2.1.0",
+		"p-all": "^2.1.0"
 	},
 	"devDependencies": {
 		"ava": "^2.1.0",
diff --git a/readme.md b/readme.md
index c06a349..9d15f63 100644
--- a/readme.md
+++ b/readme.md
@@ -94,6 +94,13 @@ const cpy = require('cpy');
 })();
 ```
 
+##### concurrency
+
+Type: `number`<br>
+Default: `(os.cpus().length || 1) * 2`
+
+Number of files being copied concurrently.
+
 
 ## Progress reporting
 
diff --git a/test.js b/test.js
index b26d710..dd67d15 100644
--- a/test.js
+++ b/test.js
@@ -49,6 +49,11 @@ test('copy array of files', async t => {
 	t.is(read('package.json'), read(t.context.tmp, 'package.json'));
 });
 
+test('throws on invalid concurrency value', async t => {
+	await t.throwsAsync(cpy(['license', 'package.json'], t.context.tmp, {concurrency: -2}));
+	await t.throwsAsync(cpy(['license', 'package.json'], t.context.tmp, {concurrency: 'foo'}));
+});
+
 test('cwd', async t => {
 	fs.mkdirSync(t.context.tmp);
 	fs.mkdirSync(path.join(t.context.tmp, 'cwd'));