41 lines
1.2 KiB
JavaScript
41 lines
1.2 KiB
JavaScript
'use strict';
|
|
|
|
var atoa = require('atoa');
|
|
var a = require('./a');
|
|
var once = require('./once');
|
|
var emitter = require('./emitter');
|
|
var debounce = require('./debounce');
|
|
|
|
module.exports = function queue (worker, concurrency) {
|
|
var q = [], load = 0, max = concurrency || 1, paused;
|
|
var qq = emitter({
|
|
push: manipulate.bind(null, 'push'),
|
|
unshift: manipulate.bind(null, 'unshift'),
|
|
pause: function pause () { paused = true; },
|
|
resume: function resume () { paused = false; debounce(labor); },
|
|
pending: q
|
|
});
|
|
if (Object.defineProperty && !Object.definePropertyPartial) {
|
|
Object.defineProperty(qq, 'length', { get: function getter () { return q.length; } });
|
|
}
|
|
function manipulate (how, task, done) {
|
|
var tasks = a(task) ? task : [task];
|
|
tasks.forEach(function insert (t) { q[how]({ t: t, done: done }); });
|
|
debounce(labor);
|
|
}
|
|
function labor () {
|
|
if (paused || load >= max) { return; }
|
|
if (!q.length) { if (load === 0) { qq.emit('drain'); } return; }
|
|
load++;
|
|
var job = q.pop();
|
|
worker(job.t, once(complete.bind(null, job)));
|
|
debounce(labor);
|
|
}
|
|
function complete (job) {
|
|
load--;
|
|
debounce(job.done, atoa(arguments, 1));
|
|
debounce(labor);
|
|
}
|
|
return qq;
|
|
};
|