| 851 | }; |
| 852 | |
| 853 | function _queue(worker, concurrency, payload) { |
| 854 | if (concurrency == null) { |
| 855 | concurrency = 1; |
| 856 | } |
| 857 | else if(concurrency === 0) { |
| 858 | throw new Error('Concurrency must not be zero'); |
| 859 | } |
| 860 | function _insert(q, data, pos, callback) { |
| 861 | if (callback != null && typeof callback !== "function") { |
| 862 | throw new Error("task callback must be a function"); |
| 863 | } |
| 864 | q.started = true; |
| 865 | if (!_isArray(data)) { |
| 866 | data = [data]; |
| 867 | } |
| 868 | if(data.length === 0 && q.idle()) { |
| 869 | // call drain immediately if there are no tasks |
| 870 | return async.setImmediate(function() { |
| 871 | q.drain(); |
| 872 | }); |
| 873 | } |
| 874 | _arrayEach(data, function(task) { |
| 875 | var item = { |
| 876 | data: task, |
| 877 | callback: callback || noop |
| 878 | }; |
| 879 | |
| 880 | if (pos) { |
| 881 | q.tasks.unshift(item); |
| 882 | } else { |
| 883 | q.tasks.push(item); |
| 884 | } |
| 885 | |
| 886 | if (q.tasks.length === q.concurrency) { |
| 887 | q.saturated(); |
| 888 | } |
| 889 | }); |
| 890 | async.setImmediate(q.process); |
| 891 | } |
| 892 | function _next(q, tasks) { |
| 893 | return function(){ |
| 894 | workers -= 1; |
| 895 | |
| 896 | var removed = false; |
| 897 | var args = arguments; |
| 898 | _arrayEach(tasks, function (task) { |
| 899 | _arrayEach(workersList, function (worker, index) { |
| 900 | if (worker === task && !removed) { |
| 901 | workersList.splice(index, 1); |
| 902 | removed = true; |
| 903 | } |
| 904 | }); |
| 905 | |
| 906 | task.callback.apply(task, args); |
| 907 | }); |
| 908 | if (q.tasks.length + workers === 0) { |
| 909 | q.drain(); |
| 910 | } |