You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

through2-concurrent.js 2.8KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101
  1. // Like through2 except execute in parallel with a set maximum
  2. // concurrency
  3. "use strict";
  4. var through2 = require('through2');
  5. function cbNoop (cb) {
  6. cb();
  7. }
  8. module.exports = function concurrentThrough (options, transform, flush) {
  9. var concurrent = 0, lastCallback = null, pendingFinish = null;
  10. if (typeof options === 'function') {
  11. flush = transform;
  12. transform = options;
  13. options = {};
  14. }
  15. var maxConcurrency = options.maxConcurrency || 16;
  16. function _transform (message, enc, callback) {
  17. var self = this;
  18. var callbackCalled = false;
  19. concurrent++;
  20. if (concurrent < maxConcurrency) {
  21. // Ask for more right away
  22. callback();
  23. } else {
  24. // We're at the concurrency limit, save the callback for
  25. // when we're ready for more
  26. lastCallback = callback;
  27. }
  28. transform.call(this, message, enc, function (err) {
  29. // Ignore multiple calls of the callback (shouldn't ever
  30. // happen, but just in case)
  31. if (callbackCalled) return;
  32. callbackCalled = true;
  33. if (err) {
  34. self.emit('error', err);
  35. } else if (arguments.length > 1) {
  36. self.push(arguments[1]);
  37. }
  38. concurrent--;
  39. if (lastCallback) {
  40. var cb = lastCallback;
  41. lastCallback = null;
  42. cb();
  43. }
  44. if (concurrent === 0 && pendingFinish) {
  45. pendingFinish();
  46. pendingFinish = null;
  47. }
  48. });
  49. }
  50. // We need to pass in final to through2 even if the caller has
  51. // not given us a final option so that it will wait for all
  52. // transform callbacks to complete before emitting a "finish"
  53. // and "end" event.
  54. if (typeof options.final !== 'function') {
  55. options.final = cbNoop;
  56. }
  57. // We also wrap flush to make sure anyone using an ancient version
  58. // of through2 without support for final will get the old behaviour.
  59. // TODO: don't wrap flush after upgrading through2 to a version with guaranteed `_final`
  60. if (typeof flush !== 'function') {
  61. flush = cbNoop;
  62. }
  63. // Flush is always called only after Final has finished
  64. // to ensure that data from Final gets processed, so we only need one pending callback at a time
  65. function callOnFinish (original) {
  66. return function (callback) {
  67. if (concurrent === 0) {
  68. original.call(this, callback);
  69. } else {
  70. pendingFinish = original.bind(this, callback);
  71. }
  72. }
  73. }
  74. options.final = callOnFinish(options.final);
  75. return through2(options, _transform, callOnFinish(flush));
  76. };
  77. module.exports.obj = function (options, transform, flush) {
  78. if (typeof options === 'function') {
  79. flush = transform;
  80. transform = options;
  81. options = {};
  82. }
  83. options.objectMode = true;
  84. if (options.highWaterMark == null) {
  85. options.highWaterMark = 16;
  86. }
  87. return module.exports(options, transform, flush);
  88. };