Index: /branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- /branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java (revision 109) +++ /branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java (revision 110) @@ -380,5 +380,5 @@ @Override - public void setUID(String uID) { + public void setUID(String uID) {System.out.println("SET UID"); this.UID = uID; } Index: /branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java =================================================================== --- /branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java (revision 109) +++ /branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java (revision 110) @@ -28,4 +28,5 @@ @Override public void spawnObject(String reference, String className, Properties env) throws Exception { + System.out.println("SPAWN broker = " + this.getUID()); RemoteObject remote = (RemoteObject) Class.forName(className).newInstance(); getBroker().bind(reference, remote, env); @@ -34,4 +35,5 @@ @Override public void spawnObject(String reference, String className) throws Exception { + System.out.println("SPAWN broker = " + this.getUID()); RemoteObject remote = (RemoteObject) Class.forName(className).newInstance(); getBroker().bind(reference, remote); @@ -50,5 +52,5 @@ @Override public HasObject hasObjectInfo(String reference) throws RetryException { - System.out.println("Hola soc un broker"+ getRef() + ", "+getUID()+ ", fil: "+Thread.currentThread().getId()); + System.out.println("Hola soc un broker" + getRef() + ", " + getUID() + ", fil: " + Thread.currentThread().getId()); if (getBroker().getRemoteObjs().containsKey(reference)) { RemoteObject r = getBroker().getRemoteObjs().get(reference); Index: anches/supervisor/src/main/java/omq/common/broker/RemoteMultiBroker.java =================================================================== --- /branches/supervisor/src/main/java/omq/common/broker/RemoteMultiBroker.java (revision 109) +++ (revision ) @@ -1,18 +1,0 @@ -package omq.common.broker; - -import omq.Remote; -import omq.client.annotation.MultiMethod; -import omq.client.annotation.SyncMethod; -import omq.exception.RetryException; - -public interface RemoteMultiBroker extends Remote { - - @MultiMethod - @SyncMethod(retry = 1, timeout = 1000) - public boolean[] hasObject(String reference) throws RetryException; - - @MultiMethod - @SyncMethod(retry = 1, timeout = 1000) - public HasObject[] hasObjectInfo(String reference) throws RetryException; - -} Index: /branches/supervisor/src/main/java/omq/supervisor/RemoteBroker.java =================================================================== --- /branches/supervisor/src/main/java/omq/supervisor/RemoteBroker.java (revision 110) +++ /branches/supervisor/src/main/java/omq/supervisor/RemoteBroker.java (revision 110) @@ -0,0 +1,33 @@ +package omq.supervisor; + +import java.io.IOException; +import java.util.Properties; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.MultiMethod; +import omq.client.annotation.SyncMethod; +import omq.common.broker.HasObject; +import omq.exception.RemoteException; +import omq.exception.RetryException; + +public interface RemoteBroker extends Remote { + + @AsyncMethod + public void spawnObject(String reference, String className, Properties env) throws Exception; + + @AsyncMethod + public void spawnObject(String reference, String className) throws Exception; + + @AsyncMethod + public void deleteObject(String reference) throws RemoteException, IOException; + + @MultiMethod + @SyncMethod(retry = 1, timeout = 1000) + public boolean[] hasObject(String reference) throws RetryException; + + @MultiMethod + @SyncMethod(retry = 1, timeout = 1000) + public HasObject[] hasObjectInfo(String reference) throws RetryException; + +} Index: /branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java =================================================================== --- /branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 109) +++ /branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 110) @@ -1,13 +1,10 @@ package omq.supervisor; -import java.lang.reflect.Proxy; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; import java.util.Set; -import omq.client.proxy.Proxymq; import omq.common.broker.HasObject; -import omq.common.broker.RemoteBroker; -import omq.common.broker.RemoteMultiBroker; import omq.exception.RemoteException; import omq.server.RemoteObject; @@ -29,11 +26,11 @@ private long sleep; private Map objectSettings; - private RemoteMultiBroker multiBroker; - private Map brokerMap; + private RemoteBroker broker; + private Set brokers; public SupervisorImpl(String brokerSet, long sleep) { this.brokerSet = brokerSet; this.sleep = sleep; - brokerMap = new HashMap(); + brokers = new HashSet(); objectSettings = new HashMap(); } @@ -42,5 +39,5 @@ public void run() { try { - multiBroker = getBroker().lookup(brokerSet, RemoteMultiBroker.class); + broker = getBroker().lookup(brokerSet, RemoteBroker.class); while (true) { try { @@ -68,13 +65,7 @@ @Override public void subscribe(String brokerSet, String brokerName) throws Exception { - if (brokerSet.equals(brokerSet) && !brokerMap.containsKey(brokerName)) { + if (brokerSet.equals(brokerSet) && !brokers.contains(brokerName)) { logger.info("Broker " + brokerName + " subscrived"); - // 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); + brokers.add(brokerName); } else { throw new Exception("blablabla"); @@ -89,6 +80,6 @@ throw new Exception("JAJAJAJAJA"); } - - HasObject[] hasList = multiBroker.hasObjectInfo(reference); + + HasObject[] hasList = broker.hasObjectInfo(reference); int minObjects = settings.getMinNumberObjects(); @@ -104,11 +95,17 @@ int i = 0; - while (numObjects <= minObjects && i < numBrokers) { + while (numObjects < minObjects && i < numBrokers) { HasObject h = hasList[i++]; - if (h.hasObject()) { - brokerMap.get(h.getBrokerName()).spawnObject(reference, settings.getClassName()); + if (!h.hasObject()) { + // Use a single broker + broker.setUID(h.getBrokerName()); + broker.spawnObject(reference, settings.getClassName()); + // Remove the UID + broker.setUID(null); numObjects++; } } + + // Once there are some objects created, put objectSettings objectSettings.put(reference, settings); @@ -117,5 +114,5 @@ @Override public void spawnObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception { - + System.out.println("FUNCIO SPAWN"); String reference = settings.getReference(); @@ -124,31 +121,40 @@ } + int numBrokers = hasList.length; + + if (numObjects < numBrokers) { + for (HasObject h : hasList) { + if (!h.hasObject()) { + // Use a single broker + broker.setUID(h.getBrokerName()); + broker.spawnObject(reference, settings.getClassName()); + // Remove the UID + broker.setUID(null); + break; + } + } + } + } + + @Override + public void unbindObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception { + System.out.println("FUNCIO UNBIND"); + + String reference = settings.getReference(); + int minObjects = settings.getMinNumberObjects(); - for (HasObject h : hasList) { - if (h.hasObject() && minObjects >= numObjects) { - brokerMap.get(h.getBrokerName()).spawnObject(reference, settings.getClassName(), settings.getProps()); - numObjects++; - if (minObjects >= numObjects) { + if (numObjects > minObjects) { + for (HasObject h : hasList) { + if (h.hasObject()) { + // Use a single broker + broker.setUID(h.getBrokerName()); + broker.deleteObject(reference); + // Remove the UID + broker.setUID(null); break; } } } - - } - - @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; - // } - // } - } @@ -166,5 +172,5 @@ int numMessages = dok.getMessageCount(); - HasObject[] hasList = multiBroker.hasObjectInfo(reference); + HasObject[] hasList = broker.hasObjectInfo(reference); for (HasObject h : hasList) { if (h.hasObject()) { @@ -177,7 +183,9 @@ if (maxMessages < numMessages || numObjects < minObjects) { logger.info("SPAWN TIME!!"); + System.out.println("SPAAAAAAAAAAAAAAAAAAAAAAAAAAWN TIME!!!!!"); spawnObject(settings, hasList, numObjects); } else if (numMessages < minMessages && minObjects < numObjects) { logger.info("Unbinding object!!!"); + System.out.println("UNBINDIN OOOOOOOOOOOOOOOOOOBJECT!!!"); unbindObject(settings, hasList, numObjects); } @@ -200,11 +208,3 @@ } - public RemoteMultiBroker getMultiBroker() { - return multiBroker; - } - - public void setMultiBroker(RemoteMultiBroker multiBroker) { - this.multiBroker = multiBroker; - } - } Index: /branches/supervisor/src/main/resources/log4j.xml =================================================================== --- /branches/supervisor/src/main/resources/log4j.xml (revision 109) +++ /branches/supervisor/src/main/resources/log4j.xml (revision 110) @@ -4,5 +4,5 @@ - + @@ -30,5 +30,5 @@ - + Index: /branches/supervisor/target/classes/log4j.xml =================================================================== --- /branches/supervisor/target/classes/log4j.xml (revision 110) +++ /branches/supervisor/target/classes/log4j.xml (revision 110) @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Index: /branches/supervisor/target/classes/log4j.xml.svntmp =================================================================== --- /branches/supervisor/target/classes/log4j.xml.svntmp (revision 110) +++ /branches/supervisor/target/classes/log4j.xml.svntmp (revision 110) @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +