Changeset 107 for branches/supervisor/src

Timestamp:

10/19/13 13:20:44 (11 years ago)

Author:

stoda

Message:

Error detected: there should be only one thread listening to the multiexchange queue.
TODO: change this. Make refactor in the invocationthread. Change remotethreadpool to achieve this behavior.
AInvocationThread <- InvocationTHread

<- MultiInvocationTHread

Location:

branches/supervisor/src

Files:

• main/java/omq/Remote.java (modified) (1 diff)
• main/java/omq/client/proxy/Proxymq.java (modified) (5 diffs)
• main/java/omq/common/broker/RemoteBroker.java (modified) (2 diffs)
• main/java/omq/common/broker/RemoteBrokerImpl.java (modified) (1 diff)
• main/java/omq/common/broker/RemoteMultiBroker.java (modified) (2 diffs)
• main/java/omq/server/InvocationThread.java (modified) (2 diffs)
• main/java/omq/server/MultiInvocationThread.java (added)
• main/java/omq/supervisor/SupervisorImpl.java (modified) (7 diffs)
• test/java/omq/test/supervisor/SleepTest.java (modified) (2 diffs)
• test/java/test (added)
• test/java/test/multi (added)
• test/java/test/multi/Beer.java (added)
• test/java/test/multi/BeerImpl.java (added)
• test/java/test/multi/BeerTest.java (added)
• test/java/test/multi/ClientBeer.java (added)
• test/java/test/multiBroker (added)
• test/java/test/multiBroker/MultiBrokerTest.java (added)
• test/java/test/queues (added)
• test/java/test/queues/Beer.java (added)
• test/java/test/queues/BeerImpl.java (added)
• test/java/test/queues/BeerTest.java (added)

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

@@ -22 +22 @@
          */
         public String getRef();
+        
+        public String getUID();
+
+        public void setUID(String uID);
 }

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

@@ -47 +47 @@
 
         private String reference;
+        private String UID;
         private transient String exchange;
         private transient String multiExchange;
@@ -72 +73 @@
          * Proxymq Constructor.
          * 
-         * This constructor uses an reference to know which object will call. It also uses
-         * Properties to set where to send the messages
+         * This constructor uses an reference to know which object will call. It
+         * also uses Properties to set where to send the messages
          * 
          * @param reference
-         *            The reference represents the unique identifier of a remote object
+         *            The reference represents the unique identifier of a remote
+         *            object
          * @param clazz
          *            It represents the real class of the remote object. With this
@@ -119 +121 @@
                                 return getRef();
                         }
+                        if (methodName.equals("getUID")) {
+                                return getUID();
+                        }
+                        if (methodName.equals("setUID")) {
+                                setUID((String) arguments[0]);
+                                return null;
+                        }
                         if (methodName.equals("equals")) {
                                 if (arguments[0] instanceof Remote) {
@@ -168 +177 @@
                 }
 
+                // TODO look this carefully
+                String appId = UID == null ? reference : UID;
+
                 // Add the correlation ID and create a replyTo property
-                BasicProperties props = new BasicProperties.Builder().appId(reference).correlationId(corrId).replyTo(replyQueueName).type(serializerType)
-                                .deliveryMode(deliveryMode).build();
+                BasicProperties props = new BasicProperties.Builder().appId(appId).correlationId(corrId).replyTo(replyQueueName)
+                                .type(serializerType).deliveryMode(deliveryMode).build();
 
                 // Publish the message
                 byte[] bytesRequest = serializer.serialize(serializerType, request);
                 broker.publishMessge(exchange, routingkey, props, bytesRequest);
-                logger.debug("Proxymq: " + reference + " invokes '" + request.getMethod() + "' , corrID: " + corrId + ", exchange: " + exchange + ", replyQueue: "
-                                + replyQueueName + ", serializerType: " + serializerType + ", multi call: " + request.isMulti() + ", async call: " + request.isAsync()
-                                + ", delivery mode: " + deliveryMode);
+                logger.debug("Proxymq: " + reference + " invokes '" + request.getMethod() + "' , corrID: " + corrId + ", exchange: " + exchange
+                                + ", replyQueue: " + replyQueueName + ", serializerType: " + serializerType + ", multi call: " + request.isMulti()
+                                + ", async call: " + request.isAsync() + ", delivery mode: " + deliveryMode);
         }
 
@@ -362 +374 @@
         }
 
+        @Override
+        public String getUID() {
+                return UID;
+        }
+
+        @Override
+        public void setUID(String uID) {
+                this.UID = uID;
+        }
+
 }

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

@@ -6 +6 @@
 
 import omq.Remote;
-import omq.client.annotation.SyncMethod;
 import omq.exception.RemoteException;
 import omq.exception.RetryException;
@@ -19 +18 @@
         public void deleteObject(String reference) throws RemoteException, IOException;
 
-        @SyncMethod(retry = 1, timeout = 1000)
         public boolean hasObject(String reference) throws RetryException;
 
-        @SyncMethod(retry = 1, timeout = 1000)
         public HasObject hasObjectInfo(String reference) throws RetryException;
 

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

