<commit_before>var events = require('events'),
    util = require('util'),
    net = require('net');

var velbus = function(host, port) {
	var that = this;
	events.EventEmitter.call(this); // Inherit EventEmitter
	this.setMaxListeners(256);

	that.host = host;
	that.port = port;
	that.retry_after_ms = 1000;

	util.log("Connecting velbus to " + that.host + ":" + that.port);
	that.socket = net.connect(that.port, that.host, function() {
		that.emit('connect');
	});

	that.socket.on('error', function (e) {
		that.emit('error', e);
	});

	that.socket.on('close', function() {
		that.emit('close');
		setTimeout(function(){
				that.socket.connect(that.host, that.port);
			}, that.retry_after_ms);
	});

	that.unprocessed_data = new Buffer(4096);
	that.n_unprocessed_data = 0;

	that.socket.on('data', function(data) {
		data.copy(that.unprocessed_data, that.n_unprocessed_data);
		that.n_unprocessed_data += data.length;
		that.process_data();
	});

	return this;
}
util.inherits(velbus, events.EventEmitter);

velbus.prototype.process_data = function() {
	while( this.n_unprocessed_data > 6 ) {
		this.process_message();<commit_msg>Bugfix in reconnect code<commit_after>var events = require('events'),
    util = require('util'),
    net = require('net');

var velbus = function(host, port) {
	var that = this;
	events.EventEmitter.call(this); // Inherit EventEmitter
	this.setMaxListeners(256);

	that.host = host;
	that.port = port;
	that.retry_after_ms = 1000;

	util.log("Connecting velbus to " + that.host + ":" + that.port);
	that.socket = net.connect(that.port, that.host, function() {
		that.emit('connect');
	});

	that.socket.on('error', function (e) {
		that.emit('error', e);
	});

	that.socket.on('close', function() {
		that.emit('close');
		setTimeout(function(){
				that.socket.connect(that.port, that.host);
			}, that.retry_after_ms);
	});

	that.unprocessed_data = new Buffer(4096);
	that.n_unprocessed_data = 0;

	that.socket.on('data', function(data) {
		data.copy(that.unprocessed_data, that.n_unprocessed_data);
		that.n_unprocessed_data += data.length;
		that.process_data();
	});

	return this;
}
util.inherits(velbus, events.EventEmitter);

velbus.prototype.process_data = function() {
	while( this.n_unprocessed_data > 6 ) {
		this.process_message();
{'fraction_non_alphanumeric': 0.1709776413853573, 'fraction_numerical': 0.011398509425690487, 'mean_word_length': 4.565853658536585, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, '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': '6666797', 'n_tokens_mistral': 875, 'n_tokens_neox': 813, 'n_words': 138}