queue.js 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204
  1. 'use strict';
  2. Object.defineProperty(exports, "__esModule", {
  3. value: true
  4. });
  5. exports.default = queue;
  6. var _baseIndexOf = require('lodash/_baseIndexOf');
  7. var _baseIndexOf2 = _interopRequireDefault(_baseIndexOf);
  8. var _isArray = require('lodash/isArray');
  9. var _isArray2 = _interopRequireDefault(_isArray);
  10. var _noop = require('lodash/noop');
  11. var _noop2 = _interopRequireDefault(_noop);
  12. var _onlyOnce = require('./onlyOnce');
  13. var _onlyOnce2 = _interopRequireDefault(_onlyOnce);
  14. var _setImmediate = require('./setImmediate');
  15. var _setImmediate2 = _interopRequireDefault(_setImmediate);
  16. var _DoublyLinkedList = require('./DoublyLinkedList');
  17. var _DoublyLinkedList2 = _interopRequireDefault(_DoublyLinkedList);
  18. var _wrapAsync = require('./wrapAsync');
  19. var _wrapAsync2 = _interopRequireDefault(_wrapAsync);
  20. function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }
  21. function queue(worker, concurrency, payload) {
  22. if (concurrency == null) {
  23. concurrency = 1;
  24. } else if (concurrency === 0) {
  25. throw new Error('Concurrency must not be zero');
  26. }
  27. var _worker = (0, _wrapAsync2.default)(worker);
  28. var numRunning = 0;
  29. var workersList = [];
  30. var processingScheduled = false;
  31. function _insert(data, insertAtFront, callback) {
  32. if (callback != null && typeof callback !== 'function') {
  33. throw new Error('task callback must be a function');
  34. }
  35. q.started = true;
  36. if (!(0, _isArray2.default)(data)) {
  37. data = [data];
  38. }
  39. if (data.length === 0 && q.idle()) {
  40. // call drain immediately if there are no tasks
  41. return (0, _setImmediate2.default)(function () {
  42. q.drain();
  43. });
  44. }
  45. for (var i = 0, l = data.length; i < l; i++) {
  46. var item = {
  47. data: data[i],
  48. callback: callback || _noop2.default
  49. };
  50. if (insertAtFront) {
  51. q._tasks.unshift(item);
  52. } else {
  53. q._tasks.push(item);
  54. }
  55. }
  56. if (!processingScheduled) {
  57. processingScheduled = true;
  58. (0, _setImmediate2.default)(function () {
  59. processingScheduled = false;
  60. q.process();
  61. });
  62. }
  63. }
  64. function _next(tasks) {
  65. return function (err) {
  66. numRunning -= 1;
  67. for (var i = 0, l = tasks.length; i < l; i++) {
  68. var task = tasks[i];
  69. var index = (0, _baseIndexOf2.default)(workersList, task, 0);
  70. if (index === 0) {
  71. workersList.shift();
  72. } else if (index > 0) {
  73. workersList.splice(index, 1);
  74. }
  75. task.callback.apply(task, arguments);
  76. if (err != null) {
  77. q.error(err, task.data);
  78. }
  79. }
  80. if (numRunning <= q.concurrency - q.buffer) {
  81. q.unsaturated();
  82. }
  83. if (q.idle()) {
  84. q.drain();
  85. }
  86. q.process();
  87. };
  88. }
  89. var isProcessing = false;
  90. var q = {
  91. _tasks: new _DoublyLinkedList2.default(),
  92. concurrency: concurrency,
  93. payload: payload,
  94. saturated: _noop2.default,
  95. unsaturated: _noop2.default,
  96. buffer: concurrency / 4,
  97. empty: _noop2.default,
  98. drain: _noop2.default,
  99. error: _noop2.default,
  100. started: false,
  101. paused: false,
  102. push: function (data, callback) {
  103. _insert(data, false, callback);
  104. },
  105. kill: function () {
  106. q.drain = _noop2.default;
  107. q._tasks.empty();
  108. },
  109. unshift: function (data, callback) {
  110. _insert(data, true, callback);
  111. },
  112. remove: function (testFn) {
  113. q._tasks.remove(testFn);
  114. },
  115. process: function () {
  116. // Avoid trying to start too many processing operations. This can occur
  117. // when callbacks resolve synchronously (#1267).
  118. if (isProcessing) {
  119. return;
  120. }
  121. isProcessing = true;
  122. while (!q.paused && numRunning < q.concurrency && q._tasks.length) {
  123. var tasks = [],
  124. data = [];
  125. var l = q._tasks.length;
  126. if (q.payload) l = Math.min(l, q.payload);
  127. for (var i = 0; i < l; i++) {
  128. var node = q._tasks.shift();
  129. tasks.push(node);
  130. workersList.push(node);
  131. data.push(node.data);
  132. }
  133. numRunning += 1;
  134. if (q._tasks.length === 0) {
  135. q.empty();
  136. }
  137. if (numRunning === q.concurrency) {
  138. q.saturated();
  139. }
  140. var cb = (0, _onlyOnce2.default)(_next(tasks));
  141. _worker(data, cb);
  142. }
  143. isProcessing = false;
  144. },
  145. length: function () {
  146. return q._tasks.length;
  147. },
  148. running: function () {
  149. return numRunning;
  150. },
  151. workersList: function () {
  152. return workersList;
  153. },
  154. idle: function () {
  155. return q._tasks.length + numRunning === 0;
  156. },
  157. pause: function () {
  158. q.paused = true;
  159. },
  160. resume: function () {
  161. if (q.paused === false) {
  162. return;
  163. }
  164. q.paused = false;
  165. (0, _setImmediate2.default)(q.process);
  166. }
  167. };
  168. return q;
  169. }
  170. module.exports = exports['default'];