Index: trunk/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 44) +++ trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 47) @@ -47,4 +47,5 @@ private String uid; + private transient String serializerType; private transient ResponseListener rListener; private transient EventDispatcher dispatcher; @@ -89,4 +90,7 @@ // this.channel = Broker.getChannel(); this.env = env; + + // set the serializer type + serializerType = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); listeners = new HashMap>(); @@ -150,8 +154,8 @@ // Add the correlation ID and create a replyTo property - BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).build(); + BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).type(serializerType).build(); // Publish the message - byte[] bytesRequest = Serializer.serialize(request); + byte[] bytesRequest = Serializer.serialize(serializerType, request); // TODO See this // channel.basicPublish(exchange, routingkey, props, bytesRequest); @@ -306,4 +310,16 @@ } + public static void stopProxy() { + proxies = new HashMap(); + } + + public static Map getProxies() { + return proxies; + } + + public static void setProxies(Map proxies) { + Proxymq.proxies = proxies; + } + @Override public String getRef() { Index: trunk/src/main/java/omq/common/broker/Broker.java =================================================================== --- trunk/src/main/java/omq/common/broker/Broker.java (revision 44) +++ trunk/src/main/java/omq/common/broker/Broker.java (revision 47) @@ -12,5 +12,4 @@ import omq.common.event.EventDispatcher; import omq.common.event.EventWrapper; -import omq.common.util.Environment; import omq.common.util.OmqConnectionFactory; import omq.common.util.ParameterQueue; @@ -33,4 +32,5 @@ private static boolean clientStarted = false; private static boolean connectionClosed = false; + private static Properties environment = null; // TODO ask Pedro if it can be only one object in the map (an object can // have multiple threads in the same broker -see environment-) @@ -44,7 +44,7 @@ */ public static synchronized void initBroker(Properties env) throws Exception { - if (Environment.isVoid()) { + if (environment == null) { remoteObjs = new HashMap(); - Environment.setEnvironment(env); + environment = env; connection = OmqConnectionFactory.getNewConnection(env); channel = connection.createChannel(); @@ -67,4 +67,5 @@ ResponseListener.stopResponseListner(); EventDispatcher.stopEventDispatcher(); + Proxymq.stopProxy(); } // Stop all the remote objects working @@ -72,6 +73,13 @@ unbind(reference); } + // Close the connection once all the listeners are died closeConnection(); + + clientStarted = false; + connectionClosed = false; + environment = null; + remoteObjs = null; + Serializer.removeSerializers(); } @@ -87,4 +95,5 @@ connectionClosed = true; connection.close(); + connectionClosed = false; } @@ -111,5 +120,4 @@ public static T lookup(String reference, Class contract) throws RemoteException { try { - Properties environment = Environment.getEnvironment(); if (!clientStarted) { @@ -132,5 +140,4 @@ public static void bind(String reference, RemoteObject remote) throws RemoteException { try { - Properties environment = Environment.getEnvironment(); remote.startRemoteObject(reference, environment); remoteObjs.put(reference, remote); @@ -244,6 +251,5 @@ } try { - Properties env = Environment.getEnvironment(); - connection = OmqConnectionFactory.getNewWorkingConnection(env); + connection = OmqConnectionFactory.getNewWorkingConnection(environment); channel = connection.createChannel(); addFaultTolerance(); @@ -265,3 +271,7 @@ } + public static Properties getEnvironment() { + return environment; + } + } Index: trunk/src/main/java/omq/common/util/Environment.java =================================================================== --- trunk/src/main/java/omq/common/util/Environment.java (revision 44) +++ (revision ) @@ -1,41 +1,0 @@ -package omq.common.util; - -import java.util.Properties; - -import omq.exception.EnvironmentException; - -/** - * - * @author Sergi Toda - * - */ -public class Environment { - private static Properties env = null; - - /** - * 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; - } - - public static boolean isVoid() { - return env == null; - } -} Index: trunk/src/main/java/omq/common/util/Log.java =================================================================== --- trunk/src/main/java/omq/common/util/Log.java (revision 44) +++ trunk/src/main/java/omq/common/util/Log.java (revision 47) @@ -7,64 +7,74 @@ import java.util.Properties; -import omq.exception.EnvironmentException; +import omq.common.broker.Broker; public class Log { public static void saveLog(String processName, byte[] bytesResponse) throws IOException { - try { - Properties env = Environment.getEnvironment(); - String debugPath = env.getProperty(ParameterQueue.DEBUGFILE, ""); - if (debugPath.length() > 0) { - long timeNow = (new Date()).getTime(); + Properties env = Broker.getEnvironment(); - File outputFolder = new File(debugPath + File.separator); - outputFolder.mkdirs(); - -// File outputFolder = new File(debugPath + File.separator + processName); -// outputFolder.mkdirs(); + String debugPath = env.getProperty(ParameterQueue.DEBUGFILE, ""); + if (debugPath.length() > 0) { + long timeNow = (new Date()).getTime(); -// File outputFileContent = new File(outputFolder.getAbsoluteFile() + File.separator + "content_" + timeNow); -// FileOutputStream outputStream = new FileOutputStream(outputFileContent); -// IOUtils.write(bytesResponse, outputStream); -// outputStream.close(); + File outputFolder = new File(debugPath + File.separator); + outputFolder.mkdirs(); - File outputFileLog = new File(debugPath + File.separator + "log"); - boolean exist = outputFileLog.exists(); - - FileWriter fw = new FileWriter(outputFileLog, true); // the true will append the new data - if(!exist){ - fw.write("#ProcessName\t\tFile\t\t\t\t\tDate\t\t\tSize\n"); - } - fw.write(processName + "\t" + "content_" + timeNow + "\t" + timeNow + "\t" + bytesResponse.length + "\tbytes\n"); - fw.close(); + // File outputFolder = new File(debugPath + File.separator + + // processName); + // outputFolder.mkdirs(); + + // File outputFileContent = new + // File(outputFolder.getAbsoluteFile() + File.separator + + // "content_" + timeNow); + // FileOutputStream outputStream = new + // FileOutputStream(outputFileContent); + // IOUtils.write(bytesResponse, outputStream); + // outputStream.close(); + + File outputFileLog = new File(debugPath + File.separator + "log"); + boolean exist = outputFileLog.exists(); + + FileWriter fw = new FileWriter(outputFileLog, true); // the true + // will + // append + // the + // new + // data + if (!exist) { + fw.write("#ProcessName\t\tFile\t\t\t\t\tDate\t\t\tSize\n"); } - } catch (EnvironmentException e) { - throw new IOException(e.getMessage(), e); + fw.write(processName + "\t" + "content_" + timeNow + "\t" + timeNow + "\t" + bytesResponse.length + "\tbytes\n"); + fw.close(); } + } - + public static void saveTimeSendRequestLog(String processName, String coorId, String method, long timeNow) throws IOException { - try { - Properties env = Environment.getEnvironment(); - String debugPath = env.getProperty(ParameterQueue.DEBUGFILE, ""); - if (debugPath.length() > 0) { - File outputFolder = new File(debugPath + File.separator + processName); - outputFolder.mkdirs(); - - File outputFileLog = new File(outputFolder + File.separator + "log"); - boolean exist = outputFileLog.exists(); - - FileWriter fw = new FileWriter(outputFileLog, true); // the true will append the new data - if(!exist){ - fw.write("#CoorId\tMethod\tDate\n"); - } - fw.write(coorId + "\t" + method + "\t" + timeNow + "\n"); - fw.close(); + Properties env = Broker.getEnvironment(); + + String debugPath = env.getProperty(ParameterQueue.DEBUGFILE, ""); + if (debugPath.length() > 0) { + File outputFolder = new File(debugPath + File.separator + processName); + outputFolder.mkdirs(); + + File outputFileLog = new File(outputFolder + File.separator + "log"); + boolean exist = outputFileLog.exists(); + + FileWriter fw = new FileWriter(outputFileLog, true); // the true + // will + // append + // the + // new + // data + if (!exist) { + fw.write("#CoorId\tMethod\tDate\n"); } - } catch (EnvironmentException e) { - throw new IOException(e.getMessage(), e); + fw.write(coorId + "\t" + method + "\t" + timeNow + "\n"); + fw.close(); } - } + + } } Index: trunk/src/main/java/omq/common/util/Serializer.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializer.java (revision 44) +++ trunk/src/main/java/omq/common/util/Serializer.java (revision 47) @@ -4,4 +4,5 @@ import java.util.Properties; +import omq.common.broker.Broker; import omq.common.event.Event; import omq.common.message.Request; @@ -11,5 +12,4 @@ import omq.common.util.Serializers.JavaImp; import omq.common.util.Serializers.KryoImp; -import omq.exception.EnvironmentException; import omq.exception.SerializerException; import omq.server.RemoteObject; @@ -21,20 +21,19 @@ */ public class Serializer { - public static String kryo = KryoImp.class.getCanonicalName(); - public static String java = JavaImp.class.getCanonicalName(); - public static String gson = GsonImp.class.getCanonicalName(); + public static String kryo = "kryo"; + public static String java = "java"; + public static String gson = "gson"; + // Client serializer public static ISerializer serializer; + // Server serializers + private static ISerializer kryoSerializer; + private static ISerializer javaSerializer; + private static ISerializer gsonSerializer; + private static Boolean getEnableCompression() { - Boolean enableCompression = false; - try { - Properties env = Environment.getEnvironment(); - enableCompression = Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false")); - } catch (EnvironmentException e) { - e.printStackTrace(); - } - - return enableCompression; + Properties env = Broker.getEnvironment(); + return Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false")); } @@ -42,5 +41,5 @@ if (serializer == null) { try { - Properties env = Environment.getEnvironment(); + Properties env = Broker.getEnvironment(); String className = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); @@ -49,5 +48,5 @@ } - serializer = (ISerializer) Class.forName(className).newInstance(); + serializer = getInstance(className); } catch (Exception ex) { throw new SerializerException(ex.getMessage(), ex); @@ -58,4 +57,40 @@ } + public static ISerializer getInstance(String type) throws SerializerException { + if (kryo.equals(type)) { + if (kryoSerializer == null) { + kryoSerializer = new KryoImp(); + } + return kryoSerializer; + } else if (gson.endsWith(type)) { + if (gsonSerializer == null) { + gsonSerializer = new GsonImp(); + } + return gsonSerializer; + } else { + if (javaSerializer == null) { + javaSerializer = new JavaImp(); + } + return javaSerializer; + } + } + + public static byte[] serialize(String type, Object obj) throws SerializerException { + ISerializer instance = getInstance(type); + + 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); + } + } + + // TODO: remove this function and think about the event serialization public static byte[] serialize(Object obj) throws SerializerException { ISerializer instance = getInstance(); @@ -74,6 +109,6 @@ } - public static Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException { - ISerializer instance = getInstance(); + public static Request deserializeRequest(String type, byte[] bytes, RemoteObject obj) throws SerializerException { + ISerializer instance = getInstance(type); Boolean enableCompression = getEnableCompression(); @@ -121,3 +156,10 @@ } } + + public static void removeSerializers() { + serializer = null; + kryoSerializer = null; + javaSerializer = null; + gsonSerializer = null; + } } Index: trunk/src/main/java/omq/server/InvocationThread.java =================================================================== --- trunk/src/main/java/omq/server/InvocationThread.java (revision 44) +++ trunk/src/main/java/omq/server/InvocationThread.java (revision 47) @@ -35,6 +35,8 @@ Delivery delivery = deliveryQueue.take(); + String serializerType = delivery.getProperties().getType(); + // Deserialize the json - Request request = Serializer.deserializeRequest(delivery.getBody(), obj); + Request request = Serializer.deserializeRequest(serializerType, delivery.getBody(), obj); // Log.saveLog("Server-Deserialize", delivery.getBody()); @@ -66,5 +68,5 @@ BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()).build(); - byte[] bytesResponse = Serializer.serialize(resp); + byte[] bytesResponse = Serializer.serialize(serializerType, resp); channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse);