gulp-concurrent.js

// Drop-in function to parallelize a stage of a Gulp pipeline.
//
// NB: order is not preserved.
//
// Before:
//
//     gulp.src('./sass/**/*.scss')
//       .pipe(sass())                // Processes files one-by-one.
//       .pipe(gulp.dest('./css'))
//
// After:
//
//     gulp.src('./sass/**/*.scss')
//       .pipe(conc(() => sass(), 8)) // Processes up to 8 SCSS files at a time.
//       .pipe(gulp.dest('./css'))

function conc(streamFactory, maxStreams) {
  maxStreams = maxStreams || 4

  var self

  var lastCallback
  var flushCallback

  var streams = []
  var availableStreams = []
  var currentStream

  function createStream() {
    var stream = streamFactory()
      .on('drain', function () {
        availableStreams.push(this)
        if (lastCallback) {
          var cb = lastCallback
          lastCallback = null
          cb()
        }
      })
      .on('data', function (obj) {
        self.push(obj)
      })
      .on('error', function () {
        self.emit('error')
      })
      .on('end', function () {
        streams.pop()
        if (streams.length === 0) {
          flushCallback()
        }
      })
    streams.push(stream)
    return stream
  }

  function write(obj) {
    if (!currentStream) {
      currentStream = availableStreams.pop() || createStream()
    }

    if (currentStream.write(obj)) {
      return true
    }

    currentStream = null

    if (streams.length < maxStreams) {
      return true
    }

    return false
  }

  self = through(function (obj, enc, cb) {
    if (write(obj)) {
      cb()
    } else {
      lastCallback = cb
    }
  }, function flush(cb) {
    streams.forEach(function (s) {
      s.end()
    })
    flushCallback = cb
  })

  return self
}