Changeset 92 for branches/supervisor/src/main/java/omq/common/broker

Timestamp:

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

Author:

stoda

Message:

TODO: delete in supervisor
check the code

Location:

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

Files:

• Broker.java (modified) (4 diffs)
• RemoteBroker.java (modified) (2 diffs)
• RemoteBrokerImpl.java (modified) (2 diffs)

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