package omq.supervisor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import omq.common.broker.HasObject;
import omq.common.broker.RemoteBroker;
import omq.common.broker.RemoteMultiBroker;
import omq.exception.RemoteException;
import omq.server.RemoteObject;

import org.apache.log4j.Logger;

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

public class SupervisorImpl extends RemoteObject implements Supervisor, Runnable {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	private static final Logger logger = Logger.getLogger(SupervisorImpl.class.getName());

	private String brokerSet;
	private long sleep;
	private Map<String, OmqSettings> objectSettings;
	// TODO: Set<?>
	private RemoteMultiBroker multiBroker;
	private Map<String, RemoteBroker> brokerMap;
	private List<RemoteBroker> brokers;

	public SupervisorImpl(String brokerSet, long sleep) {
		this.brokerSet = brokerSet;
		this.sleep = sleep;
		brokers = new ArrayList<RemoteBroker>();
		objectSettings = new HashMap<String, OmqSettings>();
	}

	@Override
	public void run() {
		try {
			multiBroker = getBroker().lookup(brokerSet, RemoteMultiBroker.class);
			while (true) {
				try {
					Set<String> keys = objectSettings.keySet();
					for (String reference : keys) {
						System.out.println("key = " + keys);
						checkObject(reference);
					}
				} catch (Exception e) {
					e.printStackTrace();
				}

				try {
					Thread.sleep(sleep);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		} catch (RemoteException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
	}

	@Override
	public void subscribe(String brokerSet, String brokerName) throws Exception {
		if (brokerSet.equals(brokerSet) && !brokerMap.containsKey(brokerName)) {
			logger.info("Broker " + brokerName + " subscrived");
			RemoteBroker broker = getBroker().lookup(brokerName, RemoteBroker.class);
			brokerMap.put(brokerSet, broker);
		} else {
			throw new Exception("blablabla");
		}
	}

	@Override
	public void spawnObject(OmqSettings settings) throws Exception {
		String reference = settings.getReference();

		if (objectSettings.containsKey(reference)) {
			throw new Exception("JAJAJAJAJA");
		}
		objectSettings.put(reference, settings);
	}

	@Override
	public void spawnObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception {

		String reference = settings.getReference();

		if (!objectSettings.containsKey(reference)) {
			objectSettings.put(reference, settings);
		}

		int minObjects = settings.getMinNumberObjects();

		for (HasObject h : hasList) {
			if (h.hasObject() && minObjects >= numObjects) {
				brokerMap.get(h.getBrokerName()).spawnObject(reference, settings.getClassName(), settings.getProps());
				numObjects++;
				if (minObjects >= numObjects) {
					break;
				}
			}
		}

	}

	@Override
	public void unbindObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception {
		String reference = settings.getReference();

		int minObjects = settings.getMinNumberObjects();

		for (RemoteBroker broker : brokers) {
			if (broker.hasObject(reference) && (numObjects - 1) >= minObjects) {
				broker.deleteObject(reference);
				break;
			}
		}

	}

	private void checkObject(String reference) throws Exception {
		OmqSettings settings = objectSettings.get(reference);

		int minObjects = settings.getMinNumberObjects();
		int maxMessages = settings.getMaxNumQueued();
		int minMessages = settings.getMinNumQueued();

		Channel channel = getBroker().getChannel();
		DeclareOk dok = channel.queueDeclarePassive(reference);

		int numObjects = 0;
		int numMessages = dok.getMessageCount();

		HasObject[] hasList = multiBroker.hasObjectInfo(reference);
		for (HasObject h : hasList) {
			if (h.hasObject()) {
				numObjects++;
			}
		}

		System.out.println("Num Consumers: " + numObjects + ", num Messages: " + numMessages);

		if (maxMessages < numMessages || numObjects < minObjects) {
			logger.info("SPAWN TIME!!");
			spawnObject(settings, hasList, numObjects);
		} else if (numMessages < minMessages && minObjects < numObjects) {
			logger.info("Unbinding object!!!");
			unbindObject(settings, hasList, numObjects);
		}
	}

	public Map<String, OmqSettings> getObjectSettings() {
		return objectSettings;
	}

	public void setObjectSettings(Map<String, OmqSettings> objectSettings) {
		this.objectSettings = objectSettings;
	}

	public List<RemoteBroker> getBrokers() {
		return brokers;
	}

	public void setBrokers(List<RemoteBroker> brokers) {
		this.brokers = brokers;
	}

	public String getBrokerSet() {
		return brokerSet;
	}

	public void setBrokerSet(String brokerSet) {
		this.brokerSet = brokerSet;
	}

	public RemoteMultiBroker getMultiBroker() {
		return multiBroker;
	}

	public void setMultiBroker(RemoteMultiBroker multiBroker) {
		this.multiBroker = multiBroker;
	}

}
