Index: trunk/src/main/java/omq/Remote.java =================================================================== --- trunk/src/main/java/omq/Remote.java (revision 71) +++ trunk/src/main/java/omq/Remote.java (revision 72) @@ -1,11 +1,5 @@ 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; /** @@ -22,11 +16,3 @@ */ public String getRef(); - - public void notifyEvent(Event event) throws IOException, SerializerException; - - public void addListener(EventListener eventListener) throws Exception; - - public void removeListener(EventListener eventListener) throws Exception; - - public Collection> getListeners() throws Exception; } Index: trunk/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 71) +++ trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 72) @@ -1,9 +1,7 @@ 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.util.Collection; import java.util.HashMap; import java.util.Map; @@ -16,7 +14,4 @@ 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; @@ -25,5 +20,4 @@ import omq.exception.OmqException; import omq.exception.RetryException; -import omq.exception.SerializerException; import omq.exception.TimeoutException; @@ -55,9 +49,7 @@ 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>(); @@ -92,5 +84,4 @@ this.broker = broker; rListener = broker.getResponseListener(); - dispatcher = broker.getEventDispatcher(); serializer = broker.getSerializer(); @@ -105,6 +96,4 @@ serializerType = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA); - listeners = new HashMap>(); - // Create a new hashmap and registry it in rListener results = new HashMap(); @@ -121,12 +110,4 @@ 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(); } } @@ -306,27 +287,3 @@ } - @Override - public void notifyEvent(Event event) throws IOException, SerializerException { - } - - @Override - public void addListener(EventListener eventListener) throws Exception { - if (eventListener.getTopic() == null) { - eventListener.setTopic(uid); - } - listeners.put(eventListener.getTopic(), eventListener); - dispatcher.addListener(eventListener); - } - - @Override - public void removeListener(EventListener eventListener) throws Exception { - listeners.remove(eventListener.getTopic()); - dispatcher.removeListener(eventListener); - } - - @Override - public Collection> getListeners() throws Exception { - return listeners.values(); - } - } Index: trunk/src/main/java/omq/common/broker/Broker.java =================================================================== --- trunk/src/main/java/omq/common/broker/Broker.java (revision 71) +++ trunk/src/main/java/omq/common/broker/Broker.java (revision 72) @@ -13,7 +13,4 @@ import omq.client.proxy.MultiProxymq; 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; @@ -21,5 +18,4 @@ import omq.exception.InitBrokerException; import omq.exception.RemoteException; -import omq.exception.SerializerException; import omq.server.RemoteObject; @@ -41,5 +37,4 @@ private Channel channel; private ResponseListener responseListener; - private EventDispatcher eventDispatcher; private Serializer serializer; private boolean clientStarted = false; @@ -75,5 +70,4 @@ if (clientStarted) { responseListener.kill(); - eventDispatcher.kill(); // TODO proxies = null; ?? } @@ -211,25 +205,4 @@ 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); } @@ -318,8 +291,4 @@ } - public EventDispatcher getEventDispatcher() { - return eventDispatcher; - } - public Serializer getSerializer() { return serializer; Index: trunk/src/main/java/omq/common/util/Serializer.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializer.java (revision 71) +++ trunk/src/main/java/omq/common/util/Serializer.java (revision 72) @@ -4,5 +4,4 @@ import java.util.Properties; -import omq.common.event.Event; import omq.common.message.Request; import omq.common.message.Response; @@ -48,9 +47,9 @@ 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) { @@ -62,5 +61,5 @@ } - public ISerializer getInstance(String type) throws SerializerException { + public ISerializer getInstance(String type) throws SerializerException { if (KRYO.equals(type)) { if (kryoSerializer == null) { @@ -79,5 +78,5 @@ return javaSerializer; } - + throw new SerializerException("Serializer not found."); } @@ -148,26 +147,3 @@ } - 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: trunk/src/main/java/omq/common/util/Serializers/GsonImp.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializers/GsonImp.java (revision 71) +++ trunk/src/main/java/omq/common/util/Serializers/GsonImp.java (revision 72) @@ -3,5 +3,4 @@ import java.util.List; -import omq.common.event.Event; import omq.common.message.Request; import omq.common.message.Response; @@ -75,24 +74,3 @@ } - @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: trunk/src/main/java/omq/common/util/Serializers/ISerializer.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializers/ISerializer.java (revision 71) +++ trunk/src/main/java/omq/common/util/Serializers/ISerializer.java (revision 72) @@ -1,5 +1,4 @@ package omq.common.util.Serializers; -import omq.common.event.Event; import omq.common.message.Request; import omq.common.message.Response; @@ -18,5 +17,3 @@ public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException; - - public Event deserializeEvent(byte[] bytes) throws SerializerException; } Index: trunk/src/main/java/omq/common/util/Serializers/JavaImp.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializers/JavaImp.java (revision 71) +++ trunk/src/main/java/omq/common/util/Serializers/JavaImp.java (revision 72) @@ -6,6 +6,4 @@ import java.io.ObjectOutputStream; -import omq.common.event.Event; -import omq.common.event.EventWrapper; import omq.common.message.Request; import omq.common.message.Response; @@ -51,10 +49,4 @@ } - @Override - public Event deserializeEvent(byte[] bytes) throws SerializerException { - EventWrapper wrapper = (EventWrapper) deserializeObject(bytes); - return wrapper.getEvent(); - } - public Object deserializeObject(byte[] bytes) throws SerializerException { try { Index: trunk/src/main/java/omq/common/util/Serializers/KryoImp.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializers/KryoImp.java (revision 71) +++ trunk/src/main/java/omq/common/util/Serializers/KryoImp.java (revision 72) @@ -3,14 +3,12 @@ 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; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; /** @@ -53,10 +51,4 @@ } - @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 { Index: trunk/src/main/java/omq/server/RemoteObject.java =================================================================== --- trunk/src/main/java/omq/server/RemoteObject.java (revision 71) +++ trunk/src/main/java/omq/server/RemoteObject.java (revision 72) @@ -4,5 +4,4 @@ import java.lang.reflect.Method; import java.util.ArrayList; -import java.util.Collection; import java.util.HashMap; import java.util.List; @@ -10,14 +9,10 @@ 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 org.apache.log4j.Logger; import com.rabbitmq.client.Channel; @@ -42,5 +37,4 @@ private Properties env; private transient Broker broker; - private transient Serializer serializer; private transient RemoteWrapper remoteWrapper; private transient Map>> params; @@ -69,5 +63,4 @@ multiQueue = UID + System.currentTimeMillis(); env = broker.getEnvironment(); - serializer = broker.getSerializer(); params = new HashMap>>(); @@ -93,5 +86,4 @@ this.broker = broker; UID = reference; - serializer = broker.getSerializer(); if (channel == null || !channel.isOpen()) { channel = broker.getChannel(); @@ -141,15 +133,4 @@ } - @Override - public void notifyEvent(Event event) throws IOException, SerializerException { - String corrID = java.util.UUID.randomUUID().toString(); - event.setTopic(UID); - event.setCorrId(corrID); - EventWrapper wrapper = new EventWrapper(event); - channel.exchangeDeclare(UID, "fanout"); - channel.basicPublish(UID, "", null, serializer.serialize(wrapper)); - logger.debug("Sending event-> topic: " + UID + ", corrID: " + corrID); - } - public void kill() throws IOException { logger.warn("Killing objectmq: " + this.getRef()); @@ -265,16 +246,3 @@ } - @Override - public void addListener(EventListener eventListener) throws Exception { - } - - @Override - public void removeListener(EventListener eventListener) throws Exception { - } - - @Override - public Collection> getListeners() throws Exception { - return null; - } - } Index: trunk/src/test/java/omq/test/calculator/Calculator.java =================================================================== --- trunk/src/test/java/omq/test/calculator/Calculator.java (revision 71) +++ trunk/src/test/java/omq/test/calculator/Calculator.java (revision 72) @@ -1,5 +1,3 @@ package omq.test.calculator; - -import java.io.IOException; import omq.Remote; @@ -7,5 +5,4 @@ import omq.client.annotation.RemoteInterface; import omq.client.annotation.SyncMethod; -import omq.exception.SerializerException; @RemoteInterface @@ -20,7 +17,4 @@ public void sendMessage(Message m); - @AsyncMethod - public void asyncDivideByZero() throws IOException, SerializerException; - @SyncMethod(timeout = 1500) public int divideByZero(); Index: trunk/src/test/java/omq/test/calculator/CalculatorImpl.java =================================================================== --- trunk/src/test/java/omq/test/calculator/CalculatorImpl.java (revision 71) +++ trunk/src/test/java/omq/test/calculator/CalculatorImpl.java (revision 72) @@ -1,21 +1,11 @@ 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; } @@ -41,11 +31,4 @@ @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()); Index: trunk/src/test/java/omq/test/calculator/CalculatorTest.java =================================================================== --- trunk/src/test/java/omq/test/calculator/CalculatorTest.java (revision 71) +++ trunk/src/test/java/omq/test/calculator/CalculatorTest.java (revision 72) @@ -120,15 +120,4 @@ @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"); Index: trunk/src/test/java/omq/test/calculator/ZeroEvent.java =================================================================== --- trunk/src/test/java/omq/test/calculator/ZeroEvent.java (revision 71) +++ (revision ) @@ -1,22 +1,0 @@ -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: trunk/src/test/java/omq/test/calculator/ZeroListener.java =================================================================== --- trunk/src/test/java/omq/test/calculator/ZeroListener.java (revision 71) +++ (revision ) @@ -1,15 +1,0 @@ -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: trunk/src/test/java/omq/test/observer/ObserverTest.java =================================================================== --- trunk/src/test/java/omq/test/observer/ObserverTest.java (revision 71) +++ trunk/src/test/java/omq/test/observer/ObserverTest.java (revision 72) @@ -23,4 +23,5 @@ private static String OBSERVER = "observer"; private static Broker broker; + private static RemoteSubjectImpl subject; private static RemoteSubject remoteSubject; @@ -71,6 +72,11 @@ env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "server_reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "server_event_queue"); + Broker broker = new Broker(env); - RemoteSubjectImpl subject = new RemoteSubjectImpl(); + subject = new RemoteSubjectImpl(broker); broker.bind(SUBJECT, subject); @@ -90,12 +96,15 @@ RemoteObserverImpl observer = new RemoteObserverImpl(); broker.bind(OBSERVER, observer); + observer.setSubject(remoteSubject); - remoteSubject.addObserver(observer); + remoteSubject.addObserver(observer.getRef()); remoteSubject.setState(expected); - remoteSubject.notifyObservers(); + // both proxies are in the same thread for this reason since subject has + // a proxy inside we need to use subject or create a new thread + subject.notifyObservers(); actual = observer.getObsState(); assertEquals(expected, actual); - remoteSubject.setState(null); + remoteSubject.setState(""); } Index: trunk/src/test/java/omq/test/observer/RemoteSubject.java =================================================================== --- trunk/src/test/java/omq/test/observer/RemoteSubject.java (revision 71) +++ trunk/src/test/java/omq/test/observer/RemoteSubject.java (revision 72) @@ -4,4 +4,5 @@ import omq.client.annotation.RemoteInterface; import omq.client.annotation.SyncMethod; +import omq.exception.RemoteException; @RemoteInterface @@ -15,8 +16,8 @@ @SyncMethod(timeout = 1000) - public void addObserver(RemoteObserver o); + public void addObserver(String ref) throws RemoteException; @SyncMethod(timeout = 1000) - public void removeObserver(RemoteObserver o); + public void removeObserver(String ref) throws RemoteException; @SyncMethod(timeout = 1000) Index: trunk/src/test/java/omq/test/observer/RemoteSubjectImpl.java =================================================================== --- trunk/src/test/java/omq/test/observer/RemoteSubjectImpl.java (revision 71) +++ trunk/src/test/java/omq/test/observer/RemoteSubjectImpl.java (revision 72) @@ -4,4 +4,6 @@ import java.util.List; +import omq.common.broker.Broker; +import omq.exception.RemoteException; import omq.server.RemoteObject; @@ -13,14 +15,22 @@ private static final long serialVersionUID = 1L; private String state; - private List list = new ArrayList(); + private Broker broker; + private List list; - @Override - public void addObserver(RemoteObserver o) { - list.add(o); + public RemoteSubjectImpl(Broker broker) { + this.broker = broker; + list = new ArrayList(); } @Override - public void removeObserver(RemoteObserver o) { - list.remove(o); + public void addObserver(String ref) throws RemoteException { + RemoteObserver obs = broker.lookup(ref, RemoteObserver.class); + list.add(obs); + } + + @Override + public void removeObserver(String ref) throws RemoteException { + RemoteObserver obs = broker.lookup(ref, RemoteObserver.class); + list.remove(obs); }