Index: branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java (revision 91) +++ branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java (revision 92) @@ -116,4 +116,11 @@ if (methodName.equals("getRef")) { return getRef(); + } + if (methodName.equals("equals")) { + if (arguments[0] instanceof Remote) { + return getRef().equals(((Remote) arguments[0]).getRef()); + } else { + return false; + } } } Index: branches/supervisor/src/main/java/omq/common/broker/Broker.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/Broker.java (revision 91) +++ branches/supervisor/src/main/java/omq/common/broker/Broker.java (revision 92) @@ -20,4 +20,5 @@ import omq.exception.RemoteException; import omq.server.RemoteObject; +import omq.supervisor.Supervisor; import org.apache.log4j.Logger; @@ -49,8 +50,10 @@ private boolean connectionClosed = false; private Properties environment = null; - private RemoteBrokerImpl remoteBrokerImpl; private Map remoteObjs; private Map proxies = new Hashtable(); private Map multiProxies = new Hashtable(); + + // Supervisor + private Supervisor supervisor; public Broker(Properties env) throws Exception { @@ -358,9 +361,4 @@ } - public void setSupervisor(String brokerSet, String brokerName) throws Exception { - remoteBrokerImpl = new RemoteBrokerImpl(); - remoteBrokerImpl.startRemoteBroker(brokerSet, brokerName, this, getEnvironment()); - } - public Properties getEnvironment() { return environment; @@ -379,3 +377,18 @@ } + /* + * Supervisor + */ + public void setSupervisor(String supervisorName, String brokerName) throws Exception { + // Create a RemoteBrokerImpl + bind(brokerName, new RemoteBrokerImpl()); + // Subscribe broker + supervisor = lookup(supervisorName, Supervisor.class); + supervisor.subscribe(brokerName); + logger.info("Supervisor set: " + supervisorName + ", BrokerName: " + brokerName); + } + + public Supervisor getSupervisor() { + return supervisor; + } } Index: branches/supervisor/src/main/java/omq/common/broker/RemoteBroker.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/RemoteBroker.java (revision 91) +++ branches/supervisor/src/main/java/omq/common/broker/RemoteBroker.java (revision 92) @@ -2,7 +2,9 @@ import java.io.IOException; +import java.util.Properties; import java.util.Set; import omq.Remote; +import omq.client.annotation.SyncMethod; import omq.exception.RemoteException; @@ -10,9 +12,12 @@ public Set getRemoteObjects(); - public void spawnObject(String reference, String className, Class parameterTypes, Object... args) throws Exception; + public void spawnObject(String reference, String className, Properties env) throws Exception; + + public void spawnObject(String reference, String className) throws Exception; public void deleteObject(String reference) throws RemoteException, IOException; - public HasObject hasObject(String reference); + @SyncMethod(retry = 1, timeout = 1000) + public boolean hasObject(String reference); } Index: branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java (revision 91) +++ branches/supervisor/src/main/java/omq/common/broker/RemoteBrokerImpl.java (revision 92) @@ -2,17 +2,9 @@ import java.io.IOException; -import java.lang.reflect.Method; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; import java.util.Properties; import java.util.Set; -import omq.common.util.ParameterQueue; import omq.exception.RemoteException; import omq.server.RemoteObject; -import omq.server.RemoteWrapper; - -import com.rabbitmq.client.QueueingConsumer; /** @@ -28,79 +20,29 @@ private static final long serialVersionUID = 1L; - // fanout broker - private String brokerSet; - // id broker - private String brokerName; - - public void startRemoteBroker(String brokerSet, String brokerName, Broker broker, Properties env) throws Exception { - this.broker = broker; - this.UID = brokerName; - this.env = env; - this.brokerSet = brokerSet; - this.brokerName = brokerName; - - this.params = new HashMap>>(); - for (Method m : this.getClass().getMethods()) { - List> list = new ArrayList>(); - for (Class clazz : m.getParameterTypes()) { - list.add(clazz); - } - this.params.put(m.getName(), list); - } - - // Get num threads to use - int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1")); - this.remoteWrapper = new RemoteWrapper(this, numThreads, broker.getSerializer()); - - startQueues(); - - // Start this listener - this.start(); - } - - private void startQueues() throws Exception { - /* - * Unique queue - */ - channel.exchangeDeclare(brokerSet, "direct"); - channel.queueDeclare(brokerName, false, true, true, null); - channel.queueBind(brokerName, brokerSet, brokerName); - - /* - * Multi queue - */ - channel.exchangeDeclare("multi#" + brokerSet, "fanout"); - channel.queueDeclare("multi#" + brokerName, false, true, true, null); - channel.queueBind("multi#" + brokerName, "multi#" + brokerSet, ""); - - /* - * Consumer - */ - - consumer = new QueueingConsumer(channel); - channel.basicConsume(brokerName, true, consumer); - channel.basicConsume(brokerName + "#multi", true, consumer); + @Override + public Set getRemoteObjects() { + return getBroker().getRemoteObjs().keySet(); } @Override - public Set getRemoteObjects() { - return this.broker.getRemoteObjs().keySet(); + public void spawnObject(String reference, String className, Properties env) throws Exception { + RemoteObject remote = (RemoteObject) Class.forName(className).newInstance(); + getBroker().bind(reference, remote, env); } @Override - public void spawnObject(String reference, String className, Class parameterTypes, Object... args) throws Exception { - RemoteObject remote = (RemoteObject) Class.forName(className).getConstructor(parameterTypes).newInstance(args); - this.broker.bind(reference, remote); + public void spawnObject(String reference, String className) throws Exception { + RemoteObject remote = (RemoteObject) Class.forName(className).newInstance(); + getBroker().bind(reference, remote); } @Override public void deleteObject(String reference) throws RemoteException, IOException { - this.broker.unbind(reference); + getBroker().unbind(reference); } @Override - public HasObject hasObject(String reference) { - boolean hasIt = this.broker.getRemoteObjs().containsKey(reference); - return new HasObject(this.brokerName, reference, hasIt); + public boolean hasObject(String reference) { + return getBroker().getRemoteObjs().containsKey(reference); } Index: branches/supervisor/src/main/java/omq/server/RemoteObject.java =================================================================== --- branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 91) +++ branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 92) @@ -38,13 +38,13 @@ private static final Logger logger = Logger.getLogger(RemoteObject.class.getName()); - protected String UID; - protected Properties env; - protected transient Broker broker; - protected transient String multiQueue; - protected transient RemoteWrapper remoteWrapper; - protected transient Map>> params; - protected transient Channel channel; - protected transient QueueingConsumer consumer; - protected transient boolean killed = false; + private String UID; + private Properties env; + private transient Broker broker; + private transient String multiQueue; + private transient RemoteWrapper remoteWrapper; + private transient Map>> params; + private transient Channel channel; + private transient QueueingConsumer consumer; + private transient boolean killed = false; private static final Map> primitiveClasses = new HashMap>(); @@ -254,4 +254,8 @@ } + public Broker getBroker() { + return broker; + } + /** * This method starts the queues using the information got in the @@ -319,4 +323,8 @@ boolean autoAck = false; + //TODO see if this is useless + int prefetchCount = 1; + channel.basicQos(prefetchCount); + // Declare a new consumer consumer = new QueueingConsumer(channel); Index: branches/supervisor/src/main/java/omq/supervisor/ISupervisor.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/ISupervisor.java (revision 91) +++ (revision ) @@ -1,12 +1,0 @@ -package omq.supervisor; - -import omq.Remote; - -public interface ISupervisor extends Remote { - - public void subscrive - - public void spawnObject(OmqSettings settings) throws Exception; - - -} Index: branches/supervisor/src/main/java/omq/supervisor/MultiBroker.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/MultiBroker.java (revision 91) +++ (revision ) @@ -1,14 +1,0 @@ -package omq.supervisor; - -import omq.Remote; -import omq.client.annotation.MultiMethod; -import omq.client.annotation.RemoteInterface; -import omq.client.annotation.SyncMethod; -import omq.common.broker.HasObject; - -@RemoteInterface -public interface MultiBroker extends Remote { - @MultiMethod(waitNum = 5) - @SyncMethod(retry = 1, timeout = 2000) - public HasObject[] hasObject(String reference); -} Index: branches/supervisor/src/main/java/omq/supervisor/OmqSettings.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/OmqSettings.java (revision 91) +++ branches/supervisor/src/main/java/omq/supervisor/OmqSettings.java (revision 92) @@ -1,8 +1,15 @@ package omq.supervisor; +import java.io.Serializable; import java.util.Properties; -public class OmqSettings { +public class OmqSettings implements Serializable { + + + /** + * + */ + private static final long serialVersionUID = 1L; private String reference; private String className; Index: branches/supervisor/src/main/java/omq/supervisor/Supervisor.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/Supervisor.java (revision 91) +++ branches/supervisor/src/main/java/omq/supervisor/Supervisor.java (revision 92) @@ -1,26 +1,11 @@ package omq.supervisor; -import java.util.Map; -import java.util.Set; +import omq.Remote; -import omq.common.broker.RemoteBroker; +public interface Supervisor extends Remote { -public class Supervisor { + public void subscribe(String brokerName) throws Exception; - private Set bindReferences; - private Map brokers; + public void spawnObject(OmqSettings settings) throws Exception; - private void checkObject() { - String reference = null; - - int numObjects = 0; - - if(minObjects > numObjects || maxMessages < numEncuats){ - spawn: - pregunta a tots i qui no té l'objecte li poses - }else if(numEncuats < minMessages && minObjects > numObjects){ - delete: - pregunta a tots i qui té l'objecte li treus - } - } } Index: branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 92) +++ branches/supervisor/src/main/java/omq/supervisor/SupervisorImpl.java (revision 92) @@ -0,0 +1,97 @@ +package omq.supervisor; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import omq.common.broker.RemoteBroker; +import omq.server.RemoteObject; + +import org.apache.log4j.Logger; + +import com.rabbitmq.client.AMQP.Queue.DeclareOk; +import com.rabbitmq.client.Channel; + +public class SupervisorImpl extends RemoteObject implements Supervisor { + + /** + * + */ + private static final long serialVersionUID = 1L; + private static final Logger logger = Logger.getLogger(SupervisorImpl.class.getName()); + private SupervisorThread thread; + private Map objectSettings; + // TODO: Set + private List brokers; + + public SupervisorImpl(long 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); + } + + @Override + public void spawnObject(OmqSettings settings) throws Exception { + + String reference = settings.getReference(); + objectSettings.put(reference, settings); + + 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) { + if (!broker.hasObject(reference) && minObjects >= numObjects) { + broker.spawnObject(reference, settings.getClassName(), settings.getProps()); + numObjects++; + if (minObjects >= numObjects) { + break; + } + } + } + + } + + public SupervisorThread getThread() { + return thread; + } + + public void setThread(SupervisorThread thread) { + this.thread = thread; + } + + public Map getObjectSettings() { + return objectSettings; + } + + public void setObjectSettings(Map objectSettings) { + this.objectSettings = objectSettings; + } + + public List getBrokers() { + return brokers; + } + + public void setBrokers(List brokers) { + this.brokers = brokers; + } + +} Index: branches/supervisor/src/main/java/omq/supervisor/SupervisorThread.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/SupervisorThread.java (revision 92) +++ branches/supervisor/src/main/java/omq/supervisor/SupervisorThread.java (revision 92) @@ -0,0 +1,67 @@ +package omq.supervisor; + +import java.util.Map; +import java.util.Set; + +import com.rabbitmq.client.AMQP.Queue.DeclareOk; +import com.rabbitmq.client.Channel; + +public class SupervisorThread extends Thread { + + 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(); + + // TODO treure merda... + Channel channel = supervisor.getBroker().getChannel(); + DeclareOk dok = channel.queueDeclarePassive(reference); + + int numConsumers = dok.getConsumerCount(); + int numMessages = dok.getMessageCount(); + + System.out.println("Num Consumers: " + numConsumers + ", num Messages: " + numMessages); + + if (maxMessages < numMessages || minObjects < numConsumers) { + System.out.println("SPAWN TIME!!"); + supervisor.spawnObject(settings); + // spawn: + // pregunta a tots i qui no té l'objecte li poses + } else if (numMessages < minMessages && minObjects > numConsumers) { + // delete: + // pregunta a tots i qui té l'objecte li treus + } + } +}