Index: branches/supervisor/src/main/java/omq/Remote.java =================================================================== --- branches/supervisor/src/main/java/omq/Remote.java (revision 106) +++ branches/supervisor/src/main/java/omq/Remote.java (revision 107) @@ -22,3 +22,7 @@ */ public String getRef(); + + public String getUID(); + + public void setUID(String uID); } Index: branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java (revision 106) +++ branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java (revision 107) @@ -47,4 +47,5 @@ private String reference; + private String UID; private transient String exchange; private transient String multiExchange; @@ -72,9 +73,10 @@ * 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,4 +121,11 @@ 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,14 +177,17 @@ } + // 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,3 +374,13 @@ } + @Override + public String getUID() { + return UID; + } + + @Override + public void setUID(String uID) { + this.UID = uID; + } + } Index: branches/supervisor/src/main/java/omq/common/broker/RemoteBroker.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/RemoteBroker.java (revision 106) +++ branches/supervisor/src/main/java/omq/common/broker/RemoteBroker.java (revision 107) @@ -6,5 +6,4 @@ import omq.Remote; -import omq.client.annotation.SyncMethod; import omq.exception.RemoteException; import omq.exception.RetryException; @@ -19,8 +18,6 @@ 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; Index: branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java (revision 106) +++ branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java (revision 107) @@ -50,10 +50,11 @@ @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); } Index: branches/supervisor/src/main/java/omq/common/broker/RemoteMultiBroker.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/RemoteMultiBroker.java (revision 106) +++ branches/supervisor/src/main/java/omq/common/broker/RemoteMultiBroker.java (revision 107) @@ -1,5 +1,3 @@ package omq.common.broker; - -import java.util.Set; import omq.Remote; @@ -9,7 +7,4 @@ public interface RemoteMultiBroker extends Remote { - @MultiMethod - @SyncMethod(retry = 1, timeout = 1000) - public Set getRemoteObjects(); @MultiMethod Index: branches/supervisor/src/main/java/omq/server/InvocationThread.java =================================================================== --- branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 106) +++ branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 107) @@ -172,8 +172,4 @@ 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,4 +238,5 @@ */ + // Disable Round Robin behavior boolean autoAck = false; Index: branches/supervisor/src/main/java/omq/server/MultiInvocationThread.java =================================================================== --- branches/supervisor/src/main/java/omq/server/MultiInvocationThread.java (revision 107) +++ branches/supervisor/src/main/java/omq/server/MultiInvocationThread.java (revision 107) @@ -0,0 +1,5 @@ +package omq.server; + +public class MultiInvocationThread { + +} Index: branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 106) +++ branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 107) @@ -1,10 +1,10 @@ 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,6 +15,6 @@ 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,13 +29,11 @@ private long sleep; private Map objectSettings; - // TODO: Set private RemoteMultiBroker multiBroker; private Map brokerMap; - private List brokers; public SupervisorImpl(String brokerSet, long sleep) { this.brokerSet = brokerSet; this.sleep = sleep; - brokers = new ArrayList(); + brokerMap = new HashMap(); objectSettings = new HashMap(); } @@ -72,5 +70,10 @@ 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,5 +89,28 @@ 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,14 +140,14 @@ @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,12 +192,4 @@ } - public List getBrokers() { - return brokers; - } - - public void setBrokers(List brokers) { - this.brokers = brokers; - } - public String getBrokerSet() { return brokerSet;