<commit_before><commit_msg>[queue] A basic queue resource, implementing configurable tick rate and concurrency<commit_after>var resource = require('resource'),
    queue = resource.define('queue');

queue.persist('memory');

queue.property('concurrency', {
  description: 'how many jobs to run at once',
  type: 'number',
  default: 1
});

queue.property('interval', {
  description: 'time interval between processing items (ms)',
  type: 'number',
  default: 5000
});

queue.property('wait', {
  description: 'wait until all running jobs are completed before executing next set',
  type: 'boolean',
  default: true
});

queue.property('queue', {
  type: 'array',
  default: []
});

//
// Basic push/shift methods for queue
//
// TODO: Implement as ring queue?
//
queue.method('push', push, {
  description: 'push an element onto the queue',
  properties: {
    queue: queue.schema,
    job: {
      properties: {
        name: {
          type: 'string'
        },
        with: {
          type: 'object',
          default: {}
        }
      }
    }
  }
});
function push(q, job) {
  return q.queue.push(job);
}

queue.method('shift', shift, {
  description: 'shift an element off the queue',
  properties: {
    options: {
      properties: queue.schema.properties
    }
  }
});
function shift(q) {
  return q.queue.shift();
}

//
// Exposing a global list of job methods
// TODO: Scope to instances
//
var jobs = {};
queue.method('jobs', addJobs, {
  description: 'expose worker methods',
  properties: {
    jobs: { type: 'object' }
  }
});
function addJobs(js) {
  jobs = js;
  return queue;
}

//
// Run a single job
//
queue.method('run', run, {
  description: 'run a job',
  type: 'object',
  properties: {
    job: {
      properties: {
        name: {
          type: 'string'
        },
        with: {
          type: 'object',
          default: {}
        }
      }
    },
    callback: {
      type: 'function'
    }
  }
});
function run(j, callback) {
  if (!jobs[j.name]) {
    return callback(new Error('could not execute job of type `' + j.method + '`'));
  }

  jobs[j.name](j.with, callback);
}

queue.method('process', process, {
  description: 'process elements off the queue',
  properties: {
    options: queue.schema,
    callback: {
      type: 'function'
    }
  }
});
function process(q, callback) {
  var async = require('async');

  //
  // TODO: Implement smarter queue push/shift methods
  //
  var js = [];
  while ((js.length < q.concurrency) && q.queue.length) {
    js.push(queue.shift(q));
  }

  //
  // Run the slice of jobs concurrently
  //
  async.forEach(js, queue.run, function (err) {
    callback(err, q);
  });
}

//
// TODO: `unload` method
//
queue.method('load', load, {
  description: 'start processing a queue',
  properties: {
    options: queue.schema
  }
});
function load(q) {
  _process();

  function _process() {
    var completed = false,
        timedOut = false;

    //
    // Process again at the end of the timeout *if* the last process step
    // completed
    //
    setTimeout(function () {
      //
      // If waiting is turned off, always process the next set of elements
      //
      if (completed || !q.wait) {
        _process();
      }
      else {
        timedOut = true;
      }
    }, q.interval);

    process(q, function (err, result) {
      if (err) {
        //
        // Remark: Emit error, keep going (for now, may change behavior)
        //
        // TODO: Attach some context to the error?
        //
        queue.emit('error', err);
      }

      completed = true;

      //
      // If the timeout occurred while the process step was still running,
      // execute it late
      //
      if (timedOut) {
        _process();
      }
    });
  }
}

exports.queue = queue;
exports.dependencies = {
  "async": "*"
};

{'fraction_non_alphanumeric': 0.15090385119203564, 'fraction_numerical': 0.0013099292638197536, 'mean_word_length': 2.2355932203389832, 'pattern_counts': {'":': 1, '<': 4, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '150620', 'n_tokens_mistral': 1208, 'n_tokens_neox': 1187, 'n_words': 336}