<commit_before>'use strict';

/**
 * Module dependencies.
 */
var fs = require('fs');
var path = require('path');
var _ = require('underscore');
var bunyan = require('bunyan');
var mkdirp = require('mkdirp');
var domain = require('domain');
var DependencyGraph = require('dependency-foo');
var cronJob = require('cron').CronJob;
var EventEmitter2 = require('eventemitter2').EventEmitter2;
    
var queue = require('./queue');
var worker = require('./worker');


/**
 * Exports
 */
module.exports = ummon;

// Can decide on the api here
function ummon(options){
  return new Ummon(options);
}
ummon.create = function(options){
  return new Ummon(options);
};

// Expose the constructor for potential inheritance
ummon.Ummon = Ummon;


/**
 * Ummon constructor
 * 
 * @param {object} options configuration options
 */
function Ummon(options){
  var self = this;

  var d = domain.create();
  
  d.on('error', function(er) {
    console.error('UMMON CONSTRUCTOR DOMAIN ERROR', er.stack);
    process.exit(1);
  });

  d.run(function() {
    if (!options) { options = {}; }

    self.dispatcher = new EventEmitter2({
      wildcard: true, // should the event emitter use wildcards.
      newListener: false, // if you want to emit the newListener event set to true.
      maxListeners: 20, // the max number of listeners that can be assigned to an event, defaults to 10.
    });

    // Load config
    var defaults = require(path.join(__dirname, '../config/config.defaults.json')); // Load config.defaults.json
    var config = {};
    
    if (options.configPath) {
      var configPath = path.normalize(options.configPath); // Get the full path to the current themes _config.json
      config = (fs.existsSync(configPath)) ? require(configPath) : {};
    }

    self.config = _.extend(defaults, config, options);
    self.MAX_WORKERS = 5;
    
    // Cleanup paths and make any folders that need to be made
    self.config.log.path = path.resolve(self.config.log.path);
    mkdirp.sync(path.dirname(self.config.log.path));


    self.log = bunyan.createLogger({
      name: 'ummon',
      streams: [
        {
          level: 'debug',
          stream: process.stdout, // log INFO and above to stdout
        },
        {
          level: self.config.log.level,
          path: self.config.log.path  // log ERROR and above to a file
        }
      ],
      serializers: bunyan.stdSerializers
    });

    self.defaults = {};
    self.tasks = {};
    self.timers = {};
    self.workers = {};
    self.dependencies = new DependencyGraph();
    self.queue = queue(self);
  

    // Setup listeners
    self.dispatcher.on('queue.new', function(r){
      self.log.info({event: 'queue.new', runid:r.id, taskid: r.task.id, collection: r.task.collection}, "A task run (%s) was added to the queue", r.id);

      self.createWorkerIfReady();
    });

    self.dispatcher.on('worker.complete', function(completedRun){
      // Cleanup
      delete self.workers[completedRun.pid];

      // Queue dependent tasks
      var refs = self.dependencies.subject(completedRun.task.id).references;
      
      // If there is dependent tasks and the previous task was successfull, GO!
      if (refs.length && completedRun.exitCode === 0) {
        for (var index = 0; index < refs.length; index++) {
          self.log.debug('Task run for %s triggered new run for %s', completedRun.task.id, refs[index]);
          self.queue.push(self.getTask(refs[index]), completedRun.id);
        }
      }
    });
  });

  // Automatically load tasks from a config file
  if (self.config.tasks) {
    self.autoLoadTasks();
  }
}

<commit_msg>Not all tasks have ids. Account for that<commit_after>'use strict';

/**
 * Module dependencies.
 */
var fs = require('fs');
var path = require('path');
var _ = require('underscore');
var bunyan = require('bunyan');
var mkdirp = require('mkdirp');
var domain = require('domain');
var DependencyGraph = require('dependency-foo');
var cronJob = require('cron').CronJob;
var EventEmitter2 = require('eventemitter2').EventEmitter2;
    
var queue = require('./queue');
var worker = require('./worker');


/**
 * Exports
 */
module.exports = ummon;

// Can decide on the api here
function ummon(options){
  return new Ummon(options);
}
ummon.create = function(options){
  return new Ummon(options);
};

// Expose the constructor for potential inheritance
ummon.Ummon = Ummon;


/**
 * Ummon constructor
 * 
 * @param {object} options configuration options
 */
function Ummon(options){
  var self = this;

  var d = domain.create();
  
  d.on('error', function(er) {
    console.error('UMMON CONSTRUCTOR DOMAIN ERROR', er.stack);
    process.exit(1);
  });

  d.run(function() {
    if (!options) { options = {}; }

    self.dispatcher = new EventEmitter2({
      wildcard: true, // should the event emitter use wildcards.
      newListener: false, // if you want to emit the newListener event set to true.
      maxListeners: 20, // the max number of listeners that can be assigned to an event, defaults to 10.
    });

    // Load config
    var defaults = require(path.join(__dirname, '../config/config.defaults.json')); // Load config.defaults.json
    var config = {};
    
    if (options.configPath) {
      var configPath = path.normalize(options.configPath); // Get the full path to the current themes _config.json
      config = (fs.existsSync(configPath)) ? require(configPath) : {};
    }

    self.config = _.extend(defaults, config, options);
    self.MAX_WORKERS = 5;
    
    // Cleanup paths and make any folders that need to be made
    self.config.log.path = path.resolve(self.config.log.path);
    mkdirp.sync(path.dirname(self.config.log.path));


    self.log = bunyan.createLogger({
      name: 'ummon',
      streams: [
        {
          level: 'debug',
          stream: process.stdout, // log INFO and above to stdout
        },
        {
          level: self.config.log.level,
          path: self.config.log.path  // log ERROR and above to a file
        }
      ],
      serializers: bunyan.stdSerializers
    });

    self.defaults = {};
    self.tasks = {};
    self.timers = {};
    self.workers = {};
    self.dependencies = new DependencyGraph();
    self.queue = queue(self);
  

    // Setup listeners
    self.dispatcher.on('queue.new', function(r){
      self.log.info({event: 'queue.new', runid:r.id, taskid: r.task.id, collection: r.task.collection}, "A task run (%s) was added to the queue", r.id);

      self.createWorkerIfReady();
    });

    self.dispatcher.on('worker.complete', function(completedRun){
      // Cleanup
      delete self.workers[completedRun.pid];

      // Queue dependent tasks
      if (completedRun.task.id) {
        var refs = self.dependencies.subject(completedRun.task.id).references;
      
        // If there is dependent tasks and the previous task was successfull, GO!
        if (refs.length && completedRun.exitCode === 0) {
          for (var index = 0; index < refs.length; index++) {
            self.log.debug('Task run for %s triggered new run for %s', completedRun.task.id, refs[index]);
            self.queue.push(self.getTask(refs[index]), completedRun.id);
          }
        }
      }
    });
  });

  // Automatically load tasks from a config file
  if (self.config.tasks) {
    self.autoLoadTasks();
  }
}


{'fraction_non_alphanumeric': 0.14023806266247094, 'fraction_numerical': 0.0032836229306334656, 'mean_word_length': 2.7660999484801647, 'pattern_counts': {'":': 0, '<': 5, '<?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': '5405198', 'n_tokens_mistral': 2257, 'n_tokens_neox': 2174, 'n_words': 597}