<commit_before>/*
 *
 * (C) 2013 <NAME>
 * MIT LICENCE
 *
 */

var net = require('net'),
    util = require('util'),
    os = require('os'),
    winston = require('winston'),
    common = require('winston/lib/winston/common');

var Logstash = exports.Logstash = function (options) {
  winston.Transport.call(this, options);
  options = options || {};

  this.name       = 'logstash';
  this.localhost  = options.localhost || os.hostname();
  this.host       = options.host || '127.0.0.1';
  this.port       = options.port || 28777;
  this.node_name  = options.node_name || process.title;
  this.pid        = options.pid || process.pid;

  // Connection state
  this.log_queue = [];
  this.connected = false;
  this.socket = null;
  this.retries = 0;

  this.connect();
};

//
// Inherit from `winston.Transport`.
//
util.inherits(Logstash, winston.Transport);

//
// Define a getter so that `winston.transports.Syslog`
// is available and thus backwards compatible.
//
winston.transports.Logstash = Logstash;


Logstash.prototype.log = function (level, msg, meta, callback) {
  var self = this,
      meta = winston.clone(meta || {}),
      log_entry;

  if (self.silent) {
    return callback(null, true);
  }

  log_entry = common.log({
    level: level,
    message: msg,
    meta: meta,
    timestamp: self.timestamp,
    json: true
  });

  if (!self.connected) {
    self.log_queue.push({
      message: log_entry,
      callback: function () {
        self.emit('logged');
        callback(null, true);
      }
    });
  } else {
    self.sendLog(log_entry, function () {
      self.emit('logged');
      callback(null, true);
    });
  }
};

Logstash.prototype.connect = function () {
  this.retries++;
  this.connecting = true;
  this.socket = new net.Socket();
  var self = this;

  this.socket.on('error', function (err) {
    self.connecting = false;
    self.connected = false;
    self.socket.destroy();
    self.socket = null;
  });

  this.socket.on('timeout', function() {
    if (self.socket.readyState !== 'open') {
      self.socket.destroy();
    }
  });

  this.socket.on('close', function (had_error) {
    self.connected = false;

    if (self.retries < 4) {
      if (!self.connecting) {
        setTimeout(function () {
          self.connect();
        }, 100);
      }
    } else {
      self.log_queue = [];
      self.silent = true;
    }
  });


  this.socket.connect(self.port, self.host, function () {
    self.announce();
  });

};

Logstash.prototype.announce = function () {
  var self = this;
  self.connected = true;
  self.flush();
};

Logstash.prototype.flush = function () {
  var self = this;

  for (var i = 0; i < self.log_queue.length; i++) {
    self.sendLog(self.log_queue[i].message, self.log_queue[i].callback);
    self.emit('logged');
  }
  self.log_queue.length = 0;
};

Logstash.prototype.sendLog = function (message, callback) {
  var self = this;
  callback = callback || function () {};

  self.socket.write(message + '\n');
  callback();
};


















<commit_msg>Added ability to configure connection retry count<commit_after>/*
 *
 * (C) 2013 <NAME>
 * MIT LICENCE
 *
 */

var net = require('net'),
    util = require('util'),
    os = require('os'),
    winston = require('winston'),
    common = require('winston/lib/winston/common');

var Logstash = exports.Logstash = function (options) {
  winston.Transport.call(this, options);
  options = options || {};

  this.name                = 'logstash';
  this.localhost           = options.localhost || os.hostname();
  this.host                = options.host || '127.0.0.1';
  this.port                = options.port || 28777;
  this.node_name           = options.node_name || process.title;
  this.pid                 = options.pid || process.pid;
  this.max_connect_retries = options.max_connect_retries || 4;

  // Connection state
  this.log_queue = [];
  this.connected = false;
  this.socket = null;
  this.retries = 0;

  this.connect();
};

//
// Inherit from `winston.Transport`.
//
util.inherits(Logstash, winston.Transport);

//
// Define a getter so that `winston.transports.Syslog`
// is available and thus backwards compatible.
//
winston.transports.Logstash = Logstash;


Logstash.prototype.log = function (level, msg, meta, callback) {
  var self = this,
      meta = winston.clone(meta || {}),
      log_entry;

  if (self.silent) {
    return callback(null, true);
  }

  log_entry = common.log({
    level: level,
    message: msg,
    meta: meta,
    timestamp: self.timestamp,
    json: true
  });

  if (!self.connected) {
    self.log_queue.push({
      message: log_entry,
      callback: function () {
        self.emit('logged');
        callback(null, true);
      }
    });
  } else {
    self.sendLog(log_entry, function () {
      self.emit('logged');
      callback(null, true);
    });
  }
};

Logstash.prototype.connect = function () {
  this.retries++;
  this.connecting = true;
  this.socket = new net.Socket();
  var self = this;

  this.socket.on('error', function (err) {
    self.connecting = false;
    self.connected = false;
    self.socket.destroy();
    self.socket = null;
  });

  this.socket.on('timeout', function() {
    if (self.socket.readyState !== 'open') {
      self.socket.destroy();
    }
  });

  this.socket.on('close', function (had_error) {
    self.connected = false;

    if (self.max_connect_retries === -1 || self.retries < self.max_connect_retries) {
      if (!self.connecting) {
        setTimeout(function () {
          self.connect();
        }, 100);
      }
    } else {
      self.log_queue = [];
      self.silent = true;
    }
  });


  this.socket.connect(self.port, self.host, function () {
    self.announce();
    self.connecting = false;
  });

};

Logstash.prototype.announce = function () {
  var self = this;
  self.connected = true;
  self.flush();
};

Logstash.prototype.flush = function () {
  var self = this;

  for (var i = 0; i < self.log_queue.length; i++) {
    self.sendLog(self.log_queue[i].message, self.log_queue[i].callback);
    self.emit('logged');
  }
  self.log_queue.length = 0;
};

Logstash.prototype.sendLog = function (message, callback) {
  var self = this;
  callback = callback || function () {};

  self.socket.write(message + '\n');
  callback();
};
{'fraction_non_alphanumeric': 0.16265732037597577, 'fraction_numerical': 0.007169029791301578, 'mean_word_length': 2.616359447004608, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '202593', 'n_tokens_mistral': 2159, 'n_tokens_neox': 2083, 'n_words': 419}