package omq.supervisor;

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

import omq.common.broker.RemoteBroker;
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 {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	private static final Logger logger = Logger.getLogger(SupervisorImpl.class.getName());
	private SupervisorThread thread;
	private Map<String, OmqSettings> objectSettings;
	// TODO: Set<?>
	private List<RemoteBroker> brokers;

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

	@Override
	public void subscribe(String brokerName) throws Exception {
		logger.info("Broker " + brokerName + " subscrived");
		RemoteBroker broker = getBroker().lookup(brokerName, RemoteBroker.class);
		brokers.add(broker);
	}

	@Override
	public void spawnObject(OmqSettings settings) throws Exception {

		String reference = settings.getReference();

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

		int minObjects = settings.getMinNumberObjects();

		Channel channel = getBroker().getNewChannel();

		int numObjects = 0;
		try {
			DeclareOk dok = channel.queueDeclarePassive(reference);
			numObjects = dok.getConsumerCount();
			channel.close();
		} catch (Exception io) {
			// The queue doesn't exist & the channel has been closed
		}

		for (RemoteBroker broker : brokers) {
			if (!broker.hasObject(reference) && minObjects >= numObjects) {
				broker.spawnObject(reference, settings.getClassName(), settings.getProps());
				numObjects++;
				if (minObjects >= numObjects) {
					break;
				}
			}
		}

	}

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

		int minObjects = settings.getMinNumberObjects();

		Channel channel = getBroker().getNewChannel();

		int numObjects = 0;
		try {
			DeclareOk dok = channel.queueDeclarePassive(reference);
			numObjects = dok.getConsumerCount();
			channel.close();

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

		} catch (Exception io) {
			// The queue doesn't exist & the channel has been closed
		}

		for (RemoteBroker broker : brokers) {
			if (!broker.hasObject(reference) && minObjects >= numObjects) {
				broker.spawnObject(reference, settings.getClassName(), settings.getProps());
				numObjects++;
				if (minObjects >= numObjects) {
					break;
				}
			}
		}
	}

	public SupervisorThread getThread() {
		return thread;
	}

	public void setThread(SupervisorThread thread) {
		this.thread = thread;
	}

	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;
	}

}
