Changeset 106 for branches/supervisor/src

Timestamp:

10/18/13 16:09:36 (11 years ago)

Author:

stoda

Message:

abans que la segueixi liant...

Location:

branches/supervisor/src

Files:

• main/java/omq/common/broker/Broker.java (modified) (3 diffs)
• main/java/omq/common/broker/RemoteMultiBroker.java (added)
• main/java/omq/server/InvocationThread.java (modified) (5 diffs)
• main/java/omq/server/RemoteObject.java (modified) (3 diffs)
• main/java/omq/supervisor/Supervisor.java (modified) (1 diff)
• main/java/omq/supervisor/SupervisorImpl.java (modified) (6 diffs)
• main/java/omq/supervisor/SupervisorThread.java (deleted)
• main/resources/log4j.xml (modified) (2 diffs)
• test/java/omq/test/supervisor/SleepTest.java (modified) (3 diffs)

branches/supervisor/src/main/java/omq/common/broker/Broker.java

@@ -240 +240 @@
         }
 
+        public void bind(String reference, String UID, RemoteObject remote) throws RemoteException, AlreadyBoundException {
+                bind(reference, UID, remote, environment);
+        }
+
         /**
          * Binds the reference to the specified remote object. This function uses
@@ -264 +268 @@
                 try {
                         remote.startRemoteObject(reference, this, env);
+                        remoteObjs.put(reference, remote);
+                } catch (Exception e) {
+                        throw new RemoteException(e);
+                }
+        }
+
+        public void bind(String reference, String UID, RemoteObject remote, Properties env) throws RemoteException, AlreadyBoundException {
+                if (remoteObjs.containsKey(reference)) {
+                        throw new AlreadyBoundException(reference);
+                }
+                // Try to start the remtoeObject listeners
+                try {
+                        remote.startRemoteObject(reference, UID, this, env);
                         remoteObjs.put(reference, remote);
                 } catch (Exception e) {
@@ -400 +417 @@
          * Supervisor
          */
-        public void setSupervisor(String supervisorName, String brokerName) throws Exception {
+        public void setSupervisor(String supervisorName, String brokerSet, String brokerName) throws Exception {
                 // Create a RemoteBrokerImpl
-                bind(brokerName, new RemoteBrokerImpl());
+                bind(brokerSet, brokerName, new RemoteBrokerImpl());
                 // Subscribe broker
                 supervisor = lookup(supervisorName, Supervisor.class);
-                supervisor.subscribe(brokerName);
-                logger.info("Supervisor set: " + supervisorName + ", BrokerName: " + brokerName);
+                supervisor.subscribe(brokerSet, brokerName);
+                logger.info("Supervisor set: " + supervisorName + ", BrokerSet: " + brokerSet + ", BrokerName: " + brokerName);
         }
 

branches/supervisor/src/main/java/omq/server/InvocationThread.java

@@ -36 +36 @@
         private RemoteObject obj;
         private String reference;
+        private String UID;
         private Properties env;
         private boolean idle;
@@ -54 +55 @@
         public InvocationThread(RemoteObject obj) throws Exception {
                 this.obj = obj;
+                this.UID = obj.getUID();
                 this.reference = obj.getRef();
                 this.env = obj.getEnv();
@@ -87 +89 @@
                                 String serializerType = delivery.getProperties().getType();
 
-                                // Deserialize the json
+                                // Deserialize the request
                                 Request request = serializer.deserializeRequest(serializerType, delivery.getBody(), obj);
                                 String methodName = request.getMethod();
@@ -196 +198 @@
 
                 /*
+                 * UID queue
+                 */
+
+                if (UID != null) {
+
+                        boolean uidDurable = false;
+                        boolean uidExclusive = true;
+                        boolean uidAutoDelete = true;
+
+                        channel.queueDeclare(UID, uidDurable, uidExclusive, uidAutoDelete, null);
+                        if (!exchange.equalsIgnoreCase("")) { // Default exchange case
+                                channel.queueBind(UID, exchange, UID);
+                        }
+                }
+
+                /*
                  * Multi queue, exclusive per each instance
                  */
@@ -233 +251 @@
                 channel.basicConsume(queue, autoAck, consumer);
                 channel.basicConsume(multiQueue, autoAck, consumer);
+                if (UID != null) {
+                        channel.basicConsume(UID, autoAck, consumer);
+                }
         }
 

branches/supervisor/src/main/java/omq/server/RemoteObject.java

@@ -31 +31 @@
 
         private String reference;
+        private String UID;
         private Properties env;
         private transient Broker broker;
@@ -83 +84 @@
                 pool = new RemoteThreadPool(minPoolThreads, maxPoolThreads, refresh, keepAliveTime, maxMessagesPerThread, this, broker);
                 pool.start();
+        }
+
+        public void startRemoteObject(String reference, String UID, Broker broker, Properties env) throws Exception {
+                this.UID = UID;
+                startRemoteObject(reference, broker, env);
         }
 
