source: branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java @ 106

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

}