<commit_before>'use strict';

var queue = require('queue-async');
var q = queue();
var sources = [];
var tilesQueue = queue(1);

global.mapOptions = JSON.parse(process.argv[4]);
var map = require(process.argv[2]);

JSON.parse(process.argv[3]).forEach(function(source) {
  q.defer(loadSource, source);
});

function loadSource(source, done) {
  var loaded = {name: source.name};
  sources.push(loaded);

  if (source.mbtiles) require('./mbtiles')(source, done);
  else if (source.url) require('./remote')(source, done);
  else throw new Error('Unknown source type');
}

q.awaitAll(function(err, results) {
  if (err) throw err;
  for (var i = 0; i < results.length; i++) sources[i].getTile = results[i];
  process.send({ready: true});
});

function processTile(tile, callback) {
  var q = queue();

  for (var i = 0; i < sources.length; i++) {
    q.defer(sources[i].getTile, tile);
  }

  q.awaitAll(gotData);

  function gotData(err, results) {
    if (err) throw err;

    var data = {};
    for (var i = 0; i < results.length; i++) {
      data[sources[i].name] = results[i];
      if (!results[i]) {
        callback(null);
        process.send({reduce: true});
        return;
      }
    }

    function gotResults(err, value) {
      if (err) throw err;
      process.send({reduce: true, value: value, tile: tile}, undefined, function() {
        callback(null);
      });
      if (process.versions.node.split('.')[0] < 4) // node/iojs prior to v4.0.0 don't have the callback to process.send above (instead the call is syncronous)
        callback(null);
    }

    map(data, tile, write, gotResults);
  }
}
<commit_msg>minor clean up of async changes<commit_after>'use strict';

var queue = require('queue-async');
var q = queue();
var sources = [];
var tilesQueue = queue(1);
var isOldNode = process.versions.node.split('.')[0] < 4;

global.mapOptions = JSON.parse(process.argv[4]);
var map = require(process.argv[2]);

JSON.parse(process.argv[3]).forEach(function(source) {
  q.defer(loadSource, source);
});

function loadSource(source, done) {
  var loaded = {name: source.name};
  sources.push(loaded);

  if (source.mbtiles) require('./mbtiles')(source, done);
  else if (source.url) require('./remote')(source, done);
  else throw new Error('Unknown source type');
}

q.awaitAll(function(err, results) {
  if (err) throw err;
  for (var i = 0; i < results.length; i++) sources[i].getTile = results[i];
  process.send({ready: true});
});

function processTile(tile, callback) {
  var q = queue();

  for (var i = 0; i < sources.length; i++) {
    q.defer(sources[i].getTile, tile);
  }

  q.awaitAll(gotData);

  function gotData(err, results) {
    if (err) throw err;

    var data = {};
    for (var i = 0; i < results.length; i++) {
      data[sources[i].name] = results[i];
      if (!results[i]) {
        callback();
        process.send({reduce: true});
        return;
      }
    }

    function gotResults(err, value) {
      if (err) throw err;
      process.send({reduce: true, value: value, tile: tile}, null, callback);
      if (isOldNode) callback(); // process.send is async since Node 4.0
    }

    map(data, tile, write, gotResults);
  }
}

{'fraction_non_alphanumeric': 0.17443324937027707, 'fraction_numerical': 0.007241813602015114, 'mean_word_length': 3.3580246913580245, 'pattern_counts': {'":': 0, '<': 11, '<?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': '2793784', 'n_tokens_mistral': 1081, 'n_tokens_neox': 1058, 'n_words': 279}