Index: /tags/0.5.4/.classpath =================================================================== --- /tags/0.5.4/.classpath (revision 65) +++ /tags/0.5.4/.classpath (revision 65) @@ -0,0 +1,36 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Index: /tags/0.5.4/.project =================================================================== --- /tags/0.5.4/.project (revision 65) +++ /tags/0.5.4/.project (revision 65) @@ -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: /tags/0.5.4/.settings/org.eclipse.jdt.core.prefs =================================================================== --- /tags/0.5.4/.settings/org.eclipse.jdt.core.prefs (revision 65) +++ /tags/0.5.4/.settings/org.eclipse.jdt.core.prefs (revision 65) @@ -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: /tags/0.5.4/.settings/org.eclipse.m2e.core.prefs =================================================================== --- /tags/0.5.4/.settings/org.eclipse.m2e.core.prefs (revision 65) +++ /tags/0.5.4/.settings/org.eclipse.m2e.core.prefs (revision 65) @@ -0,0 +1,4 @@ +activeProfiles= +eclipse.preferences.version=1 +resolveWorkspaceProjects=true +version=1 Index: /tags/0.5.4/pom.xml =================================================================== --- /tags/0.5.4/pom.xml (revision 65) +++ /tags/0.5.4/pom.xml (revision 65) @@ -0,0 +1,124 @@ + + 4.0.0 + objectmq + objectmq + 0.5.4 + 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 + + + junit + junit + 4.11 + + + log4j + log4j + 1.2.16 + + + log4j + apache-log4j-extras + 1.1 + + + + src + test + + + src/main/resources + true + + log4j.xml + example.properties + version.properties + + + + + + maven-compiler-plugin + 3.0 + + 1.6 + 1.6 + + + + org.apache.maven.plugins + maven-assembly-plugin + + + package + + attached + + + + + + jar-with-dependencies + + + + true + + + + + + org.apache.maven.plugins + maven-surefire-plugin + 2.14.1 + + true + + + + + Index: /tags/0.5.4/src/main/java/omq/Remote.java =================================================================== --- /tags/0.5.4/src/main/java/omq/Remote.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/Remote.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/client/annotation/AsyncMethod.java =================================================================== --- /tags/0.5.4/src/main/java/omq/client/annotation/AsyncMethod.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/client/annotation/AsyncMethod.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/client/annotation/MultiMethod.java =================================================================== --- /tags/0.5.4/src/main/java/omq/client/annotation/MultiMethod.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/client/annotation/MultiMethod.java (revision 65) @@ -0,0 +1,12 @@ +package omq.client.annotation; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.METHOD) +public @interface MultiMethod { + int waitNum() default 1; +} Index: /tags/0.5.4/src/main/java/omq/client/annotation/RemoteInterface.java =================================================================== --- /tags/0.5.4/src/main/java/omq/client/annotation/RemoteInterface.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/client/annotation/RemoteInterface.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/client/annotation/SyncMethod.java =================================================================== --- /tags/0.5.4/src/main/java/omq/client/annotation/SyncMethod.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/client/annotation/SyncMethod.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/client/listener/RequestInfo.java =================================================================== --- /tags/0.5.4/src/main/java/omq/client/listener/RequestInfo.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/client/listener/RequestInfo.java (revision 65) @@ -0,0 +1,45 @@ +package omq.client.listener; + +public class RequestInfo { + private String uid_request; + private boolean multi; + private int wait; + + public RequestInfo(String uid_request, boolean multi, int wait) { + this.uid_request = uid_request; + this.multi = multi; + this.wait = wait; + } + + public String getUid_request() { + return uid_request; + } + + public void setUid_request(String uid_request) { + this.uid_request = uid_request; + } + + public boolean isMulti() { + return multi; + } + + public void setMulti(boolean multi) { + this.multi = multi; + } + + public int getWait() { + return wait; + } + + public void setWait(int wait) { + this.wait = wait; + } + + @Override + public boolean equals(Object obj) { + if (obj instanceof RequestInfo) { + return uid_request.equals(((RequestInfo) obj).getUid_request()); + } + return false; + } +} Index: /tags/0.5.4/src/main/java/omq/client/listener/ResponseListener.java =================================================================== --- /tags/0.5.4/src/main/java/omq/client/listener/ResponseListener.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/client/listener/ResponseListener.java (revision 65) @@ -0,0 +1,159 @@ +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 org.apache.log4j.Logger; + +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 final Logger logger = Logger.getLogger(ResponseListener.class.getName()); + + private Broker broker; + 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 + */ + public ResponseListener(Broker broker) throws Exception { + this.broker = broker; + env = broker.getEnvironment(); + + // Init the hashtable (it's concurrent) + results = new Hashtable>(); + + startRPCQueue(); + } + + @Override + public void run() { + logger.info("ResponseListener started"); + 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(); + logger.debug("Response received -> proxy reference: " + props.getAppId() + ", corrId: " + 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) { + logger.error(i.toString(), i); + } catch (ShutdownSignalException e) { + logger.error(e.toString(), e); + 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) { + logger.error(e2.toString(), e2); + } + } + } catch (ConsumerCancelledException e) { + logger.error(e.toString(), e); + } catch (Exception e) { + logger.error(e.toString(), e); + } + } + } + + 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); + logger.info("ResponseListener creating queue: " + reply_queue + ", durable: " + durable + "TTL: " + (ttl > 0 ? ttl : "not set")); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(reply_queue, true, consumer); + } + + /** + * + * @param key + * @return whether the map has the param key + */ + public boolean containsKey(String key) { + return results.containsKey(key); + } + + /** + * Interrupt and kill the Thread + * + * @throws IOException + */ + public void kill() throws IOException { + logger.warn("Killing ResponseListener"); + interrupt(); + killed = true; + channel.close(); + } + + // Revisar això + public void registerProxy(Proxymq proxy) { + if (!results.containsKey(proxy.getRef())) { + results.put(proxy.getRef(), proxy.getResults()); + } + } +} Index: /tags/0.5.4/src/main/java/omq/client/proxy/MultiProxymq.java =================================================================== --- /tags/0.5.4/src/main/java/omq/client/proxy/MultiProxymq.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/client/proxy/MultiProxymq.java (revision 65) @@ -0,0 +1,16 @@ +package omq.client.proxy; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; + +public class MultiProxymq implements InvocationHandler { + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + + + + return null; + } + +} Index: /tags/0.5.4/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- /tags/0.5.4/src/main/java/omq/client/proxy/Proxymq.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/client/proxy/Proxymq.java (revision 65) @@ -0,0 +1,356 @@ +package omq.client.proxy; + +import java.io.IOException; +import java.lang.reflect.Array; +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.Map; +import java.util.Properties; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.MultiMethod; +import omq.client.annotation.SyncMethod; +import omq.client.listener.ResponseListener; +import omq.common.broker.Broker; +import omq.common.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.OmqException; +import omq.exception.RetryException; +import omq.exception.SerializerException; +import omq.exception.TimeoutException; + +import org.apache.log4j.Logger; + +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 final Logger logger = Logger.getLogger(Proxymq.class.getName()); + private static final String multi = "multi#"; + + private String uid; + private transient String serializerType; + private transient Broker broker; + private transient ResponseListener rListener; + private transient EventDispatcher dispatcher; + private transient Serializer serializer; + 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, Broker broker) throws Exception { + this.uid = uid; + this.broker = broker; + rListener = broker.getResponseListener(); + dispatcher = broker.getEventDispatcher(); + serializer = broker.getSerializer(); + + // TODO what is better to use a new channel or to use the same? + // this.channel = Broker.getChannel(); + env = broker.getEnvironment(); + + // set the serializer type + serializerType = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA); + + 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 { + // 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); + + Object response = null; + // Publish the request + if (request.isAsync()) { + logger.debug("Publish async request -> " + request.getId()); + publishAsyncRequest(request); + } else { + logger.debug("Publish sync request -> " + request.getId()); + response = publishSyncRequest(request, method.getReturnType()); + } + + return response; + } + + private void publishMessage(Request request, String replyQueueName) throws Exception { + String corrId = request.getId(); + + // Get the environment properties + String exchange; + String routingkey; + + if (request.isMulti()) { + exchange = multi + env.getProperty(ParameterQueue.RPC_EXCHANGE); + routingkey = ""; + } else { + exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE); + routingkey = uid; + } + + // Add the correlation ID and create a replyTo property + BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).type(serializerType).build(); + + // Publish the message + byte[] bytesRequest = serializer.serialize(serializerType, request); + broker.getChannel().basicPublish(exchange, routingkey, props, 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); + if (request.isMulti()) { + return getResults(corrId, 2, timeout, type); + } else { + return getResult(corrId, timeout, type); + } + + } catch (TimeoutException te) { + logger.error(te); + } + i++; + } + throw new RetryException(retries, timeout); + } + + private Request createRequest(Method method, Object[] arguments) { + String corrId = java.util.UUID.randomUUID().toString(); + String methodName = method.getName(); + boolean multi = false; + int wait = 0; + + if (method.getAnnotation(MultiMethod.class) != null) { + multi = true; + wait = method.getAnnotation(MultiMethod.class).waitNum(); + } + + // Since we need to know whether the method is async and if it has to + // return using an annotation, we'll only check the AsyncMethod + // annotation + if (method.getAnnotation(AsyncMethod.class) == null) { + int retries = 1; + long timeout = ParameterQueue.DEFAULT_TIMEOUT; + if (method.getAnnotation(SyncMethod.class) != null) { + SyncMethod sync = method.getAnnotation(SyncMethod.class); + retries = sync.retry(); + timeout = sync.timeout(); + } + return Request.newSyncRequest(corrId, methodName, arguments, retries, timeout, multi, wait); + } else { + return Request.newAsyncRequest(corrId, methodName, arguments, multi); + } + } + + private Object getResult(String corrId, long timeout, Class type) throws Exception { + Response resp = null; + + // Wait for the results. + long localTimeout = timeout; + long start = System.currentTimeMillis(); + synchronized (results) { + // Due to we are using notifyAll(), we need to control the real time + while (!results.containsKey(corrId) && (timeout - localTimeout) >= 0) { + results.wait(localTimeout); + localTimeout = System.currentTimeMillis() - start; + } + if ((timeout - localTimeout) <= 0) { + throw new TimeoutException("Timeout exception time: " + timeout); + } + resp = serializer.deserializeResponse(results.get(corrId), type); + + // Remove and indicate the key exists (a hashmap can contain a null + // object, using this we'll know whether a response has been + // received before) + results.put(corrId, null); + } + + if (resp.getError() != null) { + OmqException error = resp.getError(); + String name = error.getType(); + String message = error.getMessage(); + throw (Exception) Class.forName(name).getConstructor(String.class).newInstance(message); + } + + return resp.getResult(); + } + + private Object getResults(String corrId, int wait, long timeout, Class type) throws Exception { + Response resp = null; + Class actualType = type.getComponentType(); + + Object array = Array.newInstance(actualType, wait); + + int i = 0; + long localTimeout = timeout; + long start = System.currentTimeMillis(); + + while (i < wait) { + synchronized (results) { + // Due to we are using notifyAll(), we need to control the real + // time + while (!results.containsKey(corrId) && (timeout - localTimeout) >= 0) { + results.wait(localTimeout); + localTimeout = System.currentTimeMillis() - start; + } + if ((timeout - localTimeout) <= 0) { + throw new TimeoutException("Timeout exception time: " + timeout); + } + // Remove the corrId to receive new replies + resp = serializer.deserializeResponse(results.remove(corrId), actualType); + Array.set(array, i, resp.getResult()); + } + i++; + } + synchronized (results) { + results.put(corrId, null); + } + + return array; + } + + /** + * 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) { + return Proxy.newProxyInstance(loader, interfaces, proxy); + } + + /** + * 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: /tags/0.5.4/src/main/java/omq/common/broker/Broker.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/broker/Broker.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/broker/Broker.java (revision 65) @@ -0,0 +1,298 @@ +package omq.common.broker; + +import java.io.IOException; +import java.net.URL; +import java.util.HashMap; +import java.util.Hashtable; +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.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 org.apache.log4j.Logger; +import org.apache.log4j.xml.DOMConfigurator; + +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 final Logger logger = Logger.getLogger(Broker.class.getName()); + + private Connection connection; + private Channel channel; + private ResponseListener responseListener; + private EventDispatcher eventDispatcher; + private Serializer serializer; + private boolean clientStarted = false; + private boolean connectionClosed = false; + private Properties environment = null; + private Map remoteObjs; + private Map proxies = new Hashtable(); + + public Broker(Properties env) throws Exception { + // Load log4j configuration + URL log4jResource = Broker.class.getResource("/log4j.xml"); + DOMConfigurator.configure(log4jResource); + + remoteObjs = new HashMap(); + serializer = new Serializer(env); + environment = 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"); + } + } + + public void stopBroker() throws Exception { + logger.warn("Stopping broker"); + // Stop the client + if (clientStarted) { + responseListener.kill(); + eventDispatcher.kill(); + //TODO proxies = null; ?? + } + // Stop all the remote objects working + for (String reference : remoteObjs.keySet()) { + unbind(reference); + } + + // Close the connection once all the listeners are died + closeConnection(); + + clientStarted = false; + connectionClosed = false; + environment = null; + remoteObjs = null; + // Serializer.removeSerializers(); + } + + /** + * @return Broker's connection + * @throws Exception + */ + public Connection getConnection() throws Exception { + return connection; + } + + public void closeConnection() throws IOException { + logger.warn("Clossing connection"); + connectionClosed = true; + connection.close(); + connectionClosed = false; + } + + /** + * + * @return Broker's channel + * @throws Exception + */ + public Channel getChannel() throws Exception { + return channel; + } + + /** + * Creates a new channel using the Broker's connection + * + * @return newChannel + * @throws IOException + */ + public Channel getNewChannel() throws IOException { + return connection.createChannel(); + } + + @SuppressWarnings("unchecked") + public synchronized T lookup(String reference, Class contract) throws RemoteException { + try { + + if (!clientStarted) { + initClient(environment); + clientStarted = true; + } + + if (!proxies.containsKey(reference)) { + Proxymq proxy = new Proxymq(reference, contract, this); + Class[] array = { contract }; + Object newProxy = Proxymq.newProxyInstance(contract.getClassLoader(), array, proxy); + proxies.put(reference, newProxy); + return (T) newProxy; + } + return (T) proxies.get(reference); + + } catch (Exception e) { + throw new RemoteException(e); + } + } + + public void bind(String reference, RemoteObject remote) throws RemoteException { + try { + remote.startRemoteObject(reference, this); + remoteObjs.put(reference, remote); + } catch (Exception e) { + throw new RemoteException(e); + } + } + + public 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 synchronized void initClient(Properties environment) throws Exception { + if (responseListener == null) { + responseListener = new ResponseListener(this); + responseListener.start(); + } + if (eventDispatcher == null) { + eventDispatcher = new EventDispatcher(this); + eventDispatcher.start(); + } + } + + /** + * This method sends an event with its information + * + * @param event + * @throws IOException + * @throws SerializerException + */ + public void trigger(Event event) throws IOException, SerializerException { + String UID = event.getTopic(); + EventWrapper wrapper = new EventWrapper(event); + logger.debug("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); + } + + /** + * This function is used to send a ping message to see if the connection + * works + * + * @param env + * @throws Exception + */ + public 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 void addFaultTolerance() { + connection.addShutdownListener(new ShutdownListener() { + @Override + public void shutdownCompleted(ShutdownSignalException cause) { + logger.warn("Shutdown message received. Cause: " + cause.getMessage()); + if (!connectionClosed) + if (cause.isHardError()) { + if (connection.isOpen()) { + try { + connection.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + try { + connection = OmqConnectionFactory.getNewWorkingConnection(environment); + 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(); + } + } + } + } + }); + } + + public Properties getEnvironment() { + return environment; + } + + public ResponseListener getResponseListener() { + return responseListener; + } + + public EventDispatcher getEventDispatcher() { + return eventDispatcher; + } + + public Serializer getSerializer() { + return serializer; + } +} Index: /tags/0.5.4/src/main/java/omq/common/event/Event.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/event/Event.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/event/Event.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/common/event/EventDispatcher.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/event/EventDispatcher.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/event/EventDispatcher.java (revision 65) @@ -0,0 +1,186 @@ +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 org.apache.log4j.Logger; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.ConsumerCancelledException; +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; +import com.rabbitmq.client.ShutdownSignalException; + +/** + * 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 final Logger logger = Logger.getLogger(EventDispatcher.class.getName()); + + private Broker broker; + private Serializer serializer; + private Map> listeners; + private Channel channel; + private QueueingConsumer consumer; + private Properties env; + private boolean killed = false; + + public EventDispatcher(Broker broker) throws Exception { + this.broker = broker; + env = broker.getEnvironment(); + serializer = broker.getSerializer(); + + // 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); + logger.info("EventDispatcher creating queue: " + event_queue + ", durable: " + durable); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(event_queue, true, consumer); + } + + public void kill() throws Exception { + logger.warn("Stopping EventDispatcher"); + setListeners(null); + killed = true; + interrupt(); + channel.close(); + } + + @Override + public void run() { + logger.info("EventDispatcher started"); + Delivery delivery; + Event event; + + while (!killed) { + try { + // Get the delivery + delivery = consumer.nextDelivery(); + + // Get the event + event = serializer.deserializeEvent(delivery.getBody()); + + logger.info("Event received -> Topic: " + event.getTopic() + "CorrId: " + event.getCorrId()); + + // Dispatch it + dispatch(event.getTopic(), event); + } catch (InterruptedException i) { + logger.error(i); + } catch (ShutdownSignalException e) { + logger.error(e); + 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) { + logger.error(e2); + } + logger.error(e1); + } + } catch (ConsumerCancelledException e) { + logger.error(e); + } catch (Exception e) { + logger.error(e); + } + } + } + + 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(); + + logger.info("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; + } + +} Index: /tags/0.5.4/src/main/java/omq/common/event/EventListener.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/event/EventListener.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/event/EventListener.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/common/event/EventWrapper.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/event/EventWrapper.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/event/EventWrapper.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/common/message/Request.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/message/Request.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/message/Request.java (revision 65) @@ -0,0 +1,125 @@ +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 boolean multi; + private transient int wait; + 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; + } + + private Request(String id, String method, boolean async, Object[] params, boolean multi) { + this.id = id; + this.method = method; + this.async = async; + this.params = params; + this.multi = multi; + } + + 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, boolean multi, int wait) { + Request req = new Request(id, method, false, params, multi); + req.setRetries(retries); + req.setTimeout(timeout); + req.setWait(wait); + return req; + } + + public static Request newAsyncRequest(String id, String method, Object[] params, boolean multi) { + return new Request(id, method, true, params, multi); + } + + 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; + } + + public boolean isMulti() { + return multi; + } + + public void setMulti(boolean multi) { + this.multi = multi; + } + + public int getWait() { + return wait; + } + + public void setWait(int wait) { + this.wait = wait; + } +} Index: /tags/0.5.4/src/main/java/omq/common/message/Response.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/message/Response.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/message/Response.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/common/util/OmqConnectionFactory.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/util/OmqConnectionFactory.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/util/OmqConnectionFactory.java (revision 65) @@ -0,0 +1,85 @@ +package omq.common.util; + +import java.io.IOException; +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.util.Properties; + +import org.apache.log4j.Logger; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + +/** + * + * @author Sergi Toda + * + */ +public class OmqConnectionFactory { + private static final Logger logger = Logger.getLogger(OmqConnectionFactory.class.getName()); + + 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) { + logger.error(e); + long milis = 2000; + if (env != null) { + 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(); + } + + Connection connection = factory.newConnection(); + logger.info("New connection created using: username: " + username + ", host: " + host + ", port: " + port + ", connection timeout: " + + connectionTimeout + " SSL enabled: " + ssl); + return connection; + } + + public static Channel getNewChannel() throws IOException { + Channel channel = connection.createChannel(); + logger.info("New channel created using the default connection"); + return channel; + } +} Index: /tags/0.5.4/src/main/java/omq/common/util/ParameterQueue.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/util/ParameterQueue.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/util/ParameterQueue.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/common/util/Serializer.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/util/Serializer.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/util/Serializer.java (revision 65) @@ -0,0 +1,173 @@ +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.SerializerException; +import omq.server.RemoteObject; + +/** + * + * @author Sergi Toda + * + */ +public class Serializer { + // private static final Logger logger = + // Logger.getLogger(Serializer.class.getName()); + public static final String KRYO = "kryo"; + public static final String JAVA = "java"; + public static final String GSON = "gson"; + + // Client serializer + public ISerializer serializer; + + // Server serializers + private ISerializer kryoSerializer; + private ISerializer javaSerializer; + private ISerializer gsonSerializer; + + private Properties env; + + public Serializer(Properties env) { + this.env = env; + } + + private Boolean getEnableCompression() { + return Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false")); + } + + public ISerializer getInstance() throws SerializerException { + if (serializer == null) { + try { + String className = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA); + + if (className == null || className.isEmpty()) { + throw new ClassNotFoundException("Class name is null or empty."); + } + + serializer = getInstance(className); + } catch (Exception ex) { + throw new SerializerException(ex.getMessage(), ex); + } + } + + return serializer; + } + + public ISerializer getInstance(String type) throws SerializerException { + if (KRYO.equals(type)) { + if (kryoSerializer == null) { + kryoSerializer = new KryoImp(); + } + return kryoSerializer; + } else if (GSON.equals(type)) { + if (gsonSerializer == null) { + gsonSerializer = new GsonImp(); + } + return gsonSerializer; + } else if (JAVA.equals(type)) { + if (javaSerializer == null) { + javaSerializer = new JavaImp(); + } + return javaSerializer; + } + + throw new SerializerException("Serializer not found."); + } + + public byte[] serialize(String type, Object obj) throws SerializerException { + ISerializer instance = getInstance(type); + + 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); + } + } + + // TODO: remove this function and think about the event serialization + public 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 Request deserializeRequest(String type, byte[] bytes, RemoteObject obj) throws SerializerException { + ISerializer instance = getInstance(type); + + 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 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 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); + } + } + + // public static void removeSerializers() { + // logger.warn("Removing serializers"); + // serializer = null; + // kryoSerializer = null; + // javaSerializer = null; + // gsonSerializer = null; + // } +} Index: /tags/0.5.4/src/main/java/omq/common/util/Serializers/GsonImp.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/util/Serializers/GsonImp.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/util/Serializers/GsonImp.java (revision 65) @@ -0,0 +1,98 @@ +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); + 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); + + 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: /tags/0.5.4/src/main/java/omq/common/util/Serializers/ISerializer.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/util/Serializers/ISerializer.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/util/Serializers/ISerializer.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/common/util/Serializers/JavaImp.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/util/Serializers/JavaImp.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/util/Serializers/JavaImp.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/common/util/Serializers/KryoImp.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/util/Serializers/KryoImp.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/util/Serializers/KryoImp.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/common/util/Zipper.java =================================================================== --- /tags/0.5.4/src/main/java/omq/common/util/Zipper.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/common/util/Zipper.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/ConnectionException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/ConnectionException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/ConnectionException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/EnvironmentException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/EnvironmentException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/EnvironmentException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/InitBrokerException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/InitBrokerException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/InitBrokerException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/NoContainsInstanceException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/NoContainsInstanceException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/NoContainsInstanceException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/NoSuchEvObjectException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/NoSuchEvObjectException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/NoSuchEvObjectException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/NotBoundException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/NotBoundException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/NotBoundException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/ObjectAlreadyExistsException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/ObjectAlreadyExistsException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/ObjectAlreadyExistsException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/ObjectNotFoundException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/ObjectNotFoundException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/ObjectNotFoundException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/OmqException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/OmqException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/OmqException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/RegistryNotLoadedException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/RegistryNotLoadedException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/RegistryNotLoadedException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/RemoteException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/RemoteException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/RemoteException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/RetryException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/RetryException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/RetryException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/SerializerException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/SerializerException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/SerializerException.java (revision 65) @@ -0,0 +1,20 @@ +package omq.exception; + +public class SerializerException extends Exception { + + /** + * + */ + private static final long serialVersionUID = 1L; + + public SerializerException(String message) { + super(message); + } + + public SerializerException(String message, Throwable cause) { + super(message, cause); + } + + + +} Index: /tags/0.5.4/src/main/java/omq/exception/SessionException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/SessionException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/SessionException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/TimeoutException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/TimeoutException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/TimeoutException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/exception/TypeNotFoundException.java =================================================================== --- /tags/0.5.4/src/main/java/omq/exception/TypeNotFoundException.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/exception/TypeNotFoundException.java (revision 65) @@ -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: /tags/0.5.4/src/main/java/omq/server/InvocationThread.java =================================================================== --- /tags/0.5.4/src/main/java/omq/server/InvocationThread.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/server/InvocationThread.java (revision 65) @@ -0,0 +1,106 @@ +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 org.apache.log4j.Logger; + +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 static final Logger logger = Logger.getLogger(InvocationThread.class.getName()); + private RemoteObject obj; + private transient Serializer serializer; + private BlockingQueue deliveryQueue; + private boolean killed = false; + + public InvocationThread(RemoteObject obj, BlockingQueue deliveryQueue, Serializer serializer) { + this.obj = obj; + this.deliveryQueue = deliveryQueue; + this.serializer = serializer; + } + + @Override + public void run() { + while (!killed) { + try { + // Get the delivery + Delivery delivery = deliveryQueue.take(); + + String serializerType = delivery.getProperties().getType(); + + // Deserialize the json + Request request = serializer.deserializeRequest(serializerType, delivery.getBody(), obj); + String methodName = request.getMethod(); + String requestID = request.getId(); + + logger.debug("Object: " + obj.getRef() + ", method: " + methodName + " corrID: " + requestID + ", serializerType: " + serializerType); + + // Invoke the method + Object result = null; + OmqException error = null; + try { + result = obj.invokeMethod(request.getMethod(), request.getParams()); + } catch (InvocationTargetException e) { + Throwable throwable = e.getTargetException(); + logger.error("Object: " + obj.getRef() + " at method: " + methodName + ", corrID" + requestID, throwable); + error = new OmqException(throwable.getClass().getCanonicalName(), throwable.getMessage()); + } catch (NoSuchMethodException e) { + logger.error("Object: " + obj.getRef() + " cannot find method: " + methodName); + error = new OmqException(e.getClass().getCanonicalName(), e.getMessage()); + } + + // Reply if it's necessary + if (!request.isAsync()) { + Response resp = new Response(request.getId(), obj.getRef(), result, error); + + Channel channel = obj.getChannel(); + + BasicProperties props = delivery.getProperties(); + + BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()).build(); + + byte[] bytesResponse = serializer.serialize(serializerType, resp); + channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse); + logger.debug("Publish sync response -> Object: " + obj.getRef() + ", method: " + methodName + " corrID: " + requestID + " replyTo: " + + props.getReplyTo()); + } + + } catch (InterruptedException i) { + logger.error(i); + killed = true; + } catch (Exception e) { + logger.error("Object: " + obj.getRef(), e); + } + + } + } + + 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: /tags/0.5.4/src/main/java/omq/server/RemoteObject.java =================================================================== --- /tags/0.5.4/src/main/java/omq/server/RemoteObject.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/server/RemoteObject.java (revision 65) @@ -0,0 +1,268 @@ +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 org.apache.log4j.Logger; + +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 static final String multi = "multi#"; + private static final Logger logger = Logger.getLogger(RemoteObject.class.getName()); + + private String UID; + private String multiQueue; + private Properties env; + private transient Broker broker; + private transient Serializer serializer; + 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, Broker broker) throws Exception { + this.broker = broker; + UID = reference; + multiQueue = UID + System.currentTimeMillis(); + env = broker.getEnvironment(); + serializer = broker.getSerializer(); + + 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, broker.getSerializer()); + + startQueues(); + + // Start this listener + this.start(); + } + + @Override + public void run() { + while (!killed) { + try { + Delivery delivery = consumer.nextDelivery(); + + logger.debug(UID + " has received a message"); + + remoteWrapper.notifyDelivery(delivery); + } catch (InterruptedException i) { + logger.error(i); + } catch (ShutdownSignalException e) { + logger.error(e); + try { + if (channel.isOpen()) { + channel.close(); + } + startQueues(); + } catch (Exception e1) { + try { + long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000")); + Thread.sleep(milis); + } catch (InterruptedException e2) { + logger.error(e2); + } + logger.error(e1); + } + } catch (ConsumerCancelledException e) { + logger.error(e); + } catch (SerializerException e) { + logger.error(e); + } catch (Exception e) { + logger.error(e); + } + } + } + + @Override + public String getRef() { + return UID; + } + + @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 { + logger.warn("Killing objectmq: " + this.getRef()); + 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; + // Multi info + String multiExchange = multi + exchange; + + boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false")); + + // Start channel + channel = broker.getNewChannel(); + + // Declares and bindings + logger.info("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue); + channel.exchangeDeclare(exchange, "direct"); + channel.queueDeclare(queue, durable, false, false, null); + channel.queueBind(queue, exchange, routingKey); + + channel.exchangeDeclare(multiExchange, "fanout"); + channel.queueDeclare(multiQueue, durable, false, false, null); + channel.queueBind(multiQueue, multiExchange, ""); + + // Declare the event topic fanout + logger.info("RemoteObject: " + UID + " declaring fanout exchange: " + UID); + channel.exchangeDeclare(UID, "fanout"); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(queue, true, consumer); + channel.basicConsume(multiQueue, 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: /tags/0.5.4/src/main/java/omq/server/RemoteWrapper.java =================================================================== --- /tags/0.5.4/src/main/java/omq/server/RemoteWrapper.java (revision 65) +++ /tags/0.5.4/src/main/java/omq/server/RemoteWrapper.java (revision 65) @@ -0,0 +1,84 @@ +package omq.server; + +import java.util.ArrayList; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; + +import omq.common.util.Serializer; + +import org.apache.log4j.Logger; + +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; + +/** + * + * @author Sergi Toda + * + */ +public class RemoteWrapper { + private static final Logger logger = Logger.getLogger(RemoteWrapper.class.getName()); + + private RemoteObject obj; + private int numThreads; + private ArrayList invocationList; + private BlockingQueue deliveryQueue; + + public RemoteWrapper(RemoteObject obj, int numThreads, Serializer serializer) { + this.obj = obj; + this.numThreads = numThreads; + invocationList = new ArrayList(); + deliveryQueue = new LinkedBlockingDeque(); + + logger.info("Object reference: " + obj.getRef() + ", numthreads listening = " + numThreads); + + for (int i = 0; i < numThreads; i++) { + InvocationThread thread = new InvocationThread(obj, deliveryQueue, serializer); + invocationList.add(thread); + thread.start(); + } + } + + public void notifyDelivery(Delivery delivery) throws Exception { + this.deliveryQueue.put(delivery); + } + + public void stopRemoteWrapper() { + logger.warn("Stopping Invocation threads vinculed to " + obj.getRef()); + 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: /tags/0.5.4/src/main/resources/log4j.xml =================================================================== --- /tags/0.5.4/src/main/resources/log4j.xml (revision 65) +++ /tags/0.5.4/src/main/resources/log4j.xml (revision 65) @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Index: /tags/0.5.4/src/test/java/omq/test/calculator/Calculator.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/calculator/Calculator.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/calculator/Calculator.java (revision 65) @@ -0,0 +1,28 @@ +package omq.test.calculator; + +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(timeout = 1500) + public int divideByZero(); + +} Index: /tags/0.5.4/src/test/java/omq/test/calculator/CalculatorImpl.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/calculator/CalculatorImpl.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/calculator/CalculatorImpl.java (revision 65) @@ -0,0 +1,61 @@ +package omq.test.calculator; + +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; + private Broker broker; + + public CalculatorImpl() throws Exception { + super(); + } + + public CalculatorImpl(Broker broker) throws Exception { + super(); + this.broker = broker; + } + + 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: /tags/0.5.4/src/test/java/omq/test/calculator/CalculatorTest.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/calculator/CalculatorTest.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/calculator/CalculatorTest.java (revision 65) @@ -0,0 +1,142 @@ +package omq.test.calculator; + +import static org.junit.Assert.assertEquals; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.After; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; + +@RunWith(value = Parameterized.class) +public class CalculatorTest { + + private static Broker broker; + private static Calculator remoteCalc; + private static Calculator remoteCalc2; + + public CalculatorTest(String type) 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, type); + 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 = new Broker(env); + remoteCalc = broker.lookup("calculator1", Calculator.class); + remoteCalc2 = broker.lookup("calculator2", Calculator.class); + } + + @Parameters + public static Collection data() { + Object[][] data = new Object[][] { { Serializer.JAVA }, { Serializer.GSON }, { Serializer.KRYO } }; + return Arrays.asList(data); + } + + @BeforeClass + public static void server() 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.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"); + + CalculatorImpl calc = new CalculatorImpl(); + CalculatorImpl calc2 = new CalculatorImpl(); + + Broker broker = new Broker(env); + broker.bind("calculator1", calc); + broker.bind("calculator2", calc2); + + System.out.println("Server started"); + } + + @After + public void stop() throws Exception { + broker.stopBroker(); + } + + @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: /tags/0.5.4/src/test/java/omq/test/calculator/Message.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/calculator/Message.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/calculator/Message.java (revision 65) @@ -0,0 +1,39 @@ +package omq.test.calculator; + +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: /tags/0.5.4/src/test/java/omq/test/calculator/ZeroEvent.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/calculator/ZeroEvent.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/calculator/ZeroEvent.java (revision 65) @@ -0,0 +1,22 @@ +package omq.test.calculator; + +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: /tags/0.5.4/src/test/java/omq/test/calculator/ZeroListener.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/calculator/ZeroListener.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/calculator/ZeroListener.java (revision 65) @@ -0,0 +1,15 @@ +package omq.test.calculator; + +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: /tags/0.5.4/src/test/java/omq/test/exception/ClientInterface.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/exception/ClientInterface.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/exception/ClientInterface.java (revision 65) @@ -0,0 +1,33 @@ +package omq.test.exception; + +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: /tags/0.5.4/src/test/java/omq/test/exception/ExceptionTest.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/exception/ExceptionTest.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/exception/ExceptionTest.java (revision 65) @@ -0,0 +1,127 @@ +package omq.test.exception; + +import static org.junit.Assert.assertEquals; + +import java.lang.reflect.UndeclaredThrowableException; +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.After; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; + +@RunWith(value = Parameterized.class) +public class ExceptionTest { + private static Broker broker; + private static ClientInterface client; + + public ExceptionTest(String type) 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, type); + 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 = new Broker(env); + client = broker.lookup("server", ClientInterface.class); + } + + @Parameters + public static Collection data() { + Object[][] data = new Object[][] { { Serializer.JAVA }, { Serializer.GSON }, { Serializer.KRYO } }; + return Arrays.asList(data); + } + + @BeforeClass + public static void serverTest() 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.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 broker = new Broker(env); + broker.bind("server", server); + } + + @After + public void stop() throws Exception { + broker.stopBroker(); + } + + @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: /tags/0.5.4/src/test/java/omq/test/exception/OmqServerImpl.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/exception/OmqServerImpl.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/exception/OmqServerImpl.java (revision 65) @@ -0,0 +1,44 @@ +package omq.test.exception; + +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: /tags/0.5.4/src/test/java/omq/test/exception/ServerInterface.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/exception/ServerInterface.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/exception/ServerInterface.java (revision 65) @@ -0,0 +1,17 @@ +package omq.test.exception; + +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: /tags/0.5.4/src/test/java/omq/test/exception/ServerTest.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/exception/ServerTest.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/exception/ServerTest.java (revision 65) @@ -0,0 +1,36 @@ +package omq.test.exception; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +import org.junit.Test; + +public class ServerTest { + + @Test + public void test() 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.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 broker = new Broker(env); + broker.bind("server", server); + + Thread.sleep(60 * 60 * 1000); + } +} +; Index: /tags/0.5.4/src/test/java/omq/test/exception/Trailer.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/exception/Trailer.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/exception/Trailer.java (revision 65) @@ -0,0 +1,34 @@ +package omq.test.exception; + +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: /tags/0.5.4/src/test/java/omq/test/faultTolerance/FaultToleranceTest.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/faultTolerance/FaultToleranceTest.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/faultTolerance/FaultToleranceTest.java (revision 65) @@ -0,0 +1,109 @@ +package omq.test.faultTolerance; + +import static org.junit.Assert.assertEquals; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; +import omq.test.calculator.Calculator; +import omq.test.calculator.CalculatorImpl; + +import org.junit.After; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; + +@RunWith(value = Parameterized.class) +public class FaultToleranceTest { + private static Broker broker; + private static Calculator remoteCalc; + + public FaultToleranceTest(String type) 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, type); + 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 = new Broker(env); + remoteCalc = broker.lookup("calculator1", Calculator.class); + } + + @Parameters + public static Collection data() { + Object[][] data = new Object[][] { { Serializer.JAVA }, { Serializer.GSON }, { Serializer.KRYO } }; + return Arrays.asList(data); + } + + @After + public void stop() throws Exception { + broker.stopBroker(); + } + + @BeforeClass + public static void serverTest() 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.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"); + + CalculatorImpl calc = new CalculatorImpl(); + + broker = new Broker(env); + broker.bind("calculator1", calc); + + System.out.println("Server started"); + } + + @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: /tags/0.5.4/src/test/java/omq/test/multiProcess/MultiProcessTest.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/multiProcess/MultiProcessTest.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/multiProcess/MultiProcessTest.java (revision 65) @@ -0,0 +1,111 @@ +package omq.test.multiProcess; + +import static org.junit.Assert.assertEquals; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.After; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; + +@RunWith(value = Parameterized.class) +public class MultiProcessTest { + public static Broker broker; + public static NumberClient remoteNumber; + + public MultiProcessTest(String type) 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, type); + 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 = new Broker(env); + remoteNumber = broker.lookup("number", NumberClient.class); + } + + @Parameters + public static Collection data() { + Object[][] data = new Object[][] { { Serializer.JAVA }, { Serializer.GSON }, { Serializer.KRYO } }; + return Arrays.asList(data); + } + + @BeforeClass + public static void serverTest() 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.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 broker = new Broker(env); + broker.bind("number", new NumberImpl()); + + Broker broker2 = new Broker(env); + broker2.bind("number", new NumberImpl()); + } + + @After + public void stop() throws Exception { + broker.stopBroker(); + } + + @Test + public void fifoTest() { + int x = 10; + remoteNumber.setNumber(x); + int a = remoteNumber.getNumber(); + assertEquals(0, a); + int b = remoteNumber.getNumber(); + assertEquals(x, b); + remoteNumber.setNumber(0); + remoteNumber.setNumber(0); + } + + @Test + public void multiTest() throws Exception { + int x = 10; + remoteNumber.setMultiNumber(x); + Thread.sleep(200); + int a = remoteNumber.getNumber(); + int b = remoteNumber.getNumber(); + assertEquals(x, a); + assertEquals(x, b); + int[] number = remoteNumber.getMultiNumber(); + assertEquals(x, number[0]); + assertEquals(x, number[1]); + remoteNumber.setMultiNumber(0); + Thread.sleep(200); + } + +} Index: /tags/0.5.4/src/test/java/omq/test/multiProcess/Number.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/multiProcess/Number.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/multiProcess/Number.java (revision 65) @@ -0,0 +1,13 @@ +package omq.test.multiProcess; + +import omq.Remote; + +public interface Number extends Remote { + public void setNumber(int x); + + public int getNumber(); + + public void setMultiNumber(int x); + + public int getMultiNumber(); +} Index: /tags/0.5.4/src/test/java/omq/test/multiProcess/NumberClient.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/multiProcess/NumberClient.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/multiProcess/NumberClient.java (revision 65) @@ -0,0 +1,25 @@ +package omq.test.multiProcess; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.MultiMethod; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; + +@RemoteInterface +public interface NumberClient extends Remote { + @SyncMethod(timeout = 1000) + public void setNumber(int x); + + @SyncMethod(timeout = 1000) + public int getNumber(); + + @MultiMethod + @AsyncMethod + public void setMultiNumber(int x); + + @MultiMethod(waitNum = 2) + @SyncMethod(timeout = 1000) + public int[] getMultiNumber(); + +} Index: /tags/0.5.4/src/test/java/omq/test/multiProcess/NumberImpl.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/multiProcess/NumberImpl.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/multiProcess/NumberImpl.java (revision 65) @@ -0,0 +1,35 @@ +package omq.test.multiProcess; + +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; + } + + public void setNumber(int x) { + this.x = x; + } + + public int getNumber() { + return x; + } + + public void setMultiNumber(int x) { + this.x = x; + } + + public int getMultiNumber() { + return x; + } + +} Index: /tags/0.5.4/src/test/java/omq/test/serializer/CalculatorTest.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/serializer/CalculatorTest.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/serializer/CalculatorTest.java (revision 65) @@ -0,0 +1,80 @@ +package omq.test.serializer; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.test.calculator.Calculator; +import omq.test.calculator.CalculatorImpl; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class CalculatorTest { + + private static Broker broker; + private static Calculator remoteCalc; + + public CalculatorTest() 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.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 = new Broker(env); + remoteCalc = broker.lookup("calculator1", Calculator.class); + } + + @BeforeClass + public static void server() 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.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"); + + CalculatorImpl calc = new CalculatorImpl(); + CalculatorImpl calc2 = new CalculatorImpl(); + + Broker broker = new Broker(env); + broker.bind("calculator1", calc); + broker.bind("calculator2", calc2); + + System.out.println("Server started"); + } + + @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); + } +} Index: /tags/0.5.4/src/test/java/omq/test/stopBroker/BrokerKiller.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/stopBroker/BrokerKiller.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/stopBroker/BrokerKiller.java (revision 65) @@ -0,0 +1,11 @@ +package omq.test.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: /tags/0.5.4/src/test/java/omq/test/stopBroker/BrokerKillerImpl.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/stopBroker/BrokerKillerImpl.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/stopBroker/BrokerKillerImpl.java (revision 65) @@ -0,0 +1,41 @@ +package omq.test.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; + + private Broker broker; + + public BrokerKillerImpl(Broker broker) { + this.broker = broker; + } + + @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: /tags/0.5.4/src/test/java/omq/test/stopBroker/StopBrokerTest.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/stopBroker/StopBrokerTest.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/stopBroker/StopBrokerTest.java (revision 65) @@ -0,0 +1,87 @@ +package omq.test.stopBroker; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.After; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; + +@RunWith(value = Parameterized.class) +public class StopBrokerTest { + + private static Broker broker; + private static BrokerKiller bk; + + public StopBrokerTest(String type) 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, type); + 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 = new Broker(env); + bk = broker.lookup("bk", BrokerKiller.class); + } + + @Parameters + public static Collection data() { + Object[][] data = new Object[][] { { Serializer.JAVA }, { Serializer.GSON }, { Serializer.KRYO } }; + return Arrays.asList(data); + } + + @BeforeClass + public static void server() 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.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 broker = new Broker(env); + BrokerKillerImpl bki = new BrokerKillerImpl(broker); + broker.bind("bk", bki); + } + + @After + public void stop() throws Exception { + broker.stopBroker(); + } + + @Test + public void stopBroker() throws Exception { + bk.killServerBroker(); + } + +} Index: /tags/0.5.4/src/test/java/omq/test/stopBroker/UnbindTest.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/stopBroker/UnbindTest.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/stopBroker/UnbindTest.java (revision 65) @@ -0,0 +1,40 @@ +package omq.test.stopBroker; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.test.calculator.CalculatorImpl; + +import org.junit.Test; + +public class UnbindTest { + + @Test + public void serverTest() 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.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"; + CalculatorImpl calc = new CalculatorImpl(); + + Broker broker = new Broker(env); + broker.bind(reference, calc); + + broker.unbind(reference); + + broker.closeConnection(); + } + +} Index: /tags/0.5.4/src/test/java/omq/test/temporal/ProvaTest.java =================================================================== --- /tags/0.5.4/src/test/java/omq/test/temporal/ProvaTest.java (revision 65) +++ /tags/0.5.4/src/test/java/omq/test/temporal/ProvaTest.java (revision 65) @@ -0,0 +1,105 @@ +package omq.test.temporal; + +import static org.junit.Assert.assertEquals; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; +import omq.test.calculator.Calculator; +import omq.test.calculator.CalculatorImpl; + +import org.junit.After; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; + +@RunWith(value = Parameterized.class) +public class ProvaTest { + + @BeforeClass + public static void serverTest() 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.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"); + + CalculatorImpl calc = new CalculatorImpl(); + + Broker broker = new Broker(env); + broker.bind("calculator1", calc); + + System.out.println("Server started"); + } + + private static Broker broker; + private static Calculator remoteCalc; + + public ProvaTest(String type) 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, type); + 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 = new Broker(env); + remoteCalc = broker.lookup("calculator1", Calculator.class); + } + + @Parameters + public static Collection data() { + Object[][] data = new Object[][] { { Serializer.JAVA } /* + * , { + * Serializer + * .gson }, { + * Serializer + * .kryo } + */}; + return Arrays.asList(data); + } + + @After + public void stop() throws Exception { + broker.stopBroker(); + } + + @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); + } + +} Index: /tags/0.5.4/target/classes/log4j.xml =================================================================== --- /tags/0.5.4/target/classes/log4j.xml (revision 65) +++ /tags/0.5.4/target/classes/log4j.xml (revision 65) @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Index: /tags/0.5.4/target/maven-archiver/pom.properties =================================================================== --- /tags/0.5.4/target/maven-archiver/pom.properties (revision 65) +++ /tags/0.5.4/target/maven-archiver/pom.properties (revision 65) @@ -0,0 +1,5 @@ +#Generated by Maven +#Wed Jun 19 16:52:13 CEST 2013 +version=0.5.1 +groupId=objectmq +artifactId=objectmq