<commit_before>package org.ihtsdo.rvf.messaging;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.command.ActiveMQQueue;
import org.ihtsdo.rvf.autoscaling.InstanceManager;
import org.ihtsdo.rvf.execution.service.impl.ValidationRunConfig;
import org.ihtsdo.rvf.execution.service.impl.ValidationRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import com.google.gson.Gson;
import com.google.gson.JsonSyntaxException;
@Service
public class RvfValidationMessageConsumer {
	private static final String EC2_INSTANCE_ID_URL = "http://169.254.169.254/latest/meta-data/instance-id";
	private static final long TIME_TO_LIVE = 55*60*1000;
	private static final long DEFAULT_PROCESSING_TIME = 20*60*1000;
	private String queueName;
	@Autowired
	private ValidationRunner runner;
	private Logger logger = LoggerFactory.getLogger(getClass());
	@Autowired
	private ConnectionFactory connectionFactory;
	private boolean isWorker;
	@Autowired
	private InstanceManager instanceManager;
	private static long timeStart;
	
	public RvfValidationMessageConsumer( String queueName,Boolean isRvfWorker) {
		isWorker = isRvfWorker.booleanValue();
		this.queueName = queueName;
	}
	
	public void start() {
		logger.info("isRvfWorker instance:" + isWorker);
		if (isWorker) {
			Thread thread = new Thread ( new Runnable() {
				
				@Override
				public void run() {
					consumeMessage();
				}
			});
			thread.start();
			timeStart = Calendar.getInstance().getTimeInMillis();
			logger.info("RvfWorker instance started at:" + Calendar.getInstance().getTime());
		}
		
	}
	
	private void consumeMessage() {
		Connection connection = null;
		MessageConsumer consumer = null;
		Destination destination = new ActiveMQQueue(queueName + "?consumer.prefetchSize=1");
		Session session = null;
			while (!shutDown()) {
				try {
					connection =  connectionFactory.createConnection();
					connection.start();
					session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
					consumer = session.createConsumer(destination);
					Message msg = consumer.receive(30000);
					if ( msg == null) {
						logger.debug("No message received for destination:" + queueName);
						continue;<commit_msg>Change prefetch size to 0<commit_after>package org.ihtsdo.rvf.messaging;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.command.ActiveMQQueue;
import org.ihtsdo.rvf.autoscaling.InstanceManager;
import org.ihtsdo.rvf.execution.service.impl.ValidationRunConfig;
import org.ihtsdo.rvf.execution.service.impl.ValidationRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import com.google.gson.Gson;
import com.google.gson.JsonSyntaxException;
@Service
public class RvfValidationMessageConsumer {
	private static final String EC2_INSTANCE_ID_URL = "http://169.254.169.254/latest/meta-data/instance-id";
	private static final long TIME_TO_LIVE = 55*60*1000;
	private static final long DEFAULT_PROCESSING_TIME = 20*60*1000;
	private String queueName;
	@Autowired
	private ValidationRunner runner;
	private Logger logger = LoggerFactory.getLogger(getClass());
	@Autowired
	private ConnectionFactory connectionFactory;
	private boolean isWorker;
	@Autowired
	private InstanceManager instanceManager;
	private static long timeStart;
	
	public RvfValidationMessageConsumer( String queueName,Boolean isRvfWorker) {
		isWorker = isRvfWorker.booleanValue();
		this.queueName = queueName;
	}
	
	public void start() {
		logger.info("isRvfWorker instance:" + isWorker);
		if (isWorker) {
			Thread thread = new Thread ( new Runnable() {
				
				@Override
				public void run() {
					consumeMessage();
				}
			});
			thread.start();
			timeStart = Calendar.getInstance().getTimeInMillis();
			logger.info("RvfWorker instance started at:" + Calendar.getInstance().getTime());
		}
		
	}
	
	private void consumeMessage() {
		Connection connection = null;
		MessageConsumer consumer = null;
		Destination destination = new ActiveMQQueue(queueName + "?consumer.prefetchSize=0");
		Session session = null;
			while (!shutDown()) {
				try {
					connection =  connectionFactory.createConnection();
					connection.start();
					session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
					consumer = session.createConsumer(destination);
					Message msg = consumer.receive(30000);
					if ( msg == null) {
						logger.debug("No message received for destination:" + queueName);
						continue;
{'fraction_non_alphanumeric': 0.10191801429108688, 'fraction_numerical': 0.014103046257991726, 'mean_word_length': 5.023782559456398, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '1091451', 'n_tokens_mistral': 1779, 'n_tokens_neox': 1564, 'n_words': 345}