Index: /trunk/src/main/java/omq/client/proxy/MultiProxymq.java =================================================================== --- /trunk/src/main/java/omq/client/proxy/MultiProxymq.java (revision 70) +++ /trunk/src/main/java/omq/client/proxy/MultiProxymq.java (revision 70) @@ -0,0 +1,60 @@ +package omq.client.proxy; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.util.Properties; + +import com.rabbitmq.client.AMQP.BasicProperties; + +import omq.common.broker.Broker; +import omq.common.message.Request; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +/** + * TODO Aquesta classe s'eliminarà tant bon punt es faci un proxymq més + * intel·ligent + * + * @author sergi + * + */ +public class MultiProxymq implements InvocationHandler { + private static final String multi = "multi#"; + + private String uid; + private Broker broker; + private Serializer serializer; + private String replyQueueName; + private String exchange; + private static final String routingkey = ""; + private transient String serializerType; + + public MultiProxymq(String uid, Class clazz, Broker broker) throws Exception { + this.uid = uid; + this.broker = broker; + serializer = broker.getSerializer(); + + Properties env = broker.getEnvironment(); + replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE); + exchange = multi + env.getProperty(ParameterQueue.RPC_EXCHANGE); + serializerType = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA); + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + String methodName = method.getName(); + String corrId = java.util.UUID.randomUUID().toString(); + boolean multi = true; + + Request request = Request.newAsyncRequest(corrId, methodName, args, multi); + + // Add the correlation ID and create a replyTo property + BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).type(serializerType).build(); + + byte[] bytesRequest = serializer.serialize(serializerType, request); + broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest); + + return null; + } + +} Index: /trunk/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- /trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 69) +++ /trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 70) @@ -5,5 +5,4 @@ import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; -import java.lang.reflect.Proxy; import java.util.Collection; import java.util.HashMap; @@ -50,4 +49,7 @@ private String uid; + private transient String exchange; + private transient String multiExchange; + private transient String replyQueueName; private transient String serializerType; private transient Broker broker; @@ -96,4 +98,7 @@ // this.channel = Broker.getChannel(); env = broker.getEnvironment(); + exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE); + multiExchange = multi + exchange; + replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE); // set the serializer type @@ -134,5 +139,5 @@ if (request.isAsync()) { logger.debug("Publish async request -> " + request.getId()); - publishAsyncRequest(request); + publishMessage(request, replyQueueName); } else { logger.debug("Publish sync request -> " + request.getId()); @@ -151,8 +156,8 @@ if (request.isMulti()) { - exchange = multi + env.getProperty(ParameterQueue.RPC_EXCHANGE); + exchange = multiExchange; routingkey = ""; } else { - exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE); + exchange = this.exchange; routingkey = uid; } @@ -166,10 +171,4 @@ } - 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(); @@ -177,7 +176,4 @@ int retries = request.getRetries(); long timeout = request.getTimeout(); - - // Get the environment properties - String replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE); // Publish the message @@ -296,24 +292,4 @@ /** - * 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 * Index: /trunk/src/main/java/omq/common/broker/Broker.java =================================================================== --- /trunk/src/main/java/omq/common/broker/Broker.java (revision 69) +++ /trunk/src/main/java/omq/common/broker/Broker.java (revision 70) @@ -2,4 +2,5 @@ import java.io.IOException; +import java.lang.reflect.Proxy; import java.net.URL; import java.util.HashMap; @@ -10,4 +11,5 @@ import omq.Remote; import omq.client.listener.ResponseListener; +import omq.client.proxy.MultiProxymq; import omq.client.proxy.Proxymq; import omq.common.event.Event; @@ -46,4 +48,5 @@ private Map remoteObjs; private Map proxies = new Hashtable(); + private Map multiProxies = new Hashtable(); public Broker(Properties env) throws Exception { @@ -73,5 +76,5 @@ responseListener.kill(); eventDispatcher.kill(); - //TODO proxies = null; ?? + // TODO proxies = null; ?? } // Stop all the remote objects working @@ -136,5 +139,5 @@ Proxymq proxy = new Proxymq(reference, contract, this); Class[] array = { contract }; - Object newProxy = Proxymq.newProxyInstance(contract.getClassLoader(), array, proxy); + Object newProxy = Proxy.newProxyInstance(contract.getClassLoader(), array, proxy); proxies.put(reference, newProxy); return (T) newProxy; @@ -147,4 +150,21 @@ } + @SuppressWarnings("unchecked") + public synchronized T lookupMulti(String reference, Class contract) throws RemoteException { + try { + if (!multiProxies.containsKey(reference)) { + MultiProxymq proxy = new MultiProxymq(reference, contract, this); + Class[] array = { contract }; + Object newProxy = Proxy.newProxyInstance(contract.getClassLoader(), array, proxy); + multiProxies.put(reference, newProxy); + return (T) newProxy; + } + return (T) multiProxies.get(reference); + + } catch (Exception e) { + throw new RemoteException(e); + } + } + public void bind(String reference, RemoteObject remote) throws RemoteException { try { @@ -155,5 +175,5 @@ } } - + public void startTriggerEvent(String reference, RemoteObject remote) throws RemoteException { try { Index: unk/src/test/java/omq/test/event/EventTest.java =================================================================== --- /trunk/src/test/java/omq/test/event/EventTest.java (revision 69) +++ (revision ) @@ -1,77 +1,0 @@ -package omq.test.event; - -import static org.junit.Assert.assertEquals; - -import java.util.Properties; - -import omq.common.broker.Broker; -import omq.common.util.ParameterQueue; -import omq.common.util.Serializer; - -import org.junit.BeforeClass; -import org.junit.Test; - -public class EventTest { - private static EventTriggerImpl trigger; - - @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"); - - trigger = new EventTriggerImpl(); - Broker broker = new Broker(env); - - broker.startTriggerEvent("trigger", trigger); - - System.out.println("Server started"); - } - - @Test - public void eventTest() throws Exception { - String expected = "This is an event"; - String actual = null; - - Properties env = new Properties(); - env.setProperty(ParameterQueue.USER_NAME, "guest"); - env.setProperty(ParameterQueue.USER_PASS, "guest"); - - // Set host info of rabbimq (where it is) - env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); - env.setProperty(ParameterQueue.SERVER_PORT, "5672"); - env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); - env.setProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA); - env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); - - // Set info about where the message will be sent - env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); - // env.setProperty(ParameterQueue.DEBUGFILE, "c:\\middlewareDebug"); - - // Set info about the queue & the exchange where the ResponseListener - // will listen to. - env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); - env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); - - Broker broker = new Broker(env); - EventTrigger et = broker.lookup("trigger", EventTrigger.class); - - MessageListener ms = new MessageListener(); - et.addListener(ms); - trigger.triggerEvent(expected); - Thread.sleep(500); - actual = ms.getMessage(); - - assertEquals(expected, actual); - } -} Index: unk/src/test/java/omq/test/event/EventTrigger.java =================================================================== --- /trunk/src/test/java/omq/test/event/EventTrigger.java (revision 69) +++ (revision ) @@ -1,14 +1,0 @@ -package omq.test.event; - -import java.io.IOException; - -import omq.Remote; -import omq.client.annotation.AsyncMethod; -import omq.client.annotation.RemoteInterface; -import omq.exception.SerializerException; - -@RemoteInterface -public interface EventTrigger extends Remote { - @AsyncMethod - public void triggerEvent(String message) throws IOException, SerializerException; -} Index: unk/src/test/java/omq/test/event/EventTriggerImpl.java =================================================================== --- /trunk/src/test/java/omq/test/event/EventTriggerImpl.java (revision 69) +++ (revision ) @@ -1,20 +1,0 @@ -package omq.test.event; - -import java.io.IOException; - -import omq.exception.SerializerException; -import omq.server.RemoteObject; - -public class EventTriggerImpl extends RemoteObject implements EventTrigger { - - /** - * - */ - private static final long serialVersionUID = 1L; - - public void triggerEvent(String message) throws IOException, SerializerException { - MessageEvent event = new MessageEvent(message); - this.notifyEvent(event); - } - -} Index: /trunk/src/test/java/omq/test/event/Message.java =================================================================== --- /trunk/src/test/java/omq/test/event/Message.java (revision 70) +++ /trunk/src/test/java/omq/test/event/Message.java (revision 70) @@ -0,0 +1,7 @@ +package omq.test.event; + +import omq.Remote; + +public interface Message extends Remote { + public void setMessage(String message); +} Index: unk/src/test/java/omq/test/event/MessageEvent.java =================================================================== --- /trunk/src/test/java/omq/test/event/MessageEvent.java (revision 69) +++ (revision ) @@ -1,28 +1,0 @@ -package omq.test.event; - -import omq.common.event.Event; - -public class MessageEvent extends Event { - /** - * - */ - private static final long serialVersionUID = 1L; - - private String message; - - public MessageEvent() { - } - - public MessageEvent(String message) { - this.message = message; - } - - public String getMessage() { - return message; - } - - public void setMessage(String message) { - this.message = message; - } - -} Index: /trunk/src/test/java/omq/test/event/MessageImpl.java =================================================================== --- /trunk/src/test/java/omq/test/event/MessageImpl.java (revision 70) +++ /trunk/src/test/java/omq/test/event/MessageImpl.java (revision 70) @@ -0,0 +1,21 @@ +package omq.test.event; + +import omq.server.RemoteObject; + +public class MessageImpl extends RemoteObject implements Message { + + /** + * + */ + private static final long serialVersionUID = 1L; + private String message; + + @Override + public void setMessage(String message) { + this.message = message; + } + + public String getMessage() { + return message; + } +} Index: unk/src/test/java/omq/test/event/MessageListener.java =================================================================== --- /trunk/src/test/java/omq/test/event/MessageListener.java (revision 69) +++ (revision ) @@ -1,20 +1,0 @@ -package omq.test.event; - -import omq.common.event.EventListener; - -public class MessageListener extends EventListener { - private String message; - - @Override - public void notifyEvent(MessageEvent event) { - message = event.getMessage(); - } - - public String getMessage() { - return message; - } - - public void setMessage(String message) { - this.message = message; - } -} Index: /trunk/src/test/java/omq/test/event/MessageTest.java =================================================================== --- /trunk/src/test/java/omq/test/event/MessageTest.java (revision 70) +++ /trunk/src/test/java/omq/test/event/MessageTest.java (revision 70) @@ -0,0 +1,101 @@ +package omq.test.event; + +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 MessageTest { + private static final String NAME = "message"; + + private static Broker broker; + private static Message serverProxy; + private static MessageImpl clientM1; + private static MessageImpl clientM2; + + public MessageTest(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); + serverProxy = broker.lookupMulti(NAME, Message.class); + } + + @Parameters + public static Collection data() { + Object[][] data = new Object[][] { { Serializer.JAVA }, { Serializer.GSON }, { Serializer.KRYO } }; + return Arrays.asList(data); + } + + @BeforeClass + public static void client() 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); + clientM1 = new MessageImpl(); + broker.bind(NAME, clientM1); + + Broker broker2 = new Broker(env); + clientM2 = new MessageImpl(); + broker2.bind(NAME, clientM2); + } + + @After + public void stop() throws Exception { + broker.stopBroker(); + } + + @Test + public void test() throws Exception { + String expected = "Hello"; + + serverProxy.setMessage(expected); + Thread.sleep(200); + assertEquals(expected, clientM1.getMessage()); + assertEquals(expected, clientM2.getMessage()); + serverProxy.setMessage(""); + Thread.sleep(200); + } + +} Index: unk/target/classes/log4j.xml =================================================================== --- /trunk/target/classes/log4j.xml (revision 69) +++ (revision ) @@ -1,37 +1,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -