package omq.supervisor;

import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import omq.client.proxy.Proxymq;
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.AMQP.Queue.DeclareOk;
import com.rabbitmq.client.Channel;

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;
	private RemoteMultiBroker multiBroker;
	private Map<String, RemoteBroker> brokerMap;

	public SupervisorImpl(String brokerSet, long sleep) {
		this.brokerSet = brokerSet;
		this.sleep = sleep;
		brokerMap = new HashMap<String, 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(brokerSet,
			// RemoteBroker.class);
			Proxymq proxy = new Proxymq(brokerSet, RemoteBroker.class, getBroker());
			Class<?>[] array = { RemoteBroker.class };
			RemoteBroker broker = (RemoteBroker) Proxy.newProxyInstance(RemoteBroker.class.getClassLoader(), array, proxy);
			broker.setUID(brokerName);
			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");
		}
		
		HasObject[] hasList = multiBroker.hasObjectInfo(reference);

		int minObjects = settings.getMinNumberObjects();
		int numBrokers = hasList.length;
		int numObjects = 0;

		for (HasObject h : hasList) {
			if (h.hasObject()) {
				numObjects++;
			}
		}
		System.out.println("NumObjects " + numObjects + " numBrokers " + numBrokers);

		int i = 0;
		while (numObjects <= minObjects && i < numBrokers) {
			HasObject h = hasList[i++];
			if (h.hasObject()) {
				brokerMap.get(h.getBrokerName()).spawnObject(reference, settings.getClassName());
				numObjects++;
			}
		}
		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 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;
	}

}
