package omq.server;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import omq.common.broker.Broker;

import com.rabbitmq.client.AMQP.Queue.DeclareOk;
import com.rabbitmq.client.Channel;

public class RemoteThreadPool extends Thread {
	private List<InvocationThread> workers;
	private AtomicInteger busy;
	private int minPoolThreads;
	private int maxPoolThreads;
	private long refresh;
	private long keepAliveTime;
	private int maxMessagesPerThread;

	private RemoteObject obj;
	private Broker broker;
	private boolean killed = false;

	public RemoteThreadPool(int minPoolThreads, int maxPoolThreads, long refresh, long keepAliveTime, int maxMessagesPerThread, RemoteObject obj, Broker broker) {
		this.minPoolThreads = minPoolThreads;
		this.maxPoolThreads = maxPoolThreads;
		this.refresh = refresh;
		this.keepAliveTime = keepAliveTime;
		this.maxMessagesPerThread = maxMessagesPerThread;
		this.obj = obj;
		this.broker = broker;

		workers = new ArrayList<InvocationThread>(minPoolThreads);
		busy = new AtomicInteger();
	}

	@Override
	public void run() {

		// Crear aquí tots els fils?

		while (!killed) {

			try {
				Channel channel = broker.getChannel();
				DeclareOk dok = channel.queueDeclarePassive(obj.getRef());

				int numConsumers = dok.getConsumerCount();
				int numMessages = dok.getMessageCount();
				int division = numMessages / numConsumers;
				int numWorkers = workers.size();

				if (numWorkers < maxPoolThreads && division >= maxMessagesPerThread) {
					// Create a new thread
					InvocationThread worker = new InvocationThread(obj, broker);
					workers.add(worker);
					worker.start();
				} else if (numWorkers > minPoolThreads && busy.get() < numWorkers) {
					// Kill idle threads
					stopIdleThreads();
				}

				Thread.sleep(refresh);

			} catch (Exception e) {

			}
		}

	}

	private void stopIdleThreads() {
		long now = System.currentTimeMillis();

		for (InvocationThread worker : workers) {
			long lastExec = worker.getLastExecution();
			if (worker.isIdle() && (lastExec - now) > keepAliveTime) {
				// Kill thread
				try {
					worker.kill();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			}
		}
	}

	public List<InvocationThread> getWorkers() {
		return workers;
	}

	public void setWorkers(List<InvocationThread> workers) {
		this.workers = workers;
	}

	public AtomicInteger getBusy() {
		return busy;
	}

	public void setBusy(AtomicInteger busy) {
		this.busy = busy;
	}

	public int getMinPoolThreads() {
		return minPoolThreads;
	}

	public void setMinPoolThreads(int minPoolThreads) {
		this.minPoolThreads = minPoolThreads;
	}

	public int getMaxPoolThreads() {
		return maxPoolThreads;
	}

	public void setMaxPoolThreads(int maxPoolThreads) {
		this.maxPoolThreads = maxPoolThreads;
	}

	public long getRefresh() {
		return refresh;
	}

	public void setRefresh(long refresh) {
		this.refresh = refresh;
	}

	public long getKeepAliveTime() {
		return keepAliveTime;
	}

	public void setKeepAliveTime(long keepAliveTime) {
		this.keepAliveTime = keepAliveTime;
	}

	public int getMaxMessagesPerThread() {
		return maxMessagesPerThread;
	}

	public void setMaxMessagesPerThread(int maxMessagesPerThread) {
		this.maxMessagesPerThread = maxMessagesPerThread;
	}

}
