Index: /branches/supervisor/src/main/java/omq/common/broker/Broker.java =================================================================== --- /branches/supervisor/src/main/java/omq/common/broker/Broker.java (revision 105) +++ /branches/supervisor/src/main/java/omq/common/broker/Broker.java (revision 106) @@ -240,4 +240,8 @@ } + public void bind(String reference, String UID, RemoteObject remote) throws RemoteException, AlreadyBoundException { + bind(reference, UID, remote, environment); + } + /** * Binds the reference to the specified remote object. This function uses @@ -264,4 +268,17 @@ try { remote.startRemoteObject(reference, this, env); + remoteObjs.put(reference, remote); + } catch (Exception e) { + throw new RemoteException(e); + } + } + + public void bind(String reference, String UID, RemoteObject remote, Properties env) throws RemoteException, AlreadyBoundException { + if (remoteObjs.containsKey(reference)) { + throw new AlreadyBoundException(reference); + } + // Try to start the remtoeObject listeners + try { + remote.startRemoteObject(reference, UID, this, env); remoteObjs.put(reference, remote); } catch (Exception e) { @@ -400,11 +417,11 @@ * Supervisor */ - public void setSupervisor(String supervisorName, String brokerName) throws Exception { + public void setSupervisor(String supervisorName, String brokerSet, String brokerName) throws Exception { // Create a RemoteBrokerImpl - bind(brokerName, new RemoteBrokerImpl()); + bind(brokerSet, brokerName, new RemoteBrokerImpl()); // Subscribe broker supervisor = lookup(supervisorName, Supervisor.class); - supervisor.subscribe(brokerName); - logger.info("Supervisor set: " + supervisorName + ", BrokerName: " + brokerName); + supervisor.subscribe(brokerSet, brokerName); + logger.info("Supervisor set: " + supervisorName + ", BrokerSet: " + brokerSet + ", BrokerName: " + brokerName); } 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 106) @@ -0,0 +1,23 @@ +package omq.common.broker; + +import java.util.Set; + +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 Set getRemoteObjects(); + + @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/server/InvocationThread.java =================================================================== --- /branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 105) +++ /branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 106) @@ -36,4 +36,5 @@ private RemoteObject obj; private String reference; + private String UID; private Properties env; private boolean idle; @@ -54,4 +55,5 @@ public InvocationThread(RemoteObject obj) throws Exception { this.obj = obj; + this.UID = obj.getUID(); this.reference = obj.getRef(); this.env = obj.getEnv(); @@ -87,5 +89,5 @@ String serializerType = delivery.getProperties().getType(); - // Deserialize the json + // Deserialize the request Request request = serializer.deserializeRequest(serializerType, delivery.getBody(), obj); String methodName = request.getMethod(); @@ -196,4 +198,20 @@ /* + * UID queue + */ + + if (UID != null) { + + boolean uidDurable = false; + boolean uidExclusive = true; + boolean uidAutoDelete = true; + + channel.queueDeclare(UID, uidDurable, uidExclusive, uidAutoDelete, null); + if (!exchange.equalsIgnoreCase("")) { // Default exchange case + channel.queueBind(UID, exchange, UID); + } + } + + /* * Multi queue, exclusive per each instance */ @@ -233,4 +251,7 @@ channel.basicConsume(queue, autoAck, consumer); channel.basicConsume(multiQueue, autoAck, consumer); + if (UID != null) { + channel.basicConsume(UID, autoAck, consumer); + } } Index: /branches/supervisor/src/main/java/omq/server/RemoteObject.java =================================================================== --- /branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 105) +++ /branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 106) @@ -31,4 +31,5 @@ private String reference; + private String UID; private Properties env; private transient Broker broker; @@ -83,4 +84,9 @@ pool = new RemoteThreadPool(minPoolThreads, maxPoolThreads, refresh, keepAliveTime, maxMessagesPerThread, this, broker); pool.start(); + } + + public void startRemoteObject(String reference, String UID, Broker broker, Properties env) throws Exception { + this.UID = UID; + startRemoteObject(reference, broker, env); } @@ -209,3 +215,11 @@ } + public String getUID() { + return UID; + } + + public void setUID(String uID) { + UID = uID; + } + } Index: /branches/supervisor/src/main/java/omq/supervisor/Supervisor.java =================================================================== --- /branches/supervisor/src/main/java/omq/supervisor/Supervisor.java (revision 105) +++ /branches/supervisor/src/main/java/omq/supervisor/Supervisor.java (revision 106) @@ -2,12 +2,15 @@ import omq.Remote; +import omq.common.broker.HasObject; public interface Supervisor extends Remote { - public void subscribe(String brokerName) throws Exception; + public void subscribe(String brokerSet, String brokerName) throws Exception; - public void spawnObject(OmqSettings settings, int numObjects) throws Exception; + public void spawnObject(OmqSettings settings) throws Exception; - public void unbindObject(OmqSettings settings, int numObjects) throws Exception; + public void spawnObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception; + + public void unbindObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception; } Index: /branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java =================================================================== --- /branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 105) +++ /branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 106) @@ -5,11 +5,18 @@ import java.util.List; import java.util.Map; +import java.util.Set; +import omq.common.broker.HasObject; import omq.common.broker.RemoteBroker; +import omq.common.broker.RemoteMultiBroker; +import omq.exception.RemoteException; import omq.server.RemoteObject; import org.apache.log4j.Logger; -public class SupervisorImpl extends RemoteObject implements Supervisor { +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.AMQP.Queue.DeclareOk; + +public class SupervisorImpl extends RemoteObject implements Supervisor, Runnable { /** @@ -18,25 +25,70 @@ private static final long serialVersionUID = 1L; private static final Logger logger = Logger.getLogger(SupervisorImpl.class.getName()); - private SupervisorThread thread; + + private String brokerSet; + private long sleep; private Map objectSettings; // TODO: Set + private RemoteMultiBroker multiBroker; + private Map brokerMap; private List brokers; - public SupervisorImpl(long sleep) { + public SupervisorImpl(String brokerSet, long sleep) { + this.brokerSet = brokerSet; + this.sleep = sleep; brokers = new ArrayList(); objectSettings = new HashMap(); - thread = new SupervisorThread(this, sleep); - thread.start(); } @Override - public void subscribe(String brokerName) throws Exception { - logger.info("Broker " + brokerName + " subscrived"); - RemoteBroker broker = getBroker().lookup(brokerName, RemoteBroker.class); - brokers.add(broker); + public void run() { + try { + multiBroker = getBroker().lookup(brokerSet, RemoteMultiBroker.class); + while (true) { + try { + Set keys = objectSettings.keySet(); + for (String reference : keys) { + System.out.println("key = " + keys); + checkObject(reference); + } + } catch (Exception e) { + e.printStackTrace(); + } + + try { + Thread.sleep(sleep); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + } catch (RemoteException e1) { + // TODO Auto-generated catch block + e1.printStackTrace(); + } } @Override - public void spawnObject(OmqSettings settings, int numObjects) throws Exception { + public void subscribe(String brokerSet, String brokerName) throws Exception { + if (brokerSet.equals(brokerSet) && !brokerMap.containsKey(brokerName)) { + logger.info("Broker " + brokerName + " subscrived"); + RemoteBroker broker = getBroker().lookup(brokerName, RemoteBroker.class); + brokerMap.put(brokerSet, broker); + } else { + throw new Exception("blablabla"); + } + } + + @Override + public void spawnObject(OmqSettings settings) throws Exception { + String reference = settings.getReference(); + + if (objectSettings.containsKey(reference)) { + throw new Exception("JAJAJAJAJA"); + } + objectSettings.put(reference, settings); + } + + @Override + public void spawnObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception { String reference = settings.getReference(); @@ -48,7 +100,7 @@ int minObjects = settings.getMinNumberObjects(); - for (RemoteBroker broker : brokers) { - if (!broker.hasObject(reference) && minObjects >= numObjects) { - broker.spawnObject(reference, settings.getClassName(), settings.getProps()); + for (HasObject h : hasList) { + if (h.hasObject() && minObjects >= numObjects) { + brokerMap.get(h.getBrokerName()).spawnObject(reference, settings.getClassName(), settings.getProps()); numObjects++; if (minObjects >= numObjects) { @@ -61,5 +113,5 @@ @Override - public void unbindObject(OmqSettings settings, int numObjects) throws Exception { + public void unbindObject(OmqSettings settings, HasObject[] hasList, int numObjects) throws Exception { String reference = settings.getReference(); @@ -75,10 +127,33 @@ } - public SupervisorThread getThread() { - return thread; - } + private void checkObject(String reference) throws Exception { + OmqSettings settings = objectSettings.get(reference); - public void setThread(SupervisorThread thread) { - this.thread = thread; + int minObjects = settings.getMinNumberObjects(); + int maxMessages = settings.getMaxNumQueued(); + int minMessages = settings.getMinNumQueued(); + + Channel channel = getBroker().getChannel(); + DeclareOk dok = channel.queueDeclarePassive(reference); + + int numObjects = 0; + int numMessages = dok.getMessageCount(); + + HasObject[] hasList = multiBroker.hasObjectInfo(reference); + for (HasObject h : hasList) { + if (h.hasObject()) { + numObjects++; + } + } + + System.out.println("Num Consumers: " + numObjects + ", num Messages: " + numMessages); + + if (maxMessages < numMessages || numObjects < minObjects) { + logger.info("SPAWN TIME!!"); + spawnObject(settings, hasList, numObjects); + } else if (numMessages < minMessages && minObjects < numObjects) { + logger.info("Unbinding object!!!"); + unbindObject(settings, hasList, numObjects); + } } @@ -99,3 +174,19 @@ } + public String getBrokerSet() { + return brokerSet; + } + + public void setBrokerSet(String brokerSet) { + this.brokerSet = brokerSet; + } + + public RemoteMultiBroker getMultiBroker() { + return multiBroker; + } + + public void setMultiBroker(RemoteMultiBroker multiBroker) { + this.multiBroker = multiBroker; + } + } Index: anches/supervisor/src/main/java/omq/supervisor/SupervisorThread.java =================================================================== --- /branches/supervisor/src/main/java/omq/supervisor/SupervisorThread.java (revision 105) +++ (revision ) @@ -1,84 +1,0 @@ -package omq.supervisor; - -import java.util.Map; -import java.util.Set; - -import omq.common.broker.RemoteBroker; -import omq.exception.RetryException; - -import org.apache.log4j.Logger; - -import com.rabbitmq.client.AMQP.Queue.DeclareOk; -import com.rabbitmq.client.Channel; - -public class SupervisorThread extends Thread { - private static final Logger logger = Logger.getLogger(SupervisorThread.class.getName()); - - private long sleep; - private SupervisorImpl supervisor; - private Map objectSettings; - - public SupervisorThread(SupervisorImpl supervisor, long sleep) { - this.sleep = sleep; - this.supervisor = supervisor; - this.objectSettings = supervisor.getObjectSettings(); - } - - @Override - public void run() { - while (true) { - try { - Set keys = objectSettings.keySet(); - for (String reference : keys) { - checkObject(reference); - } - } catch (Exception e) { - e.printStackTrace(); - } - - try { - Thread.sleep(sleep); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - } - - private void checkObject(String reference) throws Exception { - OmqSettings settings = objectSettings.get(reference); - - int minObjects = settings.getMinNumberObjects(); - int maxMessages = settings.getMaxNumQueued(); - int minMessages = settings.getMinNumQueued(); - - Channel channel = supervisor.getBroker().getChannel(); - DeclareOk dok = channel.queueDeclarePassive(reference); - - int numObjects = getNumObjects(reference); - int numMessages = dok.getMessageCount(); - - System.out.println("Num Consumers: " + numObjects + ", num Messages: " + numMessages); - - if (maxMessages < numMessages || numObjects < minObjects) { - logger.info("SPAWN TIME!!"); - supervisor.spawnObject(settings, numObjects); - } else if (numMessages < minMessages && minObjects < numObjects) { - logger.info("Unbinding object!!!"); - 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/main/resources/log4j.xml =================================================================== --- /branches/supervisor/src/main/resources/log4j.xml (revision 105) +++ /branches/supervisor/src/main/resources/log4j.xml (revision 106) @@ -4,5 +4,5 @@ - + @@ -30,5 +30,5 @@ - + Index: /branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java =================================================================== --- /branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java (revision 105) +++ /branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java (revision 106) @@ -13,4 +13,5 @@ public class SleepTest { + private static String brokerSet = "broker"; @BeforeClass @@ -42,9 +43,11 @@ // Set supervisor - SupervisorImpl supervisor = new SupervisorImpl(500); + SupervisorImpl supervisor = new SupervisorImpl(brokerSet, 500); broker.bind("supervisor", supervisor); + Thread t = new Thread(supervisor); + t.start(); - broker.setSupervisor("supervisor", "b1"); - broker2.setSupervisor("supervisor", "b2"); + broker.setSupervisor("supervisor", brokerSet, "b1"); + broker2.setSupervisor("supervisor", brokerSet, "b2"); } @@ -63,5 +66,5 @@ OmqSettings settings = new OmqSettings("sleep", SleepImpl.class.getName(), env, 1, 20, 20); - supervisor.spawnObject(settings, 0); + supervisor.spawnObject(settings); Sleep sleep = broker.lookup("sleep", Sleep.class); Index: anches/supervisor/target/classes/log4j.xml =================================================================== --- /branches/supervisor/target/classes/log4j.xml (revision 105) +++ (revision ) @@ -1,37 +1,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - Index: anches/supervisor/target/classes/log4j.xml.svntmp =================================================================== --- /branches/supervisor/target/classes/log4j.xml.svntmp (revision 105) +++ (revision ) @@ -1,37 +1,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -