Index: branches/supervisor/src/main/java/omq/server/AInvocationThread.java =================================================================== --- branches/supervisor/src/main/java/omq/server/AInvocationThread.java (revision 108) +++ branches/supervisor/src/main/java/omq/server/AInvocationThread.java (revision 108) @@ -0,0 +1,127 @@ +package omq.server; + +import java.io.IOException; +import java.lang.reflect.InvocationTargetException; +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.message.Request; +import omq.common.message.Response; +import omq.common.util.Serializer; +import omq.exception.OmqException; + +import org.apache.log4j.Logger; + +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; + +public abstract class AInvocationThread extends Thread { + + private static final Logger logger = Logger.getLogger(AInvocationThread.class.getName()); + + // RemoteObject + protected RemoteObject obj; + protected String reference; + protected String UID; + protected Properties env; + + protected RemoteThreadPool pool; + + // Broker + protected Broker broker; + protected Serializer serializer; + + // Consumer + protected Channel channel; + protected QueueingConsumer consumer; + protected boolean killed = false; + + public AInvocationThread(RemoteObject obj) throws Exception { + this.obj = obj; + this.UID = obj.getUID(); + this.reference = obj.getRef(); + this.env = obj.getEnv(); + this.broker = obj.getBroker(); + this.pool = obj.getPool(); + this.serializer = broker.getSerializer(); + } + + @Override + public synchronized void start() { + try { + startQueues(); + super.start(); + } catch (Exception e) { + logger.error("Cannot start a remoteObject", e); + } + + } + + /** + * This method starts the queues using the information got in the + * environment. + * + * @throws Exception + */ + protected abstract void startQueues() throws Exception; + + protected void executeTask(Delivery delivery) throws Exception{ + String serializerType = delivery.getProperties().getType(); + + // Deserialize the request + Request request = serializer.deserializeRequest(serializerType, delivery.getBody(), obj); + String methodName = request.getMethod(); + String requestID = request.getId(); + + logger.debug("Object: " + obj.getRef() + ", method: " + methodName + " corrID: " + requestID + ", serializerType: " + + serializerType); + + // Invoke the method + Object result = null; + OmqException error = null; + try { + result = obj.invokeMethod(request.getMethod(), request.getParams()); + } catch (InvocationTargetException e) { + Throwable throwable = e.getTargetException(); + logger.error("Object: " + obj.getRef() + " at method: " + methodName + ", corrID" + requestID, throwable); + error = new OmqException(throwable.getClass().getCanonicalName(), throwable.getMessage()); + } catch (NoSuchMethodException e) { + logger.error("Object: " + obj.getRef() + " cannot find method: " + methodName); + error = new OmqException(e.getClass().getCanonicalName(), e.getMessage()); + } + + // Reply if it's necessary + if (!request.isAsync()) { + Response resp = new Response(request.getId(), obj.getRef(), result, error); + + BasicProperties props = delivery.getProperties(); + + BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()) + .build(); + + byte[] bytesResponse = serializer.serialize(serializerType, resp); + channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse); + logger.debug("Publish sync response -> Object: " + obj.getRef() + ", method: " + methodName + " corrID: " + requestID + + " replyTo: " + props.getReplyTo()); + } + + channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); + } + + public void kill() throws IOException { + logger.info("Killing objectmq: " + reference + " thread id"); + killed = true; + interrupt(); + channel.close(); + } + + public RemoteObject getObj() { + return obj; + } + + public void setObj(RemoteObject obj) { + this.obj = obj; + } +} Index: branches/supervisor/src/main/java/omq/server/InvocationThread.java =================================================================== --- branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 107) +++ branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 108) @@ -1,20 +1,9 @@ package omq.server; -import java.io.IOException; -import java.lang.reflect.InvocationTargetException; -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; -import com.rabbitmq.client.AMQP.BasicProperties; -import com.rabbitmq.client.Channel; import com.rabbitmq.client.ConsumerCancelledException; import com.rabbitmq.client.QueueingConsumer; @@ -28,50 +17,16 @@ * */ -public class InvocationThread extends Thread { +public class InvocationThread extends AInvocationThread { private static final Logger logger = Logger.getLogger(InvocationThread.class.getName()); - private static final String multi = "multi#"; // RemoteObject - private RemoteObject obj; - private String reference; - private String UID; - private Properties env; private boolean idle; private long lastExec; - private RemoteThreadPool pool; - - // Broker - private Broker broker; - private Serializer serializer; - - // Consumer - private Channel channel; - private QueueingConsumer consumer; - private String multiQueue; - private boolean killed = false; - public InvocationThread(RemoteObject obj) throws Exception { - this.obj = obj; - this.UID = obj.getUID(); - this.reference = obj.getRef(); - this.env = obj.getEnv(); - this.broker = obj.getBroker(); - this.pool = obj.getPool(); - this.serializer = broker.getSerializer(); + super(obj); this.lastExec = 0; this.idle = true; - } - - @Override - public synchronized void start() { - try { - startQueues(); - super.start(); - } catch (Exception e) { - logger.error("Cannot start a remoteObject", e); - } - } @@ -87,42 +42,5 @@ idle = false; - String serializerType = delivery.getProperties().getType(); - - // Deserialize the request - Request request = serializer.deserializeRequest(serializerType, delivery.getBody(), obj); - String methodName = request.getMethod(); - String requestID = request.getId(); - - logger.debug("Object: " + obj.getRef() + ", method: " + methodName + " corrID: " + requestID + ", serializerType: " + serializerType); - - // Invoke the method - Object result = null; - OmqException error = null; - try { - result = obj.invokeMethod(request.getMethod(), request.getParams()); - } catch (InvocationTargetException e) { - Throwable throwable = e.getTargetException(); - logger.error("Object: " + obj.getRef() + " at method: " + methodName + ", corrID" + requestID, throwable); - error = new OmqException(throwable.getClass().getCanonicalName(), throwable.getMessage()); - } catch (NoSuchMethodException e) { - logger.error("Object: " + obj.getRef() + " cannot find method: " + methodName); - error = new OmqException(e.getClass().getCanonicalName(), e.getMessage()); - } - - // Reply if it's necessary - if (!request.isAsync()) { - Response resp = new Response(request.getId(), obj.getRef(), result, error); - - BasicProperties props = delivery.getProperties(); - - BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()).build(); - - byte[] bytesResponse = serializer.serialize(serializerType, resp); - channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse); - logger.debug("Publish sync response -> Object: " + obj.getRef() + ", method: " + methodName + " corrID: " + requestID + " replyTo: " - + props.getReplyTo()); - } - - channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); + executeTask(delivery); // The thread is now idle @@ -168,5 +86,5 @@ * @throws Exception */ - private void startQueues() throws Exception { + protected void startQueues() throws Exception { // Start channel channel = broker.getNewChannel(); @@ -190,6 +108,6 @@ channel.queueBind(queue, exchange, routingKey); } - logger.info("RemoteObject: " + reference + " declared direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable + ", Exclusive: " - + exclusive + ", AutoDelete: " + autoDelete); + logger.info("RemoteObject: " + reference + " declared direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable + + ", Exclusive: " + exclusive + ", AutoDelete: " + autoDelete); /* @@ -207,30 +125,7 @@ channel.queueBind(UID, exchange, UID); } + // TODO logger queue + // TODO UID queue should be reference + UID } - - /* - * Multi queue, exclusive per each instance - */ - - // Get info about the multiQueue - String multiExchange = multi + reference; - // TODO:String multiExchange = multi + exchange + reference; - 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: " + reference + " declared fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + multiDurable - + ", Exclusive: " + multiExclusive + ", AutoDelete: " + multiAutoDelete); /* @@ -238,5 +133,5 @@ */ - // Disable Round Robin behavior + // Disable Round Robin behavior boolean autoAck = false; @@ -247,23 +142,7 @@ consumer = new QueueingConsumer(channel); channel.basicConsume(queue, autoAck, consumer); - channel.basicConsume(multiQueue, autoAck, consumer); if (UID != null) { channel.basicConsume(UID, autoAck, consumer); } - } - - public void kill() throws IOException { - logger.info("Killing objectmq: " + reference + " thread id"); - killed = true; - interrupt(); - channel.close(); - } - - public RemoteObject getObj() { - return obj; - } - - public void setObj(RemoteObject obj) { - this.obj = obj; } Index: branches/supervisor/src/main/java/omq/server/MultiInvocationThread.java =================================================================== --- branches/supervisor/src/main/java/omq/server/MultiInvocationThread.java (revision 107) +++ branches/supervisor/src/main/java/omq/server/MultiInvocationThread.java (revision 108) @@ -1,5 +1,107 @@ package omq.server; -public class MultiInvocationThread { +import omq.common.util.ParameterQueue; +import omq.exception.SerializerException; + +import org.apache.log4j.Logger; + +import com.rabbitmq.client.ConsumerCancelledException; +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; +import com.rabbitmq.client.ShutdownSignalException; + +public class MultiInvocationThread extends AInvocationThread { + + private static final Logger logger = Logger.getLogger(MultiInvocationThread.class.getName()); + private static final String multi = "multi#"; + + // Consumer + private String multiQueue; + + public MultiInvocationThread(RemoteObject obj) throws Exception { + super(obj); + } + + @Override + public void run() { + while (!killed) { + try { + // Get the delivery + Delivery delivery = consumer.nextDelivery(); + // This thread does not need to set busy because it's mandatory + // to exist + executeTask(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) { + e.printStackTrace(); + logger.error(e); + } + + } + logger.info("ObjectMQ ('" + obj.getRef() + "') InvocationThread " + Thread.currentThread().getId() + " is killed"); + } + + @Override + protected void startQueues() throws Exception { + // Start channel + channel = broker.getNewChannel(); + + /* + * Multi queue, exclusive per each instance + */ + + // Get info about the multiQueue + String multiExchange = multi + reference; + // TODO:String multiExchange = multi + exchange + reference; + 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: " + reference + " declared fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + + multiDurable + ", Exclusive: " + multiExclusive + ", AutoDelete: " + multiAutoDelete); + + /* + * Consumer + */ + + // Disable Round Robin behavior + boolean autoAck = false; + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(multiQueue, autoAck, consumer); + } } Index: branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java =================================================================== --- branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java (revision 107) +++ branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java (revision 108) @@ -23,4 +23,5 @@ private static final Logger logger = Logger.getLogger(RemoteThreadPool.class.getName()); private List workers; + private MultiInvocationThread multiWorker; private AtomicInteger busy; private int minPoolThreads; @@ -34,5 +35,6 @@ private boolean killed = false; - public RemoteThreadPool(int minPoolThreads, int maxPoolThreads, long refresh, long keepAliveTime, int maxMessagesPerThread, RemoteObject obj, Broker broker) { + public RemoteThreadPool(int minPoolThreads, int maxPoolThreads, long refresh, long keepAliveTime, int maxMessagesPerThread, + RemoteObject obj, Broker broker) { this.minPoolThreads = minPoolThreads; this.maxPoolThreads = maxPoolThreads; @@ -53,6 +55,14 @@ * Create and start minPoolThreads */ - logger.info("ObjectMQ reference: " + obj.getRef() + ", creating: " + minPoolThreads + ", maxPoolThreads: " + maxPoolThreads + ", refresh time: " - + refresh + ", keepAlive: " + keepAliveTime + ", maxMessagesPerThread: " + maxMessagesPerThread); + logger.info("ObjectMQ reference: " + obj.getRef() + ", creating: " + minPoolThreads + ", maxPoolThreads: " + maxPoolThreads + + ", refresh time: " + refresh + ", keepAlive: " + keepAliveTime + ", maxMessagesPerThread: " + maxMessagesPerThread); + + try { + multiWorker = new MultiInvocationThread(obj); + multiWorker.start(); + } catch (Exception e1) { + // TODO Auto-generated catch block + e1.printStackTrace(); + } for (int i = 0; i < minPoolThreads; i++) { @@ -86,5 +96,6 @@ } else if (numWorkers > minPoolThreads && busy.get() < numWorkers) { // Kill idle threads - System.out.println("Kill lazy workers, numWorkers = " + numWorkers + ", minPool = " + minPoolThreads + ", busy = " + busy.get()); + System.out.println("Kill lazy workers, numWorkers = " + numWorkers + ", minPool = " + minPoolThreads + ", busy = " + + busy.get()); stopIdleThreads(); }