Index: trunk/objectmq/src/omq/Remote.java =================================================================== --- trunk/objectmq/src/omq/Remote.java (revision 13) +++ trunk/objectmq/src/omq/Remote.java (revision 14) @@ -1,5 +1,9 @@ package omq; +import java.io.IOException; import java.io.Serializable; + +import omq.common.event.Event; +import omq.exception.SerializerException; /** @@ -16,3 +20,5 @@ */ public String getRef(); + + public void notifyEvent(Event event) throws IOException, SerializerException; } Index: trunk/objectmq/src/omq/client/proxy/Proxymq.java =================================================================== --- trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 13) +++ trunk/objectmq/src/omq/client/proxy/Proxymq.java (revision 14) @@ -14,4 +14,5 @@ import omq.client.annotation.SyncMethod; import omq.client.remote.response.ResponseListener; +import omq.common.event.Event; import omq.common.message.Request; import omq.common.message.Response; @@ -267,3 +268,8 @@ } + @Override + public void notifyEvent(Event event) throws IOException, SerializerException { + + } + } Index: trunk/objectmq/src/omq/common/broker/Broker.java =================================================================== --- trunk/objectmq/src/omq/common/broker/Broker.java (revision 13) +++ trunk/objectmq/src/omq/common/broker/Broker.java (revision 14) @@ -1,4 +1,5 @@ package omq.common.broker; +import java.io.IOException; import java.util.Properties; @@ -6,6 +7,6 @@ import omq.client.proxy.Proxymq; import omq.client.remote.response.ResponseListener; -import omq.common.remote.OmqConnectionFactory; import omq.common.util.Environment; +import omq.common.util.OmqConnectionFactory; import omq.exception.EnvironmentException; import omq.exception.RemoteException; @@ -24,5 +25,5 @@ } catch (EnvironmentException ex) { // environment not set. Environment.setEnvironment(env); - connection = OmqConnectionFactory.getConnection(env); + connection = OmqConnectionFactory.getNewConnection(env); channel = connection.createChannel(); } @@ -36,4 +37,8 @@ public static Channel getChannel() throws Exception { return channel; + } + + public static Channel getNewChannel() throws IOException { + return connection.createChannel(); } Index: trunk/objectmq/src/omq/common/event/Event.java =================================================================== --- trunk/objectmq/src/omq/common/event/Event.java (revision 14) +++ trunk/objectmq/src/omq/common/event/Event.java (revision 14) @@ -0,0 +1,40 @@ +package omq.common.event; + +import java.io.Serializable; + +/** + * + * @author Sergi Toda + * + */ +public abstract class Event implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 1L; + + private String corrId; + private String topic; + + public Event(String corrId, String topic) { + this.corrId = corrId; + this.topic = topic; + } + + public String getCorrId() { + return corrId; + } + + public void setCorrId(String corrId) { + this.corrId = corrId; + } + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } +} Index: trunk/objectmq/src/omq/common/event/EventDispatcher.java =================================================================== --- trunk/objectmq/src/omq/common/event/EventDispatcher.java (revision 14) +++ trunk/objectmq/src/omq/common/event/EventDispatcher.java (revision 14) @@ -0,0 +1,163 @@ +package omq.common.event; + +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Vector; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.ConsumerCancelledException; +import com.rabbitmq.client.QueueingConsumer; +import com.rabbitmq.client.QueueingConsumer.Delivery; +import com.rabbitmq.client.ShutdownSignalException; + +/** + * This class dispatches the events received in the client side and stores them + * into the different listeners that could exists among the different proxies + * generated + * + * @author Sergi Toda + * + */ +public class EventDispatcher extends Thread { + private static EventDispatcher dispatcher; + + private Map> listeners; + private Channel channel; + private QueueingConsumer consumer; + private Properties env; + private boolean killed = false; + + private EventDispatcher(Properties env) throws Exception { + this.env = env; + + // Declare the listeners map + listeners = new HashMap>(); + + // Get a new connection and a new channel + channel = Broker.getNewChannel(); + + String event_queue = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE); + channel.queueDeclare(event_queue, false, false, false, null); + + // Declare a new consumer + consumer = new QueueingConsumer(channel); + channel.basicConsume(event_queue, true, consumer); + } + + 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 { + 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; + } + + @Override + public void run() { + Delivery delivery; + Event event; + + while (!killed) { + try { + // Get the delivery + delivery = consumer.nextDelivery(); + + // Get the event + event = Serializer.deserializeEvent(delivery.getBody()); + + // Dispatch it + dispatch(event.getTopic(), event); + } catch (InterruptedException i) { + System.out.println("InterruptedException e: " + i); + } catch (ShutdownSignalException e) { + System.out.println("ShutdownSignalException e: " + e); + } catch (ConsumerCancelledException e) { + System.out.println("ConsumerCancelledException e: " + e); + } catch (Exception e) { + System.out.println("Exception e: " + e); + } + } + } + + public int addListener(EventListener e) throws Exception { + Vector vListeners = listeners.get(e.getTopic()); + if (vListeners == null) { + vListeners = new Vector(); + + String queueName = env.getProperty(ParameterQueue.EVENT_REPLY_QUEUE); + String reference = e.getTopic(); + channel.exchangeDeclare(reference, "fanout"); + channel.queueBind(queueName, reference, ""); + } + vListeners.add(e); + listeners.put(e.getTopic(), vListeners); + + return vListeners.size(); + } + + public int removeListener(EventListener e) { + Vector vListeners = listeners.get(e.getTopic()); + if (vListeners != null) { + // TODO: removeListener -> remove(e) override equals? + vListeners.remove(e); + } + + return vListeners.size(); + } + + /** + * This method dispatches the events. Every time an event is received, this + * method is launched. This method creates a new thread and executes the + * notifyEvent function of the listeners associated to this event + * + * @param topic + * @param event + */ + public void dispatch(String topic, final Event event) { + for (final EventListener listener : listeners.get(topic)) { + new Thread() { + public void run() { + listener.notifyEvent(event); + } + }.start(); + } + } + + public Map> getListeners() { + return listeners; + } + + public void setListeners(Map> listeners) { + this.listeners = listeners; + } + +} Index: trunk/objectmq/src/omq/common/event/EventListener.java =================================================================== --- trunk/objectmq/src/omq/common/event/EventListener.java (revision 14) +++ trunk/objectmq/src/omq/common/event/EventListener.java (revision 14) @@ -0,0 +1,43 @@ +package omq.common.event; + +/** + * Abstract class that enables to create new EventListeners. The eventListeners + * are done to execute the notifyEvent function. + * + * @author Sergi Toda + * + */ +public abstract class EventListener { + private String topic; + + public EventListener() { + topic = null; + } + + /** + * Constructor. This constructor uses a String to indicate manually which + * event we want to receive + * + * @param topic + */ + public EventListener(String topic) { + this.topic = topic; + } + + /** + * Whenever this listener it's notified of an event, will execute this + * function + * + * @param event + */ + public abstract void notifyEvent(Event event); + + public void setTopic(String topic) { + this.topic = topic; + } + + public String getTopic() { + return topic; + } + +} Index: trunk/objectmq/src/omq/common/remote/OmqConnectionFactory.java =================================================================== --- trunk/objectmq/src/omq/common/remote/OmqConnectionFactory.java (revision 13) +++ (revision ) @@ -1,42 +1,0 @@ -package omq.common.remote; - -import java.io.IOException; -import java.security.KeyManagementException; -import java.security.NoSuchAlgorithmException; -import java.util.Properties; - -import omq.common.util.ParameterQueue; - - -import com.rabbitmq.client.Connection; -import com.rabbitmq.client.ConnectionFactory; - -/** - * - * @author Sergi Toda - * - */ -public class OmqConnectionFactory { - public static Connection getConnection(Properties env) throws IOException, KeyManagementException, NoSuchAlgorithmException { - // Get login info of rabbitmq - String username = env.getProperty(ParameterQueue.USER_NAME); - String password = env.getProperty(ParameterQueue.USER_PASS); - - // Get host info of rabbimq (where it is) - String host = env.getProperty(ParameterQueue.SERVER_HOST); - int port = Integer.parseInt(env.getProperty(ParameterQueue.SERVER_PORT)); - - boolean ssl = Boolean.parseBoolean(env.getProperty(ParameterQueue.ENABLE_SSL)); - - // Start a new connection and channel - ConnectionFactory factory = new ConnectionFactory(); - factory.setUsername(username); - factory.setPassword(password); - factory.setHost(host); - factory.setPort(port); - if (ssl) { - factory.useSslProtocol(); - } - return factory.newConnection(); - } -} Index: trunk/objectmq/src/omq/common/remote/RemoteListener.java =================================================================== --- trunk/objectmq/src/omq/common/remote/RemoteListener.java (revision 13) +++ trunk/objectmq/src/omq/common/remote/RemoteListener.java (revision 14) @@ -4,4 +4,6 @@ import java.io.IOException; import java.util.Properties; + +import omq.common.util.OmqConnectionFactory; import com.rabbitmq.client.Channel; @@ -14,4 +16,5 @@ * */ +//TODO aquesta classe es pot eliminar public abstract class RemoteListener extends Thread { private static String defaultXml = "eventListener.xml"; @@ -37,5 +40,5 @@ private void startConnection(Properties env) throws Exception { - connection = OmqConnectionFactory.getConnection(env); + connection = OmqConnectionFactory.getNewConnection(env); channel = connection.createChannel(); } Index: trunk/objectmq/src/omq/common/util/OmqConnectionFactory.java =================================================================== --- trunk/objectmq/src/omq/common/util/OmqConnectionFactory.java (revision 14) +++ trunk/objectmq/src/omq/common/util/OmqConnectionFactory.java (revision 14) @@ -0,0 +1,52 @@ +package omq.common.util; + +import java.io.IOException; +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.util.Properties; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + +/** + * + * @author Sergi Toda + * + */ +public class OmqConnectionFactory { + private static Connection connection; + + public static void init(Properties env) throws KeyManagementException, NoSuchAlgorithmException, IOException { + if (connection == null) { + connection = getNewConnection(env); + } + } + + public static Connection getNewConnection(Properties env) throws IOException, KeyManagementException, NoSuchAlgorithmException { + // Get login info of rabbitmq + String username = env.getProperty(ParameterQueue.USER_NAME); + String password = env.getProperty(ParameterQueue.USER_PASS); + + // Get host info of rabbimq (where it is) + String host = env.getProperty(ParameterQueue.SERVER_HOST); + int port = Integer.parseInt(env.getProperty(ParameterQueue.SERVER_PORT)); + + boolean ssl = Boolean.parseBoolean(env.getProperty(ParameterQueue.ENABLE_SSL)); + + // Start a new connection and channel + ConnectionFactory factory = new ConnectionFactory(); + factory.setUsername(username); + factory.setPassword(password); + factory.setHost(host); + factory.setPort(port); + if (ssl) { + factory.useSslProtocol(); + } + return factory.newConnection(); + } + + public static Channel getNewChannel() throws IOException { + return connection.createChannel(); + } +} Index: trunk/objectmq/src/omq/common/util/Serializer.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializer.java (revision 13) +++ trunk/objectmq/src/omq/common/util/Serializer.java (revision 14) @@ -4,4 +4,5 @@ import java.util.Properties; +import omq.common.event.Event; import omq.common.message.Request; import omq.common.message.Response; @@ -19,26 +20,27 @@ public static ISerializer serializer; - private static Boolean getEnableCompression(){ + private static Boolean getEnableCompression() { Boolean enableCompression = false; try { Properties env = Environment.getEnvironment(); enableCompression = Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false")); - } catch (EnvironmentException e) { } - + } catch (EnvironmentException e) { + } + return enableCompression; } - - public static ISerializer getInstance() throws SerializerException { + + public static ISerializer getInstance() throws SerializerException { if (serializer == null) { - try{ - Properties env = Environment.getEnvironment(); + 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){ + } catch (Exception ex) { throw new SerializerException(ex.getMessage(), ex); } } - + return serializer; } @@ -46,7 +48,7 @@ public static byte[] serialize(Object obj) throws SerializerException { ISerializer instance = getInstance(); - - Boolean enableCompression = getEnableCompression(); - if(enableCompression){ + + Boolean enableCompression = getEnableCompression(); + if (enableCompression) { byte[] objSerialized = instance.serialize(obj); try { @@ -54,15 +56,15 @@ } catch (IOException e) { throw new SerializerException(e.getMessage(), e); - } - } else{ + } + } else { return instance.serialize(obj); - } + } } public static Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException { ISerializer instance = getInstance(); - + Boolean enableCompression = getEnableCompression(); - if(enableCompression){ + if (enableCompression) { try { byte[] unZippedBytes = Zipper.unzip(bytes); @@ -71,14 +73,14 @@ throw new SerializerException(e.getMessage(), e); } - } else{ + } else { return instance.deserializeRequest(bytes, obj); - } + } } public static Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { ISerializer instance = getInstance(); - + Boolean enableCompression = getEnableCompression(); - if(enableCompression){ + if (enableCompression) { try { byte[] unZippedBytes = Zipper.unzip(bytes); @@ -87,7 +89,23 @@ throw new SerializerException(e.getMessage(), e); } - } else{ + } else { return instance.deserializeResponse(bytes, type); - } + } + } + + public static Event deserializeEvent(byte[] bytes) throws SerializerException { + ISerializer instance = getInstance(); + + Boolean enableCompression = getEnableCompression(); + if (enableCompression) { + try { + byte[] unZippedBytes = Zipper.unzip(bytes); + return instance.deserializeEvent(unZippedBytes); + } catch (IOException e) { + throw new SerializerException(e.getMessage(), e); + } + } else { + return instance.deserializeEvent(bytes); + } } } Index: trunk/objectmq/src/omq/common/util/Serializers/GsonImp.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializers/GsonImp.java (revision 13) +++ trunk/objectmq/src/omq/common/util/Serializers/GsonImp.java (revision 14) @@ -3,4 +3,5 @@ import java.util.List; +import omq.common.event.Event; import omq.common.message.Request; import omq.common.message.Response; @@ -71,3 +72,11 @@ } + @Override + public Event deserializeEvent(byte[] unZippedBytes) throws SerializerException { + // TODO deserializeEvent class ¿? + return null; + } + + + } Index: trunk/objectmq/src/omq/common/util/Serializers/ISerializer.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializers/ISerializer.java (revision 13) +++ trunk/objectmq/src/omq/common/util/Serializers/ISerializer.java (revision 14) @@ -1,4 +1,5 @@ package omq.common.util.Serializers; +import omq.common.event.Event; import omq.common.message.Request; import omq.common.message.Response; @@ -17,3 +18,5 @@ public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException; + + public Event deserializeEvent(byte[] bytes) throws SerializerException; } Index: trunk/objectmq/src/omq/common/util/Serializers/JavaImp.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializers/JavaImp.java (revision 13) +++ trunk/objectmq/src/omq/common/util/Serializers/JavaImp.java (revision 14) @@ -6,4 +6,5 @@ import java.io.ObjectOutputStream; +import omq.common.event.Event; import omq.common.message.Request; import omq.common.message.Response; @@ -43,31 +44,28 @@ @Override public Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException { + return (Request) deserliazeObject(bytes); + } + + @Override + public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { + return (Response) deserliazeObject(bytes); + } + + @Override + public Event deserializeEvent(byte[] bytes) throws SerializerException { + return (Event) deserliazeObject(bytes); + } + + private Object deserliazeObject(byte[] bytes) throws SerializerException { try { ByteArrayInputStream input = new ByteArrayInputStream(bytes); ObjectInputStream objInput = new ObjectInputStream(input); - Object request = objInput.readObject(); + Object obj = objInput.readObject(); objInput.close(); input.close(); - return (Request) request; - } catch (Exception e) { - throw new SerializerException("Deserialize -> " + e.getMessage(), e); - } - } - - @Override - public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { - try { - ByteArrayInputStream input = new ByteArrayInputStream(bytes); - ObjectInputStream objInput = new ObjectInputStream(input); - - Object response = objInput.readObject(); - - objInput.close(); - input.close(); - - return (Response) response; + return obj; } catch (Exception e) { throw new SerializerException("Deserialize -> " + e.getMessage(), e); Index: trunk/objectmq/src/omq/common/util/Serializers/KryoImp.java =================================================================== --- trunk/objectmq/src/omq/common/util/Serializers/KryoImp.java (revision 13) +++ trunk/objectmq/src/omq/common/util/Serializers/KryoImp.java (revision 14) @@ -7,4 +7,5 @@ import com.esotericsoftware.kryo.io.Output; +import omq.common.event.Event; import omq.common.message.Request; import omq.common.message.Response; @@ -43,23 +44,24 @@ @Override public Request deserializeRequest(byte[] bytes, RemoteObject obj) throws SerializerException { - try { - Input input = new Input(bytes); - Request request = kryo.readObject(input, Request.class); - - input.close(); - return request; - } catch (Exception e) { - throw new SerializerException("Deserialize -> " + e.getMessage(), e); - } + return (Request) deserializeObject(bytes, Request.class); } @Override public Response deserializeResponse(byte[] bytes, Class type) throws SerializerException { + return (Response) deserializeObject(bytes, Response.class); + } + + @Override + public Event deserializeEvent(byte[] bytes) throws SerializerException { + return (Event) deserializeObject(bytes, Event.class); + } + + private Object deserializeObject(byte[] bytes, Class type) throws SerializerException { try { Input input = new Input(bytes); - Response response = kryo.readObject(input, Response.class); + Object obj = kryo.readObject(input, type); input.close(); - return response; + return obj; } catch (Exception e) { throw new SerializerException("Deserialize -> " + e.getMessage(), e); Index: trunk/objectmq/src/omq/server/remote/request/RemoteObject.java =================================================================== --- trunk/objectmq/src/omq/server/remote/request/RemoteObject.java (revision 13) +++ trunk/objectmq/src/omq/server/remote/request/RemoteObject.java (revision 14) @@ -11,9 +11,10 @@ import omq.Remote; import omq.common.broker.Broker; +import omq.common.event.Event; import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; import omq.exception.SerializerException; import com.rabbitmq.client.Channel; -import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConsumerCancelledException; import com.rabbitmq.client.QueueingConsumer; @@ -33,5 +34,4 @@ private transient RemoteWrapper remoteWrapper; private transient Map>> params; - private transient Connection connection; private transient Channel channel; private transient QueueingConsumer consumer; @@ -51,4 +51,9 @@ public RemoteObject() { + } + + public void start(String reference, Properties env) throws Exception { + this.UID = reference; + params = new HashMap>>(); for (Method m : this.getClass().getMethods()) { @@ -59,8 +64,4 @@ params.put(m.getName(), list); } - } - - public void start(String reference, Properties env) throws Exception { - this.UID = reference; // Get num threads to use @@ -73,7 +74,6 @@ String routingKey = UID; - // Start connection and channel - connection = Broker.getConnection(); - channel = connection.createChannel(); + // Start channel + channel = Broker.getNewChannel(); // Declares and bindings @@ -118,9 +118,15 @@ } + @Override + public void notifyEvent(Event event) throws IOException, SerializerException { + event.setTopic(UID); + channel.exchangeDeclare(UID, "fanout"); + channel.basicPublish(UID, "", null, Serializer.serialize(event)); + } + public void kill() throws IOException { interrupt(); killed = true; channel.close(); - connection.close(); remoteWrapper.stopRemoteWrapper(); }