<commit_before>var Promise     = require('bluebird')
  , _           = require('lodash')
  , request     = require('request')
  , spy         = require('through2-spy')
  , dev_null    = require('dev-null')
  , url         = require('url')
  , io          = require('socket.io-client')
  ;

request.defaults({
  pool : { maxSockets: Infinity }
});

// todo [akamel] should not add spy in first place
// function log_spy() {
//   return spy(function(chunk, enc){
//     if (can_log_spy) {
//       console.log('spy: ', (new Buffer(chunk, enc)).toString());
//     }
//   });
// }

function Worker(options) {
  this.options = options;

  this.id         = this.options.id;
  this.port       = this.options.port;
  this.dir        = this.options.dir;

  this.url = url.format({
      protocol  : this.options.protocol || 'http'
    , hostname  : this.options.host     || 'localhost'
    , port      : this.port
  });
}

Worker.prototype.connect = function(cb){
  var cb = _.once(cb);

  this.socket = io(this.url);

  this.socket.once('connect', cb);

  this.socket.on('connect', this.on_connect.bind(this));

  this.socket.on('disconnect', this.on_disconnect.bind(this));

  this.socket.on('stdout', this.on_stdout.bind(this));

  this.socket.on('stderr', this.on_stderr.bind(this));
};

Worker.prototype.on_connect = function(){
  console.log('worker connected at', this.url);
};

Worker.prototype.on_disconnect = function(){
  console.log('worker disconnected', this.id);
  // todo [akamel] should we delete all?
};

Worker.prototype.reqs = {};

Worker.prototype.on_stdout = function(id, arg){
  var i = this.reqs[id];

  if (i && !_.isUndefined(arg)) {
    i.res.stdout.write(arg);
  }
};

Worker.prototype.on_stderr = function(id, arg){
  var i = this.reqs[id];

  if (i && !_.isUndefined(arg)) {
    i.res.stderr.write(arg);
  }
};

// todo [akamel] could we pipe to wrong request stdout/err if error happens way after res is done?
Worker.prototype.handle = function(req, res, next) {
  var id      = req.task.id
    , startAt = process.hrtime()
    ;

  this.socket.emit('execution', req.task);
  this.reqs[id] = { req : req, res : res, next : next };

  req.headers = req.headers || {};

  req.headers['$originalurl'] = req.url;
  // todo [akamel] this is same as req.task.id
  req.headers['$execution-id'] = req.task.metadata.execution.id;

  // req.headers['$req-hostname'] = req.hostname;
  // req.headers['$req-port'] = req.hostname;

  res.on('end', function(){
    delete this.reqs[id];
  }.bind(this));

  req
    // .pipe(log_spy())
    .pipe(request({ url : this.url + '/execute', q: req.query, method: req.method, headers: req.headers }))
    .on('response', function(response) {
      var diff = process.hrtime(startAt)
        , time = diff[0] * 1e3 + diff[1] * 1e-6
        , val  = time.toFixed(3) + 'ms'
        ;

      response.headers['x-response-time-agent_to_worker'] = val;

      res.writeHead(response.statusCode, response.headers);
    })
    .on('error', function(err){
      var body = {
          '#system' : {<commit_msg>fix missing metadata error in case of simple http<commit_after>var Promise     = require('bluebird')
  , _           = require('lodash')
  , request     = require('request')
  , spy         = require('through2-spy')
  , dev_null    = require('dev-null')
  , url         = require('url')
  , io          = require('socket.io-client')
  ;

request.defaults({
  pool : { maxSockets: Infinity }
});

// todo [akamel] should not add spy in first place
// function log_spy() {
//   return spy(function(chunk, enc){
//     if (can_log_spy) {
//       console.log('spy: ', (new Buffer(chunk, enc)).toString());
//     }
//   });
// }

function Worker(options) {
  this.options = options;

  this.id         = this.options.id;
  this.port       = this.options.port;
  this.dir        = this.options.dir;

  this.url = url.format({
      protocol  : this.options.protocol || 'http'
    , hostname  : this.options.host     || 'localhost'
    , port      : this.port
  });
}

Worker.prototype.connect = function(cb){
  var cb = _.once(cb);

  this.socket = io(this.url);

  this.socket.once('connect', cb);

  this.socket.on('connect', this.on_connect.bind(this));

  this.socket.on('disconnect', this.on_disconnect.bind(this));

  this.socket.on('stdout', this.on_stdout.bind(this));

  this.socket.on('stderr', this.on_stderr.bind(this));
};

Worker.prototype.on_connect = function(){
  console.log('worker connected at', this.url);
};

Worker.prototype.on_disconnect = function(){
  console.log('worker disconnected', this.id);
  // todo [akamel] should we delete all?
};

Worker.prototype.reqs = {};

Worker.prototype.on_stdout = function(id, arg){
  var i = this.reqs[id];

  if (i && !_.isUndefined(arg)) {
    i.res.stdout.write(arg);
  }
};

Worker.prototype.on_stderr = function(id, arg){
  var i = this.reqs[id];

  if (i && !_.isUndefined(arg)) {
    i.res.stderr.write(arg);
  }
};

// todo [akamel] could we pipe to wrong request stdout/err if error happens way after res is done?
Worker.prototype.handle = function(req, res, next) {
  var id      = req.task.id
    , startAt = process.hrtime()
    ;

  this.socket.emit('execution', req.task);
  this.reqs[id] = { req : req, res : res, next : next };

  req.headers = req.headers || {};

  req.headers['$originalurl'] = req.url;
  req.headers['$execution-id'] = req.task.id;

  // req.headers['$req-hostname'] = req.hostname;
  // req.headers['$req-port'] = req.hostname;

  res.on('end', function(){
    delete this.reqs[id];
  }.bind(this));

  req
    // .pipe(log_spy())
    .pipe(request({ url : this.url + '/execute', q: req.query, method: req.method, headers: req.headers }))
    .on('response', function(response) {
      var diff = process.hrtime(startAt)
        , time = diff[0] * 1e3 + diff[1] * 1e-6
        , val  = time.toFixed(3) + 'ms'
        ;

      response.headers['x-response-time-agent_to_worker'] = val;

      res.writeHead(response.statusCode, response.headers);
    })
    .on('error', function(err){
      var body = {
          '#system' : {
{'fraction_non_alphanumeric': 0.18005266622778143, 'fraction_numerical': 0.0026333113890717576, 'mean_word_length': 2.8486383787207092, 'pattern_counts': {'":': 0, '<': 3, '<?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': '6567984', 'n_tokens_mistral': 2138, 'n_tokens_neox': 2091, 'n_words': 409}