Changeset 110 for branches/supervisor

Timestamp:

10/22/13 15:15:51 (11 years ago)

Author:

stoda

Message:

Supervisor seems to work

Location:

branches/supervisor

Files:

• src/main/java/omq/client/proxy/Proxymq.java (modified) (1 diff)
• src/main/java/omq/common/broker/RemoteBrokerImpl.java (modified) (3 diffs)
• src/main/java/omq/supervisor/RemoteBroker.java (moved) (moved from branches/supervisor/src/main/java/omq/common/broker/RemoteMultiBroker.java) (1 diff)
• src/main/java/omq/supervisor/SupervisorImpl.java (modified) (11 diffs)
• src/main/resources/log4j.xml (modified) (2 diffs)
• src/test/java/test/multiBroker (deleted)
• target/classes/log4j.xml (added)
• target/classes/log4j.xml.svntmp (added)

branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java

@@ -380 +380 @@
 
         @Override
-        public void setUID(String uID) {
+        public void setUID(String uID) {System.out.println("SET UID");
                 this.UID = uID;
         }

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

@@ -28 +28 @@
         @Override
         public void spawnObject(String reference, String className, Properties env) throws Exception {
+                System.out.println("SPAWN broker = " + this.getUID());
                 RemoteObject remote = (RemoteObject) Class.forName(className).newInstance();
                 getBroker().bind(reference, remote, env);
@@ -34 +35 @@
         @Override
         public void spawnObject(String reference, String className) throws Exception {
+                System.out.println("SPAWN broker = " + this.getUID());
                 RemoteObject remote = (RemoteObject) Class.forName(className).newInstance();
                 getBroker().bind(reference, remote);
@@ -50 +52 @@
         @Override
         public HasObject hasObjectInfo(String reference) throws RetryException {
-                System.out.println("Hola soc un broker"+ getRef() + ", "+getUID()+ ", fil: "+Thread.currentThread().getId());
+                System.out.println("Hola soc un broker" + getRef() + ", " + getUID() + ", fil: " + Thread.currentThread().getId());
                 if (getBroker().getRemoteObjs().containsKey(reference)) {
                         RemoteObject r = getBroker().getRemoteObjs().get(reference);

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

@@ -1 @@
-package omq.common.broker;
+package omq.supervisor;
+
+import java.io.IOException;
+import java.util.Properties;
 
 import omq.Remote;
+import omq.client.annotation.AsyncMethod;
 import omq.client.annotation.MultiMethod;
 import omq.client.annotation.SyncMethod;
+import omq.common.broker.HasObject;
+import omq.exception.RemoteException;
 import omq.exception.RetryException;
 
-public interface RemoteMultiBroker extends Remote {
+public interface RemoteBroker extends Remote {
+
+        @AsyncMethod
+        public void spawnObject(String reference, String className, Properties env) throws Exception;
+
+        @AsyncMethod
+        public void spawnObject(String reference, String className) throws Exception;
+
+        @AsyncMethod
+        public void deleteObject(String reference) throws RemoteException, IOException;
 
         @MultiMethod

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

@@ -1 +1 @@
 package omq.supervisor;
 
-import java.lang.reflect.Proxy;
 import java.util.HashMap;
+import java.util.HashSet;
 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;
@@ -29 +26 @@
         private long sleep;
         private Map<String, OmqSettings> objectSettings;
-        private RemoteMultiBroker multiBroker;
-        private Map<String, RemoteBroker> brokerMap;
+        private RemoteBroker broker;
+        private Set<String> brokers;
 
         public SupervisorImpl(String brokerSet, long sleep) {
                 this.brokerSet = brokerSet;
                 this.sleep = sleep;
-                brokerMap = new HashMap<String, RemoteBroker>();
+                brokers = new HashSet<String>();
                 objectSettings = new HashMap<String, OmqSettings>();
         }
@@ -42 +39 @@
         public void run() {
                 try {
-                        multiBroker = getBroker().lookup(brokerSet, RemoteMultiBroker.class);
+                        broker = getBroker().lookup(brokerSet, RemoteBroker.class);
                         while (true) {
                                 try {
@@ -68 +65 @@
         @Override
         public void subscribe(String brokerSet, String brokerName) throws Exception {
-                if (brokerSet.equals(brokerSet) && !brokerMap.containsKey(brokerName)) {
+                if (brokerSet.equals(brokerSet) && !brokers.contains(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);
+                        brokers.add(brokerName);
                 } else {
                         throw new Exception("blablabla");
@@ -89 +80 @@
                         throw new Exception("JAJAJAJAJA");
                 }
-                
-                HasObject[] hasList = multiBroker.hasObjectInfo(reference);
+
+                HasObject[] hasList = broker.hasObjectInfo(reference);
 
                 int minObjects = settings.getMinNumberObjects();
@@ -104 +95 @@
 
                 int i = 0;
-                while (numObjects <= minObjects && i < numBrokers) {
+                while (numObjects < minObjects && i < numBrokers) {
                         HasObject h = hasList[i++];
-                        if (h.hasObject()) {
-                                brokerMap.get(h.getBrokerName()).spawnObject(reference, settings.getClassName());
+                        if (!h.hasObject()) {
+                                // Use a single broker
+                                broker.setUID(h.getBrokerName());
+                                broker.spawnObject(reference, settings.getClassName());
+                                // Remove the UID
+                                broker.setUID(null);
                                 numObjects++;
                         }
                 }
+
+                // Once there are some objects created, put objectSettings
                 objectSettings.put(reference, settings);
 
@@ -117 +114 @@
         @Override
         public void spawnObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception {
-
+                System.out.println("FUNCIO SPAWN");
                 String reference = settings.getReference();
 
@@ -124 +121 @@
                 }
 
+                int numBrokers = hasList.length;
+
+                if (numObjects < numBrokers) {
+                        for (HasObject h : hasList) {
+                                if (!h.hasObject()) {
+                                        // Use a single broker
+                                        broker.setUID(h.getBrokerName());
+                                        broker.spawnObject(reference, settings.getClassName());
+                                        // Remove the UID
+                                        broker.setUID(null);
+                                        break;
+                                }
+                        }
+                }
+        }
+
+        @Override
+        public void unbindObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception {
+                System.out.println("FUNCIO UNBIND");
+
+                String reference = settings.getReference();
+
                 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) {
+                if (numObjects > minObjects) {
+                        for (HasObject h : hasList) {
+                                if (h.hasObject()) {
+                                        // Use a single broker
+                                        broker.setUID(h.getBrokerName());
+                                        broker.deleteObject(reference);
+                                        // Remove the UID
+                                        broker.setUID(null);
                                         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;
-                // }
-                // }
-
         }
 
@@ -166 +172 @@
                 int numMessages = dok.getMessageCount();
 
-                HasObject[] hasList = multiBroker.hasObjectInfo(reference);
+                HasObject[] hasList = broker.hasObjectInfo(reference);
                 for (HasObject h : hasList) {
                         if (h.hasObject()) {
@@ -177 +183 @@
                 if (maxMessages < numMessages || numObjects < minObjects) {
                         logger.info("SPAWN TIME!!");
+                        System.out.println("SPAAAAAAAAAAAAAAAAAAAAAAAAAAWN TIME!!!!!");
                         spawnObject(settings, hasList, numObjects);
                 } else if (numMessages < minMessages && minObjects < numObjects) {
                         logger.info("Unbinding object!!!");
+                        System.out.println("UNBINDIN OOOOOOOOOOOOOOOOOOBJECT!!!");
                         unbindObject(settings, hasList, numObjects);
                 }
@@ -200 +208 @@
         }
 
-        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="DEBUG" />
+        <param name="Threshold" value="INFO" />
         <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 ="DEBUG" /> 
+    <priority value ="INFO" /> 
     <appender-ref ref="consoleAppender" />
     <appender-ref ref="A2" />