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; Index: branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java =================================================================== --- branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java (revision 106) +++ branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java (revision 107) @@ -25,6 +25,6 @@ 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,6 +37,6 @@ 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); Index: branches/supervisor/src/test/java/test/multi/Beer.java =================================================================== --- branches/supervisor/src/test/java/test/multi/Beer.java (revision 107) +++ branches/supervisor/src/test/java/test/multi/Beer.java (revision 107) @@ -0,0 +1,17 @@ +package test.multi; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.MultiMethod; +import omq.client.annotation.SyncMethod; + +public interface Beer extends Remote { + + @MultiMethod + @SyncMethod(retry = 1, timeout = 500) + public int getCl(); + + @AsyncMethod + public void setCl(int cl); +} + Index: branches/supervisor/src/test/java/test/multi/BeerImpl.java =================================================================== --- branches/supervisor/src/test/java/test/multi/BeerImpl.java (revision 107) +++ branches/supervisor/src/test/java/test/multi/BeerImpl.java (revision 107) @@ -0,0 +1,24 @@ +package test.multi; + +import omq.server.RemoteObject; + +public class BeerImpl extends RemoteObject implements Beer { + + /** + * + */ + private static final long serialVersionUID = 1L; + + private int cl; + + @Override + public int getCl() { + return cl; + } + + @Override + public void setCl(int cl) { + this.cl = cl; + } + +} Index: branches/supervisor/src/test/java/test/multi/BeerTest.java =================================================================== --- branches/supervisor/src/test/java/test/multi/BeerTest.java (revision 107) +++ branches/supervisor/src/test/java/test/multi/BeerTest.java (revision 107) @@ -0,0 +1,55 @@ +package test.multi; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class BeerTest { + + private static Broker broker; + private static ClientBeer beer; + + @BeforeClass + public static void server() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.RABBIT_PORT, "5672"); + + Broker b1 = new Broker(env); + Broker b2 = new Broker(env); + + b1.bind("beer", new BeerImpl()); + b2.bind("beer", new BeerImpl()); + System.out.println("Server started"); + } + + + @Test + public void test() throws Exception{ + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.RABBIT_PORT, "5672"); + broker = new Broker(env); + beer = broker.lookup("beer", ClientBeer.class); + + beer.setCl(25); + Thread.sleep(500); + beer.setCl(33); + Thread.sleep(500); + assertEquals(2, beer.getCl().length); + } +} Index: branches/supervisor/src/test/java/test/multi/ClientBeer.java =================================================================== --- branches/supervisor/src/test/java/test/multi/ClientBeer.java (revision 107) +++ branches/supervisor/src/test/java/test/multi/ClientBeer.java (revision 107) @@ -0,0 +1,15 @@ +package test.multi; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.MultiMethod; +import omq.client.annotation.SyncMethod; + +public interface ClientBeer extends Remote{ + @MultiMethod + @SyncMethod(retry = 1, timeout = 500) + public int[] getCl(); + + @AsyncMethod + public void setCl(int cl); +} Index: branches/supervisor/src/test/java/test/multiBroker/MultiBrokerTest.java =================================================================== --- branches/supervisor/src/test/java/test/multiBroker/MultiBrokerTest.java (revision 107) +++ branches/supervisor/src/test/java/test/multiBroker/MultiBrokerTest.java (revision 107) @@ -0,0 +1,58 @@ +package test.multiBroker; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.broker.RemoteBroker; +import omq.common.broker.RemoteBrokerImpl; +import omq.common.broker.RemoteMultiBroker; +import omq.common.util.ParameterQueue; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class MultiBrokerTest { + + private static Broker broker; + private static RemoteMultiBroker multiBroker; + private static RemoteBroker brokerProxy; + + @BeforeClass + public static void server() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.RABBIT_PORT, "5672"); + env.setProperty(ParameterQueue.MIN_POOL_THREADS, "1"); + env.setProperty(ParameterQueue.MAX_POOL_THREADS, "1"); + + Broker b1 = new Broker(env); + Broker b2 = new Broker(env); + + b1.bind("broker", "broker1", new RemoteBrokerImpl()); + b2.bind("broker", "broker2", new RemoteBrokerImpl()); + System.out.println("Server started"); + } + + + @Test + public void test() throws Exception{ + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.RABBIT_PORT, "5672"); + broker = new Broker(env); + multiBroker = broker.lookup("broker", RemoteMultiBroker.class); +// brokerProxy = broker.lookup("broker", RemoteBroker.class); + + assertEquals(2, multiBroker.hasObjectInfo("hola").length); + } +} Index: branches/supervisor/src/test/java/test/queues/Beer.java =================================================================== --- branches/supervisor/src/test/java/test/queues/Beer.java (revision 107) +++ branches/supervisor/src/test/java/test/queues/Beer.java (revision 107) @@ -0,0 +1,8 @@ +package test.queues; + +import omq.Remote; + +public interface Beer extends Remote{ + public int getCl(); + public void setCl(int cl); +} Index: branches/supervisor/src/test/java/test/queues/BeerImpl.java =================================================================== --- branches/supervisor/src/test/java/test/queues/BeerImpl.java (revision 107) +++ branches/supervisor/src/test/java/test/queues/BeerImpl.java (revision 107) @@ -0,0 +1,25 @@ +package test.queues; + +import omq.server.RemoteObject; + +public class BeerImpl extends RemoteObject implements Beer { + + /** + * + */ + private static final long serialVersionUID = 1L; + + private int cl; + + @Override + public int getCl() { + System.out.println("Hola soc una birra" + getRef() + ", " + getUID() + ", fil: " + Thread.currentThread().getId()); + return cl; + } + + @Override + public void setCl(int cl) { + this.cl = cl; + } + +} Index: branches/supervisor/src/test/java/test/queues/BeerTest.java =================================================================== --- branches/supervisor/src/test/java/test/queues/BeerTest.java (revision 107) +++ branches/supervisor/src/test/java/test/queues/BeerTest.java (revision 107) @@ -0,0 +1,60 @@ +package test.queues; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class BeerTest { + + private static Broker broker; + + @BeforeClass + public static void server() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.RABBIT_PORT, "5672"); + + Broker b1 = new Broker(env); + Broker b2 = new Broker(env); + + b1.bind("beer", "beer1", new BeerImpl()); + b2.bind("beer", "beer2", new BeerImpl()); + System.out.println("Server started"); + } + + @Test + public void test() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.RABBIT_PORT, "5672"); + broker = new Broker(env); + Beer beer1 = broker.lookup("beer", Beer.class); + beer1.setUID("beer1"); + Beer beer2 = broker.lookup("beer", Beer.class); + beer2.setUID("beer2"); + + beer1.setCl(25); + beer2.setCl(33); + + Thread.sleep(500); + + assertEquals(25, beer1.getCl()); + assertEquals(33, beer2.getCl()); + // Thread.sleep(500000); + + } +}