<commit_before>//for inheritance stuff
var util = require('util');
//Path stuff
var path = require('path');
//Event emitter stuff
var EventEmitter = require('events').EventEmitter;

//The class
function AbstractQueue(name) {
	this.name = name;
	
	//Default invisibilityTimeout is 1 minute
	this.invisibilityTimeout = 60;
	this.pushManyInProgress = false;
	
	//Is the queue initialized?
	//Queue initialization makes sure that the queue is created
	//in case it doesn't exist already
	this.queueInitialized = false;
	
	//Have we started to listen for messages from the queue?
	this.isStarted = false;
	
	//We poll the queue for updates at this interval (milliseconds).
	//Default value is 5 secs
	this.pollingInterval = 5000;
	
	//Max dequeue count. If a message is dequed so many times, then
	//it will be deleted straight away without any processing
	//Default value is 5
	this.maxDequeueCount = 5;
	
	
	//This array stores the time messages were consumed so far
	var messagesTimes = [];
	
	//Messages older than this interval will be removed
	var throttleInterval = 0;
	
	//Max messages in interval
	var throttleValue = 0;
	
	function removeExpired() {
		var now = new Date();
		while(messagesTimes.length > 0 && now - messagesTimes[0] > throttleInterval) {
			messagesTimes.splice(0, 1);
		}
	}
	
	this.setThrottle = function(throttle) {
		var interval = throttle["throttle-unit"];
		switch (interval) {
			case 'sec': case 'second':
				throttleInterval = 1000; break;
			case 'min': case 'minute':
				throttleInterval = 1000 * 60; break;
			case 'hr': case 'hour':
				throttleInterval = 1000 * 60 * 60; break;
			case 'day':
				throttleInterval = 1000 * 60 * 60 * 24; break;
		}
		
		throttleValue = parseInt(throttle["throttle-value"]);
	};
	
	//Tells the queue that a particular message failed to process
	this.markFailed = function(message) {
		queue.log("markFailed() not implemented");
	};
	
	//Marks that the specified number of messages have been consumed from the queue
	this.markConsumed = function(numConsumed) {
		if(throttleInterval) {
			var now = new Date();
			for(var i=0; i<numConsumed; i++) {
				messagesTimes.push(now);
			}
		}
	};
	
	//Returns the number of messages we can consume within the throttle bracket
	this.getConsumable = function() {
		if(throttleInterval) {
			//Remove any expired messages
			removeExpired();
			
			var capacity = throttleValue - messagesTimes.length;
			if(capacity < 0) {
				capacity = 0;
			}
			return capacity;
		}
		else {
			return Number.MAX_VALUE;
		}
	};
	
	var queue = this;
	
	EventEmitter.call(this);
	
	var errorCodes = require(path.join(__dirname, "../errors.js")).errors;
	
	//Make sure polling-interval is defined in settings
	this.initPollingInterval = function() {
		if(!queue.settings["polling-interval"]) {
			queue.throwError("This module's settings need a polling-interval node");
		}
		
		queue.pollingInterval = parseInt(queue.settings["polling-interval"], 10);
		
		if(isNaN(queue.pollingInterval)) {
			queue.throwError("This module's settings, polling-interval must be a valid integer");
		}
	};
	
	
	//Make sure invisibility-timeout is defined in settings
	this.initInvisibilityTimeout = function() {
		if(!queue.settings["invisibility-timeout"]) {
			queue.throwError("This module's settings need a invisibility-timeout node");
		}
		
		queue.invisibilityTimeout = parseInt(queue.settings["invisibility-timeout"], 10);<commit_msg>CHANGED: Added ability to add external throttler to limit consumption rate by queue name in case there are multiple consumers across deployments.<commit_after>//for inheritance stuff
var util = require('util');
//Path stuff
var path = require('path');
//Event emitter stuff
var EventEmitter = require('events').EventEmitter;

//The class
function AbstractQueue(name) {
	var queue = this;
	
	this.name = name;
	
	//Default invisibilityTimeout is 1 minute
	this.invisibilityTimeout = 60;
	this.pushManyInProgress = false;
	
	//Is the queue initialized?
	//Queue initialization makes sure that the queue is created
	//in case it doesn't exist already
	this.queueInitialized = false;
	
	//Have we started to listen for messages from the queue?
	this.isStarted = false;
	
	//We poll the queue for updates at this interval (milliseconds).
	//Default value is 5 secs
	this.pollingInterval = 5000;
	
	//Max dequeue count. If a message is dequed so many times, then
	//it will be deleted straight away without any processing
	//Default value is 5
	this.maxDequeueCount = 5;
	
	
	//This array stores the time messages were consumed so far
	var messagesTimes = [];
	
	//Messages older than this interval will be removed
	var throttleInterval = 0;
	
	//Max messages in interval
	var throttleValue = 0;
	
	var throttler = null;
	
	function removeExpired() {
		var now = new Date();
		while(messagesTimes.length > 0 && now - messagesTimes[0] > throttleInterval) {
			messagesTimes.splice(0, 1);
		}
	}
	
	this.setThrottler = function(throttle) {
		throttler = throttle;
		throttler.setThrottleValue(throttleValue);
		throttler.setThrottleInterval(throttleInterval);
		throttler.setResourceName(queue.queueName);
	};
	
	this.setThrottle = function(throttle) {
		var interval = throttle["throttle-unit"];
		switch (interval) {
			case 'sec': case 'second':
				throttleInterval = 1000; break;
			case 'min': case 'minute':
				throttleInterval = 1000 * 60; break;
			case 'hr': case 'hour':
				throttleInterval = 1000 * 60 * 60; break;
			case 'day':
				throttleInterval = 1000 * 60 * 60 * 24; break;
		}
		
		throttleValue = parseInt(throttle["throttle-value"]);
	};
	
	//Tells the queue that a particular message failed to process
	this.markFailed = function(message) {
		queue.log("markFailed() not implemented");
	};
	
	//Marks that the specified number of messages have been consumed from the queue
	this.markConsumed = function(numConsumed) {
		if(throttleInterval) {
			if(throttler) {
				throttler.markConsumed(numConsumed);
			}
			else {
				var now = new Date();
				for(var i=0; i<numConsumed; i++) {
					messagesTimes.push(now);
				}
			}
		}
	};
	
	//Returns the number of messages we can consume within the throttle bracket
	this.getConsumable = function() {
		if(throttleInterval) {
			if(throttler) {
				return throttler.getConsumable();
			}
			else {
				//Remove any expired messages
				removeExpired();
			
				var capacity = throttleValue - messagesTimes.length;
				if(capacity < 0) {
					capacity = 0;
				}
				return capacity;
			}
		}
		else {
			return Number.MAX_VALUE;
		}
	};
	
	EventEmitter.call(this);
	
	var errorCodes = require(path.join(__dirname, "../errors.js")).errors;
	
	//Make sure polling-interval is defined in settings
	this.initPollingInterval = function() {
		if(!queue.settings["polling-interval"]) {
			queue.throwError("This module's settings need a polling-interval node");
		}
		
		queue.pollingInterval = parseInt(queue.settings["polling-interval"], 10);
		
		if(isNaN(queue.pollingInterval)) {
			queue.throwError("This module's settings, polling-interval must be a valid integer");
		}
	};
	
	
	//Make sure invisibility-timeout is defined in settings
	this.initInvisibilityTimeout = function() {
		if(!queue.settings["invisibility-timeout"]) {
			queue.throwError("This module's settings need a invisibility-timeout node");
		}
		
		queue.invisibilityTimeout = parseInt(queue.settings["invisibility-timeout"], 10);
{'fraction_non_alphanumeric': 0.1107950697548422, 'fraction_numerical': 0.01381552214546932, 'mean_word_length': 3.715197956577267, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '489935', 'n_tokens_mistral': 2626, 'n_tokens_neox': 2186, 'n_words': 685}