@@ -50 +50 @@
         @Override
         public HasObject hasObjectInfo(String reference) throws RetryException {
+                System.out.println("Hola soc un broker"+ getRef() + ", "+getUID()+ ", fil: "+Thread.currentThread().getId());
                 if (getBroker().getRemoteObjs().containsKey(reference)) {
                         RemoteObject r = getBroker().getRemoteObjs().get(reference);
                         int numThreads = r.getPool().getWorkers().size();
-                        return new HasObject(this.getRef(), reference, true, numThreads);
+                        return new HasObject(this.getUID(), reference, true, numThreads);
                 }
-                return new HasObject(this.getRef(), reference, false, 0);
+                return new HasObject(this.getUID(), reference, false, 0);
         }
 

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

@@ -1 +1 @@
 package omq.common.broker;
-
-import java.util.Set;
 
 import omq.Remote;
@@ -9 +7 @@
 
 public interface RemoteMultiBroker extends Remote {
-        @MultiMethod
-        @SyncMethod(retry = 1, timeout = 1000)
-        public Set<String> getRemoteObjects();
 
         @MultiMethod

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

@@ -172 +172 @@
                 channel = broker.getNewChannel();
 
-                /*
-                 * Default queue, Round Robin behaviour
-                 */
-
                 // Get info about which exchange and queue will use
                 String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE, "");
@@ -242 +238 @@
                  */
 
+                // Disable Round Robin behavior 
                 boolean autoAck = false;
 

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

@@ -1 +1 @@
 package omq.supervisor;
 
-import java.util.ArrayList;
+import java.lang.reflect.Proxy;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import omq.client.proxy.Proxymq;
 import omq.common.broker.HasObject;
 import omq.common.broker.RemoteBroker;
@@ -15 +15 @@
 import org.apache.log4j.Logger;
 
+import com.rabbitmq.client.AMQP.Queue.DeclareOk;
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.AMQP.Queue.DeclareOk;
 
 public class SupervisorImpl extends RemoteObject implements Supervisor, Runnable {
@@ -29 +29 @@
         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>();
+                brokerMap = new HashMap<String, RemoteBroker>();
                 objectSettings = new HashMap<String, OmqSettings>();
         }
@@ -72 +70 @@
                 if (brokerSet.equals(brokerSet) && !brokerMap.containsKey(brokerName)) {
                         logger.info("Broker " + brokerName + " subscrived");
-                        RemoteBroker broker = getBroker().lookup(brokerName, RemoteBroker.class);
+                        // 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);
                 } else {
@@ -86 +89 @@
                         throw new Exception("JAJAJAJAJA");
                 }
+                
+                HasObject[] hasList = multiBroker.hasObjectInfo(reference);
+
+                int minObjects = settings.getMinNumberObjects();
+                int numBrokers = hasList.length;
+                int numObjects = 0;
+
+                for (HasObject h : hasList) {
+                        if (h.hasObject()) {
+                                numObjects++;
+                        }
+                }
+                System.out.println("NumObjects " + numObjects + " numBrokers " + numBrokers);
+
+                int i = 0;
+                while (numObjects <= minObjects && i < numBrokers) {
+                        HasObject h = hasList[i++];
+                        if (h.hasObject()) {
+                                brokerMap.get(h.getBrokerName()).spawnObject(reference, settings.getClassName());
+                                numObjects++;
+                        }
+                }
                 objectSettings.put(reference, settings);
+
         }
 
@@ -114 +140 @@
         @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;
-                        }
-                }
+                // String reference = settings.getReference();
+                //
+                // int minObjects = settings.getMinNumberObjects();
+                //
+                // for (RemoteBroker broker : brokers) {
+                // if (broker.hasObject(reference) && (numObjects - 1) >= minObjects) {
+                // broker.deleteObject(reference);
+                // break;
+                // }
+                // }
 
         }
@@ -166 +192 @@
         }
 
-        public List<RemoteBroker> getBrokers() {
-                return brokers;
-        }
-
-        public void setBrokers(List<RemoteBroker> brokers) {
-                this.brokers = brokers;
-        }
-
         public String getBrokerSet() {
                 return brokerSet;

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

@@ -25 +25 @@
                 env1.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1");
                 env1.setProperty(ParameterQueue.RABBIT_PORT, "5672");
-                env1.setProperty(ParameterQueue.MIN_POOL_THREADS, "4");
-                env1.setProperty(ParameterQueue.MAX_POOL_THREADS, "4");
+                env1.setProperty(ParameterQueue.MIN_POOL_THREADS, "1");
+                env1.setProperty(ParameterQueue.MAX_POOL_THREADS, "1");
 
                 Broker broker = new Broker(env1);
@@ -37 +37 @@
                 env2.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1");
                 env2.setProperty(ParameterQueue.RABBIT_PORT, "5672");
-                env2.setProperty(ParameterQueue.MIN_POOL_THREADS, "4");
-                env2.setProperty(ParameterQueue.MAX_POOL_THREADS, "4");
+                env2.setProperty(ParameterQueue.MIN_POOL_THREADS, "1");
+                env2.setProperty(ParameterQueue.MAX_POOL_THREADS, "1");
 
                 Broker broker2 = new Broker(env2);