Index: branches/supervisor/src/main/java/omq/common/broker/Broker.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/Broker.java (revision 90) +++ branches/supervisor/src/main/java/omq/common/broker/Broker.java (revision 91) @@ -49,4 +49,5 @@ private boolean connectionClosed = false; private Properties environment = null; + private RemoteBrokerImpl remoteBrokerImpl; private Map remoteObjs; private Map proxies = new Hashtable(); @@ -357,4 +358,9 @@ } + public void setSupervisor(String brokerSet, String brokerName) throws Exception { + remoteBrokerImpl = new RemoteBrokerImpl(); + remoteBrokerImpl.startRemoteBroker(brokerSet, brokerName, this, getEnvironment()); + } + public Properties getEnvironment() { return environment; @@ -368,3 +374,8 @@ return serializer; } + + public Map getRemoteObjs() { + return remoteObjs; + } + } Index: branches/supervisor/src/main/java/omq/common/broker/HasObject.java =================================================================== --- branches/supervisor/src/main/java/omq/common/broker/HasObject.java (revision 91) +++ branches/supervisor/src/main/java/omq/common/broker/HasObject.java (revision 91) @@ -0,0 +1,45 @@ +package omq.common.broker; + +import java.io.Serializable; + +public class HasObject implements Serializable { + /** + * + */ + private static final long serialVersionUID = 1L; + + private String brokerName; + private String reference; + private boolean hasObject; + + public HasObject(String brokerName, String reference, boolean hasObject) { + this.brokerName = brokerName; + this.reference = reference; + this.hasObject = hasObject; + } + + public String getBrokerName() { + return brokerName; + } + + public void setBrokerName(String brokerName) { + this.brokerName = brokerName; + } + + public String getReference() { + return reference; + } + + public void setReference(String reference) { + this.reference = reference; + } + + public boolean isHasObject() { + return hasObject; + } + + public void setHasObject(boolean hasObject) { + this.hasObject = hasObject; + } + +} 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 91) @@ -0,0 +1,18 @@ +package omq.common.broker; + +import java.io.IOException; +import java.util.Set; + +import omq.Remote; +import omq.exception.RemoteException; + +public interface RemoteBroker extends Remote { + public Set getRemoteObjects(); + + public void spawnObject(String reference, String className, Class parameterTypes, Object... args) throws Exception; + + public void deleteObject(String reference) throws RemoteException, IOException; + + public HasObject 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 91) @@ -0,0 +1,107 @@ +package omq.common.broker; + +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; + +/** + * + * @author Sergi Toda + * + */ +public class RemoteBrokerImpl extends RemoteObject implements RemoteBroker { + + /** + * + */ + 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 this.broker.getRemoteObjs().keySet(); + } + + @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); + } + + @Override + public void deleteObject(String reference) throws RemoteException, IOException { + this.broker.unbind(reference); + } + + @Override + public HasObject hasObject(String reference) { + boolean hasIt = this.broker.getRemoteObjs().containsKey(reference); + return new HasObject(this.brokerName, reference, hasIt); + } + +} Index: branches/supervisor/src/main/java/omq/server/InvocationThread.java =================================================================== --- branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 90) +++ branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 91) @@ -24,11 +24,13 @@ private static final Logger logger = Logger.getLogger(InvocationThread.class.getName()); private RemoteObject obj; - private transient Serializer serializer; + private Serializer serializer; + // private RemoteWrapper wrapper; private BlockingQueue deliveryQueue; private boolean killed = false; - public InvocationThread(RemoteObject obj, BlockingQueue deliveryQueue, Serializer serializer) { + public InvocationThread(RemoteObject obj, RemoteWrapper wrapper, Serializer serializer) { this.obj = obj; - this.deliveryQueue = deliveryQueue; + // this.wrapper = wrapper; + this.deliveryQueue = wrapper.getDeliveryQueue(); this.serializer = serializer; } @@ -40,4 +42,7 @@ // Get the delivery Delivery delivery = deliveryQueue.take(); + + // // Indicate this thread is not available + // wrapper.increaseBusy(); String serializerType = delivery.getProperties().getType(); @@ -64,9 +69,13 @@ } + + Channel channel = obj.getChannel(); + + // Reply if it's necessary if (!request.isAsync()) { Response resp = new Response(request.getId(), obj.getRef(), result, error); - Channel channel = obj.getChannel(); + BasicProperties props = delivery.getProperties(); @@ -79,5 +88,9 @@ + props.getReplyTo()); } - + + channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); + + // // Indicate this thread is available + // wrapper.decreaseBusy(); } catch (InterruptedException i) { logger.error(i); Index: branches/supervisor/src/main/java/omq/server/RemoteObject.java =================================================================== --- branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 90) +++ branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 91) @@ -38,13 +38,13 @@ private static final Logger logger = Logger.getLogger(RemoteObject.class.getName()); - 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; + 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 static final Map> primitiveClasses = new HashMap>(); @@ -317,8 +317,10 @@ */ + boolean autoAck = false; + // Declare a new consumer consumer = new QueueingConsumer(channel); - channel.basicConsume(queue, true, consumer); - channel.basicConsume(multiQueue, true, consumer); + channel.basicConsume(queue, autoAck, consumer); + channel.basicConsume(multiQueue, autoAck, consumer); } Index: branches/supervisor/src/main/java/omq/server/RemoteWrapper.java =================================================================== --- branches/supervisor/src/main/java/omq/server/RemoteWrapper.java (revision 90) +++ branches/supervisor/src/main/java/omq/server/RemoteWrapper.java (revision 91) @@ -24,4 +24,6 @@ private RemoteObject obj; private int numThreads; + // private AtomicInteger busy; + private Object waitLock; private ArrayList invocationList; private BlockingQueue deliveryQueue; @@ -30,4 +32,6 @@ this.obj = obj; this.numThreads = numThreads; + // this.busy = new AtomicInteger(0); + this.waitLock = new Object(); invocationList = new ArrayList(); deliveryQueue = new LinkedBlockingDeque(); @@ -36,5 +40,5 @@ for (int i = 0; i < numThreads; i++) { - InvocationThread thread = new InvocationThread(obj, deliveryQueue, serializer); + InvocationThread thread = new InvocationThread(obj, this, serializer); invocationList.add(thread); thread.start(); @@ -51,5 +55,17 @@ */ public void notifyDelivery(Delivery delivery) throws Exception { + + // // Ensure there is at least one thread available + // while (this.busy.get() == this.numThreads) { + // System.out.println("Waiting for a thread available"); + // logger.debug("Object reference: " + obj.getRef() + " is busy"); + // + // synchronized (waitLock) { + // waitLock.wait(); + // } + // } + // Notify an available thread this.deliveryQueue.put(delivery); + } @@ -63,4 +79,16 @@ } } + + // public int increaseBusy() { + // return this.busy.incrementAndGet(); + // } + // + // public int decreaseBusy() { + // int value = this.busy.decrementAndGet(); + // synchronized (waitLock) { + // waitLock.notifyAll(); + // } + // return value; + // } public RemoteObject getObj() { @@ -95,3 +123,11 @@ this.deliveryQueue = deliveryQueue; } + + public Object getLock() { + return waitLock; + } + + public void setLock(Object lock) { + this.waitLock = lock; + } } Index: branches/supervisor/src/main/java/omq/supervisor/ISupervisor.java =================================================================== --- branches/supervisor/src/main/java/omq/supervisor/ISupervisor.java (revision 91) +++ branches/supervisor/src/main/java/omq/supervisor/ISupervisor.java (revision 91) @@ -0,0 +1,12 @@ +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) +++ branches/supervisor/src/main/java/omq/supervisor/MultiBroker.java (revision 91) @@ -0,0 +1,14 @@ +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 91) @@ -0,0 +1,71 @@ +package omq.supervisor; + +import java.util.Properties; + +public class OmqSettings { + + private String reference; + private String className; + private Properties props; + private int minNumberObjects; + private int maxNumQueued; + private int minNumQueued; + + public OmqSettings(String reference, String className, Properties props, int minNumberObjects, int maxNumQueued, int minNumQueued) { + this.reference = reference; + this.className = className; + this.props = props; + this.minNumberObjects = minNumberObjects; + this.maxNumQueued = maxNumQueued; + this.minNumQueued = minNumQueued; + } + + public String getReference() { + return reference; + } + + public void setReference(String reference) { + this.reference = reference; + } + + public String getClassName() { + return className; + } + + public void setClassName(String className) { + this.className = className; + } + + public Properties getProps() { + return props; + } + + public void setProps(Properties props) { + this.props = props; + } + + public int getMinNumberObjects() { + return minNumberObjects; + } + + public void setMinNumberObjects(int minNumberObjects) { + this.minNumberObjects = minNumberObjects; + } + + public int getMaxNumQueued() { + return maxNumQueued; + } + + public void setMaxNumQueued(int maxNumQueued) { + this.maxNumQueued = maxNumQueued; + } + + public int getMinNumQueued() { + return minNumQueued; + } + + public void setMinNumQueued(int minNumQueued) { + this.minNumQueued = minNumQueued; + } + +} 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 91) @@ -0,0 +1,26 @@ +package omq.supervisor; + +import java.util.Map; +import java.util.Set; + +import omq.common.broker.RemoteBroker; + +public class Supervisor { + + private Set bindReferences; + private Map brokers; + + 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/test/java/omq/test/lock/Sleep.java =================================================================== --- branches/supervisor/src/test/java/omq/test/lock/Sleep.java (revision 91) +++ branches/supervisor/src/test/java/omq/test/lock/Sleep.java (revision 91) @@ -0,0 +1,11 @@ +package omq.test.lock; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; + +@RemoteInterface +public interface Sleep extends Remote { + @AsyncMethod + public void sleep(); +} Index: branches/supervisor/src/test/java/omq/test/lock/SleepImpl.java =================================================================== --- branches/supervisor/src/test/java/omq/test/lock/SleepImpl.java (revision 91) +++ branches/supervisor/src/test/java/omq/test/lock/SleepImpl.java (revision 91) @@ -0,0 +1,25 @@ +package omq.test.lock; + +import omq.client.annotation.AsyncMethod; +import omq.server.RemoteObject; + +public class SleepImpl extends RemoteObject implements Sleep { + + + /** + * + */ + private static final long serialVersionUID = 1L; + + @Override + @AsyncMethod + public void sleep() { + try { + System.out.println("I'm going to sleep!!!!!!!!" + Thread.currentThread().getId()); + Thread.sleep(1000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + +} Index: branches/supervisor/src/test/java/omq/test/lock/SleepTest.java =================================================================== --- branches/supervisor/src/test/java/omq/test/lock/SleepTest.java (revision 91) +++ branches/supervisor/src/test/java/omq/test/lock/SleepTest.java (revision 91) @@ -0,0 +1,50 @@ +package omq.test.lock; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class SleepTest { + + @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.NUM_THREADS, "2"); + + SleepImpl sleep = new SleepImpl(); + + Broker broker = new Broker(env); + broker.bind("sleep", sleep); + } + + @Test + public void test() 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 broker = new Broker(env); + Sleep sleep = broker.lookup("sleep", Sleep.class); + + for (int i = 0; i < 100; i++) { + sleep.sleep(); + } + + Thread.sleep(100000); + } + +} Index: branches/supervisor/target/classes/log4j.xml =================================================================== --- branches/supervisor/target/classes/log4j.xml (revision 90) +++ (revision ) @@ -1,37 +1,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -