Changeset 16

Timestamp:

05/21/13 15:08:02 (11 years ago)

Author:

stoda

Message:

EventWrapperAdded?

Location:

trunk/objectmq/src/omq

Files:

• client/annotation/AsyncMethod.java (modified) (1 diff)
• common/event/Event.java (modified) (1 diff)
• common/event/EventDispatcher.java (modified) (1 diff)
• common/event/EventListener.java (modified) (2 diffs)
• common/event/EventWrapper.java (added)
• common/util/Serializers/GsonImp.java (modified) (1 diff)
• common/util/Serializers/KryoImp.java (modified) (2 diffs)
• server/remote/request/RemoteObject.java (modified) (2 diffs)
• ztest/calculator/Calculator.java (modified) (3 diffs)
• ztest/calculator/CalculatorImpl.java (modified) (2 diffs)
• ztest/calculator/CalculatorTest.java (modified) (2 diffs)
• ztest/calculator/ServerTest.java (modified) (1 diff)
• ztest/calculator/ZeroEvent.java (modified) (1 diff)
• ztest/calculator/ZeroListener.java (added)

trunk/objectmq/src/omq/client/annotation/AsyncMethod.java

@@ -16 +16 @@
 @Target(ElementType.METHOD)
 public @interface AsyncMethod {
-        String generateEvent() default "";
 }

trunk/objectmq/src/omq/common/event/Event.java

@@ -17 +17 @@
         private String corrId;
         private String topic;
+
+        public Event() {
+        }
+
+        public Event(String corrId, String topic) {
+                this.corrId = corrId;
+                this.topic = topic;
+        }
 
         public Event(String corrId) {

trunk/objectmq/src/omq/common/event/EventDispatcher.java

@@ -110 +110 @@
 
         public int addListener(EventListener e) throws Exception {
+                // Map<String, ArrayList<EventListener<Event>>> mListeners =
+                // listeners.get(e.getTopic());
+                // if(mListeners == null){
+                // mListeners = new HashMap<String, ArrayList<EventListener<Event>>>();
+                //
+                // String queueName = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE);
+                // String reference = e.getTopic();
+                // channel.exchangeDeclare(reference, "fanout");
+                // channel.queueBind(queueName, reference, "");
+                // }
+
                 Vector<EventListener> vListeners = listeners.get(e.getTopic());
                 if (vListeners == null) {

trunk/objectmq/src/omq/common/event/EventListener.java

@@ -8 +8 @@
  * 
  */
-public abstract class EventListener {
+public abstract class EventListener<E extends Event> {
         private String topic;
 
@@ -31 +31 @@
          * @param event
          */
-        public abstract void notifyEvent(Event event);
+        public abstract void notifyEvent(E event);
 
         public void setTopic(String topic) {

trunk/objectmq/src/omq/common/util/Serializers/GsonImp.java

@@ -73 +73 @@
 
         @Override
-        public Event deserializeEvent(byte[] unZippedBytes) throws SerializerException {
-                // TODO deserializeEvent class<?> ¿?
-                return null;
+        public Event deserializeEvent(byte[] bytes) throws SerializerException {
+                try {
+                        String json = new String(bytes);
+System.out.println(json);
+                        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());
+                }
         }
-        
-        
 
 }

trunk/objectmq/src/omq/common/util/Serializers/KryoImp.java

@@ -8 +8 @@
 
 import omq.common.event.Event;
+import omq.common.event.EventWrapper;
 import omq.common.message.Request;
 import omq.common.message.Response;
@@ -54 +55 @@
         @Override
         public Event deserializeEvent(byte[] bytes) throws SerializerException {
-                return (Event) deserializeObject(bytes, Event.class);
+                EventWrapper wrapper = (EventWrapper) deserializeObject(bytes, EventWrapper.class);
+                return wrapper.getEvent();
         }
 

trunk/objectmq/src/omq/server/remote/request/RemoteObject.java

@@ -14 +14 @@
 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;
@@ -123 +124 @@
         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(event));
+                channel.basicPublish(UID, "", null, Serializer.serialize(wrapper));
         }
 

trunk/objectmq/src/omq/ztest/calculator/Calculator.java

@@ -1 +1 @@
 package omq.ztest.calculator;
+
+import java.io.IOException;
 
 import omq.Remote;
@@ -5 +7 @@
 import omq.client.annotation.RemoteInterface;
 import omq.client.annotation.SyncMethod;
+import omq.exception.SerializerException;
 
 @RemoteInterface
@@ -15 +18 @@
         
         @AsyncMethod
-        public void divideByZero();
+        public void divideByZero() throws IOException, SerializerException;
 
 }

trunk/objectmq/src/omq/ztest/calculator/CalculatorImpl.java

@@ -1 +1 @@
 package omq.ztest.calculator;
 
+import java.io.IOException;
+
+import omq.exception.SerializerException;
 import omq.server.remote.request.RemoteObject;
 
@@ -28 +31 @@
         }
 
-        public void divideByZero() {
-                
+        public void divideByZero() throws IOException, SerializerException {
+                ZeroEvent ze = new ZeroEvent("my zero event");
+                notifyEvent(ze);
         }
 

trunk/objectmq/src/omq/ztest/calculator/CalculatorTest.java

@@ -24 +24 @@
                 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.KryoImp");
+                env.setProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.GsonImp");
                 env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false");
 
@@ -70 +71 @@
                 Thread.sleep(200);
         }
+
+        @Test
+        public void notifyEvent() throws Exception {
+                ZeroListener zL = new ZeroListener();
+
+                remoteCalc.addListener(zL);
+
+                remoteCalc.divideByZero();
+
+                Thread.sleep(200);
+        }
 }

trunk/objectmq/src/omq/ztest/calculator/ServerTest.java

@@ -18 +18 @@
                 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.KryoImp");
+                env.setProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.GsonImp");
                 env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false");
 

trunk/objectmq/src/omq/ztest/calculator/ZeroEvent.java

@@ -9 +9 @@
         private static final long serialVersionUID = 1L;
 
+        public ZeroEvent() {
+        }
+
         public ZeroEvent(String corrId) {
                 super(corrId);
         }
 
+        public String getZeroMessage() {
+                return "divition by 0";
+        }
+
 }