Index: branches/supervisor/src/main/java/omq/server/InvocationThread.java =================================================================== --- branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 95) +++ branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 96) @@ -1,11 +1,15 @@ package omq.server; +import java.io.IOException; import java.lang.reflect.InvocationTargetException; -import java.util.concurrent.BlockingQueue; - +import java.util.Properties; + +import omq.common.broker.Broker; import omq.common.message.Request; import omq.common.message.Response; +import omq.common.util.ParameterQueue; import omq.common.util.Serializer; import omq.exception.OmqException; +import omq.exception.SerializerException; import org.apache.log4j.Logger; @@ -13,5 +17,8 @@ import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; +import com.rabbitmq.client.ConsumerCancelledException; +import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.client.QueueingConsumer.Delivery; +import com.rabbitmq.client.ShutdownSignalException; /** @@ -22,16 +29,40 @@ */ public class InvocationThread extends Thread { + private static final Logger logger = Logger.getLogger(InvocationThread.class.getName()); + private static final String multi = "multi#"; + + // RemoteObject private RemoteObject obj; + private String UID; + private Properties env; + + // Broker + private Broker broker; private Serializer serializer; - // private RemoteWrapper wrapper; - private BlockingQueue deliveryQueue; + + // Consumer + private Channel channel; + private QueueingConsumer consumer; + private String multiQueue; private boolean killed = false; - public InvocationThread(RemoteObject obj, RemoteWrapper wrapper, Serializer serializer) { + public InvocationThread(RemoteObject obj, Broker broker) throws Exception { this.obj = obj; - // this.wrapper = wrapper; - this.deliveryQueue = wrapper.getDeliveryQueue(); - this.serializer = serializer; + this.UID = obj.getRef(); + this.env = obj.getEnv(); + this.broker = broker; + this.serializer = broker.getSerializer(); + } + + @Override + public synchronized void start() { + try { + startQueues(); + super.start(); + } catch (Exception e) { + logger.error("Cannot start a remoteObject", e); + } + } @@ -41,8 +72,5 @@ try { // Get the delivery - Delivery delivery = deliveryQueue.take(); - - // // Indicate this thread is not available - // wrapper.increaseBusy(); + Delivery delivery = consumer.nextDelivery(); String serializerType = delivery.getProperties().getType(); @@ -69,13 +97,7 @@ } - - Channel channel = obj.getChannel(); - - // Reply if it's necessary if (!request.isAsync()) { Response resp = new Response(request.getId(), obj.getRef(), result, error); - - BasicProperties props = delivery.getProperties(); @@ -88,17 +110,115 @@ + props.getReplyTo()); } - + channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); - - // // Indicate this thread is available - // wrapper.decreaseBusy(); } catch (InterruptedException i) { logger.error(i); - killed = true; + } catch (ShutdownSignalException e) { + logger.error(e); + try { + if (channel.isOpen()) { + channel.close(); + } + startQueues(); + } catch (Exception e1) { + try { + long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000")); + Thread.sleep(milis); + } catch (InterruptedException e2) { + logger.error(e2); + } + logger.error(e1); + } + } catch (ConsumerCancelledException e) { + logger.error(e); + } catch (SerializerException e) { + logger.error(e); } catch (Exception e) { - logger.error("Object: " + obj.getRef(), e); + logger.error(e); } } + } + + /** + * This method starts the queues using the information got in the + * environment. + * + * @throws Exception + */ + private void startQueues() throws Exception { + // Start channel + channel = broker.getNewChannel(); + + /* + * Default queue, Round Robin behaviour + */ + + // Get info about which exchange and queue will use + String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE, ""); + String queue = UID; + String routingKey = UID; + + // RemoteObject default queue + boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUE, "false")); + boolean exclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_QUEUE, "false")); + boolean autoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_QUEUE, "false")); + + // Declares and bindings + if (!exchange.equalsIgnoreCase("")) { // Default exchange case + channel.exchangeDeclare(exchange, "direct"); + } + channel.queueDeclare(queue, durable, exclusive, autoDelete, null); + if (!exchange.equalsIgnoreCase("")) { // Default exchange case + channel.queueBind(queue, exchange, routingKey); + } + logger.info("RemoteObject: " + UID + " declared direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable + ", Exclusive: " + + exclusive + ", AutoDelete: " + autoDelete); + + /* + * Multi queue, exclusive per each instance + */ + + // Get info about the multiQueue + String multiExchange = multi + UID; + // TODO:String multiExchange = multi + exchange + UID; + multiQueue = env.getProperty(ParameterQueue.MULTI_QUEUE_NAME); + + // Multi queue (exclusive queue per remoteObject) + boolean multiDurable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_MQUEUE, "false")); + boolean multiExclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_MQUEUE, "true")); + boolean multiAutoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_MQUEUE, "true")); + + // Declares and bindings + channel.exchangeDeclare(multiExchange, "fanout"); + if (multiQueue == null) { + multiQueue = channel.queueDeclare().getQueue(); + } else { + channel.queueDeclare(multiQueue, multiDurable, multiExclusive, multiAutoDelete, null); + } + channel.queueBind(multiQueue, multiExchange, ""); + logger.info("RemoteObject: " + UID + " declared fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + multiDurable + + ", Exclusive: " + multiExclusive + ", AutoDelete: " + multiAutoDelete); + + /* + * Consumer + */ + + boolean autoAck = false; + + int prefetchCount = 1; + channel.basicQos(prefetchCount); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(queue, autoAck, consumer); + channel.basicConsume(multiQueue, autoAck, consumer); + } + + public void kill() throws IOException { + logger.info("Killing objectmq: " + UID + " thread id"); + killed = true; + interrupt(); + channel.close(); } @@ -111,10 +231,3 @@ } - public BlockingQueue getDeliveryQueue() { - return deliveryQueue; - } - - public void setDeliveryQueue(BlockingQueue deliveryQueue) { - this.deliveryQueue = deliveryQueue; - } } Index: branches/supervisor/src/main/java/omq/server/RemoteObject.java =================================================================== --- branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 95) +++ branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 96) @@ -12,13 +12,6 @@ import omq.common.broker.Broker; import omq.common.util.ParameterQueue; -import omq.exception.SerializerException; import org.apache.log4j.Logger; - -import com.rabbitmq.client.Channel; -import com.rabbitmq.client.ConsumerCancelledException; -import com.rabbitmq.client.QueueingConsumer; -import com.rabbitmq.client.QueueingConsumer.Delivery; -import com.rabbitmq.client.ShutdownSignalException; /** @@ -32,8 +25,7 @@ * */ -public abstract class RemoteObject extends Thread implements Remote { +public abstract class RemoteObject implements Remote { private static final long serialVersionUID = -1778953938739846450L; - private static final String multi = "multi#"; private static final Logger logger = Logger.getLogger(RemoteObject.class.getName()); @@ -41,10 +33,6 @@ 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 transient List invocationList; private static final Map> primitiveClasses = new HashMap>(); @@ -60,7 +48,4 @@ } - public RemoteObject() { - } - /** * This method starts a remoteObject. @@ -90,47 +75,13 @@ // 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(); - } - - @Override - public void run() { - while (!killed) { - try { - Delivery delivery = consumer.nextDelivery(); - - logger.debug(UID + " has received a message, serializer: " + delivery.getProperties().getType()); - - remoteWrapper.notifyDelivery(delivery); - } catch (InterruptedException i) { - logger.error(i); - } catch (ShutdownSignalException e) { - logger.error(e); - try { - if (channel.isOpen()) { - channel.close(); - } - startQueues(); - } catch (Exception e1) { - try { - long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000")); - Thread.sleep(milis); - } catch (InterruptedException e2) { - logger.error(e2); - } - logger.error(e1); - } - } catch (ConsumerCancelledException e) { - logger.error(e); - } catch (SerializerException e) { - logger.error(e); - } catch (Exception e) { - logger.error(e); - } - } + invocationList = new ArrayList(numThreads); + + // Start invocation threads + for (int i = 0; i < numThreads; i++) { + InvocationThread iThread = new InvocationThread(this, broker); + invocationList.add(iThread); + iThread.start(); + } + } @@ -147,9 +98,8 @@ */ public void kill() throws IOException { - logger.warn("Killing objectmq: " + this.getRef()); - killed = true; - interrupt(); - channel.close(); - remoteWrapper.stopRemoteWrapper(); + logger.info("Killing objectmq: " + this.getRef()); + for (InvocationThread iThread : invocationList) { + iThread.kill(); + } } @@ -250,85 +200,10 @@ } - public Channel getChannel() { - return channel; - } - public Broker getBroker() { return broker; } - /** - * This method starts the queues using the information got in the - * environment. - * - * @throws Exception - */ - private void startQueues() throws Exception { - // Start channel - channel = broker.getNewChannel(); - - /* - * Default queue, Round Robin behaviour - */ - - // Get info about which exchange and queue will use - String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE, ""); - String queue = UID; - String routingKey = UID; - - // RemoteObject default queue - boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUE, "false")); - boolean exclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_QUEUE, "false")); - boolean autoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_QUEUE, "false")); - - // Declares and bindings - if (!exchange.equalsIgnoreCase("")) { // Default exchange case - channel.exchangeDeclare(exchange, "direct"); - } - channel.queueDeclare(queue, durable, exclusive, autoDelete, null); - if (!exchange.equalsIgnoreCase("")) { // Default exchange case - channel.queueBind(queue, exchange, routingKey); - } - logger.info("RemoteObject: " + UID + " declared direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable + ", Exclusive: " - + exclusive + ", AutoDelete: " + autoDelete); - - /* - * Multi queue, exclusive per each instance - */ - - // Get info about the multiQueue - String multiExchange = multi + UID; - multiQueue = env.getProperty(ParameterQueue.MULTI_QUEUE_NAME); - - // Multi queue (exclusive queue per remoteObject) - boolean multiDurable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_MQUEUE, "false")); - boolean multiExclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_MQUEUE, "true")); - boolean multiAutoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_MQUEUE, "true")); - - // Declares and bindings - channel.exchangeDeclare(multiExchange, "fanout"); - if (multiQueue == null) { - multiQueue = channel.queueDeclare().getQueue(); - } else { - channel.queueDeclare(multiQueue, multiDurable, multiExclusive, multiAutoDelete, null); - } - channel.queueBind(multiQueue, multiExchange, ""); - logger.info("RemoteObject: " + UID + " declared fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + multiDurable - + ", Exclusive: " + multiExclusive + ", AutoDelete: " + multiAutoDelete); - - /* - * Consumer - */ - - boolean autoAck = false; - - //TODO see if this is useless - int prefetchCount = 1; - channel.basicQos(prefetchCount); - - // Declare a new consumer - consumer = new QueueingConsumer(channel); - channel.basicConsume(queue, autoAck, consumer); - channel.basicConsume(multiQueue, autoAck, consumer); + public Properties getEnv() { + return env; } Index: branches/supervisor/src/main/java/omq/server/RemoteWrapper.java =================================================================== --- branches/supervisor/src/main/java/omq/server/RemoteWrapper.java (revision 95) +++ (revision ) @@ -1,133 +1,0 @@ -package omq.server; - -import java.util.ArrayList; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingDeque; - -import omq.common.util.Serializer; - -import org.apache.log4j.Logger; - -import com.rabbitmq.client.QueueingConsumer; -import com.rabbitmq.client.QueueingConsumer.Delivery; - -/** - * This class is used to encapsulate the invocationThreads under the - * RemoteObject. - * - * @author Sergi Toda - * - */ -public class RemoteWrapper { - private static final Logger logger = Logger.getLogger(RemoteWrapper.class.getName()); - - private RemoteObject obj; - private int numThreads; - // private AtomicInteger busy; - private Object waitLock; - private ArrayList invocationList; - private BlockingQueue deliveryQueue; - - public RemoteWrapper(RemoteObject obj, int numThreads, Serializer serializer) { - this.obj = obj; - this.numThreads = numThreads; - // this.busy = new AtomicInteger(0); - this.waitLock = new Object(); - invocationList = new ArrayList(); - deliveryQueue = new LinkedBlockingDeque(); - - logger.info("Object reference: " + obj.getRef() + ", numthreads listening = " + numThreads); - - for (int i = 0; i < numThreads; i++) { - InvocationThread thread = new InvocationThread(obj, this, serializer); - invocationList.add(thread); - thread.start(); - } - } - - /** - * This method notifies a delivery to an invocationThread using a - * blockingQueue. - * - * @param delivery - * - delivery which contains a Request to be invoked - * @throws Exception - */ - 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); - - } - - /** - * This method interrups all the invocationThreads under this remoteWrapper - */ - public void stopRemoteWrapper() { - logger.warn("Stopping Invocation threads vinculed to " + obj.getRef()); - for (InvocationThread thread : invocationList) { - thread.interrupt(); - } - } - - // public int increaseBusy() { - // return this.busy.incrementAndGet(); - // } - // - // public int decreaseBusy() { - // int value = this.busy.decrementAndGet(); - // synchronized (waitLock) { - // waitLock.notifyAll(); - // } - // return value; - // } - - public RemoteObject getObj() { - return obj; - } - - public void setObj(RemoteObject obj) { - this.obj = obj; - } - - public int getNumThreads() { - return numThreads; - } - - public void setNumThreads(int numThreads) { - this.numThreads = numThreads; - } - - public ArrayList getInvocationList() { - return invocationList; - } - - public void setInvocationList(ArrayList invocationList) { - this.invocationList = invocationList; - } - - public BlockingQueue getDeliveryQueue() { - return deliveryQueue; - } - - public void setDeliveryQueue(BlockingQueue deliveryQueue) { - this.deliveryQueue = deliveryQueue; - } - - public Object getLock() { - return waitLock; - } - - public void setLock(Object lock) { - this.waitLock = lock; - } -} Index: branches/supervisor/src/test/java/omq/test/multiThread2/MultiThreadTest.java =================================================================== --- branches/supervisor/src/test/java/omq/test/multiThread2/MultiThreadTest.java (revision 96) +++ branches/supervisor/src/test/java/omq/test/multiThread2/MultiThreadTest.java (revision 96) @@ -0,0 +1,52 @@ +package omq.test.multiThread2; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.test.supervisor.Sleep; +import omq.test.supervisor.SleepImpl; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class MultiThreadTest { + @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, "4"); + + 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 < 10; i++) { + sleep.sleep(); + } + + Thread.sleep(10000); + } + +} Index: branches/supervisor/src/test/java/omq/test/multiThread2/Sleep.java =================================================================== --- branches/supervisor/src/test/java/omq/test/multiThread2/Sleep.java (revision 96) +++ branches/supervisor/src/test/java/omq/test/multiThread2/Sleep.java (revision 96) @@ -0,0 +1,11 @@ +package omq.test.multiThread2; + +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/multiThread2/SleepImpl.java =================================================================== --- branches/supervisor/src/test/java/omq/test/multiThread2/SleepImpl.java (revision 96) +++ branches/supervisor/src/test/java/omq/test/multiThread2/SleepImpl.java (revision 96) @@ -0,0 +1,24 @@ +package omq.test.multiThread2; + +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/supervisor/SleepImpl.java =================================================================== --- branches/supervisor/src/test/java/omq/test/supervisor/SleepImpl.java (revision 95) +++ branches/supervisor/src/test/java/omq/test/supervisor/SleepImpl.java (revision 96) @@ -16,5 +16,5 @@ try { System.out.println("I'm going to sleep!!!!!!!!" + Thread.currentThread().getId()); - Thread.sleep(1000); + Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); Index: branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java =================================================================== --- branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java (revision 95) +++ branches/supervisor/src/test/java/omq/test/supervisor/SleepTest.java (revision 96) @@ -24,5 +24,5 @@ env1.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1"); env1.setProperty(ParameterQueue.RABBIT_PORT, "5672"); - env1.setProperty(ParameterQueue.NUM_THREADS, "1"); + env1.setProperty(ParameterQueue.NUM_THREADS, "4"); Broker broker = new Broker(env1);