source: branches/supervisor/src/main/java/omq/client/proxy/Proxymq.java

package omq.client.proxy;

import java.lang.reflect.Array;
import java.lang.reflect.InvocationHandler;
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.client.annotation.AsyncMethod;
import omq.client.annotation.MultiMethod;
import omq.client.annotation.SyncMethod;
import omq.client.listener.ResponseListener;
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.RetryException;
import omq.exception.TimeoutException;

import org.apache.log4j.Logger;

import com.rabbitmq.client.AMQP.BasicProperties;

/**
 * Proxymq class. This class inherits from InvocationHandler, for this reason
 * each proxymq instance has an associated invocation handler. When a method is
 * invoked on a proxymq instance, the method invocation is encoded and
 * dispatched to the invoke method of its invocation handler.
 * 
 * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
 * 
 */
public class Proxymq implements InvocationHandler, Remote {

        /**
         * 
         */
        private static final long serialVersionUID = 1L;
        private static final Logger logger = Logger.getLogger(Proxymq.class.getName());
        private static final String multi = "multi#";

        private String reference;
        private String UID;
        private transient String exchange;
        private transient String multiExchange;
        private transient String replyQueueName;
        private transient String serializerType;
        private transient Broker broker;
        private transient ResponseListener rListener;
        private transient Serializer serializer;
        private transient Properties env;
        private transient Integer deliveryMode = null;
        private transient Map<String, byte[]> results;

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

