Index: jectmq/src/omq/client/proxy/EvoProxy.java =================================================================== --- /objectmq/src/omq/client/proxy/EvoProxy.java (revision 3) +++ (revision ) @@ -1,315 +1,0 @@ -package omq.client.proxy; - -import java.lang.reflect.InvocationHandler; -import java.lang.reflect.Method; -import java.lang.reflect.Proxy; -import java.util.Collection; -import java.util.HashMap; -import java.util.Hashtable; -import java.util.Map; -import java.util.Properties; -import java.util.Vector; - -import omq.Remote; -import omq.client.annotation.AsyncMethod; -import omq.client.annotation.RemoteInterface; -import omq.client.annotation.SyncMethod; -import omq.client.remote.response.ResponseListener; -import omq.common.event.EventDispatcher; -import omq.common.event.EventListener; -import omq.common.message.request.AsyncRequest; -import omq.common.message.request.Request; -import omq.common.message.request.SyncRequest; -import omq.common.message.response.Response; -import omq.common.util.ParameterQueue; -import omq.exception.NoContainsInstanceException; - - -import com.rabbitmq.client.Channel; - -/** - * EvoProxy class. This class inherits from InvocationHandler and gives you a - * proxy with a server using an environment - * - * @author Sergi Toda - * - */ -public class EvoProxy implements InvocationHandler, Remote { - - /** - * - */ - private static final long serialVersionUID = 1L; - private static Map proxies = new Hashtable(); - - private String objUid; - private transient ResponseListener rListener; - private transient EventDispatcher dispatcher; - private transient Channel channel; - private transient Properties env; - private transient Map results; - private transient Map methodTypes; - private transient Map listeners; - - /** - * EvoProxy Constructor. - * - * This constructor uses an objUid to know which object will call. It also - * uses Properties to set where to send the messages - * - * @param objUid - * The objUid represents the unique identifier of a remote object - * @param env - * The environment is used to know where to send the messages - * @throws Exception - */ - public EvoProxy(String objUid, Properties env) throws Exception { - this.objUid = objUid; - this.rListener = ResponseListener.getRequestListener(); - this.dispatcher = EventDispatcher.getDispatcher(); - this.channel = rListener.getChannel(); - this.env = env; - - listeners = new HashMap(); - - // Without the remote object's class, we cannot know which methods have - // annotations. For this reason this map will be void. - methodTypes = new HashMap(); - - // Create a new hashmap and registry it in rListener. - results = new HashMap(); - rListener.registerProxy(this); - } - - /** - * EvoProxy Constructor. - * - * This constructor uses an objUid to know which object will call. It also - * uses Properties to set where to send the messages - * - * @param objUid - * The objUid 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 EvoProxy(String objUid, Class clazz, Properties env) throws Exception { - this.objUid = objUid; - this.rListener = ResponseListener.getRequestListener(); - this.dispatcher = EventDispatcher.getDispatcher(); - this.channel = rListener.getChannel(); - this.env = env; - - listeners = new HashMap(); - - // Get annotated Methods - methodTypes = getAnnotatedMethods(clazz); - - // Create a new hashmap and registry it in rListener - results = new HashMap(); - 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(); - } else if (methodName.equals("addListener")) { - addListener((EventListener) arguments[0]); - return null; - } else if (methodName.equals("removeListener")) { - removeListener((EventListener) arguments[0]); - return null; - } else if (methodName.equals("getListeners")) { - return getListeners(); - } - // else if (methodName.equals("notify")) { - // // notify fanout - // } - } - - // If the method was not a local method, it'll be sent in a queue - // Publish the method we want to invoke - Request request = createRequest(method, arguments); - - return request.publishRequest(results, env, channel); - } - - /** - * This method creates a new request using a method and it's arguments - * - * @param method - * Method to be called - * @param arguments - * Arguments of this method - * @return request - */ - private Request createRequest(Method method, Object[] arguments) { - String corrId = java.util.UUID.randomUUID().toString(); - String methodName = method.getName(); - Vector args = new Vector(); - - if (arguments != null) { - for (Object o : arguments) { - args.add(o); - } - } - - // 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 new SyncRequest(this.objUid, corrId, methodName, args, timeout, retries); - } else { - String topic = method.getAnnotation(AsyncMethod.class).generateEvent(); - return new AsyncRequest(this.objUid, corrId, methodName, args, topic); - } - } - - /** - * This method searches annotations in a RemoteInterface. Every time it - * finds an annotation saves it in a Map. - * - * @param clazz - * @return annotatedMethods - */ - private Map getAnnotatedMethods(Class clazz) { - Class[] interfaces = clazz.getInterfaces(); - Map methodTypes = new HashMap(); - - // We look for the dermi interface. Once it's found, we read its methods - // and its annotations and we save them into a Map - for (Class inter : interfaces) { - if (inter.getAnnotation(RemoteInterface.class) != null) { - // Read the methods - for (Method method : inter.getMethods()) { - if (method.getAnnotation(AsyncMethod.class) != null) { - methodTypes.put(method, CallType.ASYNC); - } else if (method.getAnnotation(SyncMethod.class) != null) { - methodTypes.put(method, CallType.SYNC); - } - } - - break; - } - } - return methodTypes; - } - - /** - * - * @param reference - * RemoteObject reference - * @return true if the proxy has been created before or false in the other - * case - */ - public static boolean containsProxy(String reference) { - return proxies.containsKey(reference); - } - - /** - * - * @param reference - * RemoteObject reference - * @return a proxy instance - * @throws NoContainsInstanceException - */ - public static Object getInstance(String reference) throws NoContainsInstanceException { - if (!containsProxy(reference)) { - throw new NoContainsInstanceException(reference); - } - return proxies.get(reference); - } - - /** - * Returns an instance of a proxy class for the specified interfaces that - * dispatches method invocations to the specified invocation handler. * @param - * loader - * - * @param loader - * the class loader to define the proxy class - * - * @param interfaces - * the list of interfaces for the proxy class to implement - * @param proxy - * the invocation handler to dispatch method invocations to - * @return a proxy instance with the specified invocation handler of a proxy - * class that is defined by the specified class loader and that - * implements the specified interfaces - */ - public static Object newProxyInstance(ClassLoader loader, Class[] interfaces, EvoProxy proxy) { - if (proxies.containsKey(proxy.getRef())) { - System.out.println("Proxy trobat"); - return proxies.get(proxy.getRef()); - } - Object value = Proxy.newProxyInstance(loader, interfaces, proxy); - proxies.put(proxy.getRef(), value); - return value; - } - - /** - * 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 getResults() { - return results; - } - - /** - * Local method that enables to addListeners to this Proxy - */ - public void addListener(EventListener eventListener) throws Exception { - if (eventListener.getTopic() == null) { - eventListener.setTopic(objUid); - } - listeners.put(eventListener.getTopic(), eventListener); - dispatcher.addListener(eventListener); - } - - /** - * Local method to remove listeners of this proxy - */ - public void removeListener(EventListener eventListener) throws Exception { - listeners.remove(eventListener.getTopic()); - dispatcher.removeListener(eventListener); - } - - @Override - public String getRef() { - return objUid; - } - - @Override - public Response invokeMethod(String methodName, Vector args) throws Exception { - return null; - } - - @Override - public Collection getListeners() throws Exception { - return listeners.values(); - } - - @Override - public void notify(Object obj) { - - } - -} Index: /objectmq/src/omq/client/proxy/Proxymq.java =================================================================== --- /objectmq/src/omq/client/proxy/Proxymq.java (revision 4) +++ /objectmq/src/omq/client/proxy/Proxymq.java (revision 4) @@ -0,0 +1,311 @@ +package omq.client.proxy; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.Collection; +import java.util.HashMap; +import java.util.Hashtable; +import java.util.Map; +import java.util.Properties; +import java.util.Vector; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; +import omq.client.remote.response.ResponseListener; +import omq.common.event.EventDispatcher; +import omq.common.event.EventListener; +import omq.common.message.request.AsyncRequest; +import omq.common.message.request.Request; +import omq.common.message.request.SyncRequest; +import omq.common.message.response.Response; +import omq.common.util.ParameterQueue; +import omq.exception.NoContainsInstanceException; + +import com.rabbitmq.client.Channel; + +/** + * EvoProxy class. This class inherits from InvocationHandler and gives you a + * proxy with a server using an environment + * + * @author Sergi Toda + * + */ +public class Proxymq implements InvocationHandler, Remote { + + /** + * + */ + private static final long serialVersionUID = 1L; + private static Map proxies = new Hashtable(); + + private String uid; + private transient ResponseListener rListener; + private transient EventDispatcher dispatcher; + private transient Channel channel; + private transient Properties env; + private transient Map results; + private transient Map methodTypes; + private transient Map listeners; + + /** + * EvoProxy Constructor. + * + * This constructor uses an uid to know which object will call. It also uses + * Properties to set where to send the messages + * + * @param uid + * The uid represents the unique identifier of a remote object + * @param env + * The environment is used to know where to send the messages + * @throws Exception + */ + public Proxymq(String uid, Properties env) throws Exception { + this.uid = uid; + this.rListener = ResponseListener.getRequestListener(); + this.dispatcher = EventDispatcher.getDispatcher(); + this.channel = rListener.getChannel(); + this.env = env; + + listeners = new HashMap(); + + // Without the remote object's class, we cannot know which methods have + // annotations. For this reason this map will be void. + methodTypes = new HashMap(); + + // Create a new hashmap and registry it in rListener. + results = new HashMap(); + rListener.registerProxy(this); + } + + /** + * EvoProxy Constructor. + * + * This constructor uses an uid to know which object will call. It also uses + * Properties to set where to send the messages + * + * @param uid + * The uid 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 uid, Class clazz, Properties env) throws Exception { + this.uid = uid; + this.rListener = ResponseListener.getRequestListener(); + this.dispatcher = EventDispatcher.getDispatcher(); + this.channel = rListener.getChannel(); + this.env = env; + + listeners = new HashMap(); + + // Get annotated Methods + methodTypes = getAnnotatedMethods(clazz); + + // Create a new hashmap and registry it in rListener + results = new HashMap(); + 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(); + } else if (methodName.equals("addListener")) { + addListener((EventListener) arguments[0]); + return null; + } else if (methodName.equals("removeListener")) { + removeListener((EventListener) arguments[0]); + return null; + } else if (methodName.equals("getListeners")) { + return getListeners(); + } + } + + // If the method was not a local method, it'll be sent in a queue + // Publish the method we want to invoke + Request request = createRequest(method, arguments); + + return request.publishRequest(results, env, channel); + } + + /** + * This method creates a new request using a method and it's arguments + * + * @param method + * Method to be called + * @param arguments + * Arguments of this method + * @return request + */ + private Request createRequest(Method method, Object[] arguments) { + String corrId = java.util.UUID.randomUUID().toString(); + String methodName = method.getName(); + Vector args = new Vector(); + + if (arguments != null) { + for (Object o : arguments) { + args.add(o); + } + } + + // 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 new SyncRequest(this.uid, corrId, methodName, args, timeout, retries); + } else { + String topic = method.getAnnotation(AsyncMethod.class).generateEvent(); + return new AsyncRequest(this.uid, corrId, methodName, args, topic); + } + } + + /** + * This method searches annotations in a RemoteInterface. Every time it + * finds an annotation saves it in a Map. + * + * @param clazz + * @return annotatedMethods + */ + private Map getAnnotatedMethods(Class clazz) { + Class[] interfaces = clazz.getInterfaces(); + Map methodTypes = new HashMap(); + + // We look for the dermi interface. Once it's found, we read its methods + // and its annotations and we save them into a Map + for (Class inter : interfaces) { + if (inter.getAnnotation(RemoteInterface.class) != null) { + // Read the methods + for (Method method : inter.getMethods()) { + if (method.getAnnotation(AsyncMethod.class) != null) { + methodTypes.put(method, CallType.ASYNC); + } else if (method.getAnnotation(SyncMethod.class) != null) { + methodTypes.put(method, CallType.SYNC); + } + } + + break; + } + } + return methodTypes; + } + + /** + * + * @param reference + * RemoteObject reference + * @return true if the proxy has been created before or false in the other + * case + */ + public static boolean containsProxy(String reference) { + return proxies.containsKey(reference); + } + + /** + * + * @param reference + * RemoteObject reference + * @return a proxy instance + * @throws NoContainsInstanceException + */ + public static Object getInstance(String reference) throws NoContainsInstanceException { + if (!containsProxy(reference)) { + throw new NoContainsInstanceException(reference); + } + return proxies.get(reference); + } + + /** + * Returns an instance of a proxy class for the specified interfaces that + * dispatches method invocations to the specified invocation handler. * @param + * loader + * + * @param loader + * the class loader to define the proxy class + * + * @param interfaces + * the list of interfaces for the proxy class to implement + * @param proxy + * the invocation handler to dispatch method invocations to + * @return a proxy instance with the specified invocation handler of a proxy + * class that is defined by the specified class loader and that + * implements the specified interfaces + */ + public static Object newProxyInstance(ClassLoader loader, Class[] interfaces, Proxymq proxy) { + if (proxies.containsKey(proxy.getRef())) { + System.out.println("Proxy trobat"); + return proxies.get(proxy.getRef()); + } + Object value = Proxy.newProxyInstance(loader, interfaces, proxy); + proxies.put(proxy.getRef(), value); + return value; + } + + /** + * 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 getResults() { + return results; + } + + /** + * Local method that enables to addListeners to this Proxy + */ + public void addListener(EventListener eventListener) throws Exception { + if (eventListener.getTopic() == null) { + eventListener.setTopic(uid); + } + listeners.put(eventListener.getTopic(), eventListener); + dispatcher.addListener(eventListener); + } + + /** + * Local method to remove listeners of this proxy + */ + public void removeListener(EventListener eventListener) throws Exception { + listeners.remove(eventListener.getTopic()); + dispatcher.removeListener(eventListener); + } + + @Override + public String getRef() { + return uid; + } + + @Override + public Response invokeMethod(String methodName, Vector args) throws Exception { + return null; + } + + @Override + public Collection getListeners() throws Exception { + return listeners.values(); + } + + @Override + public void notify(Object obj) { + + } + +} Index: /objectmq/src/omq/client/remote/response/ResponseListener.java =================================================================== --- /objectmq/src/omq/client/remote/response/ResponseListener.java (revision 3) +++ /objectmq/src/omq/client/remote/response/ResponseListener.java (revision 4) @@ -5,5 +5,5 @@ import java.util.Properties; -import omq.client.proxy.EvoProxy; +import omq.client.proxy.Proxymq; import omq.common.message.response.Response; import omq.common.remote.RemoteListener; @@ -157,5 +157,5 @@ // Revisar això - public void registerProxy(EvoProxy proxy) { + public void registerProxy(Proxymq proxy) { if (!results.containsKey(proxy.getRef())) { results.put(proxy.getRef(), proxy.getResults()); Index: /objectmq/src/omq/common/broker/Broker.java =================================================================== --- /objectmq/src/omq/common/broker/Broker.java (revision 4) +++ /objectmq/src/omq/common/broker/Broker.java (revision 4) @@ -0,0 +1,59 @@ +package omq.common.broker; + +import java.util.Properties; + +import omq.Remote; +import omq.client.proxy.Proxymq; +import omq.common.remote.OmqConnectionFactory; +import omq.exception.RemoteException; +import omq.server.remote.request.RemoteObject; + +import com.rabbitmq.client.Connection; + +public class Broker { + private static Connection connection; + private static Properties environment; + + public static void initBroker(Properties env) throws Exception { + if (environment != null) { + connection = OmqConnectionFactory.getConnection(env); + } + } + + // TODO: what happens if the connection is not set + public static Connection getConnection() throws Exception { + return connection; + } + + public static Remote lookup(String reference, Class contract) throws RemoteException { + try { + if (Proxymq.containsProxy(reference)) { + Proxymq proxy = new Proxymq(reference, contract, null); + Class[] array = { contract }; + return (Remote) Proxymq.newProxyInstance(contract.getClassLoader(), array, proxy); + } + return (Remote) Proxymq.getInstance(reference); + + } catch (Exception e) { + throw new RemoteException(e); + } + } + + public static void bind(String reference, Remote obj) throws RemoteException { + try { + RemoteObject remote = (RemoteObject) obj; + remote.start(reference, environment); + } catch (Exception e) { + throw new RemoteException(e); + } + } + + public static void unbind(String reference) throws RemoteException { + + } + + public void rebind(String name, Remote obj) throws RemoteException { + + } + +} Index: /objectmq/src/omq/common/event/EventDispatcher.java =================================================================== --- /objectmq/src/omq/common/event/EventDispatcher.java (revision 3) +++ /objectmq/src/omq/common/event/EventDispatcher.java (revision 4) @@ -6,5 +6,5 @@ import java.util.Vector; -import omq.common.remote.RevoConnectionFactory; +import omq.common.remote.OmqConnectionFactory; import omq.common.util.ParameterQueue; import omq.common.util.Serializer; @@ -43,5 +43,5 @@ // Get a new connection and a new channel - connection = RevoConnectionFactory.getConnection(env); + connection = OmqConnectionFactory.getConnection(env); channel = connection.createChannel(); Index: /objectmq/src/omq/common/event/EventTrigger.java =================================================================== --- /objectmq/src/omq/common/event/EventTrigger.java (revision 3) +++ /objectmq/src/omq/common/event/EventTrigger.java (revision 4) @@ -6,5 +6,5 @@ import java.util.Properties; -import omq.common.remote.RevoConnectionFactory; +import omq.common.remote.OmqConnectionFactory; import omq.common.util.RevoEnvironment; import omq.common.util.Serializer; @@ -28,5 +28,5 @@ private EventTrigger(Properties env) throws IOException, KeyManagementException, NoSuchAlgorithmException { - connection = RevoConnectionFactory.getConnection(env); + connection = OmqConnectionFactory.getConnection(env); channel = connection.createChannel(); } Index: /objectmq/src/omq/common/message/response/ProxyResponse.java =================================================================== --- /objectmq/src/omq/common/message/response/ProxyResponse.java (revision 3) +++ /objectmq/src/omq/common/message/response/ProxyResponse.java (revision 4) @@ -3,5 +3,5 @@ import java.util.Properties; -import omq.client.proxy.EvoProxy; +import omq.client.proxy.Proxymq; @@ -31,11 +31,11 @@ @Override public Object getResp(Properties env) throws Exception { - if (!EvoProxy.containsProxy(reference)) { + if (!Proxymq.containsProxy(reference)) { Class clazz = Class.forName(remoteInterface); - EvoProxy evoProxy = new EvoProxy(reference, clazz, env); + Proxymq evoProxy = new Proxymq(reference, clazz, env); Class[] array = { clazz }; - return EvoProxy.newProxyInstance(clazz.getClassLoader(), array, evoProxy); + return Proxymq.newProxyInstance(clazz.getClassLoader(), array, evoProxy); } - return EvoProxy.getInstance(reference); + return Proxymq.getInstance(reference); } Index: /objectmq/src/omq/common/remote/OmqConnectionFactory.java =================================================================== --- /objectmq/src/omq/common/remote/OmqConnectionFactory.java (revision 4) +++ /objectmq/src/omq/common/remote/OmqConnectionFactory.java (revision 4) @@ -0,0 +1,42 @@ +package omq.common.remote; + +import java.io.IOException; +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.util.Properties; + +import omq.common.util.ParameterQueue; + + +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + +/** + * + * @author Sergi Toda + * + */ +public class OmqConnectionFactory { + public static Connection getConnection(Properties env) throws IOException, KeyManagementException, NoSuchAlgorithmException { + // Get login info of rabbitmq + String username = env.getProperty(ParameterQueue.USER_NAME); + String password = env.getProperty(ParameterQueue.USER_PASS); + + // Get host info of rabbimq (where it is) + String host = env.getProperty(ParameterQueue.SERVER_HOST); + int port = Integer.parseInt(env.getProperty(ParameterQueue.SERVER_PORT)); + + boolean ssl = Boolean.parseBoolean(env.getProperty(ParameterQueue.ENABLE_SSL)); + + // Start a new connection and channel + ConnectionFactory factory = new ConnectionFactory(); + factory.setUsername(username); + factory.setPassword(password); + factory.setHost(host); + factory.setPort(port); + if (ssl) { + factory.useSslProtocol(); + } + return factory.newConnection(); + } +} Index: /objectmq/src/omq/common/remote/RemoteListener.java =================================================================== --- /objectmq/src/omq/common/remote/RemoteListener.java (revision 3) +++ /objectmq/src/omq/common/remote/RemoteListener.java (revision 4) @@ -37,5 +37,5 @@ private void startConnection(Properties env) throws Exception { - connection = RevoConnectionFactory.getConnection(env); + connection = OmqConnectionFactory.getConnection(env); channel = connection.createChannel(); } Index: jectmq/src/omq/common/remote/RevoConnectionFactory.java =================================================================== --- /objectmq/src/omq/common/remote/RevoConnectionFactory.java (revision 3) +++ (revision ) @@ -1,42 +1,0 @@ -package omq.common.remote; - -import java.io.IOException; -import java.security.KeyManagementException; -import java.security.NoSuchAlgorithmException; -import java.util.Properties; - -import omq.common.util.ParameterQueue; - - -import com.rabbitmq.client.Connection; -import com.rabbitmq.client.ConnectionFactory; - -/** - * - * @author Sergi Toda - * - */ -public class RevoConnectionFactory { - public static Connection getConnection(Properties env) throws IOException, KeyManagementException, NoSuchAlgorithmException { - // Get login info of rabbitmq - String username = env.getProperty(ParameterQueue.USER_NAME); - String password = env.getProperty(ParameterQueue.USER_PASS); - - // Get host info of rabbimq (where it is) - String host = env.getProperty(ParameterQueue.SERVER_HOST); - int port = Integer.parseInt(env.getProperty(ParameterQueue.SERVER_PORT)); - - boolean ssl = Boolean.parseBoolean(env.getProperty(ParameterQueue.ENABLE_SSL)); - - // Start a new connection and channel - ConnectionFactory factory = new ConnectionFactory(); - factory.setUsername(username); - factory.setPassword(password); - factory.setHost(host); - factory.setPort(port); - if (ssl) { - factory.useSslProtocol(); - } - return factory.newConnection(); - } -} Index: /objectmq/src/omq/common/util/ParameterQueue.java =================================================================== --- /objectmq/src/omq/common/util/ParameterQueue.java (revision 3) +++ /objectmq/src/omq/common/util/ParameterQueue.java (revision 4) @@ -35,4 +35,6 @@ public static String RPC_TYPE = "direct"; + public static String NUM_THREADS = "omq.num_threads"; + public static String REGISTRY_NAME = "REGISTRY"; Index: /objectmq/src/omq/server/remote/request/RemoteObject.java =================================================================== --- /objectmq/src/omq/server/remote/request/RemoteObject.java (revision 3) +++ /objectmq/src/omq/server/remote/request/RemoteObject.java (revision 4) @@ -1,23 +1,29 @@ package omq.server.remote.request; +import java.io.IOException; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; -import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.Map; +import java.util.Properties; import java.util.Vector; import omq.Remote; +import omq.common.broker.Broker; import omq.common.event.EventListener; import omq.common.event.EventTrigger; -import omq.common.message.response.CollectionResponse; import omq.common.message.response.DefaultResponse; import omq.common.message.response.ExceptionResponse; -import omq.common.message.response.ProxyResponse; import omq.common.message.response.Response; - +import omq.common.util.ParameterQueue; +import omq.exception.SerializerException; import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConsumerCancelledException; +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; +import com.rabbitmq.client.ShutdownSignalException; /** @@ -26,9 +32,14 @@ * */ -public abstract class RemoteObject implements Remote { +public abstract class RemoteObject extends Thread implements Remote { private static final long serialVersionUID = -1778953938739846450L; private String UID; + private RemoteWrapper remoteWrapper; + private Connection connection; + private Channel channel; + private QueueingConsumer consumer; + private boolean killed = false; private static final Map> primitiveClasses = new HashMap>(); @@ -44,32 +55,54 @@ } - public RemoteObject(String UID) throws Exception { - this.UID = UID; - registerToRequestListener(); - declareEventTopic(); - } - - public RemoteObject() throws Exception { - this.UID = java.util.UUID.randomUUID().toString(); - System.out.println("New Object its UID is " + this.UID); - registerToRequestListener(); - declareEventTopic(); - } - - /** - * Registers this object in the RequestListener - * - * @throws Exception - */ - private void registerToRequestListener() throws Exception { - RequestListener rListener = RequestListener.getRequestListener(); - rListener.addObj(this); - } - - private void declareEventTopic() throws Exception { - RequestListener rListener = RequestListener.getRequestListener(); - Channel channel = rListener.getChannel(); + public void start(String reference, Properties env) throws Exception { + this.UID = reference; + + // Get num threads to use + int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS)); + remoteWrapper = new RemoteWrapper(this, numThreads); + + // Get info about which exchange and queue will use + String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE); + String queue = UID; + String routingKey = UID; + + // Start connection and channel + connection = Broker.getConnection(); + channel = connection.createChannel(); + + // Declares and bindings + channel.exchangeDeclare(exchange, "direct"); + channel.queueDeclare(queue, false, false, false, null); + channel.queueBind(queue, exchange, routingKey); + + // Declare the event topic fanout channel.exchangeDeclare(UID, "fanout"); - channel.close(); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(queue, true, consumer); + + // Start this listener + this.start(); + } + + @Override + public void run() { + while (!killed) { + try { + Delivery delivery = consumer.nextDelivery(); + remoteWrapper.notifyDelivery(delivery); + } catch (InterruptedException i) { + i.printStackTrace(); + } catch (ShutdownSignalException e) { + e.printStackTrace(); + } catch (ConsumerCancelledException e) { + e.printStackTrace(); + } catch (SerializerException e) { + e.printStackTrace(); + } catch (Exception e) { + e.printStackTrace(); + } + } } @@ -79,15 +112,19 @@ } + public void kill() throws IOException { + interrupt(); + killed = true; + channel.close(); + connection.close(); + remoteWrapper.stopRemoteWrapper(); + } + @Override public Response invokeMethod(String methodName, Vector args) throws Exception { - Response resp = null; - Object[] arguments = new Object[args.size()]; for (int i = 0; i < args.size(); i++) { Object arg = args.get(i); - if (arg instanceof Remote) { - arg = RequestListener.getRequestListener().getObj(((Remote) arg).getRef()); - } + // TODO: what happens if the object is a remote object? arguments[i] = arg; } @@ -98,39 +135,8 @@ try { Object result = method.invoke(this, arguments); - // TODO see if a result is a collection and if it has some remote - // objects - if (result instanceof Remote) { - resp = getProxyResponse((Remote) result); - } else if (result instanceof Collection) { - Collection collection = (Collection) result; - boolean containsRemote = false; - for (Object o : collection) { - if (o instanceof Remote) { - containsRemote = true; - break; - } - } - if (containsRemote) { - Collection responses = new ArrayList(); - for (Object o : collection) { - if (o instanceof Remote) { - responses.add(getProxyResponse((Remote) o)); - } else { - responses.add(new DefaultResponse(this.getRef(), o)); - } - } - String collectionType = collection.getClass().getCanonicalName(); - resp = new CollectionResponse(this.getRef(), collectionType, responses); - } else { - resp = new DefaultResponse(this.getRef(), result); - } - } else { - resp = new DefaultResponse(this.getRef(), result); - } + return new DefaultResponse(this.getRef(), result); } catch (InvocationTargetException e) { - resp = new ExceptionResponse(this.getRef(), e.getTargetException()); - } - - return resp; + return new ExceptionResponse(this.getRef(), e.getTargetException()); + } } @@ -186,10 +192,4 @@ } - private Response getProxyResponse(Remote r) { - String reference = r.getRef(); - String remoteInterface = r.getClass().getInterfaces()[0].getCanonicalName(); - return new ProxyResponse(this.getRef(), reference, remoteInterface); - } - @Override public void notify(Object obj) throws Exception { Index: jectmq/src/omq/server/remote/request/RequestListener.java =================================================================== --- /objectmq/src/omq/server/remote/request/RequestListener.java (revision 3) +++ (revision ) @@ -1,124 +1,0 @@ -package omq.server.remote.request; - -import java.util.Collection; -import java.util.Hashtable; -import java.util.Properties; - -import omq.Remote; -import omq.common.remote.RemoteListener; -import omq.common.util.ParameterQueue; -import omq.exception.ObjectAlreadyExistsException; -import omq.exception.ObjectNotFoundException; -import omq.exception.SerializerException; - - -import com.rabbitmq.client.ConsumerCancelledException; -import com.rabbitmq.client.QueueingConsumer; -import com.rabbitmq.client.ShutdownSignalException; -import com.rabbitmq.client.QueueingConsumer.Delivery; - -/** - * - * @author Sergi Toda - * - */ -public class RequestListener extends RemoteListener { - private static int numThreads = 4; - private static RequestListener rListener; - private Hashtable remoteWrappers; - - private RequestListener(Properties env) throws Exception { - super(env); - // Init the hashtable - this.remoteWrappers = new Hashtable(); - - // Get info about the queue & the exchange where the RemoteListener will - // listen to. - String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE); - String queue = env.getProperty(ParameterQueue.RPC_QUEUE); - String key = env.getProperty(ParameterQueue.RPC_ROUTING_KEY); - - // Declares and bindings - channel.exchangeDeclare(exchange, ParameterQueue.RPC_TYPE); - channel.queueDeclare(queue, false, false, false, null); - channel.queueBind(queue, exchange, key); - - // Declare a new consumer - consumer = new QueueingConsumer(channel); - channel.basicConsume(queue, true, key, consumer); - } - - @Override - public void run() { - System.out.println("waiting for requests"); - while (!killed) { - try { - Delivery delivery = consumer.nextDelivery(); - String key = delivery.getProperties().getAppId(); - RemoteWrapper remoteWrapper = remoteWrappers.get(key); - remoteWrapper.notifyDelivery(delivery); - } catch (InterruptedException i) { - i.printStackTrace(); - } catch (ShutdownSignalException e) { - e.printStackTrace(); - } catch (ConsumerCancelledException e) { - e.printStackTrace(); - } catch (SerializerException e) { - e.printStackTrace(); - } catch (Exception e) { - e.printStackTrace(); - } - } - } - - public static void stopRequestListener() throws Exception { - rListener.kill(); - Hashtable hashTable = rListener.getRemoteWrappers(); - Collection collection = hashTable.values(); - for (RemoteWrapper rW : collection) { - rW.stopRemoteWrapper(); - } - rListener = null; - } - - public static void init(Properties env) throws Exception { - if (rListener == null) { - rListener = new RequestListener(env); - rListener.start(); - } else { - throw new Exception("Cannot init because it already exists"); - } - } - - public static RequestListener getRequestListener() throws Exception { - if (rListener == null) { - throw new Exception("Request listener not initialized"); - } - return rListener; - } - - public synchronized void addObj(Remote obj) throws ObjectAlreadyExistsException { - if (!remoteWrappers.containsKey(obj.getRef())) { - RemoteWrapper remoteWrapper = new RemoteWrapper(obj, numThreads); - remoteWrappers.put(obj.getRef(), remoteWrapper); - } else { - throw new ObjectAlreadyExistsException(obj.getRef()); - } - } - - public synchronized void removeObj(Remote obj) throws ObjectNotFoundException { - if (remoteWrappers.containsKey(obj.getRef())) { - remoteWrappers.remove(obj.getRef()); - } else { - throw new ObjectNotFoundException(obj.getRef()); - } - } - - public synchronized Remote getObj(String key) { - return remoteWrappers.get(key).getObj(); - } - - public Hashtable getRemoteWrappers() { - return remoteWrappers; - } -}