<commit_before>var when = require('when');
var QueueProcessor = require('./QueueProcessor').QueueProcessor;
var Writable = require('stream').Writable;

function WritableQueue(processorFunction, processorOptions) {
	var self = this;
	self._pendingTasks = 0;
	self._maximumTasks = processorOptions.concurrencyLimit || 1;
	
	self._processorFunction = function(taskWrapper) {
		return when.try(processorFunction, taskWrapper.data).then(function() {
			self._pendingTasks -= 1;
			if (self._pendingTasks === 0) {
				self.emit('drain');
			}
			return taskWrapper.callback();
		}, function(error) {
			self.emit('error', error);
			return taskWrapper.callback(error);
		});
	};
	
	Writable.call(self, {
		objectMode: true
	});
	
	this._queue = new QueueProcessor(processorOptions);
	this._queue.setProcessorFunction(self._processorFunction);

	this._queue.start();
}
WritableQueue.prototype = Object.create(Writable.prototype);

WritableQueue.prototype._write = function write(chunk, encoding, callback) {
	callback = callback || function() {};
	this._queue.push({ data: chunk, callback: callback });
	this._pendingTasks += 1;
	return this._pendingTasks < this._maximumTasks;
};

WritableQueue.prototype._end = function end(chunk, encoding, callback) {
	if (chunk) {
		this.write(chunk, encoding, callback);
	}
};

module.exports.WritableQueue = WritableQueue;<commit_msg>Revert "Utils/WritableQueue: changed inheritance from EventEmmiter to WritableStream due to some additional underlaying optimalizations (fixed big memory leaks)"<commit_after>var when = require('when');
var EventEmitter = require('events').EventEmitter;
var QueueProcessor = require('./QueueProcessor').QueueProcessor;

function WritableQueue(processorFunction, processorOptions) {
	var self = this;
	self._pendingTasks = 0;
	self._maximumTasks = processorOptions.concurrencyLimit || 1;
	
	self._processorFunction = function(taskWrapper) {
		return when.try(processorFunction, taskWrapper.data).then(function() {
			self._pendingTasks -= 1;
			if (self._pendingTasks === 0) {
				self.emit('drain');
			}
			taskWrapper.callback();
		}, function(error) {
			self.emit('error', error);
			taskWrapper.callback(error);
		});
	};
	
	this._queue = new QueueProcessor(processorOptions);
	this._queue.setProcessorFunction(self._processorFunction);
	this._queue.start();
}
WritableQueue.prototype = Object.create(EventEmitter.prototype);

WritableQueue.prototype.write = function write(chunk, encoding, callback) {
	callback = callback || function() {};
	this._queue.push({ data: chunk, callback: callback });
	this._pendingTasks += 1;
	return this._pendingTasks < this._maximumTasks;
};

WritableQueue.prototype.end = function end(chunk, encoding, callback) {
	if (chunk) {
		this.write(chunk, encoding, callback);
	}
	
};

module.exports.WritableQueue = WritableQueue;
{'fraction_non_alphanumeric': 0.13356890459363957, 'fraction_numerical': 0.0035335689045936395, 'mean_word_length': 5.493119266055046, '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': '2813299', 'n_tokens_mistral': 875, 'n_tokens_neox': 809, 'n_words': 167}