source: branches/supervisor/src/main/java/omq/server/InvocationThread.java @ 100

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;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;

/**
 * An invocationThread waits for requests an invokes them.
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
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;
        private boolean idle;
        private long lastExec;

        private RemoteThreadPool pool; // TODO posar això bé

        // 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, Broker broker) throws Exception {
                this.obj = obj;
                this.UID = obj.getRef();
                this.env = obj.getEnv();
                this.broker = broker;
                this.serializer = broker.getSerializer();
                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);
                }

        }

        @Override
        public void run() {
                while (!killed) {
                        try {
                                // Get the delivery
                                Delivery delivery = consumer.nextDelivery();

                                // This thread gets busy
                                pool.getBusy().incrementAndGet();
                                idle = false;

                                String serializerType = delivery.getProperties().getType();

                                // Deserialize the json
                                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);

                                // The thread is now idle
                                lastExec = System.currentTimeMillis();
                                idle = true;
                                pool.getBusy().decrementAndGet();

                        } 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);
                        }

                }
        }

        /**
         * 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();
        }

        public RemoteObject getObj() {
                return obj;
        }

        public void setObj(RemoteObject obj) {
                this.obj = obj;
        }

        public long getLastExecution() {
                return lastExec;
        }

        public boolean isIdle() {
                return idle;
        }

}