        /**
         * Proxymq Constructor.
         * 
         * This constructor uses an reference to know which object will call. It
         * also uses Properties to set where to send the messages
         * 
         * @param reference
         *            The reference represents the unique identifier of a remote
         *            object
         * @param clazz
         *            It represents the real class of the remote object. With this
         *            class the system can know the remoteInterface used and it can
         *            also see which annotations are used
         * @param env
         *            The environment is used to know where to send the messages
         * @throws Exception
         */
        public Proxymq(String reference, Class<?> clazz, Broker broker) throws Exception {
                this.reference = reference;
                this.broker = broker;
                rListener = broker.getResponseListener();
                serializer = broker.getSerializer();

                // TODO what is better to use a new channel or to use the same?
                // this.channel = Broker.getChannel();
                env = broker.getEnvironment();
                exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE, "");
                multiExchange = multi + reference;
                replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);

                // set the serializer type
                serializerType = env.getProperty(ParameterQueue.PROXY_SERIALIZER, Serializer.JAVA);
                if (env.getProperty(ParameterQueue.DELIVERY_MODE) != null) {
                        deliveryMode = Integer.parseInt(env.getProperty(ParameterQueue.DELIVERY_MODE));
                }

                // Create a new hashmap and registry it in rListener
                results = new HashMap<String, byte[]>();
                rListener.registerProxy(this);
        }

        @Override
        public Object invoke(Object proxy, Method method, Object[] arguments) throws Throwable {
                // Local methods only
                String methodName = method.getName();

                // The local methods will be invoked here
                if (method.getDeclaringClass().equals(Remote.class)) {
                        if (methodName.equals("getRef")) {
                                return getRef();
                        }
                        if (methodName.equals("getUID")) {
                                return getUID();
                        }
                        if (methodName.equals("setUID")) {
                                setUID((String) arguments[0]);
                                return null;
                        }
                        if (methodName.equals("equals")) {
                                if (arguments[0] instanceof Remote) {
                                        return getRef().equals(((Remote) arguments[0]).getRef());
                                } else {
                                        return false;
                                }
                        }
                }

                // Create the request
                Request request = createRequest(method, arguments);

                Object response = null;
                // Publish the request
                if (request.isAsync()) {
                        publishMessage(request, replyQueueName);
                } else {
                        response = publishSyncRequest(request, method.getReturnType());
                }

                return response;
        }

        /**
         * This method publishes a request
         * 
         * @param request
         *            - this request contains which method and which params will be
         *            invoked in the server side.
         * @param replyQueueName
         *            - this param indicates where the responseListener will be
         *            listen to.
         * @throws Exception
         */
        private void publishMessage(Request request, String replyQueueName) throws Exception {
                String corrId = request.getId();

                // Get the environment properties
                String exchange;
                String routingkey;

                if (request.isMulti()) {
                        exchange = multiExchange;
                        routingkey = "";
                } else {
                        exchange = this.exchange;
                        routingkey = reference;
                }

                // TODO look this carefully
                String appId = UID == null ? reference : UID;

                // Add the correlation ID and create a replyTo property
                BasicProperties props = new BasicProperties.Builder().appId(appId).correlationId(corrId).replyTo(replyQueueName)
                                .type(serializerType).deliveryMode(deliveryMode).build();

                // Publish the message
                byte[] bytesRequest = serializer.serialize(serializerType, request);
                broker.publishMessge(exchange, routingkey, props, bytesRequest);
                logger.debug("Proxymq: " + reference + " invokes '" + request.getMethod() + "' , corrID: " + corrId + ", exchange: " + exchange
                                + ", replyQueue: " + replyQueueName + ", serializerType: " + serializerType + ", multi call: " + request.isMulti()
                                + ", async call: " + request.isAsync() + ", delivery mode: " + deliveryMode);
        }

        /**
         * This method publishes a synchronous request
         * 
         * @param request
         *            - this request contains which method and which params will be
         *            invoked in the server side.
         * @param type
         *            - indicates which return type we are waiting for
         * @return serverResponse
         * @throws Exception
         */
        private Object publishSyncRequest(Request request, Class<?> type) throws Exception {
                String corrId = request.getId();

                int retries = request.getRetries();
                long timeout = request.getTimeout();

                // Publish the message
                int i = 0;
                while (i < retries) {
                        try {
                                publishMessage(request, replyQueueName);
                                if (request.isMulti()) {
                                        return getResults(corrId, timeout, type);
                                } else {
                                        return getResult(corrId, timeout, type);
                                }

                        } catch (TimeoutException te) {
                                logger.error(te);
                        }
                        i++;
                }
                throw new RetryException(retries, timeout);
        }

        /**
         * This method creates a request using the annotations of the Remote
         * interface
         * 
         * @param method
         *            - method to invoke in the server side
         * @param arguments
         *            - arguments of the method
         * @return new Request
         */
        private Request createRequest(Method method, Object[] arguments) {
                String corrId = java.util.UUID.randomUUID().toString();
                String methodName = method.getName();
                boolean multi = false;

                if (method.getAnnotation(MultiMethod.class) != null) {
                        multi = true;
                }

                // Since we need to know whether the method is async and if it has to
                // return using an annotation, we'll only check the AsyncMethod
                // annotation
                if (method.getAnnotation(AsyncMethod.class) == null) {
                        int retries = 1;
                        long timeout = ParameterQueue.DEFAULT_TIMEOUT;
                        if (method.getAnnotation(SyncMethod.class) != null) {
                                SyncMethod sync = method.getAnnotation(SyncMethod.class);
                                retries = sync.retry();
                                timeout = sync.timeout();
                        }
                        return Request.newSyncRequest(corrId, methodName, arguments, retries, timeout, multi);
                } else {
                        return Request.newAsyncRequest(corrId, methodName, arguments, multi);
                }
        }

        private Object getResult(String corrId, long timeout, Class<?> type) throws Exception {
                Response resp = null;

                // Wait for the results.
                long localTimeout = 0;
                long start = System.currentTimeMillis();
                synchronized (results) {
                        // Due to we are using notifyAll(), we need to control the real time
                        while (!results.containsKey(corrId) && (timeout - localTimeout) >= 0) {
                                results.wait(localTimeout);
                                localTimeout = System.currentTimeMillis() - start;
                        }
                        if ((timeout - localTimeout) <= 0) {
                                throw new TimeoutException("Timeout exception time: " + timeout);
                        }
                        resp = serializer.deserializeResponse(results.get(corrId), type);

                        // Remove and indicate the key exists (a hashmap can contain a null
                        // object, using this we'll know whether a response has been
                        // received before)
                        results.put(corrId, null);
                }

                if (resp.getError() != null) {
                        OmqException error = resp.getError();
                        String name = error.getType();
                        String message = error.getMessage();
                        throw (Exception) Class.forName(name).getConstructor(String.class).newInstance(message);
                }

                return resp.getResult();
        }

        /**
         * This method returns an array with length @MultiMethod.waitNum() with all
         * the responses received.
         * 
         * @param corrId
         *            - Correlation Id of the request
         * @param timeout
         *            - Timeout read in @SyncMethod.timeout(). If the timeout is set
         *            in 2 seconds, the system will wait 2 seconds for the arriving
         *            of all the responses.
         * @param type
         *            - Must be an Array type
         * @return resultArray
         * @throws Exception
         */
        private Object getResults(String corrId, long timeout, Class<?> type) throws Exception {
                Response resp = null;
                // Get the component type of an array
                Class<?> actualType = type.getComponentType();

                List<Object> list = new ArrayList<Object>();

                int i = 0;
                long localTimeout = 0;
                long start = System.currentTimeMillis();

                while (true) {
                        synchronized (results) {
                                // Due to we are using notifyAll(), we need to control the real
                                // time
                                while (!results.containsKey(corrId) && (timeout - localTimeout) >= 0) {
                                        results.wait(localTimeout);
                                        localTimeout = System.currentTimeMillis() - start;
                                }
                                if ((timeout - localTimeout) <= 0) {
                                        break;
                                }
                                // Remove the corrId to receive new replies
                                resp = serializer.deserializeResponse(results.remove(corrId), actualType);
                                list.add(resp.getResult());
                        }
                        i++;
                }

                if (i == 0) {
                        results.remove(corrId);
                        throw new TimeoutException("Timeout exception time: " + timeout);
                }

                synchronized (results) {
                        results.put(corrId, null);
                }

                Object array = Array.newInstance(actualType, i);
                i = 0;
                for (Object o : list) {
                        Array.set(array, i++, o);
                }

                return array;
        }

        /**
         * Gets the Map used internally to retreive the response of the server
         * 
         * @return a map with all the keys processed. Every key is a correlation id
         *         of a method invoked remotely
         */
        public Map<String, byte[]> getResults() {
                return results;
        }

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

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

        @Override
        public void setUID(String uID) {
                this.UID = uID;
        }

}