source: branches/supervisor/src/main/java/omq/server/RemoteObject.java @ 92

package omq.server;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import omq.Remote;
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;

/**
 * A RemoteObject when it's started will be waiting for requests and will invoke
 * them. When a RemoteObject is started it listens two queues, the first one has
 * the same name as its reference and the second one is its multiqueue -this
 * name can be set using a property, be aware to use a name not used by another
 * object!!!-.
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
public abstract class RemoteObject extends Thread implements Remote {

        private static final long serialVersionUID = -1778953938739846450L;
        private static final String multi = "multi#";
        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<String, List<Class<?>>> params;
        private transient Channel channel;
        private transient QueueingConsumer consumer;
        private transient boolean killed = false;

        private static final Map<String, Class<?>> primitiveClasses = new HashMap<String, Class<?>>();

        static {
                primitiveClasses.put("byte", Byte.class);
                primitiveClasses.put("short", Short.class);
                primitiveClasses.put("char", Character.class);
                primitiveClasses.put("int", Integer.class);
                primitiveClasses.put("long", Long.class);
                primitiveClasses.put("float", Float.class);
                primitiveClasses.put("double", Double.class);
        }

        public RemoteObject() {
        }

        /**
         * This method starts a remoteObject.
         * 
         * @param reference
         *            - broker's binding referece
         * @param broker
         *            - broker that binds this remoteObject
         * @param env
         *            - properties of this remoteObject
         * @throws Exception
         */
        public void startRemoteObject(String reference, Broker broker, Properties env) throws Exception {
                this.broker = broker;
                this.UID = reference;
                this.env = env;

                this.params = new HashMap<String, List<Class<?>>>();
                for (Method m : this.getClass().getMethods()) {
                        List<Class<?>> list = new ArrayList<Class<?>>();
                        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();
        }

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

        @Override
        public String getRef() {
                return UID;
        }

        /**
         * This method kills all the threads waiting for requests
         * 
         * @throws IOException
         *             - If an operation failed.
         */
        public void kill() throws IOException {
                logger.warn("Killing objectmq: " + this.getRef());
                killed = true;
                interrupt();
                channel.close();
                remoteWrapper.stopRemoteWrapper();
        }

        /**
         * This method invokes the method specified by methodName and arguments
         * 
         * @param methodName
         * @param arguments
         * @return result
         * @throws Exception
         */
        public Object invokeMethod(String methodName, Object[] arguments) throws Exception {

                // Get the specific method identified by methodName and its arguments
                Method method = loadMethod(methodName, arguments);

                return method.invoke(this, arguments);
        }

        /**
         * This method loads the method specified by methodName and args
         * 
         * @param methodName
         * @param args
         * @return method
         * @throws NoSuchMethodException
         *             - If the method cannot be found
         */
        private Method loadMethod(String methodName, Object[] args) throws NoSuchMethodException {
                Method m = null;

                // Obtain the class reference
                Class<?> clazz = this.getClass();
                Class<?>[] argArray = null;

                if (args != null) {
                        argArray = new Class<?>[args.length];
                        for (int i = 0; i < args.length; i++) {
                                argArray[i] = args[i].getClass();
                        }
                }

                try {
                        m = clazz.getMethod(methodName, argArray);
                } catch (NoSuchMethodException nsm) {
                        m = loadMethodWithPrimitives(methodName, argArray);
                }
                return m;
        }

        /**
         * This method loads a method which uses primitives as arguments
         * 
         * @param methodName
         *            - name of the method wanted to invoke
         * @param argArray
         *            - arguments
         * @return method
         * @throws NoSuchMethodException
         *             - If the method cannot be found
         */
        private Method loadMethodWithPrimitives(String methodName, Class<?>[] argArray) throws NoSuchMethodException {
                if (argArray != null) {
                        Method[] methods = this.getClass().getMethods();
                        int length = argArray.length;

                        for (Method method : methods) {
                                String name = method.getName();
                                int argsLength = method.getParameterTypes().length;

                                if (name.equals(methodName) && length == argsLength) {
                                        // This array can have primitive types inside
                                        Class<?>[] params = method.getParameterTypes();

                                        boolean found = true;

                                        for (int i = 0; i < length; i++) {
                                                if (params[i].isPrimitive()) {
                                                        Class<?> paramWrapper = primitiveClasses.get(params[i].getName());

                                                        if (!paramWrapper.equals(argArray[i])) {
                                                                found = false;
                                                                break;
                                                        }
                                                }
                                        }
                                        if (found) {
                                                return method;
                                        }
                                }
                        }
                }
                throw new NoSuchMethodException(methodName);
        }

        public List<Class<?>> getParams(String methodName) {
                return params.get(methodName);
        }

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

}