Index: /trunk/.classpath =================================================================== --- /trunk/.classpath (revision 52) +++ /trunk/.classpath (revision 53) @@ -23,5 +23,5 @@ - + Index: /trunk/src/main/java/omq/client/listener/ResponseListener.java =================================================================== --- /trunk/src/main/java/omq/client/listener/ResponseListener.java (revision 52) +++ /trunk/src/main/java/omq/client/listener/ResponseListener.java (revision 53) @@ -28,7 +28,7 @@ */ public class ResponseListener extends Thread { - private static final Logger logger = Logger.getLogger(ResponseListener.class.getName()); - private static ResponseListener rListener; + private final Logger logger = Logger.getLogger(ResponseListener.class.getName()); + private Broker broker; private Channel channel; private QueueingConsumer consumer; @@ -43,9 +43,10 @@ * @throws Exception */ - protected ResponseListener(Properties env) throws Exception { - this.env = env; + public ResponseListener(Broker broker) throws Exception { + this.broker = broker; + env = broker.getEnvironment(); // Init the hashtable (it's concurrent) - this.results = new Hashtable>(); + results = new Hashtable>(); startRPCQueue(); @@ -107,5 +108,5 @@ private void startRPCQueue() throws Exception { - channel = Broker.getNewChannel(); + channel = broker.getNewChannel(); Map args = null; @@ -128,56 +129,4 @@ /** - * Static function which initializes the ResponseListener - * - * @param env - * @throws Exception - */ - public static void init(Properties env) throws Exception { - if (rListener == null) { - rListener = new ResponseListener(env); - rListener.start(); - } else { - throw new Exception("Cannot init because it already exists"); - } - } - - /** - * Method to retrieve the unique ResponseListener, this function can also - * initialize a ResponseListener using and environment - * - * @param env - * @return unique ResponseListener - * @throws Exception - */ - public static ResponseListener getRequestListener(Properties env) throws Exception { - if (rListener == null) { - rListener = new ResponseListener(env); - rListener.start(); - } else { - // TODO: create a new exception to indicate that a response listener - // cannot be init - throw new Exception("Cannot init because it already exists"); - } - return rListener; - } - - public static boolean isVoid() { - return rListener == null; - } - - /** - * Method to retrieve the unique ResponseListener - * - * @return - * @throws Exception - */ - public static ResponseListener getRequestListener() throws Exception { - if (rListener == null) { - throw new Exception("Request listener not initialized"); - } - return rListener; - } - - /** * * @param key @@ -186,14 +135,4 @@ public boolean containsKey(String key) { return results.containsKey(key); - } - - /** - * This method is used to kill the unique responseListener in the system - * - * @throws Exception - */ - public static void stopResponseListner() throws Exception { - rListener.kill(); - rListener = null; } Index: /trunk/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- /trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 52) +++ /trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 53) @@ -51,6 +51,8 @@ private String uid; private transient String serializerType; + private transient Broker broker; private transient ResponseListener rListener; private transient EventDispatcher dispatcher; + private transient Serializer serializer; // private transient Channel channel; private transient Properties env; @@ -85,12 +87,14 @@ * @throws Exception */ - public Proxymq(String uid, Class clazz, Properties env) throws Exception { + public Proxymq(String uid, Class clazz, Broker broker) throws Exception { this.uid = uid; - this.rListener = ResponseListener.getRequestListener(); - this.dispatcher = EventDispatcher.getDispatcher(); + this.broker = broker; + rListener = broker.getResponseListener(); + dispatcher = broker.getEventDispatcher(); + serializer = broker.getSerializer(); // TODO what is better to use a new channel or to use the same? // this.channel = Broker.getChannel(); - this.env = env; + env = broker.getEnvironment(); // set the serializer type @@ -160,8 +164,8 @@ // Publish the message - byte[] bytesRequest = Serializer.serialize(serializerType, request); + byte[] bytesRequest = serializer.serialize(serializerType, request); // TODO See this // channel.basicPublish(exchange, routingkey, props, bytesRequest); - Broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest); + broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest); // Log.saveLog("Client-Serialize", bytesRequest); } @@ -232,5 +236,5 @@ throw new TimeoutException("Timeout exception time: " + timeout); } - resp = Serializer.deserializeResponse(results.get(corrId), type); + resp = serializer.deserializeResponse(results.get(corrId), type); // Log.saveLog("Client-Deserialize", results.get(corrId)); Index: /trunk/src/main/java/omq/common/broker/Broker.java =================================================================== --- /trunk/src/main/java/omq/common/broker/Broker.java (revision 52) +++ /trunk/src/main/java/omq/common/broker/Broker.java (revision 53) @@ -35,50 +35,40 @@ private static final Logger logger = Logger.getLogger(Broker.class.getName()); - private static Connection connection; - private static Channel channel; - 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-) - private static Map remoteObjs; - - /** - * Initializes a new Broker with the environment called by reference - * - * @param env - * @throws Exception - */ - public static synchronized void initBroker(Properties env) throws Exception { - if (environment == null) { - - // Load log4j configuration - URL log4jResource = Broker.class.getResource("/log4j.xml"); - DOMConfigurator.configure(log4jResource); - - remoteObjs = new HashMap(); - environment = env; - connection = OmqConnectionFactory.getNewConnection(env); - channel = connection.createChannel(); - addFaultTolerance(); - try { - tryConnection(env); - } catch (Exception e) { - channel.close(); - connection.close(); - throw new InitBrokerException("The connection didn't work"); - } - } else { - logger.error("Broker is already started"); - throw new InitBrokerException("Broker is already started"); - } - } - - public static void stopBroker() throws Exception { + private Connection connection; + private Channel channel; + private ResponseListener responseListener; + private EventDispatcher eventDispatcher; + private Serializer serializer; + private boolean clientStarted = false; + private boolean connectionClosed = false; + private Properties environment = null; + private Map remoteObjs; + + public Broker(Properties env) throws Exception { + // Load log4j configuration + URL log4jResource = Broker.class.getResource("/log4j.xml"); + DOMConfigurator.configure(log4jResource); + + remoteObjs = new HashMap(); + serializer = new Serializer(env); + environment = env; + connection = OmqConnectionFactory.getNewConnection(env); + channel = connection.createChannel(); + addFaultTolerance(); + try { + tryConnection(env); + } catch (Exception e) { + channel.close(); + connection.close(); + throw new InitBrokerException("The connection didn't work"); + } + } + + public void stopBroker() throws Exception { logger.warn("Stopping broker"); // Stop the client if (clientStarted) { - ResponseListener.stopResponseListner(); - EventDispatcher.stopEventDispatcher(); + responseListener.kill(); + eventDispatcher.kill(); Proxymq.stopProxy(); } @@ -95,5 +85,5 @@ environment = null; remoteObjs = null; - Serializer.removeSerializers(); + // Serializer.removeSerializers(); } @@ -102,9 +92,9 @@ * @throws Exception */ - public static Connection getConnection() throws Exception { + public Connection getConnection() throws Exception { return connection; } - public static void closeConnection() throws IOException { + public void closeConnection() throws IOException { logger.warn("Clossing connection"); connectionClosed = true; @@ -118,5 +108,5 @@ * @throws Exception */ - public static Channel getChannel() throws Exception { + public Channel getChannel() throws Exception { return channel; } @@ -128,10 +118,10 @@ * @throws IOException */ - public static Channel getNewChannel() throws IOException { + public Channel getNewChannel() throws IOException { return connection.createChannel(); } @SuppressWarnings("unchecked") - public static T lookup(String reference, Class contract) throws RemoteException { + public T lookup(String reference, Class contract) throws RemoteException { try { @@ -142,5 +132,5 @@ if (!Proxymq.containsProxy(reference)) { - Proxymq proxy = new Proxymq(reference, contract, environment); + Proxymq proxy = new Proxymq(reference, contract, this); Class[] array = { contract }; return (T) Proxymq.newProxyInstance(contract.getClassLoader(), array, proxy); @@ -153,7 +143,7 @@ } - public static void bind(String reference, RemoteObject remote) throws RemoteException { + public void bind(String reference, RemoteObject remote) throws RemoteException { try { - remote.startRemoteObject(reference, environment); + remote.startRemoteObject(reference, this); remoteObjs.put(reference, remote); } catch (Exception e) { @@ -162,5 +152,5 @@ } - public static void unbind(String reference) throws RemoteException, IOException { + public void unbind(String reference) throws RemoteException, IOException { if (remoteObjs.containsKey(reference)) { RemoteObject remote = remoteObjs.get(reference); @@ -183,10 +173,10 @@ * @throws Exception */ - private static synchronized void initClient(Properties environment) throws Exception { - if (ResponseListener.isVoid()) { - ResponseListener.init(environment); - } - if (EventDispatcher.isVoid()) { - EventDispatcher.init(environment); + private synchronized void initClient(Properties environment) throws Exception { + if (responseListener == null) { + responseListener = new ResponseListener(this); + } + if (eventDispatcher == null) { + eventDispatcher = new EventDispatcher(this); } } @@ -199,5 +189,5 @@ * @throws SerializerException */ - public static void trigger(Event event) throws IOException, SerializerException { + public void trigger(Event event) throws IOException, SerializerException { String UID = event.getTopic(); EventWrapper wrapper = new EventWrapper(event); @@ -205,5 +195,5 @@ channel.exchangeDeclare(UID, "fanout"); - byte[] bytesResponse = Serializer.serialize(wrapper); + byte[] bytesResponse = serializer.serialize(wrapper); channel.basicPublish(UID, "", null, bytesResponse); @@ -218,5 +208,5 @@ * @throws Exception */ - public static void tryConnection(Properties env) throws Exception { + public void tryConnection(Properties env) throws Exception { Channel channel = connection.createChannel(); String message = "ping"; @@ -252,5 +242,5 @@ * have the listener. */ - private static void addFaultTolerance() { + private void addFaultTolerance() { connection.addShutdownListener(new ShutdownListener() { @Override @@ -287,7 +277,18 @@ } - public static Properties getEnvironment() { + public Properties getEnvironment() { return environment; } + public ResponseListener getResponseListener() { + return responseListener; + } + + public EventDispatcher getEventDispatcher() { + return eventDispatcher; + } + + public Serializer getSerializer() { + return serializer; + } } Index: /trunk/src/main/java/omq/common/event/EventDispatcher.java =================================================================== --- /trunk/src/main/java/omq/common/event/EventDispatcher.java (revision 52) +++ /trunk/src/main/java/omq/common/event/EventDispatcher.java (revision 53) @@ -29,6 +29,7 @@ public class EventDispatcher extends Thread { private static final Logger logger = Logger.getLogger(EventDispatcher.class.getName()); - private static EventDispatcher dispatcher; + private Broker broker; + private Serializer serializer; private Map> listeners; private Channel channel; @@ -37,6 +38,8 @@ private boolean killed = false; - private EventDispatcher(Properties env) throws Exception { - this.env = env; + public EventDispatcher(Broker broker) throws Exception { + this.broker = broker; + env = broker.getEnvironment(); + serializer = broker.getSerializer(); // Declare the listeners map @@ -44,10 +47,9 @@ startEventQueue(); - } private void startEventQueue() throws Exception { // Get a new connection and a new channel - channel = Broker.getNewChannel(); + channel = broker.getNewChannel(); String event_queue = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE); @@ -60,35 +62,10 @@ } - public static void init(Properties env) throws Exception { - if (dispatcher == null) { - dispatcher = new EventDispatcher(env); - dispatcher.start(); - } else { - throw new Exception("Already initialized"); - } - } - - public static void stopEventDispatcher() throws Exception { + public void kill() throws Exception { logger.warn("Stopping EventDispatcher"); - dispatcher.setListeners(null); - dispatcher.killed = true; - dispatcher.interrupt(); - dispatcher.channel.close(); - dispatcher = null; - } - - public static EventDispatcher getDispatcher(Properties env) throws Exception { - if (dispatcher == null) { - dispatcher = new EventDispatcher(env); - dispatcher.start(); - } - return dispatcher; - } - - public static EventDispatcher getDispatcher() throws Exception { - if (dispatcher == null) { - throw new Exception("EventDispatcher not initialized"); - } - return dispatcher; + setListeners(null); + killed = true; + interrupt(); + channel.close(); } @@ -104,5 +81,5 @@ // Get the event - event = Serializer.deserializeEvent(delivery.getBody()); + event = serializer.deserializeEvent(delivery.getBody()); logger.info("Event received -> Topic: " + event.getTopic() + "CorrId: " + event.getCorrId()); @@ -210,7 +187,3 @@ } - public static boolean isVoid() { - return dispatcher == null; - } - } Index: unk/src/main/java/omq/common/util/Log.java =================================================================== --- /trunk/src/main/java/omq/common/util/Log.java (revision 52) +++ (revision ) @@ -1,80 +1,0 @@ -package omq.common.util; - -import java.io.File; -import java.io.FileWriter; -import java.io.IOException; -import java.util.Date; -import java.util.Properties; - -import omq.common.broker.Broker; - -public class Log { - - public static void saveLog(String processName, byte[] bytesResponse) throws IOException { - - Properties env = Broker.getEnvironment(); - - String debugPath = env.getProperty(ParameterQueue.DEBUGFILE, ""); - if (debugPath.length() > 0) { - long timeNow = (new Date()).getTime(); - - File outputFolder = new File(debugPath + File.separator); - outputFolder.mkdirs(); - - // 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"); - } - 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 { - - 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"); - } - 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 52) +++ /trunk/src/main/java/omq/common/util/Serializer.java (revision 53) @@ -4,7 +4,4 @@ import java.util.Properties; -import org.apache.log4j.Logger; - -import omq.common.broker.Broker; import omq.common.event.Event; import omq.common.message.Request; @@ -23,26 +20,31 @@ */ public class Serializer { - private static final Logger logger = Logger.getLogger(Serializer.class.getName()); - public static String kryo = "kryo"; - public static String java = "java"; - public static String gson = "gson"; + // private static final Logger logger = + // Logger.getLogger(Serializer.class.getName()); + public static final String kryo = "kryo"; + public static final String java = "java"; + public static final String gson = "gson"; // Client serializer - public static ISerializer serializer; + public ISerializer serializer; // Server serializers - private static ISerializer kryoSerializer; - private static ISerializer javaSerializer; - private static ISerializer gsonSerializer; + private ISerializer kryoSerializer; + private ISerializer javaSerializer; + private ISerializer gsonSerializer; - private static Boolean getEnableCompression() { - Properties env = Broker.getEnvironment(); + private Properties env; + + public Serializer(Properties env) { + this.env = env; + } + + private Boolean getEnableCompression() { return Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false")); } - public static ISerializer getInstance() throws SerializerException { + public ISerializer getInstance() throws SerializerException { if (serializer == null) { try { - Properties env = Broker.getEnvironment(); String className = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.java); @@ -60,5 +62,5 @@ } - public static ISerializer getInstance(String type) throws SerializerException { + public ISerializer getInstance(String type) throws SerializerException { if (kryo.equals(type)) { if (kryoSerializer == null) { @@ -79,5 +81,5 @@ } - public static byte[] serialize(String type, Object obj) throws SerializerException { + public byte[] serialize(String type, Object obj) throws SerializerException { ISerializer instance = getInstance(type); @@ -96,5 +98,5 @@ // TODO: remove this function and think about the event serialization - public static byte[] serialize(Object obj) throws SerializerException { + public byte[] serialize(Object obj) throws SerializerException { ISerializer instance = getInstance(); @@ -112,5 +114,5 @@ } - public static Request deserializeRequest(String type, byte[] bytes, RemoteObject obj) throws SerializerException { + public Request deserializeRequest(String type, byte[] bytes, RemoteObject obj) throws SerializerException { ISerializer instance = getInstance(type); @@ -128,5 +130,5 @@ } - public static Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { + public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { ISerializer instance = getInstance(); @@ -144,5 +146,5 @@ } - public static Event deserializeEvent(byte[] bytes) throws SerializerException { + public Event deserializeEvent(byte[] bytes) throws SerializerException { ISerializer instance = getInstance(); @@ -160,10 +162,10 @@ } - public static void removeSerializers() { - logger.warn("Removing serializers"); - serializer = null; - kryoSerializer = null; - javaSerializer = null; - gsonSerializer = null; - } + // public static void removeSerializers() { + // logger.warn("Removing serializers"); + // 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 52) +++ /trunk/src/main/java/omq/server/InvocationThread.java (revision 53) @@ -23,10 +23,12 @@ private static final Logger logger = Logger.getLogger(InvocationThread.class.getName()); private RemoteObject obj; + private transient Serializer serializer; private BlockingQueue deliveryQueue; private boolean killed = false; - public InvocationThread(RemoteObject obj, BlockingQueue deliveryQueue) { + public InvocationThread(RemoteObject obj, BlockingQueue deliveryQueue, Serializer serializer) { this.obj = obj; this.deliveryQueue = deliveryQueue; + this.serializer = serializer; } @@ -41,5 +43,5 @@ // Deserialize the json - Request request = Serializer.deserializeRequest(serializerType, delivery.getBody(), obj); + Request request = serializer.deserializeRequest(serializerType, delivery.getBody(), obj); // Log.saveLog("Server-Deserialize", delivery.getBody()); @@ -73,5 +75,5 @@ BasicProperties replyProps = new BasicProperties.Builder().appId(obj.getRef()).correlationId(props.getCorrelationId()).build(); - byte[] bytesResponse = Serializer.serialize(serializerType, resp); + byte[] bytesResponse = serializer.serialize(serializerType, resp); channel.basicPublish("", props.getReplyTo(), replyProps, bytesResponse); Index: /trunk/src/main/java/omq/server/RemoteObject.java =================================================================== --- /trunk/src/main/java/omq/server/RemoteObject.java (revision 52) +++ /trunk/src/main/java/omq/server/RemoteObject.java (revision 53) @@ -39,4 +39,6 @@ private String UID; private Properties env; + private transient Broker broker; + private transient Serializer serializer; private transient RemoteWrapper remoteWrapper; private transient Map>> params; @@ -60,7 +62,9 @@ } - public void startRemoteObject(String reference, Properties env) throws Exception { - this.UID = reference; - this.env = env; + public void startRemoteObject(String reference, Broker broker) throws Exception { + this.broker = broker; + UID = reference; + env = broker.getEnvironment(); + serializer = broker.getSerializer(); params = new HashMap>>(); @@ -75,5 +79,5 @@ // Get num threads to use int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1")); - remoteWrapper = new RemoteWrapper(this, numThreads); + remoteWrapper = new RemoteWrapper(this, numThreads, broker.getSerializer()); startQueues(); @@ -130,5 +134,5 @@ EventWrapper wrapper = new EventWrapper(event); channel.exchangeDeclare(UID, "fanout"); - channel.basicPublish(UID, "", null, Serializer.serialize(wrapper)); + channel.basicPublish(UID, "", null, serializer.serialize(wrapper)); } @@ -221,5 +225,5 @@ // Start channel - channel = Broker.getNewChannel(); + channel = broker.getNewChannel(); // Declares and bindings Index: /trunk/src/main/java/omq/server/RemoteWrapper.java =================================================================== --- /trunk/src/main/java/omq/server/RemoteWrapper.java (revision 52) +++ /trunk/src/main/java/omq/server/RemoteWrapper.java (revision 53) @@ -4,4 +4,6 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingDeque; + +import omq.common.util.Serializer; import org.apache.log4j.Logger; @@ -23,5 +25,5 @@ private BlockingQueue deliveryQueue; - public RemoteWrapper(RemoteObject obj, int numThreads) { + public RemoteWrapper(RemoteObject obj, int numThreads, Serializer serializer) { this.obj = obj; this.numThreads = numThreads; @@ -32,5 +34,5 @@ for (int i = 0; i < numThreads; i++) { - InvocationThread thread = new InvocationThread(obj, deliveryQueue); + InvocationThread thread = new InvocationThread(obj, deliveryQueue, serializer); invocationList.add(thread); thread.start(); Index: /trunk/target/maven-archiver/pom.properties =================================================================== --- /trunk/target/maven-archiver/pom.properties (revision 53) +++ /trunk/target/maven-archiver/pom.properties (revision 53) @@ -0,0 +1,5 @@ +#Generated by Maven +#Wed Jun 19 16:52:13 CEST 2013 +version=0.5.1 +groupId=objectmq +artifactId=objectmq