@@ -209 +215 @@
         }
 
+        public String getUID() {
+                return UID;
+        }
+
+        public void setUID(String uID) {
+                UID = uID;
+        }
+
 }

branches/supervisor/src/main/java/omq/supervisor/Supervisor.java

@@ -2 +2 @@
 
 import omq.Remote;
+import omq.common.broker.HasObject;
 
 public interface Supervisor extends Remote {
 
-        public void subscribe(String brokerName) throws Exception;
+        public void subscribe(String brokerSet, String brokerName) throws Exception;
 
-        public void spawnObject(OmqSettings settings, int numObjects) throws Exception;
+        public void spawnObject(OmqSettings settings) throws Exception;
 
-        public void unbindObject(OmqSettings settings, int numObjects) throws Exception;
+        public void spawnObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception;
+
+        public void unbindObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception;
 
 }

branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java

@@ -5 +5 @@
 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;
 
-public class SupervisorImpl extends RemoteObject implements Supervisor {
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.AMQP.Queue.DeclareOk;
+
+public class SupervisorImpl extends RemoteObject implements Supervisor, Runnable {
 
         /**
@@ -18 +25 @@
         private static final long serialVersionUID = 1L;
         private static final Logger logger = Logger.getLogger(SupervisorImpl.class.getName());
-        private SupervisorThread thread;
+
+        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(long sleep) {
+        public SupervisorImpl(String brokerSet, long sleep) {
+                this.brokerSet = brokerSet;
+                this.sleep = 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);
+        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 spawnObject(OmqSettings settings, int numObjects) throws Exception {
+        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();
@@ -48 +100 @@
                 int minObjects = settings.getMinNumberObjects();
 
-                for (RemoteBroker broker : brokers) {
-                        if (!broker.hasObject(reference) && minObjects >= numObjects) {
-                                broker.spawnObject(reference, settings.getClassName(), settings.getProps());
+                for (HasObject h : hasList) {
+                        if (h.hasObject() && minObjects >= numObjects) {
+                                brokerMap.get(h.getBrokerName()).spawnObject(reference, settings.getClassName(), settings.getProps());
                                 numObjects++;
                                 if (minObjects >= numObjects) {
@@ -61 +113 @@
 
         @Override
-        public void unbindObject(OmqSettings settings, int numObjects) throws Exception {
+        public void unbindObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception {
                 String reference = settings.getReference();
 
@@ -75 +127 @@
         }
 
-        public SupervisorThread getThread() {
-                return thread;
-        }
+        private void checkObject(String reference) throws Exception {
+                OmqSettings settings = objectSettings.get(reference);
 
-        public void setThread(SupervisorThread thread) {
-                this.thread = thread;
+                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);
+                }
         }
 
@@ -99 +174 @@
         }
 
+        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;
+        }
+
 }

branches/supervisor/src/main/resources/log4j.xml

@@ -4 +4 @@
 <log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
     <appender name="consoleAppender" class="org.apache.log4j.ConsoleAppender">
-        <param name="Threshold" value="INFO" />
+        <param name="Threshold" value="DEBUG" />
         <layout class="org.apache.log4j.PatternLayout">
             <param name="ConversionPattern" value="%d{[yyyy-MM-dd HH:mm:ss]} %-5p %c:%L - %m%n" />
@@ -30 +30 @@
 
   <root> 
-    <priority value ="INFO" /> 
+    <priority value ="DEBUG" /> 
     <appender-ref ref="consoleAppender" />
     <appender-ref ref="A2" />  

branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java

@@ -13 +13 @@
 
 public class SleepTest {
+        private static String brokerSet = "broker";
 
         @BeforeClass
@@ -42 +43 @@
 
                 // Set supervisor
-                SupervisorImpl supervisor = new SupervisorImpl(500);
+                SupervisorImpl supervisor = new SupervisorImpl(brokerSet, 500);
                 broker.bind("supervisor", supervisor);
+                Thread t = new Thread(supervisor);
+                t.start();
 
-                broker.setSupervisor("supervisor", "b1");
-                broker2.setSupervisor("supervisor", "b2");
+                broker.setSupervisor("supervisor", brokerSet, "b1");
+                broker2.setSupervisor("supervisor", brokerSet, "b2");
         }
 
@@ -63 +66 @@
 
                 OmqSettings settings = new OmqSettings("sleep", SleepImpl.class.getName(), env, 1, 20, 20);
-                supervisor.spawnObject(settings, 0);
+                supervisor.spawnObject(settings);
 
                 Sleep sleep = broker.lookup("sleep", Sleep.class);