Index: trunk/.classpath =================================================================== --- trunk/.classpath (revision 44) +++ trunk/.classpath (revision 44) @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Index: trunk/.project =================================================================== --- trunk/.project (revision 44) +++ trunk/.project (revision 44) @@ -0,0 +1,23 @@ + + + objectmq + + + + + + org.eclipse.jdt.core.javabuilder + + + + + org.eclipse.m2e.core.maven2Builder + + + + + + org.eclipse.jdt.core.javanature + org.eclipse.m2e.core.maven2Nature + + Index: trunk/.settings/org.eclipse.jdt.core.prefs =================================================================== --- trunk/.settings/org.eclipse.jdt.core.prefs (revision 44) +++ trunk/.settings/org.eclipse.jdt.core.prefs (revision 44) @@ -0,0 +1,12 @@ +eclipse.preferences.version=1 +org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled +org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.6 +org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve +org.eclipse.jdt.core.compiler.compliance=1.6 +org.eclipse.jdt.core.compiler.debug.lineNumber=generate +org.eclipse.jdt.core.compiler.debug.localVariable=generate +org.eclipse.jdt.core.compiler.debug.sourceFile=generate +org.eclipse.jdt.core.compiler.problem.assertIdentifier=error +org.eclipse.jdt.core.compiler.problem.enumIdentifier=error +org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning +org.eclipse.jdt.core.compiler.source=1.6 Index: trunk/.settings/org.eclipse.m2e.core.prefs =================================================================== --- trunk/.settings/org.eclipse.m2e.core.prefs (revision 44) +++ trunk/.settings/org.eclipse.m2e.core.prefs (revision 44) @@ -0,0 +1,4 @@ +activeProfiles= +eclipse.preferences.version=1 +resolveWorkspaceProjects=true +version=1 Index: trunk/pom.xml =================================================================== --- trunk/pom.xml (revision 44) +++ trunk/pom.xml (revision 44) @@ -0,0 +1,99 @@ + + 4.0.0 + objectmq + objectmq + 0.5 + objectmq + Middleware based on AMQP + + + Sonatype repository + Sonatype's Maven repository + http://oss.sonatype.org/content/groups/public + + + Sonatype RSO + Sonatype's Forge repository + https://repository.sonatype.org/content/groups/forge + + + + + com.rabbitmq + amqp-client + 3.1.1 + + + com.rabbitmq + amqp-client + 3.1.1 + javadoc + + + commons-cli + commons-cli + 1.2 + + + commons-io + commons-io + 2.1 + + + com.google.code.gson + gson + 2.2.4 + + + com.esotericsoftware.kryo + kryo + 2.21 + + + + src + test + + + maven-compiler-plugin + 3.0 + + 1.6 + 1.6 + + + + org.apache.maven.plugins + maven-assembly-plugin + + + package + + attached + + + + + + jar-with-dependencies + + + + true + lib/ + + + + + + org.apache.maven.plugins + maven-surefire-plugin + 2.14.1 + + true + + + + + Index: trunk/src/main/java/omq/Remote.java =================================================================== --- trunk/src/main/java/omq/Remote.java (revision 44) +++ trunk/src/main/java/omq/Remote.java (revision 44) @@ -0,0 +1,32 @@ +package omq; + +import java.io.IOException; +import java.io.Serializable; +import java.util.Collection; + +import omq.common.event.Event; +import omq.common.event.EventListener; +import omq.exception.SerializerException; + +/** + * + * @author Sergi Toda + * + */ +public interface Remote extends Serializable { + + /** + * Returns the UID of a RemoteObject + * + * @return UID + */ + public String getRef(); + + public void notifyEvent(Event event) throws IOException, SerializerException; + + public void addListener(EventListener eventListener) throws Exception; + + public void removeListener(EventListener eventListener) throws Exception; + + public Collection> getListeners() throws Exception; +} Index: trunk/src/main/java/omq/client/annotation/AsyncMethod.java =================================================================== --- trunk/src/main/java/omq/client/annotation/AsyncMethod.java (revision 44) +++ trunk/src/main/java/omq/client/annotation/AsyncMethod.java (revision 44) @@ -0,0 +1,18 @@ +package omq.client.annotation; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Annotation which indicates a method as Asynchronous. + * + * @author Sergi Toda + * + */ + +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.METHOD) +public @interface AsyncMethod { +} Index: trunk/src/main/java/omq/client/annotation/RemoteInterface.java =================================================================== --- trunk/src/main/java/omq/client/annotation/RemoteInterface.java (revision 44) +++ trunk/src/main/java/omq/client/annotation/RemoteInterface.java (revision 44) @@ -0,0 +1,21 @@ +package omq.client.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Annotation which indicates which is the remote interface that can have + * asynchmethods or syncmethods. By default every method without an annotation + * will be classified as a SyncMethod + * + * @author Sergi Toda + * + */ +@Documented +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.TYPE) +public @interface RemoteInterface { +} Index: trunk/src/main/java/omq/client/annotation/SyncMethod.java =================================================================== --- trunk/src/main/java/omq/client/annotation/SyncMethod.java (revision 44) +++ trunk/src/main/java/omq/client/annotation/SyncMethod.java (revision 44) @@ -0,0 +1,22 @@ +package omq.client.annotation; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Annotation which indicates a method as Synchronous. It can have two + * parameters: timeout and retry which will give you how long you have to wait a + * synchronous method and how many times you'll wait for the response. + * + * @author Sergi Toda + * + */ +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.METHOD) +public @interface SyncMethod { + long timeout() default 60000L; + + int retry() default 1; +} Index: trunk/src/main/java/omq/client/listener/ResponseListener.java =================================================================== --- trunk/src/main/java/omq/client/listener/ResponseListener.java (revision 44) +++ trunk/src/main/java/omq/client/listener/ResponseListener.java (revision 44) @@ -0,0 +1,214 @@ +package omq.client.listener; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Hashtable; +import java.util.Map; +import java.util.Properties; + +import omq.client.proxy.Proxymq; +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +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; + +/** + * Class that inherits from RemoteListener. It's used in the server side. This + * class gets the deliveries from the server and stores them into the proxies + * + * @author Sergi Toda + * + */ +public class ResponseListener extends Thread { + private static ResponseListener rListener; + + private Channel channel; + private QueueingConsumer consumer; + private boolean killed = false; + private Map> results; + private Properties env; + + /** + * Protected constructor used by the singleton pattern + * + * @param env + * @throws Exception + */ + protected ResponseListener(Properties env) throws Exception { + this.env = env; + + // Init the hashtable (it's concurrent) + this.results = new Hashtable>(); + + startRPCQueue(); + } + + @Override + public void run() { + Delivery delivery; + String uid_request; + + while (!killed) { + try { + // Get the delivery + + delivery = consumer.nextDelivery(); + + BasicProperties props = delivery.getProperties(); + + // Get the response with its uid + uid_request = delivery.getProperties().getCorrelationId(); + System.out.println("Response received -> " + uid_request); + + // Stores the new response + Map proxyResults = results.get(props.getAppId()); + + // Put the result into the proxy results and notify him + synchronized (proxyResults) { + // If we haven't received this response before, we store it + if (!proxyResults.containsKey(uid_request)) { + proxyResults.put(uid_request, delivery.getBody()); + proxyResults.notifyAll(); + } + } + } catch (InterruptedException i) { + i.printStackTrace(); + } catch (ShutdownSignalException e) { + e.printStackTrace(); + try { + if (channel.isOpen()) { + channel.close(); + } + startRPCQueue(); + } catch (Exception e1) { + e1.printStackTrace(); + try { + long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000")); + Thread.sleep(milis); + } catch (InterruptedException e2) { + e2.printStackTrace(); + } + } + } catch (ConsumerCancelledException e) { + e.printStackTrace(); + } catch (Exception e) { + e.printStackTrace(); + } + } + } + + private void startRPCQueue() throws Exception { + channel = Broker.getNewChannel(); + + Map args = null; + + String reply_queue = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE); + boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false")); + + int ttl = Integer.parseInt(env.getProperty(ParameterQueue.MESSAGE_TTL_IN_QUEUES, "-1")); + if (ttl > 0) { + args = new HashMap(); + args.put("x-message-ttl", ttl); + } + + channel.queueDeclare(reply_queue, durable, false, false, args); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(reply_queue, true, consumer); + } + + /** + * Static function which initializes the ResponseListener + * + * @param env + * @throws Exception + */ + public static void init(Properties env) throws Exception { + if (rListener == null) { + rListener = new ResponseListener(env); + rListener.start(); + } else { + throw new Exception("Cannot init because it already exists"); + } + } + + /** + * Method to retrieve the unique ResponseListener, this function can also + * initialize a ResponseListener using and environment + * + * @param env + * @return unique ResponseListener + * @throws Exception + */ + public static ResponseListener getRequestListener(Properties env) throws Exception { + if (rListener == null) { + rListener = new ResponseListener(env); + rListener.start(); + } else { + // TODO: create a new exception to indicate that a response listener + // cannot be init + throw new Exception("Cannot init because it already exists"); + } + return rListener; + } + + public static boolean isVoid() { + return rListener == null; + } + + /** + * Method to retrieve the unique ResponseListener + * + * @return + * @throws Exception + */ + public static ResponseListener getRequestListener() throws Exception { + if (rListener == null) { + throw new Exception("Request listener not initialized"); + } + return rListener; + } + + /** + * + * @param key + * @return whether the map has the param key + */ + public boolean containsKey(String key) { + return results.containsKey(key); + } + + /** + * This method is used to kill the unique responseListener in the system + * + * @throws Exception + */ + public static void stopResponseListner() throws Exception { + rListener.kill(); + rListener = null; + } + + /** + * Interrupt and kill the Thread + * + * @throws IOException + */ + public void kill() throws IOException { + interrupt(); + killed = true; + channel.close(); + } + + // Revisar això + public void registerProxy(Proxymq proxy) { + if (!results.containsKey(proxy.getRef())) { + results.put(proxy.getRef(), proxy.getResults()); + } + } +} Index: trunk/src/main/java/omq/client/proxy/CallType.java =================================================================== --- trunk/src/main/java/omq/client/proxy/CallType.java (revision 44) +++ trunk/src/main/java/omq/client/proxy/CallType.java (revision 44) @@ -0,0 +1,12 @@ +package omq.client.proxy; + +/** + * CallType is an enumeration with the call types you can do: synchronous calls + * and asynchronous calls + * + * @author Sergi Toda + * + */ +public enum CallType { + SYNC, ASYNC; +} Index: trunk/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 44) +++ trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 44) @@ -0,0 +1,337 @@ +package omq.client.proxy; + +import java.io.IOException; +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 omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.SyncMethod; +import omq.client.listener.ResponseListener; +import omq.common.broker.Broker; +import omq.common.event.Event; +import omq.common.event.EventDispatcher; +import omq.common.event.EventListener; +import omq.common.message.Request; +import omq.common.message.Response; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; +import omq.exception.NoContainsInstanceException; +import omq.exception.OmqException; +import omq.exception.RetryException; +import omq.exception.SerializerException; +import omq.exception.TimeoutException; + +import com.rabbitmq.client.AMQP.BasicProperties; + +/** + * 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> listeners; + + private static final Map> primitiveClasses = new HashMap>(); + 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); + } + + /** + * 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(); + + // TODO what is better to use a new channel or to use the same? + // this.channel = Broker.getChannel(); + this.env = env; + + listeners = new HashMap>(); + + // 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 { + // long timeStart = (new Date()).getTime(); + + // 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(); + } + } + + // Create the request + Request request = createRequest(method, arguments); + + // Log.saveTimeSendRequestLog("Client-time-request", request.getId(), + // method.getName(), timeStart); + + Object response = null; + // Publish the request + if (request.isAsync()) { + System.out.println("Publish async request -> " + request.getId()); + publishAsyncRequest(request); + } else { + System.out.println("Publish sync request -> " + request.getId()); + response = publishSyncRequest(request, method.getReturnType()); + + // long timeEnd = (new Date()).getTime(); + // Log.saveTimeSendRequestLog("Client-time-response", + // request.getId(), method.getName(), timeEnd); + } + + return response; + } + + private void publishMessage(Request request, String replyQueueName) throws Exception { + String corrId = request.getId(); + + // Get the environment properties + String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE); + String routingkey = this.uid; + + // Add the correlation ID and create a replyTo property + BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).build(); + + // Publish the message + byte[] bytesRequest = Serializer.serialize(request); + // TODO See this + // channel.basicPublish(exchange, routingkey, props, bytesRequest); + Broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest); + // Log.saveLog("Client-Serialize", bytesRequest); + } + + private void publishAsyncRequest(Request request) throws Exception { + // Get the environment properties + String replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE); + publishMessage(request, replyQueueName); + } + + private Object publishSyncRequest(Request request, Class type) throws Exception { + String corrId = request.getId(); + + int retries = request.getRetries(); + long timeout = request.getTimeout(); + + // Get the environment properties + String replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE); + + // Publish the message + int i = 0; + while (i < retries) { + try { + publishMessage(request, replyQueueName); + return getResult(corrId, timeout, type); + } catch (TimeoutException te) { + System.out.println("Timeout exception catched " + te); + te.printStackTrace(); + } + i++; + } + throw new RetryException(retries, timeout); + } + + private Request createRequest(Method method, Object[] arguments) { + String corrId = java.util.UUID.randomUUID().toString(); + String methodName = method.getName(); + + // 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); + } else { + return Request.newAsyncRequest(corrId, methodName, arguments); + } + } + + 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(timeout); + localTimeout = System.currentTimeMillis() - start; + } + if ((timeout - localTimeout) <= 0) { + throw new TimeoutException("Timeout exception time: " + timeout); + } + resp = Serializer.deserializeResponse(results.get(corrId), type); + // Log.saveLog("Client-Deserialize", results.get(corrId)); + + // 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(); + } + + /** + * + * @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; + } + + @Override + public String getRef() { + return uid; + } + + @Override + public void notifyEvent(Event event) throws IOException, SerializerException { + } + + @Override + public void addListener(EventListener eventListener) throws Exception { + if (eventListener.getTopic() == null) { + eventListener.setTopic(uid); + } + listeners.put(eventListener.getTopic(), eventListener); + dispatcher.addListener(eventListener); + } + + @Override + public void removeListener(EventListener eventListener) throws Exception { + listeners.remove(eventListener.getTopic()); + dispatcher.removeListener(eventListener); + } + + @Override + public Collection> getListeners() throws Exception { + return listeners.values(); + } + +} Index: trunk/src/main/java/omq/common/broker/Broker.java =================================================================== --- trunk/src/main/java/omq/common/broker/Broker.java (revision 44) +++ trunk/src/main/java/omq/common/broker/Broker.java (revision 44) @@ -0,0 +1,267 @@ +package omq.common.broker; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; + +import omq.Remote; +import omq.client.listener.ResponseListener; +import omq.client.proxy.Proxymq; +import omq.common.event.Event; +import omq.common.event.EventDispatcher; +import omq.common.event.EventWrapper; +import omq.common.util.Environment; +import omq.common.util.OmqConnectionFactory; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; +import omq.exception.InitBrokerException; +import omq.exception.RemoteException; +import omq.exception.SerializerException; +import omq.server.RemoteObject; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; +import com.rabbitmq.client.ShutdownListener; +import com.rabbitmq.client.ShutdownSignalException; + +public class Broker { + private static Connection connection; + private static Channel channel; + private static boolean clientStarted = false; + private static boolean connectionClosed = false; + // TODO ask Pedro if it can be only one object in the map (an object can + // have multiple threads in the same broker -see environment-) + private static Map remoteObjs; + + /** + * Initializes a new Broker with the environment called by reference + * + * @param env + * @throws Exception + */ + public static synchronized void initBroker(Properties env) throws Exception { + if (Environment.isVoid()) { + remoteObjs = new HashMap(); + Environment.setEnvironment(env); + connection = OmqConnectionFactory.getNewConnection(env); + channel = connection.createChannel(); + addFaultTolerance(); + try { + tryConnection(env); + } catch (Exception e) { + channel.close(); + connection.close(); + throw new InitBrokerException("The connection didn't work"); + } + } else { + throw new InitBrokerException("Broker already started"); + } + } + + public static void stopBroker() throws Exception { + // Stop the client + if (clientStarted) { + ResponseListener.stopResponseListner(); + EventDispatcher.stopEventDispatcher(); + } + // Stop all the remote objects working + for (String reference : remoteObjs.keySet()) { + unbind(reference); + } + // Close the connection once all the listeners are died + closeConnection(); + } + + /** + * @return Broker's connection + * @throws Exception + */ + public static Connection getConnection() throws Exception { + return connection; + } + + public static void closeConnection() throws IOException { + connectionClosed = true; + connection.close(); + } + + /** + * + * @return Broker's channel + * @throws Exception + */ + public static Channel getChannel() throws Exception { + return channel; + } + + /** + * Creates a new channel using the Broker's connection + * + * @return newChannel + * @throws IOException + */ + public static Channel getNewChannel() throws IOException { + return connection.createChannel(); + } + + @SuppressWarnings("unchecked") + public static T lookup(String reference, Class contract) throws RemoteException { + try { + Properties environment = Environment.getEnvironment(); + + if (!clientStarted) { + initClient(environment); + clientStarted = true; + } + + if (!Proxymq.containsProxy(reference)) { + Proxymq proxy = new Proxymq(reference, contract, environment); + Class[] array = { contract }; + return (T) Proxymq.newProxyInstance(contract.getClassLoader(), array, proxy); + } + return (T) Proxymq.getInstance(reference); + + } catch (Exception e) { + throw new RemoteException(e); + } + } + + public static void bind(String reference, RemoteObject remote) throws RemoteException { + try { + Properties environment = Environment.getEnvironment(); + remote.startRemoteObject(reference, environment); + remoteObjs.put(reference, remote); + } catch (Exception e) { + throw new RemoteException(e); + } + } + + public static void unbind(String reference) throws RemoteException, IOException { + if (remoteObjs.containsKey(reference)) { + RemoteObject remote = remoteObjs.get(reference); + remote.kill(); + } else { + throw new RemoteException("The object referenced by 'reference' does not exist in the Broker"); + } + + } + + public void rebind(String name, Remote obj) throws RemoteException { + + } + + /** + * This method ensures the client will have only one ResponseListener and + * only one EventDispatcher. Both with the same environment. + * + * @param environment + * @throws Exception + */ + private static synchronized void initClient(Properties environment) throws Exception { + if (ResponseListener.isVoid()) { + ResponseListener.init(environment); + } + if (EventDispatcher.isVoid()) { + EventDispatcher.init(environment); + } + } + + /** + * This method sends an event with its information + * + * @param event + * @throws IOException + * @throws SerializerException + */ + public static void trigger(Event event) throws IOException, SerializerException { + String UID = event.getTopic(); + EventWrapper wrapper = new EventWrapper(event); + System.out.println("EventTrigger fanout exchange: " + UID + " Event topic: " + UID + " Event corrID: " + event.getCorrId()); + channel.exchangeDeclare(UID, "fanout"); + + byte[] bytesResponse = Serializer.serialize(wrapper); + channel.basicPublish(UID, "", null, bytesResponse); + + // Log.saveLog("Server-Serialize", bytesResponse); + } + + /** + * This function is used to send a ping message to see if the connection + * works + * + * @param env + * @throws Exception + */ + public static void tryConnection(Properties env) throws Exception { + Channel channel = connection.createChannel(); + String message = "ping"; + + String exchange = env.getProperty(ParameterQueue.USER_NAME) + "ping"; + String queueName = exchange; + String routingKey = "routingKey"; + + channel.exchangeDeclare(exchange, "direct"); + channel.queueDeclare(queueName, false, false, false, null); + channel.queueBind(queueName, exchange, routingKey); + + channel.basicPublish(exchange, routingKey, null, message.getBytes()); + + QueueingConsumer consumer = new QueueingConsumer(channel); + + channel.basicConsume(queueName, true, consumer); + Delivery delivery = consumer.nextDelivery(1000); + + channel.exchangeDelete(exchange); + channel.queueDelete(queueName); + + channel.close(); + + if (!message.equalsIgnoreCase(new String(delivery.getBody()))) { + throw new IOException("Ping initialitzation has failed"); + } + } + + /** + * This method adds a ShutdownListener to the Broker's connection. When this + * connection falls, a new connection will be created and this will also + * have the listener. + */ + private static void addFaultTolerance() { + connection.addShutdownListener(new ShutdownListener() { + @Override + public void shutdownCompleted(ShutdownSignalException cause) { + if (!connectionClosed) + if (cause.isHardError()) { + if (connection.isOpen()) { + try { + connection.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + try { + Properties env = Environment.getEnvironment(); + connection = OmqConnectionFactory.getNewWorkingConnection(env); + channel = connection.createChannel(); + addFaultTolerance(); + } catch (Exception e) { + e.printStackTrace(); + } + } else { + Channel channel = (Channel) cause.getReference(); + if (channel.isOpen()) { + try { + channel.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + } + }); + } + +} Index: trunk/src/main/java/omq/common/event/Event.java =================================================================== --- trunk/src/main/java/omq/common/event/Event.java (revision 44) +++ trunk/src/main/java/omq/common/event/Event.java (revision 44) @@ -0,0 +1,47 @@ +package omq.common.event; + +import java.io.Serializable; + +/** + * + * @author Sergi Toda + * + */ +public abstract class Event implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 1L; + + private String corrId; + private String topic; + + public Event() { + } + + public Event(String corrId, String topic) { + this.corrId = corrId; + this.topic = topic; + } + + public Event(String corrId) { + this.corrId = corrId; + } + + public String getCorrId() { + return corrId; + } + + public void setCorrId(String corrId) { + this.corrId = corrId; + } + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } +} Index: trunk/src/main/java/omq/common/event/EventDispatcher.java =================================================================== --- trunk/src/main/java/omq/common/event/EventDispatcher.java (revision 44) +++ trunk/src/main/java/omq/common/event/EventDispatcher.java (revision 44) @@ -0,0 +1,216 @@ +package omq.common.event; + +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Vector; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +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; + +/** + * This class dispatches the events received in the client side and stores them + * into the different listeners that could exists among the different proxies + * generated + * + * @author Sergi Toda + * + */ +@SuppressWarnings("rawtypes") +public class EventDispatcher extends Thread { + private static EventDispatcher dispatcher; + + private Map> listeners; + private Channel channel; + private QueueingConsumer consumer; + private Properties env; + private boolean killed = false; + + private EventDispatcher(Properties env) throws Exception { + this.env = env; + + // Declare the listeners map + listeners = new HashMap>(); + + startEventQueue(); + + } + + private void startEventQueue() throws Exception { + // Get a new connection and a new channel + channel = Broker.getNewChannel(); + + String event_queue = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE); + boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false")); + channel.queueDeclare(event_queue, durable, false, false, null); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(event_queue, true, consumer); + } + + public static void init(Properties env) throws Exception { + if (dispatcher == null) { + dispatcher = new EventDispatcher(env); + dispatcher.start(); + } else { + throw new Exception("Already initialized"); + } + } + + public static void stopEventDispatcher() throws Exception { + dispatcher.setListeners(null); + dispatcher.killed = true; + dispatcher.interrupt(); + dispatcher.channel.close(); + dispatcher = null; + } + + public static EventDispatcher getDispatcher(Properties env) throws Exception { + if (dispatcher == null) { + dispatcher = new EventDispatcher(env); + dispatcher.start(); + } + return dispatcher; + } + + public static EventDispatcher getDispatcher() throws Exception { + if (dispatcher == null) { + throw new Exception("EventDispatcher not initialized"); + } + return dispatcher; + } + + @Override + public void run() { + Delivery delivery; + Event event; + + while (!killed) { + try { + // Get the delivery + delivery = consumer.nextDelivery(); + + // Get the event + event = Serializer.deserializeEvent(delivery.getBody()); + + System.out.println("Event received -> Topic: " + event.getTopic() + "CorrId: " + event.getCorrId()); + // Log.saveLog("Client-Deserialize", delivery.getBody()); + + // long timeEnd = (new Date()).getTime(); + // Log.saveTimeSendRequestLog("Client-time-response", + // event.getCorrId(), "Event!", timeEnd); + + // Dispatch it + dispatch(event.getTopic(), event); + } catch (InterruptedException i) { + System.out.println("InterruptedException e: " + i); + i.printStackTrace(); + } catch (ShutdownSignalException e) { + System.out.println("ShutdownSignalException e: " + e); + e.printStackTrace(); + try { + if (channel.isOpen()) { + channel.close(); + } + startEventQueue(); + } catch (Exception e1) { + try { + long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000")); + Thread.sleep(milis); + } catch (InterruptedException e2) { + e2.printStackTrace(); + } + e1.printStackTrace(); + } + } catch (ConsumerCancelledException e) { + System.out.println("ConsumerCancelledException e: " + e); + e.printStackTrace(); + } catch (Exception e) { + System.out.println("Exception e: " + e); + e.printStackTrace(); + } + } + } + + public int addListener(EventListener e) throws Exception { + // Map>> mListeners = + // listeners.get(e.getTopic()); + // if(mListeners == null){ + // mListeners = new HashMap>>(); + // + // String queueName = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE); + // String reference = e.getTopic(); + // channel.exchangeDeclare(reference, "fanout"); + // channel.queueBind(queueName, reference, ""); + // } + + Vector vListeners = listeners.get(e.getTopic()); + if (vListeners == null) { + vListeners = new Vector(); + + String queueName = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE); + String reference = e.getTopic(); + + System.out.println("EventDispatcher declaring fanout -> " + reference + " Binding with: " + queueName); + + channel.exchangeDeclare(reference, "fanout"); + channel.queueBind(queueName, reference, ""); + } + vListeners.add(e); + listeners.put(e.getTopic(), vListeners); + + return vListeners.size(); + } + + public int removeListener(EventListener e) { + Vector vListeners = listeners.get(e.getTopic()); + if (vListeners != null) { + // TODO: removeListener -> remove(e) override equals? + vListeners.remove(e); + } + + return vListeners.size(); + } + + /** + * This method dispatches the events. Every time an event is received, this + * method is launched. This method creates a new thread and executes the + * notifyEvent function of the listeners associated to this event + * + * @param topic + * @param event + */ + public void dispatch(String topic, final Event event) { + if (listeners.containsKey(topic)) { + for (final EventListener listener : listeners.get(topic)) { + new Thread() { + @SuppressWarnings("unchecked") + public void run() { + listener.notifyEvent(event); + } + }.start(); + } + } + } + + public Map> getListeners() { + return listeners; + } + + public void setListeners(Map> listeners) { + this.listeners = listeners; + } + + public static boolean isVoid() { + return dispatcher == null; + } + +} Index: trunk/src/main/java/omq/common/event/EventListener.java =================================================================== --- trunk/src/main/java/omq/common/event/EventListener.java (revision 44) +++ trunk/src/main/java/omq/common/event/EventListener.java (revision 44) @@ -0,0 +1,43 @@ +package omq.common.event; + +/** + * Abstract class that enables to create new EventListeners. The eventListeners + * are done to execute the notifyEvent function. + * + * @author Sergi Toda + * + */ +public abstract class EventListener { + private String topic; + + public EventListener() { + topic = null; + } + + /** + * Constructor. This constructor uses a String to indicate manually which + * event we want to receive + * + * @param topic + */ + public EventListener(String topic) { + this.topic = topic; + } + + /** + * Whenever this listener it's notified of an event, will execute this + * function + * + * @param event + */ + public abstract void notifyEvent(E event); + + public void setTopic(String topic) { + this.topic = topic; + } + + public String getTopic() { + return topic; + } + +} Index: trunk/src/main/java/omq/common/event/EventWrapper.java =================================================================== --- trunk/src/main/java/omq/common/event/EventWrapper.java (revision 44) +++ trunk/src/main/java/omq/common/event/EventWrapper.java (revision 44) @@ -0,0 +1,41 @@ +package omq.common.event; + +import java.io.Serializable; + +public class EventWrapper implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 1L; + + private String type; + private Event event; + + public EventWrapper() { + } + + // TODO change to simpleName + public EventWrapper(Event event) { + this.event = event; + type = event.getClass().getCanonicalName(); + // type = event.getClass().getSimpleName(); + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public Event getEvent() { + return event; + } + + public void setEvent(Event event) { + this.event = event; + } + +} Index: trunk/src/main/java/omq/common/message/Request.java =================================================================== --- trunk/src/main/java/omq/common/message/Request.java (revision 44) +++ trunk/src/main/java/omq/common/message/Request.java (revision 44) @@ -0,0 +1,99 @@ +package omq.common.message; + +import java.io.Serializable; + +public class Request implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 6366255840200365083L; + + private String method; + private Object[] params; + private String id; + private boolean async = false; + + private transient long timeout; + private transient int retries; + + public Request() { + } + + public Request(String id, String method, Object[] params) { + this.id = id; + this.method = method; + this.params = params; + } + + private Request(String id, String method, boolean async, Object[] params) { + this.id = id; + this.method = method; + this.async = async; + this.params = params; + } + + public static Request newSyncRequest(String id, String method, Object[] params) { + return new Request(id, method, false, params); + } + + public static Request newSyncRequest(String id, String method, Object[] params, int retries, long timeout) { + Request req = new Request(id, method, false, params); + req.setRetries(retries); + req.setTimeout(timeout); + return req; + } + + public static Request newAsyncRequest(String id, String method, Object[] params) { + return new Request(id, method, true, params); + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getMethod() { + return method; + } + + public void setMethod(String method) { + this.method = method; + } + + public Object[] getParams() { + return params; + } + + public void setParams(Object[] params) { + this.params = params; + } + + public boolean isAsync() { + return async; + } + + public void setAsync(boolean async) { + this.async = async; + } + + public long getTimeout() { + return timeout; + } + + public void setTimeout(long timeout) { + this.timeout = timeout; + } + + public int getRetries() { + return retries; + } + + public void setRetries(int retries) { + this.retries = retries; + } + +} Index: trunk/src/main/java/omq/common/message/Response.java =================================================================== --- trunk/src/main/java/omq/common/message/Response.java (revision 44) +++ trunk/src/main/java/omq/common/message/Response.java (revision 44) @@ -0,0 +1,61 @@ +package omq.common.message; + +import java.io.Serializable; + +import omq.exception.OmqException; + +public class Response implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 3368363997012527189L; + + private Object result; + private OmqException error; + private String id; + private String idOmq; + + public Response() { + } + + public Response(String id, String idOmq, Object result, OmqException error) { + this.id = id; + this.idOmq = idOmq; + this.result = result; + this.error = error; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getIdOmq() { + return idOmq; + } + + public void setIdOmq(String idOmq) { + this.idOmq = idOmq; + } + + public Object getResult() { + return result; + } + + public void setResult(Object result) { + this.result = result; + } + + public OmqException getError() { + return error; + } + + public void setError(OmqException error) { + this.error = error; + } + +} Index: trunk/src/main/java/omq/common/util/Environment.java =================================================================== --- trunk/src/main/java/omq/common/util/Environment.java (revision 44) +++ trunk/src/main/java/omq/common/util/Environment.java (revision 44) @@ -0,0 +1,41 @@ +package omq.common.util; + +import java.util.Properties; + +import omq.exception.EnvironmentException; + +/** + * + * @author Sergi Toda + * + */ +public class Environment { + private static Properties env = null; + + /** + * This method return the environment of the middleware + * + * @return Environment to use in the middleware + * @throws EnvironmentException + * will be thrown whether the environment is null + */ + public static Properties getEnvironment() throws EnvironmentException { + if (env == null) { + throw new EnvironmentException(); + } + return env; + } + + /** + * This method sets the environment of the middleware + * + * @param environment + */ + public static void setEnvironment(Properties environment) { + env = environment; + } + + public static boolean isVoid() { + return env == null; + } +} Index: trunk/src/main/java/omq/common/util/Log.java =================================================================== --- trunk/src/main/java/omq/common/util/Log.java (revision 44) +++ trunk/src/main/java/omq/common/util/Log.java (revision 44) @@ -0,0 +1,70 @@ +package omq.common.util; + +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.util.Date; +import java.util.Properties; + +import omq.exception.EnvironmentException; + +public class Log { + + public static void saveLog(String processName, byte[] bytesResponse) throws IOException { + try { + Properties env = Environment.getEnvironment(); + + String debugPath = env.getProperty(ParameterQueue.DEBUGFILE, ""); + if (debugPath.length() > 0) { + long timeNow = (new Date()).getTime(); + + File outputFolder = new File(debugPath + File.separator); + outputFolder.mkdirs(); + +// File outputFolder = new File(debugPath + File.separator + processName); +// outputFolder.mkdirs(); + +// File outputFileContent = new File(outputFolder.getAbsoluteFile() + File.separator + "content_" + timeNow); +// FileOutputStream outputStream = new FileOutputStream(outputFileContent); +// IOUtils.write(bytesResponse, outputStream); +// outputStream.close(); + + File outputFileLog = new File(debugPath + File.separator + "log"); + boolean exist = outputFileLog.exists(); + + FileWriter fw = new FileWriter(outputFileLog, true); // the true will append the new data + if(!exist){ + fw.write("#ProcessName\t\tFile\t\t\t\t\tDate\t\t\tSize\n"); + } + fw.write(processName + "\t" + "content_" + timeNow + "\t" + timeNow + "\t" + bytesResponse.length + "\tbytes\n"); + fw.close(); + } + } catch (EnvironmentException e) { + throw new IOException(e.getMessage(), e); + } + } + + public static void saveTimeSendRequestLog(String processName, String coorId, String method, long timeNow) throws IOException { + try { + Properties env = Environment.getEnvironment(); + + String debugPath = env.getProperty(ParameterQueue.DEBUGFILE, ""); + if (debugPath.length() > 0) { + File outputFolder = new File(debugPath + File.separator + processName); + outputFolder.mkdirs(); + + File outputFileLog = new File(outputFolder + File.separator + "log"); + boolean exist = outputFileLog.exists(); + + FileWriter fw = new FileWriter(outputFileLog, true); // the true will append the new data + if(!exist){ + fw.write("#CoorId\tMethod\tDate\n"); + } + fw.write(coorId + "\t" + method + "\t" + timeNow + "\n"); + fw.close(); + } + } catch (EnvironmentException e) { + throw new IOException(e.getMessage(), e); + } + } +} Index: trunk/src/main/java/omq/common/util/OmqConnectionFactory.java =================================================================== --- trunk/src/main/java/omq/common/util/OmqConnectionFactory.java (revision 44) +++ trunk/src/main/java/omq/common/util/OmqConnectionFactory.java (revision 44) @@ -0,0 +1,72 @@ +package omq.common.util; + +import java.io.IOException; +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.util.Properties; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + +/** + * + * @author Sergi Toda + * + */ +public class OmqConnectionFactory { + private static Connection connection; + private static int connectionTimeout = 2 * 1000; + + public static void init(Properties env) throws KeyManagementException, NoSuchAlgorithmException, IOException { + if (connection == null) { + connection = getNewConnection(env); + } + } + + public static Connection getNewWorkingConnection(Properties env) throws Exception { + Connection connection = null; + boolean working = false; + + while (!working) { + try { + connection = getNewConnection(env); + working = true; + } catch (Exception e) { + e.printStackTrace(); + long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000")); + Thread.sleep(milis); + } + } + + return connection; + } + + public static Connection getNewConnection(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); + factory.setConnectionTimeout(connectionTimeout); + if (ssl) { + factory.useSslProtocol(); + } + return factory.newConnection(); + } + + public static Channel getNewChannel() throws IOException { + return connection.createChannel(); + } +} Index: trunk/src/main/java/omq/common/util/ParameterQueue.java =================================================================== --- trunk/src/main/java/omq/common/util/ParameterQueue.java (revision 44) +++ trunk/src/main/java/omq/common/util/ParameterQueue.java (revision 44) @@ -0,0 +1,98 @@ +package omq.common.util; + +/** + * + * @author Sergi Toda + * + */ +public class ParameterQueue { + + /* + * Properties environment + */ + + public static String SERIALIZER_NAME = "omq.serializer"; + + /** + * Set whether the messages must be compressed or not + */ + public static String ENABLECOMPRESSION = "omq.compression"; + + /** + * Set the ip where the rabbitmq server is. + */ + public static String SERVER_HOST = "omq.host"; + + /** + * Set the port that rabbitmq uses. + */ + public static String SERVER_PORT = "omq.port"; + + /** + * Set the clients username + */ + public static String USER_NAME = "omq.username"; + + /** + * Set the clients password + */ + public static String USER_PASS = "omq.pass"; + + /** + * Set the exchange where the objectmq are listening + */ + public static String RPC_EXCHANGE = "omq.rpc_exchange"; + + /** + * Set the clients reply queue. Every client must have a different queue + * name. + */ + public static String RPC_REPLY_QUEUE = "omq.reply_queue_rpc"; + + /** + * Set the clients event queue. Every client must have a different queue + * name. + */ + public static String EVENT_REPLY_QUEUE = "omq.reply_queue_event"; + + /** + * Set if the queues must be durable. The queues won't be lost when rabbitmq + * crashes if DURABLE_QUEUES is set trues. + */ + public static String DURABLE_QUEUES = "omq.durable_queue"; + + /** + * The MESSAGE_TTL_IN_QUEUES controls for how long a message published to + * the queues can live before it is discarded. A message that has been in + * the queue for longer than the configured TTL is said to be dead. + * + * This property must be a non-negative 32 bit integer (0 <= n <= 2^32-1), + * describing the TTL period in milliseconds. + */ + public static String MESSAGE_TTL_IN_QUEUES = "omq.message_ttl_queue"; + + // TODO persisten messages? the messages will be saved in the disk if this + // flag is set true + + public static String ENABLE_SSL = "omq.enable_ssl"; + public static String DEBUGFILE = "omq.debug_file"; + + public static String RETRY_TIME_CONNECTION = "omq.retry_connection"; + + /* + * Values + */ + + // Change this!!! + public static String RPC_TYPE = "direct"; + + public static String NUM_THREADS = "omq.num_threads"; + + public static String REGISTRY_NAME = "REGISTRY"; + + /** + * Time in milis + */ + public static long DEFAULT_TIMEOUT = 1 * 1000 * 60; + +} Index: trunk/src/main/java/omq/common/util/Serializer.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializer.java (revision 44) +++ trunk/src/main/java/omq/common/util/Serializer.java (revision 44) @@ -0,0 +1,123 @@ +package omq.common.util; + +import java.io.IOException; +import java.util.Properties; + +import omq.common.event.Event; +import omq.common.message.Request; +import omq.common.message.Response; +import omq.common.util.Serializers.GsonImp; +import omq.common.util.Serializers.ISerializer; +import omq.common.util.Serializers.JavaImp; +import omq.common.util.Serializers.KryoImp; +import omq.exception.EnvironmentException; +import omq.exception.SerializerException; +import omq.server.RemoteObject; + +/** + * + * @author Sergi Toda + * + */ +public class Serializer { + public static String kryo = KryoImp.class.getCanonicalName(); + public static String java = JavaImp.class.getCanonicalName(); + public static String gson = GsonImp.class.getCanonicalName(); + + public static ISerializer serializer; + + private static Boolean getEnableCompression() { + Boolean enableCompression = false; + try { + Properties env = Environment.getEnvironment(); + enableCompression = Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false")); + } catch (EnvironmentException e) { + e.printStackTrace(); + } + + return enableCompression; + } + + public static ISerializer getInstance() throws SerializerException { + if (serializer == null) { + try { + Properties env = Environment.getEnvironment(); + String className = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + + if (className == null || className.isEmpty()) { + throw new ClassNotFoundException("Class name is null or empty."); + } + + serializer = (ISerializer) Class.forName(className).newInstance(); + } catch (Exception ex) { + throw new SerializerException(ex.getMessage(), ex); + } + } + + return serializer; + } + + public static byte[] serialize(Object obj) throws SerializerException { + ISerializer instance = getInstance(); + + Boolean enableCompression = getEnableCompression(); + if (enableCompression) { + byte[] objSerialized = instance.serialize(obj); + try { + return Zipper.zip(objSerialized); + } catch (IOException e) { + throw new SerializerException(e.getMessage(), e); + } + } else { + return instance.serialize(obj); + } + } + + public static Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException { + ISerializer instance = getInstance(); + + Boolean enableCompression = getEnableCompression(); + if (enableCompression) { + try { + byte[] unZippedBytes = Zipper.unzip(bytes); + return instance.deserializeRequest(unZippedBytes, obj); + } catch (IOException e) { + throw new SerializerException(e.getMessage(), e); + } + } else { + return instance.deserializeRequest(bytes, obj); + } + } + + public static Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { + ISerializer instance = getInstance(); + + Boolean enableCompression = getEnableCompression(); + if (enableCompression) { + try { + byte[] unZippedBytes = Zipper.unzip(bytes); + return instance.deserializeResponse(unZippedBytes, type); + } catch (IOException e) { + throw new SerializerException(e.getMessage(), e); + } + } else { + return instance.deserializeResponse(bytes, type); + } + } + + public static Event deserializeEvent(byte[] bytes) throws SerializerException { + ISerializer instance = getInstance(); + + Boolean enableCompression = getEnableCompression(); + if (enableCompression) { + try { + byte[] unZippedBytes = Zipper.unzip(bytes); + return instance.deserializeEvent(unZippedBytes); + } catch (IOException e) { + throw new SerializerException(e.getMessage(), e); + } + } else { + return instance.deserializeEvent(bytes); + } + } +} Index: trunk/src/main/java/omq/common/util/Serializers/GsonImp.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializers/GsonImp.java (revision 44) +++ trunk/src/main/java/omq/common/util/Serializers/GsonImp.java (revision 44) @@ -0,0 +1,100 @@ +package omq.common.util.Serializers; + +import java.util.List; + +import omq.common.event.Event; +import omq.common.message.Request; +import omq.common.message.Response; +import omq.exception.OmqException; +import omq.exception.SerializerException; +import omq.server.RemoteObject; + +import com.google.gson.Gson; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; + +public class GsonImp implements ISerializer { + private final Gson gson = new Gson(); + + @Override + public byte[] serialize(Object obj) throws SerializerException { + String json = gson.toJson(obj); + System.out.println(json); + return json.getBytes(); + } + + @Override + public Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException { + String json = new String(bytes); + + JsonParser parser = new JsonParser(); + JsonObject jsonObj = parser.parse(json).getAsJsonObject(); + + String id = jsonObj.get("id").getAsString(); + String method = jsonObj.get("method").getAsString(); + + List> types = obj.getParams(method); + + try { + JsonArray jsonArgs = (JsonArray) jsonObj.get("params"); + + // TODO: if (jsonArgs.size() == types.size()) + int length = jsonArgs.size(); + Object[] arguments = new Object[length]; + + int i = 0; + for (JsonElement element : jsonArgs) { + arguments[i] = gson.fromJson(element, types.get(i)); + i++; + } + + return new Request(id, method, arguments); + } catch (NullPointerException e) { + return new Request(id, method, null); + } + } + + @Override + public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { + String json = new String(bytes); + + JsonParser parser = new JsonParser(); + JsonObject jsonObj = parser.parse(json).getAsJsonObject(); + + String id = jsonObj.get("id").getAsString(); + String idOmq = jsonObj.get("idOmq").getAsString(); + + JsonElement jsonElement = jsonObj.get("result"); + Object result = gson.fromJson(jsonElement, type); + + JsonElement jsonError = jsonObj.get("error"); + OmqException error = gson.fromJson(jsonError, OmqException.class); + + return new Response(id, idOmq, result, error); + } + + @Override + public Event deserializeEvent(byte[] bytes) throws SerializerException { + try { + String json = new String(bytes); + System.out.println(json); + + JsonParser parser = new JsonParser(); + JsonObject jsonObj = parser.parse(json).getAsJsonObject(); + + String type = jsonObj.get("type").getAsString(); + + JsonElement jsonElement = jsonObj.get("event"); + Event event; + + event = (Event) gson.fromJson(jsonElement, Class.forName(type)); + + return event; + } catch (Exception e) { + throw new SerializerException("Deserialize event", e.getCause()); + } + } + +} Index: trunk/src/main/java/omq/common/util/Serializers/ISerializer.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializers/ISerializer.java (revision 44) +++ trunk/src/main/java/omq/common/util/Serializers/ISerializer.java (revision 44) @@ -0,0 +1,22 @@ +package omq.common.util.Serializers; + +import omq.common.event.Event; +import omq.common.message.Request; +import omq.common.message.Response; +import omq.exception.SerializerException; +import omq.server.RemoteObject; + +/** + * + * @author Sergi Toda + * + */ +public interface ISerializer { + public byte[] serialize(Object obj) throws SerializerException; + + public Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException; + + public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException; + + public Event deserializeEvent(byte[] bytes) throws SerializerException; +} Index: trunk/src/main/java/omq/common/util/Serializers/JavaImp.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializers/JavaImp.java (revision 44) +++ trunk/src/main/java/omq/common/util/Serializers/JavaImp.java (revision 44) @@ -0,0 +1,75 @@ +package omq.common.util.Serializers; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; + +import omq.common.event.Event; +import omq.common.event.EventWrapper; +import omq.common.message.Request; +import omq.common.message.Response; +import omq.exception.SerializerException; +import omq.server.RemoteObject; + +/** + * + * @author Sergi Toda + * + */ +public class JavaImp implements ISerializer { + + @Override + public byte[] serialize(Object obj) throws SerializerException { + try { + ByteArrayOutputStream stream = new ByteArrayOutputStream(); + ObjectOutputStream output = new ObjectOutputStream(stream); + output.writeObject(obj); + + output.flush(); + output.close(); + + byte[] bArray = stream.toByteArray(); + + stream.flush(); + stream.close(); + + return bArray; + } catch (Exception e) { + throw new SerializerException("Serialize -> " + e.getMessage(), e); + } + } + + @Override + public Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException { + return (Request) deserializeObject(bytes); + } + + @Override + public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { + return (Response) deserializeObject(bytes); + } + + @Override + public Event deserializeEvent(byte[] bytes) throws SerializerException { + EventWrapper wrapper = (EventWrapper) deserializeObject(bytes); + return wrapper.getEvent(); + } + + public Object deserializeObject(byte[] bytes) throws SerializerException { + try { + ByteArrayInputStream input = new ByteArrayInputStream(bytes); + ObjectInputStream objInput = new ObjectInputStream(input); + + Object obj = objInput.readObject(); + + objInput.close(); + input.close(); + + return obj; + } catch (Exception e) { + throw new SerializerException("Deserialize -> " + e.getMessage(), e); + } + } + +} Index: trunk/src/main/java/omq/common/util/Serializers/KryoImp.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializers/KryoImp.java (revision 44) +++ trunk/src/main/java/omq/common/util/Serializers/KryoImp.java (revision 44) @@ -0,0 +1,73 @@ +package omq.common.util.Serializers; + +import java.io.ByteArrayOutputStream; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; + +import omq.common.event.Event; +import omq.common.event.EventWrapper; +import omq.common.message.Request; +import omq.common.message.Response; +import omq.exception.SerializerException; +import omq.server.RemoteObject; + +/** + * + * @author Sergi Toda + * + */ +public class KryoImp implements ISerializer { + private final Kryo kryo = new Kryo(); + + @Override + public byte[] serialize(Object obj) throws SerializerException { + try { + ByteArrayOutputStream stream = new ByteArrayOutputStream(); + Output output = new Output(stream); + kryo.writeObject(output, obj); + + output.flush(); + output.close(); + + byte[] bArray = stream.toByteArray(); + + stream.flush(); + stream.close(); + + return bArray; + } catch (Exception e) { + throw new SerializerException("Serialize -> " + e.getMessage(), e); + } + } + + @Override + public Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException { + return (Request) deserializeObject(bytes, Request.class); + } + + @Override + public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { + return (Response) deserializeObject(bytes, Response.class); + } + + @Override + public Event deserializeEvent(byte[] bytes) throws SerializerException { + EventWrapper wrapper = (EventWrapper) deserializeObject(bytes, EventWrapper.class); + return wrapper.getEvent(); + } + + public Object deserializeObject(byte[] bytes, Class type) throws SerializerException { + try { + Input input = new Input(bytes); + Object obj = kryo.readObject(input, type); + + input.close(); + return obj; + } catch (Exception e) { + throw new SerializerException("Deserialize -> " + e.getMessage(), e); + } + } + +} Index: trunk/src/main/java/omq/common/util/Zipper.java =================================================================== --- trunk/src/main/java/omq/common/util/Zipper.java (revision 44) +++ trunk/src/main/java/omq/common/util/Zipper.java (revision 44) @@ -0,0 +1,53 @@ +package omq.common.util; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; + +public class Zipper { + + public static byte[] zip(byte[] b) throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + + GZIPOutputStream zos = null; + try { + zos = new GZIPOutputStream(baos); + zos.write(b); + } finally{ + if(zos != null){ + zos.close(); + } + + baos.close(); + } + + return baos.toByteArray(); + } + + public static byte[] unzip(byte[] b) throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ByteArrayInputStream bais = new ByteArrayInputStream(b); + + GZIPInputStream zis = null; + try { + zis = new GZIPInputStream(bais); + + byte[] tmpBuffer = new byte[256]; + int n; + while ((n = zis.read(tmpBuffer)) >= 0) { + baos.write(tmpBuffer, 0, n); + } + } finally { + if(zis != null){ + zis.close(); + } + + bais.close(); + baos.close(); + } + + return baos.toByteArray(); + } +} Index: trunk/src/main/java/omq/exception/ConnectionException.java =================================================================== --- trunk/src/main/java/omq/exception/ConnectionException.java (revision 44) +++ trunk/src/main/java/omq/exception/ConnectionException.java (revision 44) @@ -0,0 +1,36 @@ +/***************************************************************************************** + * EVO : an Event-Based Object Distributed Middleware + * 2003-2011 AST Research Group + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + *****************************************************************************************/ +package omq.exception; + +/** + + * + * A ConnectionException is thrown if an Exception occurs in the Event middleware. + * + */ +public class ConnectionException extends RemoteException { + + private static final long serialVersionUID = -5408761207745105350L; + + public ConnectionException(String msg) + { + super(msg); + } + +} Index: trunk/src/main/java/omq/exception/EnvironmentException.java =================================================================== --- trunk/src/main/java/omq/exception/EnvironmentException.java (revision 44) +++ trunk/src/main/java/omq/exception/EnvironmentException.java (revision 44) @@ -0,0 +1,14 @@ +package omq.exception; + +public class EnvironmentException extends Exception { + + /** + * + */ + private static final long serialVersionUID = 1L; + + public EnvironmentException() { + super("Environment not found"); + } + +} Index: trunk/src/main/java/omq/exception/InitBrokerException.java =================================================================== --- trunk/src/main/java/omq/exception/InitBrokerException.java (revision 44) +++ trunk/src/main/java/omq/exception/InitBrokerException.java (revision 44) @@ -0,0 +1,13 @@ +package omq.exception; + +public class InitBrokerException extends Exception { + + /** + * + */ + private static final long serialVersionUID = 1L; + + public InitBrokerException(String string) { + super(string); + } +} Index: trunk/src/main/java/omq/exception/NoContainsInstanceException.java =================================================================== --- trunk/src/main/java/omq/exception/NoContainsInstanceException.java (revision 44) +++ trunk/src/main/java/omq/exception/NoContainsInstanceException.java (revision 44) @@ -0,0 +1,14 @@ +package omq.exception; + +public class NoContainsInstanceException extends Exception { + + public NoContainsInstanceException(String reference) { + super("Reference: " + reference + " not found"); + } + + /** + * + */ + private static final long serialVersionUID = 1L; + +} Index: trunk/src/main/java/omq/exception/NoSuchEvObjectException.java =================================================================== --- trunk/src/main/java/omq/exception/NoSuchEvObjectException.java (revision 44) +++ trunk/src/main/java/omq/exception/NoSuchEvObjectException.java (revision 44) @@ -0,0 +1,28 @@ +/***************************************************************************************** + * EVO : an Event-Based Object Distributed Middleware + * 2003-2011 AST Research Group + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + *****************************************************************************************/ +package omq.exception; + +public class NoSuchEvObjectException extends Exception { + + private static final long serialVersionUID = 7483302654634875430L; + +public NoSuchEvObjectException (String msg) { + super (msg); + } +} Index: trunk/src/main/java/omq/exception/NotBoundException.java =================================================================== --- trunk/src/main/java/omq/exception/NotBoundException.java (revision 44) +++ trunk/src/main/java/omq/exception/NotBoundException.java (revision 44) @@ -0,0 +1,28 @@ +/***************************************************************************************** + * EVO : an Event-Based Object Distributed Middleware + * 2003-2011 AST Research Group + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + *****************************************************************************************/ +package omq.exception; + +public class NotBoundException extends RemoteException { + + private static final long serialVersionUID = -686502784080766852L; + + public NotBoundException (String msg) { + super(msg); + } +} Index: trunk/src/main/java/omq/exception/ObjectAlreadyExistsException.java =================================================================== --- trunk/src/main/java/omq/exception/ObjectAlreadyExistsException.java (revision 44) +++ trunk/src/main/java/omq/exception/ObjectAlreadyExistsException.java (revision 44) @@ -0,0 +1,14 @@ +package omq.exception; + +public class ObjectAlreadyExistsException extends Exception { + + /** + * + */ + private static final long serialVersionUID = 1L; + + public ObjectAlreadyExistsException(String ref) { + super(ref); + } + +} Index: trunk/src/main/java/omq/exception/ObjectNotFoundException.java =================================================================== --- trunk/src/main/java/omq/exception/ObjectNotFoundException.java (revision 44) +++ trunk/src/main/java/omq/exception/ObjectNotFoundException.java (revision 44) @@ -0,0 +1,14 @@ +package omq.exception; + +public class ObjectNotFoundException extends Exception { + + /** + * + */ + private static final long serialVersionUID = 1L; + + public ObjectNotFoundException(String ref) { + super(ref); + } + +} Index: trunk/src/main/java/omq/exception/OmqException.java =================================================================== --- trunk/src/main/java/omq/exception/OmqException.java (revision 44) +++ trunk/src/main/java/omq/exception/OmqException.java (revision 44) @@ -0,0 +1,38 @@ +package omq.exception; + +import java.io.Serializable; + +public class OmqException implements Serializable { + /** + * + */ + private static final long serialVersionUID = 1L; + + private String type; + private String message; + + public OmqException() { + } + + public OmqException(String type, String message) { + this.type = type; + this.message = message; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } + +} Index: trunk/src/main/java/omq/exception/RegistryNotLoadedException.java =================================================================== --- trunk/src/main/java/omq/exception/RegistryNotLoadedException.java (revision 44) +++ trunk/src/main/java/omq/exception/RegistryNotLoadedException.java (revision 44) @@ -0,0 +1,28 @@ +/***************************************************************************************** + * EVO : an Event-Based Object Distributed Middleware + * 2003-2011 AST Research Group + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + *****************************************************************************************/ +package omq.exception; + +public class RegistryNotLoadedException extends RemoteException { + + private static final long serialVersionUID = 5721550693129237244L; + +public RegistryNotLoadedException (String msg) { + super (msg); + } +} Index: trunk/src/main/java/omq/exception/RemoteException.java =================================================================== --- trunk/src/main/java/omq/exception/RemoteException.java (revision 44) +++ trunk/src/main/java/omq/exception/RemoteException.java (revision 44) @@ -0,0 +1,39 @@ +/***************************************************************************************** + * EVO : an Event-Based Object Distributed Middleware + * 2003-2011 AST Research Group + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + *****************************************************************************************/ +package omq.exception; + +import java.io.Serializable; + +/** + * A RemoteException is the common superclass for a wide number of communication-related + * exceptions that may occur during the execution of a remote method call. + * Each method of a remote interface must list RemoteException in its throws clause. + */ +public class RemoteException extends Exception implements Serializable { + + static final long serialVersionUID = 10L; + + public RemoteException (Exception ex) { + super (ex); + } + + public RemoteException (String msg) { + super (msg); + } +} Index: trunk/src/main/java/omq/exception/RetryException.java =================================================================== --- trunk/src/main/java/omq/exception/RetryException.java (revision 44) +++ trunk/src/main/java/omq/exception/RetryException.java (revision 44) @@ -0,0 +1,16 @@ +package omq.exception; + +public class RetryException extends Exception { + + /** + * + */ + private static final long serialVersionUID = 5450662697539597010L; + + public RetryException(int retries, long timeout) { + super("RetyException num retries="+retries+" timeout per retry="+timeout); + } + + + +} Index: trunk/src/main/java/omq/exception/SerializerException.java =================================================================== --- trunk/src/main/java/omq/exception/SerializerException.java (revision 44) +++ trunk/src/main/java/omq/exception/SerializerException.java (revision 44) @@ -0,0 +1,16 @@ +package omq.exception; + +public class SerializerException extends Exception { + + /** + * + */ + private static final long serialVersionUID = 1L; + + public SerializerException(String message, Throwable cause) { + super(message, cause); + } + + + +} Index: trunk/src/main/java/omq/exception/SessionException.java =================================================================== --- trunk/src/main/java/omq/exception/SessionException.java (revision 44) +++ trunk/src/main/java/omq/exception/SessionException.java (revision 44) @@ -0,0 +1,44 @@ +/***************************************************************************************** + * EVO : an Event-Based Object Distributed Middleware + * 2003-2011 AST Research Group + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + *****************************************************************************************/ +package omq.exception; + +/** + * Wrapper exception class to different exceptions + * It simply stores the incoming exception and provides a method for obtaining the "real" exception. + * + */ +public class SessionException extends Exception { + + private static final long serialVersionUID = 8873640230153201413L; + private Exception ex; + + public SessionException (Exception ex) { + super (ex.getMessage()); + ex.printStackTrace(); + this.ex = ex; + } + + /** + * This method returns the exception produced in the notification service. + * @return Exception The exception occured in the underlying notification service. + */ + public Exception getException() { + return ex; + } +} Index: trunk/src/main/java/omq/exception/TimeoutException.java =================================================================== --- trunk/src/main/java/omq/exception/TimeoutException.java (revision 44) +++ trunk/src/main/java/omq/exception/TimeoutException.java (revision 44) @@ -0,0 +1,32 @@ +/***************************************************************************************** + * EVO : an Event-Based Object Distributed Middleware + * 2003-2011 AST Research Group + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + *****************************************************************************************/ +package omq.exception; + +/** + * TimeoutException is thrown if the timeout expires in a remote call. + * + */ +public class TimeoutException extends RemoteException { + + private static final long serialVersionUID = 5155578659802352754L; + +public TimeoutException (String msg) { + super (msg); + } +} Index: trunk/src/main/java/omq/exception/TypeNotFoundException.java =================================================================== --- trunk/src/main/java/omq/exception/TypeNotFoundException.java (revision 44) +++ trunk/src/main/java/omq/exception/TypeNotFoundException.java (revision 44) @@ -0,0 +1,32 @@ +/***************************************************************************************** + * EVO : an Event-Based Object Distributed Middleware + * 2003-2011 AST Research Group + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + *****************************************************************************************/ +package omq.exception; + +/** + * TypeNotFoundException is thrown if a type is unknown when generating + * stubs and skeletons. + * + */ +public class TypeNotFoundException extends Exception { + static final long serialVersionUID = 10L; + + public TypeNotFoundException (String msg) { + super (msg); + } +} Index: trunk/src/main/java/omq/server/InvocationThread.java =================================================================== --- trunk/src/main/java/omq/server/InvocationThread.java (revision 44) +++ trunk/src/main/java/omq/server/InvocationThread.java (revision 44) @@ -0,0 +1,100 @@ +package omq.server; + +import java.lang.reflect.InvocationTargetException; +import java.util.concurrent.BlockingQueue; + +import omq.common.message.Request; +import omq.common.message.Response; +import omq.common.util.Serializer; +import omq.exception.OmqException; + +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.QueueingConsumer.Delivery; + +/** + * + * @author Sergi Toda + * + */ +public class InvocationThread extends Thread { + private RemoteObject obj; + private BlockingQueue deliveryQueue; + private boolean killed = false; + + public InvocationThread(RemoteObject obj, BlockingQueue deliveryQueue) { + this.obj = obj; + this.deliveryQueue = deliveryQueue; + } + + @Override + public void run() { + while (!killed) { + try { + // Get the delivery + Delivery delivery = deliveryQueue.take(); + + // Deserialize the json + Request request = Serializer.deserializeRequest(delivery.getBody(), obj); + // Log.saveLog("Server-Deserialize", delivery.getBody()); + + String methodName = request.getMethod(); + String requestID = request.getId(); + + System.out.println("Invoke method: " + methodName + " CorrID: " + requestID); + + // Invoke the method + Object result = null; + OmqException error = null; + try { + result = obj.invokeMethod(request.getMethod(), request.getParams()); + } catch (InvocationTargetException e) { + Throwable throwable = e.getTargetException(); + error = new OmqException(throwable.getClass().getCanonicalName(), throwable.getMessage()); + } catch (NoSuchMethodException e) { + 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); + + Channel channel = obj.getChannel(); + + BasicProperties props = delivery.getProperties(); + + BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()).build(); + + byte[] bytesResponse = Serializer.serialize(resp); + channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse); + + // Log.saveLog("Server-Serialize", bytesResponse); + } + + } catch (InterruptedException i) { + i.printStackTrace(); + killed = true; + } catch (Exception e) { + System.out.println("Error a l'Invocation Thread \nException: " + e); + e.printStackTrace(); + } + + } + } + + public RemoteObject getObj() { + return obj; + } + + public void setObj(RemoteObject obj) { + this.obj = obj; + } + + public BlockingQueue getDeliveryQueue() { + return deliveryQueue; + } + + public void setDeliveryQueue(BlockingQueue deliveryQueue) { + this.deliveryQueue = deliveryQueue; + } +} Index: trunk/src/main/java/omq/server/RemoteObject.java =================================================================== --- trunk/src/main/java/omq/server/RemoteObject.java (revision 44) +++ trunk/src/main/java/omq/server/RemoteObject.java (revision 44) @@ -0,0 +1,247 @@ +package omq.server; + +import java.io.IOException; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Collection; +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.event.Event; +import omq.common.event.EventListener; +import omq.common.event.EventWrapper; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; +import omq.exception.SerializerException; + +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; + +/** + * + * @author Sergi Toda + * + */ +public abstract class RemoteObject extends Thread implements Remote { + + private static final long serialVersionUID = -1778953938739846450L; + + private String UID; + private Properties env; + private transient RemoteWrapper remoteWrapper; + private transient Map>> params; + private transient Channel channel; + private transient QueueingConsumer consumer; + private transient boolean killed = false; + + private static final Map> primitiveClasses = new HashMap>(); + + 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() { + } + + public void startRemoteObject(String reference, Properties env) throws Exception { + this.UID = reference; + this.env = env; + + params = new HashMap>>(); + for (Method m : this.getClass().getMethods()) { + List> list = new ArrayList>(); + for (Class clazz : m.getParameterTypes()) { + list.add(clazz); + } + params.put(m.getName(), list); + } + + // Get num threads to use + int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1")); + remoteWrapper = new RemoteWrapper(this, numThreads); + + startQueues(); + + // Start this listener + this.start(); + } + + @Override + public void run() { + while (!killed) { + try { + Delivery delivery = consumer.nextDelivery(); + System.out.println("RemoteObject: " + UID + " has received a message"); + remoteWrapper.notifyDelivery(delivery); + } catch (InterruptedException i) { + i.printStackTrace(); + } catch (ShutdownSignalException e) { + e.printStackTrace(); + 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) { + e2.printStackTrace(); + } + e1.printStackTrace(); + } + } catch (ConsumerCancelledException e) { + e.printStackTrace(); + } catch (SerializerException e) { + e.printStackTrace(); + } catch (Exception e) { + e.printStackTrace(); + } + } + } + + @Override + public String getRef() { + return UID; + } + + @Override + public void notifyEvent(Event event) throws IOException, SerializerException { + event.setTopic(UID); + EventWrapper wrapper = new EventWrapper(event); + channel.exchangeDeclare(UID, "fanout"); + channel.basicPublish(UID, "", null, Serializer.serialize(wrapper)); + } + + public void kill() throws IOException { + killed = true; + interrupt(); + channel.close(); + remoteWrapper.stopRemoteWrapper(); + } + + 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); + } + + 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; + } + + 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> getParams(String methodName) { + return params.get(methodName); + } + + public Channel getChannel() { + return channel; + } + + private void startQueues() throws Exception { + // Get info about which exchange and queue will use + String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE); + String queue = UID; + String routingKey = UID; + boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false")); + + // Start channel + channel = Broker.getNewChannel(); + + // Declares and bindings + System.out.println("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue); + channel.exchangeDeclare(exchange, "direct"); + channel.queueDeclare(queue, durable, false, false, null); + channel.queueBind(queue, exchange, routingKey); + + // Declare the event topic fanout + System.out.println("RemoteObject: " + UID + " declaring fanout exchange: " + UID); + channel.exchangeDeclare(UID, "fanout"); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(queue, true, consumer); + } + + @Override + public void addListener(EventListener eventListener) throws Exception { + } + + @Override + public void removeListener(EventListener eventListener) throws Exception { + } + + @Override + public Collection> getListeners() throws Exception { + return null; + } + +} Index: trunk/src/main/java/omq/server/RemoteWrapper.java =================================================================== --- trunk/src/main/java/omq/server/RemoteWrapper.java (revision 44) +++ trunk/src/main/java/omq/server/RemoteWrapper.java (revision 44) @@ -0,0 +1,77 @@ +package omq.server; + +import java.util.ArrayList; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; + +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; + +/** + * + * @author Sergi Toda + * + */ +public class RemoteWrapper { + private RemoteObject obj; + private int numThreads; + private ArrayList invocationList; + private BlockingQueue deliveryQueue; + + public RemoteWrapper(RemoteObject obj, int numThreads) { + this.obj = obj; + this.numThreads = numThreads; + invocationList = new ArrayList(); + deliveryQueue = new LinkedBlockingDeque(); + + System.out.println("RemoteWrapper -> Object: " + obj.getRef() + ", numthreads listening = " + numThreads); + + for (int i = 0; i < numThreads; i++) { + InvocationThread thread = new InvocationThread(obj, deliveryQueue); + invocationList.add(thread); + thread.start(); + } + } + + public void notifyDelivery(Delivery delivery) throws Exception { + this.deliveryQueue.put(delivery); + } + + public void stopRemoteWrapper() { + for (InvocationThread thread : invocationList) { + thread.interrupt(); + } + } + + public RemoteObject getObj() { + return obj; + } + + public void setObj(RemoteObject obj) { + this.obj = obj; + } + + public int getNumThreads() { + return numThreads; + } + + public void setNumThreads(int numThreads) { + this.numThreads = numThreads; + } + + public ArrayList getInvocationList() { + return invocationList; + } + + public void setInvocationList(ArrayList invocationList) { + this.invocationList = invocationList; + } + + public BlockingQueue getDeliveryQueue() { + return deliveryQueue; + } + + public void setDeliveryQueue(BlockingQueue deliveryQueue) { + this.deliveryQueue = deliveryQueue; + } +} Index: trunk/src/test/java/calculatorTest/Calculator.java =================================================================== --- trunk/src/test/java/calculatorTest/Calculator.java (revision 44) +++ trunk/src/test/java/calculatorTest/Calculator.java (revision 44) @@ -0,0 +1,28 @@ +package calculatorTest; + +import java.io.IOException; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; +import omq.exception.SerializerException; + +@RemoteInterface +public interface Calculator extends Remote { + @SyncMethod(timeout = 1500) + public int add(int x, int y); + + @AsyncMethod + public void mult(int x, int y); + + @AsyncMethod + public void sendMessage(Message m); + + @AsyncMethod + public void asyncDivideByZero() throws IOException, SerializerException; + + @SyncMethod + public int divideByZero(); + +} Index: trunk/src/test/java/calculatorTest/CalculatorImpl.java =================================================================== --- trunk/src/test/java/calculatorTest/CalculatorImpl.java (revision 44) +++ trunk/src/test/java/calculatorTest/CalculatorImpl.java (revision 44) @@ -0,0 +1,55 @@ +package calculatorTest; + +import java.io.IOException; + +import omq.common.broker.Broker; +import omq.exception.SerializerException; +import omq.server.RemoteObject; + +public class CalculatorImpl extends RemoteObject implements Calculator { + private int mult = 0; + + public CalculatorImpl() throws Exception { + super(); + } + + private static final long serialVersionUID = 1L; + + @Override + public int add(int x, int y) { + return x + y; + } + + @Override + public void mult(int x, int y) { + mult = x * y; + } + + public int getMult() { + return mult; + } + + public void setMult(int mult) { + this.mult = mult; + } + + @Override + public void asyncDivideByZero() throws IOException, SerializerException { + ZeroEvent ze = new ZeroEvent("my zero event", "zero-event"); + Broker.trigger(ze); + //notifyEvent(ze); + } + + @Override + public void sendMessage(Message m) { + System.out.println("Code = "+m.getCode()); + System.out.println("Message = "+m.getMessage()); + } + + @Override + public int divideByZero() { + int x = 2 / 0; + return x; + } + +} Index: trunk/src/test/java/calculatorTest/ClientTest.java =================================================================== --- trunk/src/test/java/calculatorTest/ClientTest.java (revision 44) +++ trunk/src/test/java/calculatorTest/ClientTest.java (revision 44) @@ -0,0 +1,97 @@ +package calculatorTest; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class ClientTest { + private static Calculator remoteCalc; + private static Calculator remoteCalc2; + + @BeforeClass + public static void startClient() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + // env.setProperty(ParameterQueue.DEBUGFILE, "c:\\middlewareDebug"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + + Broker.initBroker(env); + remoteCalc = (Calculator) Broker.lookup("calculator1", Calculator.class); + remoteCalc2 = (Calculator) Broker.lookup("calculator2", Calculator.class); + } + + @Test + public void add() throws Exception { + int x = 10; + int y = 20; + + int sync = remoteCalc.add(x, y); + int sum = x + y; + + assertEquals(sum, sync); + } + + @Test + public void add2() throws Exception { + int x = 10; + int y = 20; + + int sync = remoteCalc2.add(x, y); + int sum = x + y; + + assertEquals(sum, sync); + } + + @Test + public void mult() throws Exception { + int x = 5; + int y = 15; + + remoteCalc.mult(x, y); + Thread.sleep(200); + } + + @Test + public void notifyEvent() throws Exception { + ZeroListener zL = new ZeroListener("zero-event"); + + remoteCalc.addListener(zL); + + remoteCalc.asyncDivideByZero(); + + Thread.sleep(200); + } + + @Test + public void sendMessage() throws Exception { + Message m = new Message(2334, "Hello objectmq"); + remoteCalc.sendMessage(m); + } + + @Test(expected = ArithmeticException.class) + public void divideByZero() { + remoteCalc.divideByZero(); + } +} Index: trunk/src/test/java/calculatorTest/Message.java =================================================================== --- trunk/src/test/java/calculatorTest/Message.java (revision 44) +++ trunk/src/test/java/calculatorTest/Message.java (revision 44) @@ -0,0 +1,39 @@ +package calculatorTest; + +import java.io.Serializable; + +public class Message implements Serializable { + + + + /** + * + */ + private static final long serialVersionUID = 1L; + private int code; + private String message; + + public Message() { + } + + public Message(int code, String message) { + this.code = code; + this.message = message; + } + + public int getCode() { + return code; + } + + public void setCode(int code) { + this.code = code; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } +} Index: trunk/src/test/java/calculatorTest/ServerTest.java =================================================================== --- trunk/src/test/java/calculatorTest/ServerTest.java (revision 44) +++ trunk/src/test/java/calculatorTest/ServerTest.java (revision 44) @@ -0,0 +1,38 @@ +package calculatorTest; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +public class ServerTest { + private static CalculatorImpl calc; + private static CalculatorImpl calc2; + + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "true"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + calc = new CalculatorImpl(); + calc2 = new CalculatorImpl(); + + Broker.initBroker(env); + Broker.bind("calculator1", calc); + Broker.bind("calculator2", calc2); + + System.out.println("Server started"); + } +} Index: trunk/src/test/java/calculatorTest/ZeroEvent.java =================================================================== --- trunk/src/test/java/calculatorTest/ZeroEvent.java (revision 44) +++ trunk/src/test/java/calculatorTest/ZeroEvent.java (revision 44) @@ -0,0 +1,22 @@ +package calculatorTest; + +import omq.common.event.Event; + +public class ZeroEvent extends Event { + /** + * + */ + private static final long serialVersionUID = 1L; + + public ZeroEvent() { + } + + public ZeroEvent(String corrId, String topic) { + super(corrId, topic); + } + + public String getZeroMessage() { + return "divition by 0"; + } + +} Index: trunk/src/test/java/calculatorTest/ZeroListener.java =================================================================== --- trunk/src/test/java/calculatorTest/ZeroListener.java (revision 44) +++ trunk/src/test/java/calculatorTest/ZeroListener.java (revision 44) @@ -0,0 +1,15 @@ +package calculatorTest; + +import omq.common.event.EventListener; + +public class ZeroListener extends EventListener { + + public ZeroListener(String topic) { + super(topic); + } + + @Override + public void notifyEvent(ZeroEvent event) { + System.out.println(event.getZeroMessage()); + } +} Index: trunk/src/test/java/exceptionTest/ClientInterface.java =================================================================== --- trunk/src/test/java/exceptionTest/ClientInterface.java (revision 44) +++ trunk/src/test/java/exceptionTest/ClientInterface.java (revision 44) @@ -0,0 +1,33 @@ +package exceptionTest; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; + +@RemoteInterface +public interface ClientInterface extends Remote { + @AsyncMethod + public void addWheels(int numWheels); + + @AsyncMethod + public void addHp(int hp); + + @AsyncMethod + public void addTrailer(Trailer t); + + @AsyncMethod + public void setPrice(double price); + + @SyncMethod(timeout = 1000) + public Trailer getTrailer(); + + @SyncMethod(timeout = 1000) + public int getHp(); + + @SyncMethod(timeout = 1000) + public int getWheels(); + + @SyncMethod(timeout = 1000) + public double getPrice(); +} Index: trunk/src/test/java/exceptionTest/ClientTest.java =================================================================== --- trunk/src/test/java/exceptionTest/ClientTest.java (revision 44) +++ trunk/src/test/java/exceptionTest/ClientTest.java (revision 44) @@ -0,0 +1,87 @@ +package exceptionTest; + +import static org.junit.Assert.assertEquals; + +import java.lang.reflect.UndeclaredThrowableException; +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class ClientTest { + private static ClientInterface client; + + @BeforeClass + public static void startClient() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.kryo); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + + Broker.initBroker(env); + client = (ClientInterface) Broker.lookup("server", ClientInterface.class); + } + + @Test + public void addWheels() throws Exception { + int wheels = 4; + client.addWheels(wheels); + Thread.sleep(200); + int result = client.getWheels(); + + assertEquals(wheels, result); + } + + @Test + public void addHp() throws Exception { + int hp = 200; + client.addHp(hp); + Thread.sleep(200); + int result = client.getHp(); + + assertEquals(hp, result); + } + + @Test + public void addTrailer() throws Exception { + Trailer t = new Trailer(1200); + client.addTrailer(t); + Thread.sleep(200); + } + + @Test(expected = UndeclaredThrowableException.class) + // This exception will be caused by java.lang.NoSuchMethodException + public void getTrailer() throws Exception { + client.getTrailer(); + } + + @Test + public void setPrice() throws Exception { + double price = 4999.99; + client.setPrice(price); + Thread.sleep(200); + } + + @Test(expected = ClassCastException.class) + public void getPrice() throws Exception { + client.getPrice(); + } +} Index: trunk/src/test/java/exceptionTest/OmqServerImpl.java =================================================================== --- trunk/src/test/java/exceptionTest/OmqServerImpl.java (revision 44) +++ trunk/src/test/java/exceptionTest/OmqServerImpl.java (revision 44) @@ -0,0 +1,44 @@ +package exceptionTest; + +import omq.server.RemoteObject; + +public class OmqServerImpl extends RemoteObject implements ServerInterface { + /** + * + */ + private static final long serialVersionUID = 1L; + private int numWheels; + private int hp; + private int price; + + @Override + public void addWheels(int numWheels) { + this.numWheels = numWheels; + } + + @Override + public void addHp(int hp) { + this.hp = hp; + } + + @Override + public int getHp() { + return hp; + } + + @Override + public int getWheels() { + return numWheels; + } + + @Override + public void setPrice(int price) { + this.price = price; + } + + @Override + public int getPrice() { + return price; + } + +} Index: trunk/src/test/java/exceptionTest/ServerInterface.java =================================================================== --- trunk/src/test/java/exceptionTest/ServerInterface.java (revision 44) +++ trunk/src/test/java/exceptionTest/ServerInterface.java (revision 44) @@ -0,0 +1,17 @@ +package exceptionTest; + +import omq.Remote; + +public interface ServerInterface extends Remote { + public void addWheels(int numWheels); + + public void addHp(int hp); + + public int getHp(); + + public int getWheels(); + + public void setPrice(int price); + + public int getPrice(); +} Index: trunk/src/test/java/exceptionTest/ServerTest.java =================================================================== --- trunk/src/test/java/exceptionTest/ServerTest.java (revision 44) +++ trunk/src/test/java/exceptionTest/ServerTest.java (revision 44) @@ -0,0 +1,32 @@ +package exceptionTest; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +public class ServerTest { + + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.kryo); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + OmqServerImpl server = new OmqServerImpl(); + + Broker.initBroker(env); + Broker.bind("server", server); + } +} Index: trunk/src/test/java/exceptionTest/Trailer.java =================================================================== --- trunk/src/test/java/exceptionTest/Trailer.java (revision 44) +++ trunk/src/test/java/exceptionTest/Trailer.java (revision 44) @@ -0,0 +1,34 @@ +package exceptionTest; + +import java.io.Serializable; + +public class Trailer implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 1L; + + private int kg; + + public Trailer(int kg) { + this.kg = kg; + } + + public int getKg() { + return kg; + } + + public void setKg(int kg) { + this.kg = kg; + } + + @Override + public boolean equals(Object obj) { + if (obj instanceof Trailer) { + Trailer t = (Trailer) obj; + return kg == t.getKg(); + } + return false; + } +} Index: trunk/src/test/java/faultToleranceTest/ClientTest.java =================================================================== --- trunk/src/test/java/faultToleranceTest/ClientTest.java (revision 44) +++ trunk/src/test/java/faultToleranceTest/ClientTest.java (revision 44) @@ -0,0 +1,67 @@ +package faultToleranceTest; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.BeforeClass; +import org.junit.Test; + +import calculatorTest.Calculator; + +public class ClientTest { + private static Calculator remoteCalc; + + @BeforeClass + public static void startClient() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + // env.setProperty(ParameterQueue.DEBUGFILE, "c:\\middlewareDebug"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "5000"); + + Broker.initBroker(env); + remoteCalc = (Calculator) Broker.lookup("calculator1", Calculator.class); + } + + @Test + public void toleranceTest() throws Exception { + int x = 10; + int y = 20; + int sum = 10 + 20; + + int sync = remoteCalc.add(x, y); + + String password = "unpc"; + String[] command = { "/bin/bash", "-c", "echo " + password + " | sudo -S service rabbitmq-server restart" }; + + Runtime runtime = Runtime.getRuntime(); + runtime.exec(command); + + Thread.sleep(15000); + int resp = remoteCalc.add(x, y); + + assertEquals(sum, sync); + assertEquals(sum, resp); + } + +} Index: trunk/src/test/java/faultToleranceTest/ServerTest.java =================================================================== --- trunk/src/test/java/faultToleranceTest/ServerTest.java (revision 44) +++ trunk/src/test/java/faultToleranceTest/ServerTest.java (revision 44) @@ -0,0 +1,36 @@ +package faultToleranceTest; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; +import calculatorTest.CalculatorImpl; + +public class ServerTest { + private static CalculatorImpl calc; + + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + calc = new CalculatorImpl(); + + Broker.initBroker(env); + Broker.bind("calculator1", calc); + + System.out.println("Server started"); + } +} Index: trunk/src/test/java/multiProcessTest/ClientTest.java =================================================================== --- trunk/src/test/java/multiProcessTest/ClientTest.java (revision 44) +++ trunk/src/test/java/multiProcessTest/ClientTest.java (revision 44) @@ -0,0 +1,53 @@ +package multiProcessTest; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class ClientTest { + // Execute ServerTest.java 2 times before start this test + public static Number remoteNumber; + + @BeforeClass + public static void startClient() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + + Broker.initBroker(env); + remoteNumber = Broker.lookup("number", Number.class); + } + + @Test + public void test() { + int x = 10; + remoteNumber.setNumber(x); + int a = remoteNumber.getNumer(); + assertEquals(a, 0); + int b = remoteNumber.getNumer(); + assertEquals(x, b); + } + +} Index: trunk/src/test/java/multiProcessTest/Number.java =================================================================== --- trunk/src/test/java/multiProcessTest/Number.java (revision 44) +++ trunk/src/test/java/multiProcessTest/Number.java (revision 44) @@ -0,0 +1,14 @@ +package multiProcessTest; + +import omq.Remote; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; + +@RemoteInterface +public interface Number extends Remote { + @SyncMethod(timeout = 1000) + public void setNumber(int x); + + @SyncMethod(timeout = 1000) + public int getNumer(); +} Index: trunk/src/test/java/multiProcessTest/NumberImpl.java =================================================================== --- trunk/src/test/java/multiProcessTest/NumberImpl.java (revision 44) +++ trunk/src/test/java/multiProcessTest/NumberImpl.java (revision 44) @@ -0,0 +1,32 @@ +package multiProcessTest; + +import omq.client.annotation.SyncMethod; +import omq.server.RemoteObject; + +public class NumberImpl extends RemoteObject implements Number { + /** + * + */ + private static final long serialVersionUID = 1L; + private int x = 0; + + public NumberImpl() { + } + + public NumberImpl(int x) { + this.x = x; + } + + @Override + @SyncMethod + public void setNumber(int x) { + this.x = x; + } + + @Override + @SyncMethod(timeout = 1000) + public int getNumer() { + return x; + } + +} Index: trunk/src/test/java/multiProcessTest/ServerTest.java =================================================================== --- trunk/src/test/java/multiProcessTest/ServerTest.java (revision 44) +++ trunk/src/test/java/multiProcessTest/ServerTest.java (revision 44) @@ -0,0 +1,29 @@ +package multiProcessTest; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +public class ServerTest { + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + Broker.initBroker(env); + Broker.bind("number", new NumberImpl()); + } +} Index: trunk/src/test/java/multiThreadTest/Car.java =================================================================== --- trunk/src/test/java/multiThreadTest/Car.java (revision 44) +++ trunk/src/test/java/multiThreadTest/Car.java (revision 44) @@ -0,0 +1,37 @@ +package multiThreadTest; + +import java.util.List; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; +import omq.exception.RemoteException; + +@RemoteInterface +public interface Car extends Remote { + @AsyncMethod + public void setPlate(String plate); + + @SyncMethod(timeout = 1500) + public String getPlate(); + + @AsyncMethod + public void setHP(int hp); + + @SyncMethod(timeout = 1500) + public int getHP(); + + @AsyncMethod + public void setRims(List rims); + + @SyncMethod(timeout = 3000) + public List getRims(); + + @AsyncMethod + public void setMobile(String mobile) throws RemoteException; + + @SyncMethod + public String getMobile(); + +} Index: trunk/src/test/java/multiThreadTest/CarImpl.java =================================================================== --- trunk/src/test/java/multiThreadTest/CarImpl.java (revision 44) +++ trunk/src/test/java/multiThreadTest/CarImpl.java (revision 44) @@ -0,0 +1,60 @@ +package multiThreadTest; + +import java.util.List; + +import omq.common.broker.Broker; +import omq.exception.RemoteException; +import omq.server.RemoteObject; + +public class CarImpl extends RemoteObject implements Car { + + /** + * + */ + private static final long serialVersionUID = 1L; + private String plate; + private int hp; + private List rims; + private Mobile mobile; + + @Override + public void setPlate(String plate) { + this.plate = plate; + } + + @Override + public String getPlate() { + return plate; + } + + @Override + public void setHP(int hp) { + this.hp = hp; + } + + @Override + public int getHP() { + return hp; + } + + @Override + public void setRims(List rims) { + this.rims = rims; + } + + @Override + public List getRims() { + return rims; + } + + @Override + public void setMobile(String mobile) throws RemoteException { + this.mobile = (Mobile) Broker.lookup(mobile, Mobile.class); + } + + @Override + public String getMobile() { + return mobile.getRef(); + } + +} Index: trunk/src/test/java/multiThreadTest/CarThread.java =================================================================== --- trunk/src/test/java/multiThreadTest/CarThread.java (revision 44) +++ trunk/src/test/java/multiThreadTest/CarThread.java (revision 44) @@ -0,0 +1,51 @@ +package multiThreadTest; + +import java.util.ArrayList; +import java.util.List; + +import omq.common.broker.Broker; + +public class CarThread extends Thread { + private String name; + private String tfn; + private Object lock; + + public CarThread(String name, String tfn, Object lock) { + this.name = name; + this.tfn = tfn; + this.lock = lock; + } + + @Override + public void run() { + try { + Car car = (Car) Broker.lookup(name, Car.class); + car.setHP(1001); + car.setPlate("California 125"); + List rims = new ArrayList(); + rims.add(new Rim("asdf", 17)); + rims.add(new Rim("qwer", 21)); + car.setRims(rims); + + Thread.sleep(1000); + + System.out.println("HP -> " + car.getHP()); + System.out.println("Plate -> " + car.getPlate()); + for (Rim r : car.getRims()) { + System.out.println("Rim -> " + r.getModel() + ", " + r.getInch()); + } + + Mobile mobile = (Mobile) Broker.lookup(tfn, Mobile.class); + synchronized (lock) { + lock.wait(2000); + List messages = mobile.getMessages(); + for (String m : messages) { + System.out.println("Message -> " + m); + } + } + + } catch (Exception e) { + + } + } +} Index: trunk/src/test/java/multiThreadTest/ClientTest.java =================================================================== --- trunk/src/test/java/multiThreadTest/ClientTest.java (revision 44) +++ trunk/src/test/java/multiThreadTest/ClientTest.java (revision 44) @@ -0,0 +1,41 @@ +package multiThreadTest; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +public class ClientTest { + + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + // env.setProperty(ParameterQueue.SERIALIZERNAME, + // "omq.common.util.Serializers.GsonImp"); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + + Broker.initBroker(env); + + String car = "audi"; + String tfn = "aifon"; + Object lock = new Object(); + CarThread t1 = new CarThread(car, tfn, lock); + MobileThread t2 = new MobileThread(tfn, lock); + + t1.start(); + t2.start(); + + } +} Index: trunk/src/test/java/multiThreadTest/Mobile.java =================================================================== --- trunk/src/test/java/multiThreadTest/Mobile.java (revision 44) +++ trunk/src/test/java/multiThreadTest/Mobile.java (revision 44) @@ -0,0 +1,19 @@ +package multiThreadTest; + +import java.util.List; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; + +@RemoteInterface +public interface Mobile extends Remote { + @AsyncMethod + public void sendMessage(String message); + + @SyncMethod + public List getMessages(); + + +} Index: trunk/src/test/java/multiThreadTest/MobileImpl.java =================================================================== --- trunk/src/test/java/multiThreadTest/MobileImpl.java (revision 44) +++ trunk/src/test/java/multiThreadTest/MobileImpl.java (revision 44) @@ -0,0 +1,26 @@ +package multiThreadTest; + +import java.util.ArrayList; +import java.util.List; + +import omq.server.RemoteObject; + +public class MobileImpl extends RemoteObject implements Mobile { + /** + * + */ + private static final long serialVersionUID = 1L; + private List messages = new ArrayList(); + + @Override + public synchronized void sendMessage(String message) { + System.out.println("Message received " + this.getRef() + " -> " + message); + messages.add(message); + } + + @Override + public List getMessages() { + return messages; + } + +} Index: trunk/src/test/java/multiThreadTest/MobileThread.java =================================================================== --- trunk/src/test/java/multiThreadTest/MobileThread.java (revision 44) +++ trunk/src/test/java/multiThreadTest/MobileThread.java (revision 44) @@ -0,0 +1,28 @@ +package multiThreadTest; + +import omq.common.broker.Broker; + +public class MobileThread extends Thread { + private String name; + private Object lock; + + public MobileThread(String name, Object lock) { + this.name = name; + this.lock = lock; + } + + @Override + public void run() { + try { + Mobile mobile = (Mobile) Broker.lookup(name, Mobile.class); + synchronized (lock) { + mobile.sendMessage("Hello this is Sergi"); + mobile.sendMessage("and this is a test"); + Thread.sleep(1000); + lock.notify(); + } + } catch (Exception e) { + + } + } +} Index: trunk/src/test/java/multiThreadTest/Rim.java =================================================================== --- trunk/src/test/java/multiThreadTest/Rim.java (revision 44) +++ trunk/src/test/java/multiThreadTest/Rim.java (revision 44) @@ -0,0 +1,39 @@ +package multiThreadTest; + +import java.io.Serializable; + +public class Rim implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 1L; + + private String model; + private int inch; + + public Rim() { + } + + public Rim(String model, int inch) { + this.model = model; + this.inch = inch; + } + + public String getModel() { + return model; + } + + public void setModel(String model) { + this.model = model; + } + + public int getInch() { + return inch; + } + + public void setInch(int inch) { + this.inch = inch; + } + +} Index: trunk/src/test/java/multiThreadTest/ServerTest.java =================================================================== --- trunk/src/test/java/multiThreadTest/ServerTest.java (revision 44) +++ trunk/src/test/java/multiThreadTest/ServerTest.java (revision 44) @@ -0,0 +1,36 @@ +package multiThreadTest; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +public class ServerTest { + + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + // env.setProperty(ParameterQueue.SERIALIZERNAME, + // "omq.common.util.Serializers.GsonImp"); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + env.setProperty(ParameterQueue.NUM_THREADS, "4"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + String car = "audi"; + String tfn = "aifon"; + + Broker.initBroker(env); + Broker.bind(car, new CarImpl()); + Broker.bind(tfn, new MobileImpl()); + + System.out.println("Server started"); + } +} Index: trunk/src/test/java/stopBroker/BrokerKiller.java =================================================================== --- trunk/src/test/java/stopBroker/BrokerKiller.java (revision 44) +++ trunk/src/test/java/stopBroker/BrokerKiller.java (revision 44) @@ -0,0 +1,11 @@ +package stopBroker; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; + +@RemoteInterface +public interface BrokerKiller extends Remote { + @AsyncMethod + public void killServerBroker() throws Exception; +} Index: trunk/src/test/java/stopBroker/BrokerKillerImpl.java =================================================================== --- trunk/src/test/java/stopBroker/BrokerKillerImpl.java (revision 44) +++ trunk/src/test/java/stopBroker/BrokerKillerImpl.java (revision 44) @@ -0,0 +1,35 @@ +package stopBroker; + +import omq.client.annotation.AsyncMethod; +import omq.common.broker.Broker; +import omq.server.RemoteObject; + +public class BrokerKillerImpl extends RemoteObject implements BrokerKiller { + + /** + * + */ + private static final long serialVersionUID = 1L; + + @Override + @AsyncMethod + public void killServerBroker() throws Exception { + System.out.println("Kill broker"); + + // A remote method cannot stop the Broker because the stop method is + // thought to wait for the methods finish before it stops. For this + // reason it actually cannot stop itself + new Thread() { + public void run() { + try { + Thread.sleep(1000); + Broker.stopBroker(); + } catch (Exception e) { + e.printStackTrace(); + } + }; + }.start(); + + } + +} Index: trunk/src/test/java/stopBroker/ClientTest.java =================================================================== --- trunk/src/test/java/stopBroker/ClientTest.java (revision 44) +++ trunk/src/test/java/stopBroker/ClientTest.java (revision 44) @@ -0,0 +1,44 @@ +package stopBroker; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +public class ClientTest { + + /** + * @param args + * @throws Exception + */ + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + // env.setProperty(ParameterQueue.DEBUGFILE, "c:\\middlewareDebug"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "5000"); + + Broker.initBroker(env); + BrokerKiller bk = (BrokerKiller) Broker.lookup("bk", BrokerKiller.class); + + bk.killServerBroker(); + Broker.stopBroker(); + } + +} Index: trunk/src/test/java/stopBroker/ServerTest.java =================================================================== --- trunk/src/test/java/stopBroker/ServerTest.java (revision 44) +++ trunk/src/test/java/stopBroker/ServerTest.java (revision 44) @@ -0,0 +1,36 @@ +package stopBroker; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +public class ServerTest { + + /** + * @param args + */ + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + BrokerKillerImpl bki = new BrokerKillerImpl(); + + Broker.initBroker(env); + Broker.bind("bk", bki); + } + +} Index: trunk/src/test/java/stopBroker/UnbindTest.java =================================================================== --- trunk/src/test/java/stopBroker/UnbindTest.java (revision 44) +++ trunk/src/test/java/stopBroker/UnbindTest.java (revision 44) @@ -0,0 +1,43 @@ +package stopBroker; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; +import calculatorTest.CalculatorImpl; + +public class UnbindTest { + private static CalculatorImpl calc; + + /** + * @param args + */ + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + String reference = "calculator1"; + calc = new CalculatorImpl(); + + Broker.initBroker(env); + Broker.bind(reference, calc); + + Broker.unbind(reference); + + Broker.closeConnection(); + } + +} Index: trunk/src/test/java/test/Client.java =================================================================== --- trunk/src/test/java/test/Client.java (revision 44) +++ trunk/src/test/java/test/Client.java (revision 44) @@ -0,0 +1,28 @@ +package test; + +import java.util.Set; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; +import omq.exception.RemoteException; + +@RemoteInterface +public interface Client extends Remote { + + @SyncMethod(retry = 1, timeout = 1500) + public String getID(); + + @SyncMethod(retry = 1, timeout = 1500) + public String getProfileInfo(); + + @SyncMethod(retry = 1, timeout = 1500) + public Set getFriends(); + + @AsyncMethod + public void sendMessage(String message); + + @AsyncMethod + public void addContact(String contact) throws RemoteException; +} Index: trunk/src/test/java/test/ClientImpl.java =================================================================== --- trunk/src/test/java/test/ClientImpl.java (revision 44) +++ trunk/src/test/java/test/ClientImpl.java (revision 44) @@ -0,0 +1,62 @@ +package test; + +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.SyncMethod; +import omq.common.broker.Broker; +import omq.exception.RemoteException; +import omq.server.RemoteObject; + +public class ClientImpl extends RemoteObject implements Client { + + /** + * + */ + private static final long serialVersionUID = 1L; + private String id; + private String profileInfo; + private Map friendList; + + public ClientImpl(String id, String profileInfo) { + this.id = id; + this.profileInfo = profileInfo; + this.friendList = new HashMap(); + } + + @Override + @SyncMethod(retry = 1, timeout = 1500) + public String getID() { + return id; + } + + @Override + @SyncMethod(retry = 1, timeout = 1500) + public String getProfileInfo() { + return profileInfo; + } + + @Override + @SyncMethod(retry = 1, timeout = 1500) + public Set getFriends() { + return friendList.keySet(); + } + + @Override + @AsyncMethod + public void sendMessage(String message) { + System.out.println("" + message); + } + + @Override + @AsyncMethod + public void addContact(String contact) throws RemoteException { + if (!id.equalsIgnoreCase(contact) && !friendList.containsKey(contact)) { + Client client = (Client) Broker.lookup(contact, Client.class); + friendList.put(contact, client); + } + } + +} Index: trunk/src/test/java/test/ClientTest.java =================================================================== --- trunk/src/test/java/test/ClientTest.java (revision 44) +++ trunk/src/test/java/test/ClientTest.java (revision 44) @@ -0,0 +1,40 @@ +package test; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.BeforeClass; + +public class ClientTest { + // private static Client user; + + @BeforeClass + public static void startClient() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + // env.setProperty(ParameterQueue.DEBUGFILE, "c:\\middlewareDebug"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "5000"); + + Broker.initBroker(env); + } + +} Index: trunk/src/test/java/test/ServerTest.java =================================================================== --- trunk/src/test/java/test/ServerTest.java (revision 44) +++ trunk/src/test/java/test/ServerTest.java (revision 44) @@ -0,0 +1,36 @@ +package test; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +public class ServerTest { + + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + ClientImpl c1 = new ClientImpl("Miki", "Hey, this is Miky"); + ClientImpl c2 = new ClientImpl("Jack", "This is Jack"); + + Broker.initBroker(env); + + Broker.bind(c1.getID(), c1); + Broker.bind(c2.getID(), c2); + } + +}