Changeset 92

Timestamp:

10/01/13 12:02:41 (11 years ago)

Author:

stoda

Message:

TODO: delete in supervisor
check the code

Location:

branches/supervisor

Files:

• src/main/java/omq/client/proxy/Proxymq.java (modified) (1 diff)
• src/main/java/omq/common/broker/Broker.java (modified) (4 diffs)
• src/main/java/omq/common/broker/RemoteBroker.java (modified) (2 diffs)
• src/main/java/omq/common/broker/RemoteBrokerImpl.java (modified) (2 diffs)
• src/main/java/omq/server/RemoteObject.java (modified) (3 diffs)
• src/main/java/omq/supervisor/ISupervisor.java (deleted)
• src/main/java/omq/supervisor/MultiBroker.java (deleted)
• src/main/java/omq/supervisor/OmqSettings.java (modified) (1 diff)
• src/main/java/omq/supervisor/Supervisor.java (modified) (1 diff)
• src/main/java/omq/supervisor/SupervisorImpl.java (added)
• src/main/java/omq/supervisor/SupervisorThread.java (added)
• src/test/java/omq/test/lock/SleepTest.java (modified) (1 diff)
• src/test/java/omq/test/supervisor (added)
• src/test/java/omq/test/supervisor/Sleep.java (added)
• src/test/java/omq/test/supervisor/SleepImpl.java (added)
• src/test/java/omq/test/supervisor/SleepTest.java (added)
• target/classes/log4j.xml (added)

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

@@ -116 +116 @@
                         if (methodName.equals("getRef")) {
                                 return getRef();
+                        }
+                        if (methodName.equals("equals")) {
+                                if (arguments[0] instanceof Remote) {
+                                        return getRef().equals(((Remote) arguments[0]).getRef());
+                                } else {
+                                        return false;
+                                }
                         }
                 }

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

@@ -20 +20 @@
 import omq.exception.RemoteException;
 import omq.server.RemoteObject;
+import omq.supervisor.Supervisor;
 
 import org.apache.log4j.Logger;
@@ -49 +50 @@
         private boolean connectionClosed = false;
         private Properties environment = null;
-        private RemoteBrokerImpl remoteBrokerImpl;
         private Map<String, RemoteObject> remoteObjs;
         private Map<String, Object> proxies = new Hashtable<String, Object>();
         private Map<String, Object> multiProxies = new Hashtable<String, Object>();
+
+        // Supervisor
+        private Supervisor supervisor;
 
         public Broker(Properties env) throws Exception {
@@ -358 +361 @@
         }
 
-        public void setSupervisor(String brokerSet, String brokerName) throws Exception {
-                remoteBrokerImpl = new RemoteBrokerImpl();
-                remoteBrokerImpl.startRemoteBroker(brokerSet, brokerName, this, getEnvironment());
-        }
-
         public Properties getEnvironment() {
                 return environment;
@@ -379 +377 @@
         }
 
+        /*
+         * Supervisor
+         */
+        public void setSupervisor(String supervisorName, String brokerName) throws Exception {
+                // Create a RemoteBrokerImpl
+                bind(brokerName, new RemoteBrokerImpl());
+                // Subscribe broker
+                supervisor = lookup(supervisorName, Supervisor.class);
+                supervisor.subscribe(brokerName);
+                logger.info("Supervisor set: " + supervisorName + ", BrokerName: " + brokerName);
+        }
+
+        public Supervisor getSupervisor() {
+                return supervisor;
+        }
 }

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

@@ -2 +2 @@
 
 import java.io.IOException;
+import java.util.Properties;
 import java.util.Set;
 
 import omq.Remote;
+import omq.client.annotation.SyncMethod;
 import omq.exception.RemoteException;
 
@@ -10 +12 @@
         public Set<String> getRemoteObjects();
 
-        public void spawnObject(String reference, String className, Class<?> parameterTypes, Object... args) throws Exception;
+        public void spawnObject(String reference, String className, Properties env) throws Exception;
+
+        public void spawnObject(String reference, String className) throws Exception;
 
         public void deleteObject(String reference) throws RemoteException, IOException;
 
-        public HasObject hasObject(String reference);
+        @SyncMethod(retry = 1, timeout = 1000)
+        public boolean hasObject(String reference);
 
 }

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

@@ -2 +2 @@
 
 import java.io.IOException;
-import java.lang.reflect.Method;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
 import java.util.Properties;
 import java.util.Set;
 
-import omq.common.util.ParameterQueue;
 import omq.exception.RemoteException;
 import omq.server.RemoteObject;
