Changeset 11

Timestamp:

05/17/13 15:41:11 (12 years ago)

Author:

gguerrero

Message:

Fixed dynamic Serializer (by default is JavaImp?).
Added compression parameter (by default is false).

Location:

trunk/objectmq/src/omq

Files:

• common/broker/Broker.java (modified) (4 diffs)
• common/util/Environment.java (moved) (moved from trunk/objectmq/src/omq/common/util/RevoEnvironment.java) (1 diff)
• common/util/ParameterQueue.java (modified) (1 diff)
• common/util/Serializer.java (modified) (3 diffs)
• common/util/Zipper.java (added)
• ztest/calculator/CalculatorTest.java (modified) (3 diffs)

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

@@ -7 +7 @@
 import omq.client.remote.response.ResponseListener;
 import omq.common.remote.OmqConnectionFactory;
+import omq.common.util.Environment;
+import omq.exception.EnvironmentException;
 import omq.exception.RemoteException;
 import omq.server.remote.request.RemoteObject;
@@ -16 +18 @@
         private static Connection connection;
         private static Channel channel;
-        private static Properties environment;
 
         public static void initBroker(Properties env) throws Exception {
-                if (environment == null) {
-                        environment = env;
+                try {
+                        Environment.getEnvironment();
+                } catch (EnvironmentException ex) { // environment not set.
+                        Environment.setEnvironment(env);
                         connection = OmqConnectionFactory.getConnection(env);
                         channel = connection.createChannel();
@@ -37 +40 @@
         public static Remote lookup(String reference, Class<?> contract) throws RemoteException {
                 try {
+                        Properties environment = Environment.getEnvironment();
+
                         if (ResponseListener.isVoid()) {
                                 ResponseListener.init(environment);
@@ -54 +59 @@
         public static void bind(String reference, RemoteObject remote) throws RemoteException {
                 try {
+                        Properties environment = Environment.getEnvironment();
                         remote.start(reference, environment);
                 } catch (Exception e) {

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

@@ -11 +11 @@
  * 
  */
-public class RevoEnvironment {
+public class Environment {
         private static Properties env;
 

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

@@ -12 +12 @@
          */
 
+        public static String SERIALIZERNAME = "revo.serializer";
+        public static String ENABLECOMPRESSION = "revo.compression";
+        
         public static String SERVER_HOST = "revo.host";
         public static String SERVER_PORT = "revo.port";

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

@@ -1 +1 @@
 package omq.common.util;
+
+import java.io.IOException;
+import java.util.Properties;
 
 import omq.common.message.Request;
 import omq.common.message.Response;
-import omq.common.util.Serializers.GsonImp;
 import omq.common.util.Serializers.ISerializer;
+import omq.exception.EnvironmentException;
 import omq.exception.SerializerException;
 import omq.server.remote.request.RemoteObject;
@@ -16 +19 @@
         public static ISerializer serializer;
 
-        public static ISerializer getInstance() {
+        private static Boolean getEnableCompression(){
+                Boolean enableCompression = false;
+                try {
+                        Properties env = Environment.getEnvironment();
+                        enableCompression = Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false"));
+                } catch (EnvironmentException e) { }
+                
+                return enableCompression;
+        }
+        
+        public static ISerializer getInstance() throws SerializerException {            
                 if (serializer == null) {
-                        serializer = new GsonImp();
-                        // serializer = new JavaImp();
-                        // serializer = new KryoImp();
+                        try{
+                                Properties env = Environment.getEnvironment();                          
+                                String className = env.getProperty(ParameterQueue.SERIALIZERNAME, "omq.common.util.Serializers.JavaImp");
+                                
+                                serializer = (ISerializer) Class.forName(className).newInstance();
+                        } catch (Exception ex){
+                                throw new SerializerException(ex.getMessage(), ex);
+                        }
                 }
+                
                 return serializer;
         }
@@ -27 +46 @@
         public static byte[] serialize(Object obj) throws SerializerException {
                 ISerializer instance = getInstance();
-                return instance.serialize(obj);
+                
+                Boolean enableCompression = getEnableCompression();             
+                if(enableCompression){
+                        byte[] objSerialized = instance.serialize(obj);
+                        try {
+                                return Zipper.zip(objSerialized);
+                        } catch (IOException e) {
+                                throw new SerializerException(e.getMessage(), e);
+                        }                       
+                } else{
+                        return instance.serialize(obj);
+                }               
         }
 
         public static Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException {
                 ISerializer instance = getInstance();
-                return instance.deserializeRequest(bytes, obj);
+                                
+                Boolean enableCompression = getEnableCompression();
+                if(enableCompression){
+                        try {
+                                byte[] unZippedBytes = Zipper.unzip(bytes);
+                                return instance.deserializeRequest(unZippedBytes, obj);
+                        } catch (IOException e) {
+                                throw new SerializerException(e.getMessage(), e);
+                        }
+                } else{
+                        return instance.deserializeRequest(bytes, obj);
+                }                       
         }
 
         public static Response deserializeResponse(byte[] bytes, Class<?> type) throws SerializerException {
                 ISerializer instance = getInstance();
-                return instance.deserializeResponse(bytes, type);
+                
+                Boolean enableCompression = getEnableCompression();
+                if(enableCompression){
+                        try {
+                                byte[] unZippedBytes = Zipper.unzip(bytes);
+                                return instance.deserializeResponse(unZippedBytes, type);
+                        } catch (IOException e) {
+                                throw new SerializerException(e.getMessage(), e);
+                        }
+                } else{
+                        return instance.deserializeResponse(bytes, type);
+                }                       
         }
 }

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

@@ -22 +22 @@
 
                 // 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, "omq.common.util.Serializers.KryoImp");
+                env.setProperty(ParameterQueue.ENABLECOMPRESSION, "true");
 
                 // Get info about the queue & the exchange where the RemoteListener will
@@ -44 +46 @@
 
                 // 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, "omq.common.util.Serializers.KryoImp");
+                env.setProperty(ParameterQueue.ENABLECOMPRESSION, "true");
+                
                 // Set info about where the message will be sent
                 env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange");
@@ -57 +61 @@
 
                 Broker.initBroker(env);
-
                 remoteCalc = (Calculator) Broker.lookup(Calculator.class.getSimpleName(), Calculator.class);
         }