import isArray from 'lodash/isArray'; import noop from 'lodash/noop'; import setImmediate from './setImmediate'; import queue from './queue'; /** * The same as [async.queue]{@link module:ControlFlow.queue} only tasks are assigned a priority and * completed in ascending priority order. * * @name priorityQueue * @static * @memberOf module:ControlFlow * @method * @see [async.queue]{@link module:ControlFlow.queue} * @category Control Flow * @param {Function} worker - An asynchronous function for processing a queued * task, which must call its `callback(err)` argument when finished, with an * optional `error` as an argument. If you want to handle errors from an * individual task, pass a callback to `q.push()`. Invoked with * (task, callback). * @param {number} concurrency - An `integer` for determining how many `worker` * functions should be run in parallel. If omitted, the concurrency defaults to * `1`. If the concurrency is `0`, an error is thrown. * @returns {module:ControlFlow.QueueObject} A priorityQueue object to manage the tasks. There are two * differences between `queue` and `priorityQueue` objects: * * `push(task, priority, [callback])` - `priority` should be a number. If an * array of `tasks` is given, all tasks will be assigned the same priority. * * The `unshift` method was removed. */ export default function(worker, concurrency) { // Start with a normal queue var q = queue(worker, concurrency); // Override push to accept second parameter representing priority q.push = function(data, priority, callback) { if (callback == null) callback = noop; if (typeof callback !== 'function') { throw new Error('task callback must be a function'); } q.started = true; if (!isArray(data)) { data = [data]; } if (data.length === 0) { // call drain immediately if there are no tasks return setImmediate(function() { q.drain(); }); } priority = priority || 0; var nextNode = q._tasks.head; while (nextNode && priority >= nextNode.priority) { nextNode = nextNode.next; } for (var i = 0, l = data.length; i < l; i++) { var item = { data: data[i], priority: priority, callback: callback }; if (nextNode) { q._tasks.insertBefore(nextNode, item); } else { q._tasks.push(item); } } setImmediate(q.process); }; // Remove unshift function delete q.unshift; return q; }