lib/jobqueue.js
function () {
    var self = this;

    // See how much capacity we have left to fill.
    var launchCount = this.concurrency - Object.keys(this.active).length;

    // Identify up to launchCount eligible tasks, giving priority to those earlier in the queue.
    var offset = 0;
    var launchTasks = [];
    while (launchTasks.length < launchCount && this.tasks.length > offset) {
      var task = this.tasks[offset];

      if (task.key in this.active) {
        // This task cannot run right now, so skip it.
        offset += 1;
      } else {
        // This task is eligible to run. Remove it from the queue and prepare it to launch.
        this.tasks.splice(offset, 1);
        launchTasks.push(task);
      }
    }

    // Create a task completion callback. Remove the task from the active set, invoke the tasks'
    // push() callback, then drain() again to see if another task is ready to run.
    function makeTaskHandler(task) {
      return function (err) {
        delete self.active[task.key];

        task.callback(err);

        // Defer the next drain() call again in case the task's callback was synchronous.
        process.nextTick(self.drain.bind(self));
      };
    }

    // Launch the queue handler for each chosen task.
    for (var i = 0; i < launchTasks.length; i++) {
      var each = launchTasks[i];

      this.active[each.key] = each;

      this.handler(each.job, each.config, makeTaskHandler(each));
    }

    // Fire and unset the drain callback if one has been registered.
    if (this.drainCallback) {
      var lastCallback = this.drainCallback;
      this.drainCallback = null;
      lastCallback();
    }
  }
function () {
    var self = this;

    // See how much capacity we have left to fill.
    var launchCount = this.concurrency - Object.keys(this.active).length;

    // Identify up to launchCount eligible tasks, giving priority to those earlier in the queue.
    var offset = 0;
    var launchTasks = [];
    while (launchTasks.length < launchCount && this.tasks.length > offset) {
      var task = this.tasks[offset];

      if (task.key in this.active) {
        // This task cannot run right now, so skip it.
        offset += 1;
      } else {
        // This task is eligible to run. Remove it from the queue and prepare it to launch.
        this.tasks.splice(offset, 1);
        launchTasks.push(task);
      }
    }

    // Create a task completion callback. Remove the task from the active set, invoke the tasks'
    // push() callback, then drain() again to see if another task is ready to run.
    function makeTaskHandler(task) {
      return function (err) {
        delete self.active[task.key];

        task.callback(err);

        // Defer the next drain() call again in case the task's callback was synchronous.
        process.nextTick(self.drain.bind(self));
      };
    }

    // Launch the queue handler for each chosen task.
    for (var i = 0; i < launchTasks.length; i++) {
      var each = launchTasks[i];

      this.active[each.key] = each;

      this.handler(each.job, each.config, makeTaskHandler(each));
    }

    // Fire and unset the drain callback if one has been registered.
    if (this.drainCallback) {
      var lastCallback = this.drainCallback;
      this.drainCallback = null;
      lastCallback();
    }
  }
function ( ) { var self = this ; // See how much capacity we have left to fill. var launchCount = this . concurrency - Object . keys ( this . active ) . length ; // Identify up to launchCount eligible tasks, giving priority to those earlier in the queue. var offset = 0 ; var launchTasks = [ ] ; while ( launchTasks . length < launchCount && this . tasks . length > offset ) { var task = this . tasks [ offset ] ; if ( task . key in this . active ) { // This task cannot run right now, so skip it. offset += 1 ; } else { // This task is eligible to run. Remove it from the queue and prepare it to launch. this . tasks . splice ( offset , 1 ) ; launchTasks . push ( task ) ; } } // Create a task completion callback. Remove the task from the active set, invoke the tasks' // push() callback, then drain() again to see if another task is ready to run. function makeTaskHandler ( task ) { return function ( err ) { delete self . active [ task . key ] ; task . callback ( err ) ; // Defer the next drain() call again in case the task's callback was synchronous. process . nextTick ( self . drain . bind ( self ) ) ; } ; } // Launch the queue handler for each chosen task. for ( var i = 0 ; i < launchTasks . length ; i ++ ) { var each = launchTasks [ i ] ; this . active [ each . key ] = each ; this . handler ( each . job , each . config , makeTaskHandler ( each ) ) ; } // Fire and unset the drain callback if one has been registered. if ( this . drainCallback ) { var lastCallback = this . drainCallback ; this . drainCallback = null ; lastCallback ( ) ; } }
Launch the asynchronous handler function for each eligible waiting task until the queue is saturated.
Launch the asynchronous handler function for each eligible waiting task until the queue is saturated .