-import omq.server.RemoteWrapper;
-
-import com.rabbitmq.client.QueueingConsumer;
 
 /**
@@ -28 +20 @@
         private static final long serialVersionUID = 1L;
 
-        // fanout broker
-        private String brokerSet;
-        // id broker
-        private String brokerName;
-
-        public void startRemoteBroker(String brokerSet, String brokerName, Broker broker, Properties env) throws Exception {
-                this.broker = broker;
-                this.UID = brokerName;
-                this.env = env;
-                this.brokerSet = brokerSet;
-                this.brokerName = brokerName;
-
-                this.params = new HashMap<String, List<Class<?>>>();
-                for (Method m : this.getClass().getMethods()) {
-                        List<Class<?>> list = new ArrayList<Class<?>>();
-                        for (Class<?> clazz : m.getParameterTypes()) {
-                                list.add(clazz);
-                        }
-                        this.params.put(m.getName(), list);
-                }
-
-                // Get num threads to use
-                int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1"));
-                this.remoteWrapper = new RemoteWrapper(this, numThreads, broker.getSerializer());
-
-                startQueues();
-
-                // Start this listener
-                this.start();
-        }
-
-        private void startQueues() throws Exception {
-                /*
-                 * Unique queue
-                 */
-                channel.exchangeDeclare(brokerSet, "direct");
-                channel.queueDeclare(brokerName, false, true, true, null);
-                channel.queueBind(brokerName, brokerSet, brokerName);
-
-                /*
-                 * Multi queue
-                 */
-                channel.exchangeDeclare("multi#" + brokerSet, "fanout");
-                channel.queueDeclare("multi#" + brokerName, false, true, true, null);
-                channel.queueBind("multi#" + brokerName, "multi#" + brokerSet, "");
-
-                /*
-                 * Consumer
-                 */
-
-                consumer = new QueueingConsumer(channel);
-                channel.basicConsume(brokerName, true, consumer);
-                channel.basicConsume(brokerName + "#multi", true, consumer);
+        @Override
+        public Set<String> getRemoteObjects() {
+                return getBroker().getRemoteObjs().keySet();
         }
 
         @Override
-        public Set<String> getRemoteObjects() {
-                return this.broker.getRemoteObjs().keySet();
+        public void spawnObject(String reference, String className, Properties env) throws Exception {
+                RemoteObject remote = (RemoteObject) Class.forName(className).newInstance();
+                getBroker().bind(reference, remote, env);
         }
 
         @Override
-        public void spawnObject(String reference, String className, Class<?> parameterTypes, Object... args) throws Exception {
-                RemoteObject remote = (RemoteObject) Class.forName(className).getConstructor(parameterTypes).newInstance(args);
-                this.broker.bind(reference, remote);
+        public void spawnObject(String reference, String className) throws Exception {
+                RemoteObject remote = (RemoteObject) Class.forName(className).newInstance();
+                getBroker().bind(reference, remote);
         }
 
         @Override
         public void deleteObject(String reference) throws RemoteException, IOException {
-                this.broker.unbind(reference);
+                getBroker().unbind(reference);
         }
 
         @Override
-        public HasObject hasObject(String reference) {
-                boolean hasIt = this.broker.getRemoteObjs().containsKey(reference);
-                return new HasObject(this.brokerName, reference, hasIt);
+        public boolean hasObject(String reference) {
+                return getBroker().getRemoteObjs().containsKey(reference);
         }
 

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

@@ -38 +38 @@
         private static final Logger logger = Logger.getLogger(RemoteObject.class.getName());
 
-        protected String UID;
-        protected Properties env;
-        protected transient Broker broker;
-        protected transient String multiQueue;
-        protected transient RemoteWrapper remoteWrapper;
-        protected transient Map<String, List<Class<?>>> params;
-        protected transient Channel channel;
-        protected transient QueueingConsumer consumer;
-        protected transient boolean killed = false;
+        private String UID;
+        private Properties env;
+        private transient Broker broker;
+        private transient String multiQueue;
+        private transient RemoteWrapper remoteWrapper;
+        private transient Map<String, List<Class<?>>> params;
+        private transient Channel channel;
+        private transient QueueingConsumer consumer;
+        private transient boolean killed = false;
 
         private static final Map<String, Class<?>> primitiveClasses = new HashMap<String, Class<?>>();
@@ -254 +254 @@
         }
 
+        public Broker getBroker() {
+                return broker;
+        }
+
         /**
          * This method starts the queues using the information got in the
@@ -319 +323 @@
                 boolean autoAck = false;
 
+                //TODO see if this is useless
+                int prefetchCount = 1;
+                channel.basicQos(prefetchCount);
+
                 // Declare a new consumer
                 consumer = new QueueingConsumer(channel);

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

@@ -1 +1 @@
 package omq.supervisor;
 
+import java.io.Serializable;
 import java.util.Properties;
 
-public class OmqSettings {
+public class OmqSettings implements Serializable {
+        
+        
 
+        /**
+         * 
+         */
+        private static final long serialVersionUID = 1L;
         private String reference;
         private String className;

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

@@ -1 +1 @@
 package omq.supervisor;
 
-import java.util.Map;
-import java.util.Set;
+import omq.Remote;
 
-import omq.common.broker.RemoteBroker;
+public interface Supervisor extends Remote {
 
-public class Supervisor {
+        public void subscribe(String brokerName) throws Exception;
 
-        private Set<String> bindReferences;
-        private Map<String, RemoteBroker> brokers;
+        public void spawnObject(OmqSettings settings) throws Exception;
 
-        private void checkObject() {
-                String reference = null;
-
-                int numObjects = 0;
-
-                if(minObjects > numObjects || maxMessages < numEncuats){
-                        spawn:
-                                pregunta a tots i qui no té l'objecte li poses
-                }else if(numEncuats < minMessages && minObjects > numObjects){
-                        delete:
-                                pregunta a tots i qui té l'objecte li treus
-                }
-        }
 }

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

@@ -20 +20 @@
                 env.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1");
                 env.setProperty(ParameterQueue.RABBIT_PORT, "5672");
-                env.setProperty(ParameterQueue.NUM_THREADS, "2");
+                env.setProperty(ParameterQueue.NUM_THREADS, "1");
 
-                SleepImpl sleep = new SleepImpl();
+                for (int i = 0; i < 4; i++) {
+                        Broker broker = new Broker(env);
+                        broker.bind("sleep", new SleepImpl());
+                }
 
-                Broker broker = new Broker(env);
-                broker.bind("sleep", sleep);
         }