Index: trunk/objectmq/.classpath =================================================================== --- trunk/objectmq/.classpath (revision 31) +++ trunk/objectmq/.classpath (revision 34) @@ -9,5 +9,5 @@ - + Index: trunk/objectmq/src/omq/client/listener/ResponseListener.java =================================================================== --- trunk/objectmq/src/omq/client/listener/ResponseListener.java (revision 34) +++ trunk/objectmq/src/omq/client/listener/ResponseListener.java (revision 34) @@ -0,0 +1,192 @@ +package omq.client.listener; + +import java.io.IOException; +import java.util.Hashtable; +import java.util.Map; +import java.util.Properties; + +import omq.client.proxy.Proxymq; +import omq.common.util.OmqConnectionFactory; +import omq.common.util.ParameterQueue; + +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConsumerCancelledException; +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; +import com.rabbitmq.client.ShutdownSignalException; + +/** + * Class that inherits from RemoteListener. It's used in the server side. This + * class gets the deliveries from the server and stores them into the proxies + * + * @author Sergi Toda + * + */ +public class ResponseListener extends Thread { + private static ResponseListener rListener; + + private Connection connection; + private Channel channel; + private QueueingConsumer consumer; + private boolean killed = false; + private Map> results; + + /** + * Protected constructor used by the singleton pattern + * + * @param env + * @throws Exception + */ + protected ResponseListener(Properties env) throws Exception { + connection = OmqConnectionFactory.getNewConnection(env); + channel = connection.createChannel(); + + // Init the hashtable (it's concurrent) + this.results = new Hashtable>(); + + String reply_queue = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE); + channel.queueDeclare(reply_queue, false, false, false, null); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(reply_queue, true, consumer); + } + + @Override + public void run() { + Delivery delivery; + String uid_request; + + while (!killed) { + try { + // Get the delivery + + delivery = consumer.nextDelivery(); + + BasicProperties props = delivery.getProperties(); + + // Get the response with its uid + uid_request = delivery.getProperties().getCorrelationId(); + System.out.println("Response received -> " + uid_request); + + // Stores the new response + Map proxyResults = results + .get(props.getAppId()); + + // Put the result into the proxy results and notify him + synchronized (proxyResults) { + // If we haven't received this response before, we store it + if (!proxyResults.containsKey(uid_request)) { + proxyResults.put(uid_request, delivery.getBody()); + proxyResults.notifyAll(); + } + } + } catch (InterruptedException i) { + i.printStackTrace(); + } catch (ShutdownSignalException e) { + e.printStackTrace(); + } catch (ConsumerCancelledException e) { + e.printStackTrace(); + } catch (Exception e) { + e.printStackTrace(); + } + } + } + + /** + * Static function which initializes the ResponseListener + * + * @param env + * @throws Exception + */ + public static void init(Properties env) throws Exception { + if (rListener == null) { + rListener = new ResponseListener(env); + rListener.start(); + } else { + throw new Exception("Cannot init because it already exists"); + } + } + + /** + * Method to retrieve the unique ResponseListener, this function can also + * initialize a ResponseListener using and environment + * + * @param env + * @return unique ResponseListener + * @throws Exception + */ + public static ResponseListener getRequestListener(Properties env) + throws Exception { + if (rListener == null) { + rListener = new ResponseListener(env); + rListener.start(); + } else { + // TODO: create a new exception to indicate that a response listener + // cannot be init + throw new Exception("Cannot init because it already exists"); + } + return rListener; + } + + public static boolean isVoid() { + return rListener == null; + } + + /** + * Method to retrieve the unique ResponseListener + * + * @return + * @throws Exception + */ + public static ResponseListener getRequestListener() throws Exception { + if (rListener == null) { + throw new Exception("Request listener not initialized"); + } + return rListener; + } + + public synchronized Channel getChannel() throws Exception { + return connection.createChannel(); + } + + /** + * + * @param key + * @return whether the map has the param key + */ + public boolean containsKey(String key) { + return results.containsKey(key); + } + + /** + * This method is used to kill the unique responseListener in the system + * + * @throws Exception + */ + public static void stopResponseListner() throws Exception { + rListener.kill(); + rListener = null; + } + + /** + * Interrupt and kill the Thread + * + * @throws IOException + */ + public void kill() throws IOException { + interrupt(); + killed = true; + channel.close(); + connection.close(); + } + + // Revisar això + public void registerProxy(Proxymq proxy) { + if (!results.containsKey(proxy.getRef())) { + results.put(proxy.getRef(), proxy.getResults()); + } + } +} Index: trunk/objectmq/src/omq/client/proxy/Proxymq.java =================================================================== --- trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 31) +++ trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 34) @@ -6,5 +6,4 @@ import java.lang.reflect.Proxy; import java.util.Collection; -import java.util.Date; import java.util.HashMap; import java.util.Hashtable; @@ -15,5 +14,5 @@ import omq.client.annotation.AsyncMethod; import omq.client.annotation.SyncMethod; -import omq.client.remote.response.ResponseListener; +import omq.client.listener.ResponseListener; import omq.common.event.Event; import omq.common.event.EventDispatcher; Index: trunk/objectmq/src/omq/common/broker/Broker.java =================================================================== --- trunk/objectmq/src/omq/common/broker/Broker.java (revision 31) +++ trunk/objectmq/src/omq/common/broker/Broker.java (revision 34) @@ -5,6 +5,6 @@ import omq.Remote; +import omq.client.listener.ResponseListener; import omq.client.proxy.Proxymq; -import omq.client.remote.response.ResponseListener; import omq.common.event.Event; import omq.common.event.EventDispatcher; @@ -18,5 +18,5 @@ import omq.exception.RemoteException; import omq.exception.SerializerException; -import omq.server.remote.request.RemoteObject; +import omq.server.RemoteObject; import com.rabbitmq.client.Channel; Index: trunk/objectmq/src/omq/common/util/ParameterQueue.java =================================================================== --- trunk/objectmq/src/omq/common/util/ParameterQueue.java (revision 31) +++ trunk/objectmq/src/omq/common/util/ParameterQueue.java (revision 34) @@ -12,23 +12,21 @@ */ - public static String SERIALIZERNAME = "revo.serializer"; - public static String ENABLECOMPRESSION = "revo.compression"; + public static String SERIALIZERNAME = "omq.serializer"; + public static String ENABLECOMPRESSION = "omq.compression"; - public static String SERVER_HOST = "revo.host"; - public static String SERVER_PORT = "revo.port"; - public static String SERVER_REGISTRY = "revo.registry"; + public static String SERVER_HOST = "omq.host"; + public static String SERVER_PORT = "omq.port"; + public static String SERVER_REGISTRY = "omq.registry"; - public static String USER_NAME = "revo.username"; - public static String USER_PASS = "revo.pass"; + public static String USER_NAME = "omq.username"; + public static String USER_PASS = "omq.pass"; - public static String RPC_EXCHANGE = "revo.rpc_exchange"; - public static String RPC_QUEUE = "revo.rpc_queue"; - public static String RPC_ROUTING_KEY = "revo.routing_key"; + public static String RPC_EXCHANGE = "omq.rpc_exchange"; - public static String RPC_REPLY_QUEUE = "revo.reply_queue_rpc"; - public static String EVENT_REPLY_QUEUE = "revo.reply_queue_event"; + public static String RPC_REPLY_QUEUE = "omq.reply_queue_rpc"; + public static String EVENT_REPLY_QUEUE = "omq.reply_queue_event"; - public static String ENABLE_SSL = "revo.enable_ssl"; - public static String DEBUGFILE = "revo.debug_file"; + public static String ENABLE_SSL = "omq.enable_ssl"; + public static String DEBUGFILE = "omq.debug_file"; public static String RETRY_TIME_CONNECTION = "omq.retry_connection"; Index: trunk/objectmq/src/omq/common/util/Serializer.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializer.java (revision 31) +++ trunk/objectmq/src/omq/common/util/Serializer.java (revision 34) @@ -10,5 +10,5 @@ import omq.exception.EnvironmentException; import omq.exception.SerializerException; -import omq.server.remote.request.RemoteObject; +import omq.server.RemoteObject; /** Index: trunk/objectmq/src/omq/common/util/Serializers/GsonImp.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializers/GsonImp.java (revision 31) +++ trunk/objectmq/src/omq/common/util/Serializers/GsonImp.java (revision 34) @@ -7,5 +7,5 @@ import omq.common.message.Response; import omq.exception.SerializerException; -import omq.server.remote.request.RemoteObject; +import omq.server.RemoteObject; import com.google.gson.Gson; Index: trunk/objectmq/src/omq/common/util/Serializers/ISerializer.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializers/ISerializer.java (revision 31) +++ trunk/objectmq/src/omq/common/util/Serializers/ISerializer.java (revision 34) @@ -5,5 +5,5 @@ import omq.common.message.Response; import omq.exception.SerializerException; -import omq.server.remote.request.RemoteObject; +import omq.server.RemoteObject; /** Index: trunk/objectmq/src/omq/common/util/Serializers/JavaImp.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializers/JavaImp.java (revision 31) +++ trunk/objectmq/src/omq/common/util/Serializers/JavaImp.java (revision 34) @@ -11,5 +11,5 @@ import omq.common.message.Response; import omq.exception.SerializerException; -import omq.server.remote.request.RemoteObject; +import omq.server.RemoteObject; /** Index: trunk/objectmq/src/omq/common/util/Serializers/KryoImp.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializers/KryoImp.java (revision 31) +++ trunk/objectmq/src/omq/common/util/Serializers/KryoImp.java (revision 34) @@ -12,5 +12,5 @@ import omq.common.message.Response; import omq.exception.SerializerException; -import omq.server.remote.request.RemoteObject; +import omq.server.RemoteObject; /** Index: trunk/objectmq/src/omq/server/InvocationThread.java =================================================================== --- trunk/objectmq/src/omq/server/InvocationThread.java (revision 34) +++ trunk/objectmq/src/omq/server/InvocationThread.java (revision 34) @@ -0,0 +1,100 @@ +package omq.server; + +import java.util.concurrent.BlockingQueue; + +import omq.common.message.Request; +import omq.common.message.Response; +import omq.common.util.Serializer; + +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.QueueingConsumer.Delivery; + +/** + * + * @author Sergi Toda + * + */ +public class InvocationThread extends Thread { + private RemoteObject obj; + private BlockingQueue deliveryQueue; + private boolean killed = false; + + public InvocationThread(RemoteObject obj, BlockingQueue deliveryQueue) { + this.obj = obj; + this.deliveryQueue = deliveryQueue; + } + + @Override + public void run() { + while (!killed) { + try { + // Get the delivery + Delivery delivery = deliveryQueue.take(); + + // Deserialize the json + Request request = Serializer.deserializeRequest(delivery.getBody(), obj); + //Log.saveLog("Server-Deserialize", delivery.getBody()); + + String methodName = request.getMethod(); + String requestID = request.getId(); + + System.out.println("Invoke method: " + methodName + " CorrID: " + requestID); + + // Changed --------------------------------------- + Object result = null; + if ("commit".equalsIgnoreCase(methodName)) { + Object[] arguments = request.getArguments(); + arguments[1] = ((String) arguments[1]) + "@@" + requestID; + result = obj.invokeMethod(methodName, arguments); + } else { + result = obj.invokeMethod(request.getMethod(), request.getArguments()); + } + // ----------------------------------------------- + + // // Invoke the method + // Object result = obj.invokeMethod(request.getMethod(), + // request.getArguments()); + + if (!request.isAsync()) { + Response resp = new Response(request.getId(), obj.getRef(), result); + + Channel channel = obj.getChannel(); + + BasicProperties props = delivery.getProperties(); + + BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()).build(); + + byte[] bytesResponse = Serializer.serialize(resp); + channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse); + + //Log.saveLog("Server-Serialize", bytesResponse); + } + + } catch (InterruptedException i) { + i.printStackTrace(); + killed = true; + } catch (Exception e) { + System.out.println("Error a l'Invocation Thread \nException: " + e); + e.printStackTrace(); + } + + } + } + + public RemoteObject getObj() { + return obj; + } + + public void setObj(RemoteObject obj) { + this.obj = obj; + } + + public BlockingQueue getDeliveryQueue() { + return deliveryQueue; + } + + public void setDeliveryQueue(BlockingQueue deliveryQueue) { + this.deliveryQueue = deliveryQueue; + } +} Index: trunk/objectmq/src/omq/server/RemoteObject.java =================================================================== --- trunk/objectmq/src/omq/server/RemoteObject.java (revision 34) +++ trunk/objectmq/src/omq/server/RemoteObject.java (revision 34) @@ -0,0 +1,245 @@ +package omq.server; + +import java.io.IOException; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import omq.Remote; +import omq.common.broker.Broker; +import omq.common.event.Event; +import omq.common.event.EventListener; +import omq.common.event.EventWrapper; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; +import omq.exception.SerializerException; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.ConsumerCancelledException; +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; +import com.rabbitmq.client.ShutdownSignalException; + +/** + * + * @author Sergi Toda + * + */ +public abstract class RemoteObject extends Thread implements Remote { + + private static final long serialVersionUID = -1778953938739846450L; + + private String UID; + private Properties env; + private transient RemoteWrapper remoteWrapper; + private transient Map>> params; + private transient Channel channel; + private transient QueueingConsumer consumer; + private transient boolean killed = false; + + private static final Map> primitiveClasses = new HashMap>(); + + static { + primitiveClasses.put("byte", Byte.class); + primitiveClasses.put("short", Short.class); + primitiveClasses.put("char", Character.class); + primitiveClasses.put("int", Integer.class); + primitiveClasses.put("long", Long.class); + primitiveClasses.put("float", Float.class); + primitiveClasses.put("double", Double.class); + } + + public RemoteObject() { + } + + public void startRemoteObject(String reference, Properties env) throws Exception { + this.UID = reference; + this.env = env; + + params = new HashMap>>(); + for (Method m : this.getClass().getMethods()) { + List> list = new ArrayList>(); + for (Class clazz : m.getParameterTypes()) { + list.add(clazz); + } + params.put(m.getName(), list); + } + + // Get num threads to use + int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1")); + remoteWrapper = new RemoteWrapper(this, numThreads); + + startQueues(); + + // Start this listener + this.start(); + } + + @Override + public void run() { + while (!killed) { + try { + Delivery delivery = consumer.nextDelivery(); + System.out.println("RemoteObject: " + UID + " has received a message"); + remoteWrapper.notifyDelivery(delivery); + } catch (InterruptedException i) { + i.printStackTrace(); + } catch (ShutdownSignalException e) { + e.printStackTrace(); + try { + if (channel.isOpen()) { + channel.close(); + } + startQueues(); + } catch (Exception e1) { + try { + long milis = Long.parseLong(env.getProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000")); + Thread.sleep(milis); + } catch (InterruptedException e2) { + // TODO Auto-generated catch block + e2.printStackTrace(); + } + e1.printStackTrace(); + } + } catch (ConsumerCancelledException e) { + e.printStackTrace(); + } catch (SerializerException e) { + e.printStackTrace(); + } catch (Exception e) { + e.printStackTrace(); + } + } + } + + @Override + public String getRef() { + return UID; + } + + @Override + public void notifyEvent(Event event) throws IOException, SerializerException { + event.setTopic(UID); + EventWrapper wrapper = new EventWrapper(event); + channel.exchangeDeclare(UID, "fanout"); + channel.basicPublish(UID, "", null, Serializer.serialize(wrapper)); + } + + public void kill() throws IOException { + interrupt(); + killed = true; + 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 { + 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; + + // Start channel + channel = Broker.getNewChannel(); + + // Declares and bindings + System.out.println("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue); + channel.exchangeDeclare(exchange, "direct"); + channel.queueDeclare(queue, false, false, false, null); + channel.queueBind(queue, exchange, routingKey); + + // Declare the event topic fanout + System.out.println("RemoteObject: " + UID + " declaring fanout exchange: " + UID); + channel.exchangeDeclare(UID, "fanout"); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(queue, true, consumer); + } + + @Override + public void addListener(EventListener eventListener) throws Exception { + } + + @Override + public void removeListener(EventListener eventListener) throws Exception { + } + + @Override + public Collection getListeners() throws Exception { + return null; + } + +} Index: trunk/objectmq/src/omq/server/RemoteWrapper.java =================================================================== --- trunk/objectmq/src/omq/server/RemoteWrapper.java (revision 34) +++ trunk/objectmq/src/omq/server/RemoteWrapper.java (revision 34) @@ -0,0 +1,77 @@ +package omq.server; + +import java.util.ArrayList; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; + +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; + +/** + * + * @author Sergi Toda + * + */ +public class RemoteWrapper { + private RemoteObject obj; + private int numThreads; + private ArrayList invocationList; + private BlockingQueue deliveryQueue; + + public RemoteWrapper(RemoteObject obj, int numThreads) { + this.obj = obj; + this.numThreads = numThreads; + invocationList = new ArrayList(); + deliveryQueue = new LinkedBlockingDeque(); + + System.out.println("RemoteWrapper -> Object: " + obj.getRef() + ", numthreads listening = " + numThreads); + + for (int i = 0; i < numThreads; i++) { + InvocationThread thread = new InvocationThread(obj, deliveryQueue); + invocationList.add(thread); + thread.start(); + } + } + + public void notifyDelivery(Delivery delivery) throws Exception { + this.deliveryQueue.put(delivery); + } + + public void stopRemoteWrapper() { + for (InvocationThread thread : invocationList) { + thread.interrupt(); + } + } + + public RemoteObject getObj() { + return obj; + } + + public void setObj(RemoteObject obj) { + this.obj = obj; + } + + public int getNumThreads() { + return numThreads; + } + + public void setNumThreads(int numThreads) { + this.numThreads = numThreads; + } + + public ArrayList getInvocationList() { + return invocationList; + } + + public void setInvocationList(ArrayList invocationList) { + this.invocationList = invocationList; + } + + public BlockingQueue getDeliveryQueue() { + return deliveryQueue; + } + + public void setDeliveryQueue(BlockingQueue deliveryQueue) { + this.deliveryQueue = deliveryQueue; + } +} Index: trunk/objectmq/test/calculatorTest/Calculator.java =================================================================== --- trunk/objectmq/test/calculatorTest/Calculator.java (revision 34) +++ trunk/objectmq/test/calculatorTest/Calculator.java (revision 34) @@ -0,0 +1,25 @@ +package calculatorTest; + +import java.io.IOException; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; +import omq.exception.SerializerException; + +@RemoteInterface +public interface Calculator extends Remote { + @SyncMethod(timeout = 1500) + public int add(int x, int y); + + @AsyncMethod + public void mult(int x, int y); + + @AsyncMethod + public void sendMessage(Message m); + + @AsyncMethod + public void divideByZero() throws IOException, SerializerException; + +} Index: trunk/objectmq/test/calculatorTest/CalculatorImpl.java =================================================================== --- trunk/objectmq/test/calculatorTest/CalculatorImpl.java (revision 34) +++ trunk/objectmq/test/calculatorTest/CalculatorImpl.java (revision 34) @@ -0,0 +1,50 @@ +package calculatorTest; + +import java.io.IOException; + +import omq.client.annotation.AsyncMethod; +import omq.common.broker.Broker; +import omq.exception.SerializerException; +import omq.server.RemoteObject; + +public class CalculatorImpl extends RemoteObject implements Calculator { + private int mult = 0; + + public CalculatorImpl() throws Exception { + super(); + } + + private static final long serialVersionUID = 1L; + + public int add(int x, int y) { + return x + y; + } + + public void mult(int x, int y) { + mult = x * y; + } + + public int getMult() { + return mult; + } + + public void setMult(int mult) { + this.mult = mult; + } + + public void divideByZero() throws IOException, SerializerException { + ZeroEvent ze = new ZeroEvent("my zero event", "zero-event"); + Broker.trigger(ze); + //notifyEvent(ze); + } + + @Override + @AsyncMethod + public void sendMessage(Message m) { + System.out.println("Code = "+m.getCode()); + System.out.println("Message = "+m.getMessage()); + } + + + +} Index: trunk/objectmq/test/calculatorTest/ClientTest.java =================================================================== --- trunk/objectmq/test/calculatorTest/ClientTest.java (revision 34) +++ trunk/objectmq/test/calculatorTest/ClientTest.java (revision 34) @@ -0,0 +1,92 @@ +package calculatorTest; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class ClientTest { + private static Calculator remoteCalc; + private static Calculator remoteCalc2; + + @BeforeClass + public static void startClient() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + // env.setProperty(ParameterQueue.SERIALIZERNAME, + // "omq.common.util.Serializers.KryoImp"); + env.setProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.GsonImp"); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + // env.setProperty(ParameterQueue.DEBUGFILE, "c:\\middlewareDebug"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + + Broker.initBroker(env); + remoteCalc = (Calculator) Broker.lookup("calculator1", Calculator.class); + remoteCalc2 = (Calculator) Broker.lookup("calculator2", Calculator.class); + } + + @Test + public void add() throws Exception { + int x = 10; + int y = 20; + + int sync = remoteCalc.add(x, y); + int sum = x + y; + + assertEquals(sum, sync); + } + + @Test + public void add2() throws Exception { + int x = 10; + int y = 20; + + int sync = remoteCalc2.add(x, y); + int sum = x + y; + + assertEquals(sum, sync); + } + + @Test + public void mult() throws Exception { + int x = 5; + int y = 15; + + remoteCalc.mult(x, y); + Thread.sleep(200); + } + + @Test + public void notifyEvent() throws Exception { + ZeroListener zL = new ZeroListener("zero-event"); + + remoteCalc.addListener(zL); + + remoteCalc.divideByZero(); + + Thread.sleep(200); + } + + @Test + public void sendMessage() throws Exception { + Message m = new Message(2334, "Hello objectmq"); + remoteCalc.sendMessage(m); + } +} Index: trunk/objectmq/test/calculatorTest/Message.java =================================================================== --- trunk/objectmq/test/calculatorTest/Message.java (revision 34) +++ trunk/objectmq/test/calculatorTest/Message.java (revision 34) @@ -0,0 +1,39 @@ +package calculatorTest; + +import java.io.Serializable; + +public class Message implements Serializable { + + + + /** + * + */ + private static final long serialVersionUID = 1L; + private int code; + private String message; + + public Message() { + } + + public Message(int code, String message) { + this.code = code; + this.message = message; + } + + public int getCode() { + return code; + } + + public void setCode(int code) { + this.code = code; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } +} Index: trunk/objectmq/test/calculatorTest/ServerTest.java =================================================================== --- trunk/objectmq/test/calculatorTest/ServerTest.java (revision 34) +++ trunk/objectmq/test/calculatorTest/ServerTest.java (revision 34) @@ -0,0 +1,38 @@ +package calculatorTest; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +public class ServerTest { + private static CalculatorImpl calc; + private static CalculatorImpl calc2; + + public static void main(String[] args) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + // env.setProperty(ParameterQueue.SERIALIZERNAME, + // "omq.common.util.Serializers.KryoImp"); + env.setProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.GsonImp"); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + calc = new CalculatorImpl(); + calc2 = new CalculatorImpl(); + + Broker.initBroker(env); + Broker.bind("calculator1", calc); + Broker.bind("calculator2", calc2); + + System.out.println("Server started"); + } +} Index: trunk/objectmq/test/calculatorTest/ZeroEvent.java =================================================================== --- trunk/objectmq/test/calculatorTest/ZeroEvent.java (revision 34) +++ trunk/objectmq/test/calculatorTest/ZeroEvent.java (revision 34) @@ -0,0 +1,22 @@ +package calculatorTest; + +import omq.common.event.Event; + +public class ZeroEvent extends Event { + /** + * + */ + private static final long serialVersionUID = 1L; + + public ZeroEvent() { + } + + public ZeroEvent(String corrId, String topic) { + super(corrId, topic); + } + + public String getZeroMessage() { + return "divition by 0"; + } + +} Index: trunk/objectmq/test/calculatorTest/ZeroListener.java =================================================================== --- trunk/objectmq/test/calculatorTest/ZeroListener.java (revision 34) +++ trunk/objectmq/test/calculatorTest/ZeroListener.java (revision 34) @@ -0,0 +1,15 @@ +package calculatorTest; + +import omq.common.event.EventListener; + +public class ZeroListener extends EventListener { + + public ZeroListener(String topic) { + super(topic); + } + + @Override + public void notifyEvent(ZeroEvent event) { + System.out.println(event.getZeroMessage()); + } +} Index: trunk/objectmq/test/farmTest/Animal.java =================================================================== --- trunk/objectmq/test/farmTest/Animal.java (revision 34) +++ trunk/objectmq/test/farmTest/Animal.java (revision 34) @@ -0,0 +1,9 @@ +package farmTest; + +import java.io.Serializable; + +public interface Animal extends Serializable { + public String getName(); + + public void setName(String name); +} Index: trunk/objectmq/test/farmTest/Cow.java =================================================================== --- trunk/objectmq/test/farmTest/Cow.java (revision 34) +++ trunk/objectmq/test/farmTest/Cow.java (revision 34) @@ -0,0 +1,29 @@ +package farmTest; + +public class Cow implements Animal { + + + /** + * + */ + private static final long serialVersionUID = 1L; + private String name; + + public Cow() { + } + + public Cow(String name) { + this.name = name; + } + + @Override + public String getName() { + return name; + } + + @Override + public void setName(String name) { + this.name = name; + } + +} Index: trunk/objectmq/test/farmTest/Farm.java =================================================================== --- trunk/objectmq/test/farmTest/Farm.java (revision 34) +++ trunk/objectmq/test/farmTest/Farm.java (revision 34) @@ -0,0 +1,23 @@ +package farmTest; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; + +@RemoteInterface +public interface Farm extends Remote { + + @AsyncMethod + public void setPig(Pig pig); + + @AsyncMethod + public void setCow(Cow cow); + + @SyncMethod(timeout = 1500) + public Cow getCow(); + + @SyncMethod(timeout = 1500) + public Pig getPig(); + +} Index: trunk/objectmq/test/farmTest/FarmImpl.java =================================================================== --- trunk/objectmq/test/farmTest/FarmImpl.java (revision 34) +++ trunk/objectmq/test/farmTest/FarmImpl.java (revision 34) @@ -0,0 +1,30 @@ +package farmTest; + +import omq.server.RemoteObject; + +public class FarmImpl extends RemoteObject implements Farm { + /** + * + */ + private static final long serialVersionUID = 1L; + + private Cow cow; + private Pig pig; + + public Cow getCow() { + return cow; + } + + public void setCow(Cow cow) { + this.cow = cow; + } + + public Pig getPig() { + return pig; + } + + public void setPig(Pig pig) { + this.pig = pig; + } + +} Index: trunk/objectmq/test/farmTest/FarmTest.java =================================================================== --- trunk/objectmq/test/farmTest/FarmTest.java (revision 34) +++ trunk/objectmq/test/farmTest/FarmTest.java (revision 34) @@ -0,0 +1,87 @@ +package farmTest; + +import static org.junit.Assert.assertEquals; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class FarmTest { + private static FarmImpl farm; + private static Farm remoteFarm; + + @BeforeClass + public static void startServer() 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"); + + // Get info about the queue & the exchange where the RemoteListener will + // listen to. + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + + farm = new FarmImpl(); + + Broker.initBroker(env); + Broker.bind(Farm.class.getSimpleName(), farm); + } + + @BeforeClass + public static void startClient() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + + Broker.initBroker(env); + + remoteFarm = (Farm) Broker.lookup(Farm.class.getSimpleName(), Farm.class); + } + + @Test + public void addCow() throws Exception { + String name = "muu"; + + Cow cow = new Cow(); + cow.setName(name); + + remoteFarm.setCow(cow); + Thread.sleep(100); + + assertEquals(name, farm.getCow().getName()); + } + + @Test + public void getPig() throws Exception { + String name = "oing"; + + Pig pig = new Pig(); + pig.setName(name); + + remoteFarm.setPig(pig); + Thread.sleep(100); + + pig = remoteFarm.getPig(); + + assertEquals(name, pig.getName()); + } +} Index: trunk/objectmq/test/farmTest/Pig.java =================================================================== --- trunk/objectmq/test/farmTest/Pig.java (revision 34) +++ trunk/objectmq/test/farmTest/Pig.java (revision 34) @@ -0,0 +1,27 @@ +package farmTest; + +public class Pig implements Animal { + + + /** + * + */ + private static final long serialVersionUID = 1L; + private String name; + + public Pig() { + } + + public Pig(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + +} Index: trunk/objectmq/test/test/ClientImpl.java =================================================================== --- trunk/objectmq/test/test/ClientImpl.java (revision 31) +++ trunk/objectmq/test/test/ClientImpl.java (revision 34) @@ -9,5 +9,5 @@ import omq.common.broker.Broker; import omq.exception.RemoteException; -import omq.server.remote.request.RemoteObject; +import omq.server.RemoteObject; public class ClientImpl extends RemoteObject implements Client { Index: trunk/objectmq/test/test2/CarImpl.java =================================================================== --- trunk/objectmq/test/test2/CarImpl.java (revision 31) +++ trunk/objectmq/test/test2/CarImpl.java (revision 34) @@ -5,5 +5,5 @@ import omq.common.broker.Broker; import omq.exception.RemoteException; -import omq.server.remote.request.RemoteObject; +import omq.server.RemoteObject; public class CarImpl extends RemoteObject implements Car { Index: trunk/objectmq/test/test2/MobileImpl.java =================================================================== --- trunk/objectmq/test/test2/MobileImpl.java (revision 31) +++ trunk/objectmq/test/test2/MobileImpl.java (revision 34) @@ -4,5 +4,5 @@ import java.util.List; -import omq.server.remote.request.RemoteObject; +import omq.server.RemoteObject; public class MobileImpl extends RemoteObject implements Mobile {