add async pool

Author: qieguo2016

js/stack_queue.js

@@ -38,37 +38,52 @@ function pairTest(str) {
 
 console.log(pairTest('sdf\<asdsdfeesf{sdfefi{esadf{aefw}sdfw}sd}\>'));
 
-function waterfall(tasks, callback) {
-
-	// var err = null;
-	// var res = null;
-	// var current;
-	// while (fnArray.length !== 0) {
-	// 	current = fnArray.shift();
-	// 	current.apply(null, args);
-	// }
-
-	if (!tasks.length) return callback();
-	var taskIndex = 0;
+/**
+ * 异步操作池
+ * @param {Array<Promise>} tasks
+ * @param {Number} limit 最大并发数，默认为1表示串行
+ * 两种错误处理：1、其中一个任务出错就停止所有；2、执行完所有任务后收集所有错误
+ */
+function asyncPoolByThunk(tasks, limit = 1, cb = () => {}, options = {}) {
+	const stopImmediate = options.stopImmediate;
+	let index = limit - 1;
+	let errs = [];
 
-	function nextTask(args) {
-		if (taskIndex === tasks.length) {
-			return callback.apply(null, [null].concat(args));
+	function next(err) {
+		if (err) {
+			errs.push(err);
+			if (stopImmediate) {
+				cb(errs);
+				return;
+			}
 		}
-
-		var taskCallback = onlyOnce(rest(function(err, args) {
-			if (err) {
-				return callback.apply(null, [err].concat(args));
+		if (index > tasks.length) {
+			cb(errs);
+			return;
+		}
+		const current = tasks[++index];
+		if (typeof current === 'function') {
+			if (stopImmediate && errs.length > 0) {
+				return;
 			}
-			nextTask(args);
-		}));
-
-		args.push(taskCallback);
+			current(next);
+		}
+	}
 
-		var task = tasks[taskIndex++];
-		task.apply(null, args);
+	for (let k = 0; k < limit; k++) {
+		tasks[k](next);
 	}
 
-	nextTask([]);
+}
 
-}
+function asyncFn(params) {
+	return function (next) {
+		setTimeout(() => {
+			console.log('async', params);
+			next(params  === '4' ? '5555' : null);
+		}, 1000);
+	}
+}
+asyncPoolByThunk('123456789'.split('').map(el => asyncFn(el)), 3, function (errs) {
+	console.log('errs', errs);
+}, { stopImmediate: true });