Index: trunk/src/main/java/omq/Remote.java =================================================================== --- trunk/src/main/java/omq/Remote.java (revision 70) +++ 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 70) +++ 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 70) +++ 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 70) +++ 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 70) +++ 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 70) +++ 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 70) +++ 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 70) +++ 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 70) +++ 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; - } - }