priorityQueue.js 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103
  1. 'use strict';
  2. Object.defineProperty(exports, "__esModule", {
  3. value: true
  4. });
  5. exports.default = function (worker, concurrency) {
  6. // Start with a normal queue
  7. var q = (0, _queue2.default)(worker, concurrency);
  8. // Override push to accept second parameter representing priority
  9. q.push = function (data, priority, callback) {
  10. if (callback == null) callback = _noop2.default;
  11. if (typeof callback !== 'function') {
  12. throw new Error('task callback must be a function');
  13. }
  14. q.started = true;
  15. if (!(0, _isArray2.default)(data)) {
  16. data = [data];
  17. }
  18. if (data.length === 0) {
  19. // call drain immediately if there are no tasks
  20. return (0, _setImmediate2.default)(function () {
  21. q.drain();
  22. });
  23. }
  24. priority = priority || 0;
  25. var nextNode = q._tasks.head;
  26. while (nextNode && priority >= nextNode.priority) {
  27. nextNode = nextNode.next;
  28. }
  29. (0, _arrayEach2.default)(data, function (task) {
  30. var item = {
  31. data: task,
  32. priority: priority,
  33. callback: callback
  34. };
  35. if (nextNode) {
  36. q._tasks.insertBefore(nextNode, item);
  37. } else {
  38. q._tasks.push(item);
  39. }
  40. });
  41. (0, _setImmediate2.default)(q.process);
  42. };
  43. // Remove unshift function
  44. delete q.unshift;
  45. return q;
  46. };
  47. var _arrayEach = require('lodash/_arrayEach');
  48. var _arrayEach2 = _interopRequireDefault(_arrayEach);
  49. var _isArray = require('lodash/isArray');
  50. var _isArray2 = _interopRequireDefault(_isArray);
  51. var _noop = require('lodash/noop');
  52. var _noop2 = _interopRequireDefault(_noop);
  53. var _setImmediate = require('./setImmediate');
  54. var _setImmediate2 = _interopRequireDefault(_setImmediate);
  55. var _queue = require('./queue');
  56. var _queue2 = _interopRequireDefault(_queue);
  57. function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }
  58. module.exports = exports['default'];
  59. /**
  60. * The same as [async.queue]{@link module:ControlFlow.queue} only tasks are assigned a priority and
  61. * completed in ascending priority order.
  62. *
  63. * @name priorityQueue
  64. * @static
  65. * @memberOf module:ControlFlow
  66. * @method
  67. * @see [async.queue]{@link module:ControlFlow.queue}
  68. * @category Control Flow
  69. * @param {Function} worker - An asynchronous function for processing a queued
  70. * task, which must call its `callback(err)` argument when finished, with an
  71. * optional `error` as an argument. If you want to handle errors from an
  72. * individual task, pass a callback to `q.push()`. Invoked with
  73. * (task, callback).
  74. * @param {number} concurrency - An `integer` for determining how many `worker`
  75. * functions should be run in parallel. If omitted, the concurrency defaults to
  76. * `1`. If the concurrency is `0`, an error is thrown.
  77. * @returns {module:ControlFlow.QueueObject} A priorityQueue object to manage the tasks. There are two
  78. * differences between `queue` and `priorityQueue` objects:
  79. * * `push(task, priority, [callback])` - `priority` should be a number. If an
  80. * array of `tasks` is given, all tasks will be assigned the same priority.
  81. * * The `unshift` method was removed.
  82. */