Index: branches/supervisor/src/main/java/omq/common/broker/HasObject.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/HasObject.java (revision 102) +++ branches/supervisor/src/main/java/omq/common/broker/HasObject.java (revision 103) @@ -37,5 +37,5 @@ } - public boolean isHasObject() { + public boolean hasObject() { return hasObject; } Index: branches/supervisor/src/main/java/omq/common/broker/RemoteBroker.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/RemoteBroker.java (revision 102) +++ branches/supervisor/src/main/java/omq/common/broker/RemoteBroker.java (revision 103) @@ -8,4 +8,5 @@ import omq.client.annotation.SyncMethod; import omq.exception.RemoteException; +import omq.exception.RetryException; public interface RemoteBroker extends Remote { @@ -19,5 +20,8 @@ @SyncMethod(retry = 1, timeout = 1000) - public boolean hasObject(String reference); + 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 102) +++ branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java (revision 103) @@ -6,4 +6,5 @@ import omq.exception.RemoteException; +import omq.exception.RetryException; import omq.server.RemoteObject; @@ -43,7 +44,17 @@ @Override - public boolean hasObject(String reference) { + public boolean hasObject(String reference) throws RetryException { return getBroker().getRemoteObjs().containsKey(reference); } + @Override + public HasObject hasObjectInfo(String reference) throws RetryException { + 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.getRef(), reference, false, 0); + } + } Index: branches/supervisor/src/main/java/omq/supervisor/Supervisor.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/Supervisor.java (revision 102) +++ branches/supervisor/src/main/java/omq/supervisor/Supervisor.java (revision 103) @@ -7,7 +7,7 @@ public void subscribe(String brokerName) throws Exception; - public void spawnObject(OmqSettings settings) throws Exception; + public void spawnObject(OmqSettings settings, int numObjects) throws Exception; - public void unbindObject(OmqSettings settings) throws Exception; + public void unbindObject(OmqSettings settings, int numObjects) throws Exception; } Index: branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 102) +++ branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 103) @@ -10,7 +10,4 @@ import org.apache.log4j.Logger; - -import com.rabbitmq.client.AMQP.Queue.DeclareOk; -import com.rabbitmq.client.Channel; public class SupervisorImpl extends RemoteObject implements Supervisor { @@ -41,5 +38,5 @@ @Override - public void spawnObject(OmqSettings settings) throws Exception { + public void spawnObject(OmqSettings settings, int numObjects) throws Exception { String reference = settings.getReference(); @@ -50,15 +47,4 @@ int minObjects = settings.getMinNumberObjects(); - - Channel channel = getBroker().getNewChannel(); - - int numObjects = 0; - try { - DeclareOk dok = channel.queueDeclarePassive(reference); - numObjects = dok.getConsumerCount(); - channel.close(); - } catch (Exception io) { - // The queue doesn't exist & the channel has been closed - } for (RemoteBroker broker : brokers) { @@ -75,27 +61,16 @@ @Override - public void unbindObject(OmqSettings settings) throws Exception { + public void unbindObject(OmqSettings settings, int numObjects) throws Exception { String reference = settings.getReference(); int minObjects = settings.getMinNumberObjects(); - Channel channel = getBroker().getNewChannel(); + for (RemoteBroker broker : brokers) { + if (broker.hasObject(reference) && (numObjects - 1) >= minObjects) { + broker.deleteObject(reference); + break; + } + } - int numObjects = 0; - try { - DeclareOk dok = channel.queueDeclarePassive(reference); - numObjects = dok.getConsumerCount(); - channel.close(); - - for (RemoteBroker broker : brokers) { - if (broker.hasObject(reference) && (numObjects - 1) >= minObjects) { - broker.deleteObject(reference); - break; - } - } - - } catch (Exception io) { - // The queue doesn't exist & the channel has been closed - } } Index: branches/supervisor/src/main/java/omq/supervisor/SupervisorThread.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/SupervisorThread.java (revision 102) +++ branches/supervisor/src/main/java/omq/supervisor/SupervisorThread.java (revision 103) @@ -3,4 +3,7 @@ import java.util.Map; import java.util.Set; + +import omq.common.broker.RemoteBroker; +import omq.exception.RetryException; import org.apache.log4j.Logger; @@ -49,20 +52,33 @@ int minMessages = settings.getMinNumQueued(); - Channel channel = supervisor.getBroker().getChannel(); DeclareOk dok = channel.queueDeclarePassive(reference); - int numConsumers = dok.getConsumerCount(); + int numObjects = getNumObjects(reference); int numMessages = dok.getMessageCount(); - System.out.println("Num Consumers: " + numConsumers + ", num Messages: " + numMessages); + System.out.println("Num Consumers: " + numObjects + ", num Messages: " + numMessages); - if (maxMessages < numMessages || numConsumers < minObjects) { + if (maxMessages < numMessages || numObjects < minObjects) { logger.info("SPAWN TIME!!"); - supervisor.spawnObject(settings); - } else if (numMessages < minMessages && minObjects < numConsumers) { + supervisor.spawnObject(settings, numObjects); + } else if (numMessages < minMessages && minObjects < numObjects) { logger.info("Unbinding object!!!"); - supervisor.unbindObject(settings); + supervisor.unbindObject(settings, numObjects); } } + + private int getNumObjects(String reference) { + int num = 0; + for (RemoteBroker broker : supervisor.getBrokers()) { + try { + if (broker.hasObject(reference)) { + num++; + } + } catch (RetryException e) { + e.printStackTrace(); + } + } + return num; + } } Index: branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java =================================================================== --- branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java (revision 102) +++ branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java (revision 103) @@ -63,5 +63,5 @@ OmqSettings settings = new OmqSettings("sleep", SleepImpl.class.getName(), env, 1, 20, 20); - supervisor.spawnObject(settings); + supervisor.spawnObject(settings, 0); Sleep sleep = broker.lookup("sleep", Sleep.class); Index: branches/supervisor/target/classes/log4j.xml =================================================================== --- branches/supervisor/target/classes/log4j.xml (revision 102) +++ (revision ) @@ -1,37 +1,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -