Index: trunk/objectmq/src/omq/common/broker/Broker.java =================================================================== --- trunk/objectmq/src/omq/common/broker/Broker.java (revision 10) +++ trunk/objectmq/src/omq/common/broker/Broker.java (revision 11) @@ -7,4 +7,6 @@ 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,9 +18,10 @@ 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,4 +40,6 @@ public static Remote lookup(String reference, Class contract) throws RemoteException { try { + Properties environment = Environment.getEnvironment(); + if (ResponseListener.isVoid()) { ResponseListener.init(environment); @@ -54,4 +59,5 @@ public static void bind(String reference, RemoteObject remote) throws RemoteException { try { + Properties environment = Environment.getEnvironment(); remote.start(reference, environment); } catch (Exception e) { Index: trunk/objectmq/src/omq/common/util/Environment.java =================================================================== --- trunk/objectmq/src/omq/common/util/Environment.java (revision 11) +++ trunk/objectmq/src/omq/common/util/Environment.java (revision 11) @@ -0,0 +1,38 @@ +package omq.common.util; + +import java.util.Properties; + +import omq.exception.EnvironmentException; + + +/** + * + * @author Sergi Toda + * + */ +public class Environment { + private static Properties env; + + /** + * This method return the environment of the middleware + * + * @return Environment to use in the middleware + * @throws EnvironmentException + * will be thrown whether the environment is null + */ + public static Properties getEnvironment() throws EnvironmentException { + if (env == null) { + throw new EnvironmentException(); + } + return env; + } + + /** + * This method sets the environment of the middleware + * + * @param environment + */ + public static void setEnvironment(Properties environment) { + env = environment; + } +} Index: trunk/objectmq/src/omq/common/util/ParameterQueue.java =================================================================== --- trunk/objectmq/src/omq/common/util/ParameterQueue.java (revision 10) +++ trunk/objectmq/src/omq/common/util/ParameterQueue.java (revision 11) @@ -12,4 +12,7 @@ */ + 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"; Index: trunk/objectmq/src/omq/common/util/RevoEnvironment.java =================================================================== --- trunk/objectmq/src/omq/common/util/RevoEnvironment.java (revision 10) +++ (revision ) @@ -1,38 +1,0 @@ -package omq.common.util; - -import java.util.Properties; - -import omq.exception.EnvironmentException; - - -/** - * - * @author Sergi Toda - * - */ -public class RevoEnvironment { - private static Properties env; - - /** - * This method return the environment of the middleware - * - * @return Environment to use in the middleware - * @throws EnvironmentException - * will be thrown whether the environment is null - */ - public static Properties getEnvironment() throws EnvironmentException { - if (env == null) { - throw new EnvironmentException(); - } - return env; - } - - /** - * This method sets the environment of the middleware - * - * @param environment - */ - public static void setEnvironment(Properties environment) { - env = environment; - } -} Index: trunk/objectmq/src/omq/common/util/Serializer.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializer.java (revision 10) +++ trunk/objectmq/src/omq/common/util/Serializer.java (revision 11) @@ -1,8 +1,11 @@ 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,10 +19,26 @@ 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,15 +46,48 @@ 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); + } } } Index: trunk/objectmq/src/omq/common/util/Zipper.java =================================================================== --- trunk/objectmq/src/omq/common/util/Zipper.java (revision 11) +++ trunk/objectmq/src/omq/common/util/Zipper.java (revision 11) @@ -0,0 +1,52 @@ +package omq.common.util; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; + +public class Zipper { + + public static byte[] zip(byte[] b) throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + + GZIPOutputStream zos = null; + try { + zos = new GZIPOutputStream(baos); + zos.write(b); + } catch (IOException e) { + throw new IOException(e.getMessage(), e); + } finally{ + if(zos != null){ + zos.close(); + } + } + + return baos.toByteArray(); + } + + public static byte[] unzip(byte[] b) throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ByteArrayInputStream bais = new ByteArrayInputStream(b); + + GZIPInputStream zis = null; + try { + zis = new GZIPInputStream(bais); + + byte[] tmpBuffer = new byte[256]; + int n; + while ((n = zis.read(tmpBuffer)) >= 0) { + baos.write(tmpBuffer, 0, n); + } + } catch (IOException e) { + throw new IOException(e.getMessage(), e); + } finally { + if(zis != null){ + zis.close(); + } + } + + return baos.toByteArray(); + } +} Index: trunk/objectmq/src/omq/ztest/calculator/CalculatorTest.java =================================================================== --- trunk/objectmq/src/omq/ztest/calculator/CalculatorTest.java (revision 10) +++ trunk/objectmq/src/omq/ztest/calculator/CalculatorTest.java (revision 11) @@ -22,6 +22,8 @@ // 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,7 +46,9 @@ // 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,5 +61,4 @@ Broker.initBroker(env); - remoteCalc = (Calculator) Broker.lookup(Calculator.class.getSimpleName(), Calculator.class); }