Changeset 18

Timestamp:

05/23/13 15:05:54 (11 years ago)

Author:

gguerrero

Message:

Logs and new event.

Location:

trunk/objectmq/src/omq

Files:

• client/proxy/Proxymq.java (modified) (6 diffs)
• common/broker/Broker.java (modified) (2 diffs)
• common/util/Log.java (added)
• common/util/ParameterQueue.java (modified) (1 diff)
• common/util/Serializer.java (modified) (1 diff)
• server/remote/request/InvocationThread.java (modified) (3 diffs)
• ztest/calculator/CalculatorImpl.java (modified) (2 diffs)
• ztest/calculator/CalculatorTest.java (modified) (3 diffs)
• ztest/calculator/ServerTest.java (modified) (1 diff)
• ztest/calculator/ZeroEvent.java (modified) (1 diff)
• ztest/calculator/ZeroListener.java (modified) (1 diff)

trunk/objectmq/src/omq/client/proxy/Proxymq.java

@@ -20 +20 @@
 import omq.common.message.Request;
 import omq.common.message.Response;
+import omq.common.util.Log;
 import omq.common.util.ParameterQueue;
 import omq.common.util.Serializer;
@@ -126 +127 @@
                 }
         }
-
-        private void publishAsyncRequest(Request request) throws IOException, SerializerException {
+        
+        private void publishMessage(Request request, String replyQueueName) throws IOException, SerializerException{
                 String corrId = request.getId();
 
                 // Get the environment properties
                 String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
-                String replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
                 String routingkey = this.uid;
 
@@ -139 +139 @@
 
                 // Publish the message
-                channel.basicPublish(exchange, routingkey, props, Serializer.serialize(request));
+                byte[] bytesRequest = Serializer.serialize(request);                            
+                channel.basicPublish(exchange, routingkey, props, bytesRequest);
+                Log.saveLog("Client-Serialize", bytesRequest);          
+        }
+
+        private void publishAsyncRequest(Request request) throws IOException, SerializerException {
+                // Get the environment properties
+                String replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
+                publishMessage(request, replyQueueName);
         }
 
@@ -149 +157 @@
 
                 // Get the environment properties
-                String exchange = env.getProperty(ParameterQueue.RPC_EXCHANGE);
                 String replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE);
-                String routingkey = this.uid;
-
-                // Add the correlation ID and create a replyTo property
-                BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).build();
 
                 // Publish the message
@@ -160 +163 @@
                 while (i < retries) {
                         try {
-                                channel.basicPublish(exchange, routingkey, props, Serializer.serialize(request));
+                                publishMessage(request, replyQueueName);
                                 return getResult(corrId, timeout, type);
                         } catch (TimeoutException te) {
@@ -208 +211 @@
                         }
                         resp = Serializer.deserializeResponse(results.get(corrId), type);
+                        Log.saveLog("Client-Deserialize", results.get(corrId));
                         // Remove and indicate the key exists (a hashmap can contain a null
                         // object, using this we'll know whether a response has been

trunk/objectmq/src/omq/common/broker/Broker.java

@@ -7 +7 @@
 import omq.client.proxy.Proxymq;
 import omq.client.remote.response.ResponseListener;
+import omq.common.event.Event;
 import omq.common.event.EventDispatcher;
+import omq.common.event.EventWrapper;
 import omq.common.util.Environment;
 import omq.common.util.OmqConnectionFactory;
+import omq.common.util.Serializer;
 import omq.exception.EnvironmentException;
 import omq.exception.RemoteException;
+import omq.exception.SerializerException;
 import omq.server.remote.request.RemoteObject;
 
@@ -91 +95 @@
                 }
         }
+        
+        
+        public static void trigger(Event event) throws IOException, SerializerException{
+                String UID = event.getTopic();
+                EventWrapper wrapper = new EventWrapper(event);
+                channel.exchangeDeclare(UID, "fanout");
+                channel.basicPublish(UID, "", null, Serializer.serialize(wrapper));
+        }
 
 }

trunk/objectmq/src/omq/common/util/ParameterQueue.java

@@ -30 +30 @@
 
         public static String ENABLE_SSL = "revo.enable_ssl";
+        public static String DEBUGFILE = "revo.debug_file";
+        
 
         /*

trunk/objectmq/src/omq/common/util/Serializer.java

@@ -37 +37 @@
                                 String className = env.getProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.JavaImp");
 
+                                if (className == null || className.isEmpty()) {
+                                        throw new ClassNotFoundException("Class name is null or empty.");
+                                }
+                                
                                 serializer = (ISerializer) Class.forName(className).newInstance();
                         } catch (Exception ex) {

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

@@ -5 +5 @@
 import omq.common.message.Request;
 import omq.common.message.Response;
+import omq.common.util.Log;
 import omq.common.util.Serializer;
 
@@ -35 +36 @@
                                 // Deserialize the json
                                 Request request = Serializer.deserializeRequest(delivery.getBody(), obj);
+                                Log.saveLog("Server-Deserialize", delivery.getBody());
 
                                 // Invoke the method
@@ -46 +48 @@
                                         BasicProperties props = delivery.getProperties();
 
-                                        BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()).build();
+                                        BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef())
+                                                        .correlationId(props.getCorrelationId()).build();
 
-                                        channel.basicPublish("", props.getReplyTo(), replyProps, Serializer.serialize(resp));
+                                        byte[] bytesResponse = Serializer.serialize(resp);
+                                        channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse);
+
+                                        Log.saveLog("Server-Serialize", bytesResponse);
                                 }
 

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

@@ -4 +4 @@
 
 import omq.client.annotation.AsyncMethod;
+import omq.common.broker.Broker;
 import omq.exception.SerializerException;
 import omq.server.remote.request.RemoteObject;
@@ -33 +34 @@
 
         public void divideByZero() throws IOException, SerializerException {
-                ZeroEvent ze = new ZeroEvent("my zero event");
-                notifyEvent(ze);
+                ZeroEvent ze = new ZeroEvent("my zero event", "zero-event");
+                Broker.trigger(ze);
+                //notifyEvent(ze);
         }
 

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

@@ -22 +22 @@
 
                 // Set host info of rabbimq (where it is)
-                env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1");
+                env.setProperty(ParameterQueue.SERVER_HOST, "10.30.239.228");
                 env.setProperty(ParameterQueue.SERVER_PORT, "5672");
                 // env.setProperty(ParameterQueue.SERIALIZERNAME,
@@ -31 +31 @@
                 // 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
@@ -75 +76 @@
         @Test
         public void notifyEvent() throws Exception {
-                ZeroListener zL = new ZeroListener();
+                ZeroListener zL = new ZeroListener("zero-event");
 
                 remoteCalc.addListener(zL);

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

@@ -16 +16 @@
 
                 // Get host info of rabbimq (where it is)
-                env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1");
+                env.setProperty(ParameterQueue.SERVER_HOST, "10.30.239.228");
                 env.setProperty(ParameterQueue.SERVER_PORT, "5672");
                 // env.setProperty(ParameterQueue.SERIALIZERNAME,

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

@@ -12 +12 @@
         }
 
-        public ZeroEvent(String corrId) {
-                super(corrId);
+        public ZeroEvent(String corrId, String topic) {
+                super(corrId, topic);
         }
 

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

@@ -5 +5 @@
 public class ZeroListener extends EventListener<ZeroEvent> {
 
+        public ZeroListener(String topic){
+                super(topic);
+        }
+        
         @Override
         public void notifyEvent(ZeroEvent event) {