Index: trunk/src/main/java/omq/client/proxy/MultiProxymq.java =================================================================== --- trunk/src/main/java/omq/client/proxy/MultiProxymq.java (revision 76) +++ trunk/src/main/java/omq/client/proxy/MultiProxymq.java (revision 77) @@ -41,5 +41,5 @@ replyQueueName = env.getProperty(ParameterQueue.RPC_REPLY_QUEUE); exchange = multi + uid; - serializerType = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA); + serializerType = env.getProperty(ParameterQueue.PROXY_SERIALIZER, Serializer.JAVA); } Index: trunk/src/main/java/omq/client/proxy/Proxymq.java =================================================================== --- trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 76) +++ trunk/src/main/java/omq/client/proxy/Proxymq.java (revision 77) @@ -51,4 +51,5 @@ private transient Serializer serializer; private transient Properties env; + private transient Integer deliveryMode = null; private transient Map results; @@ -94,5 +95,8 @@ // set the serializer type - serializerType = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA); + serializerType = env.getProperty(ParameterQueue.PROXY_SERIALIZER, Serializer.JAVA); + if (env.getProperty(ParameterQueue.DELIVERY_MODE) != null) { + deliveryMode = Integer.parseInt(env.getProperty(ParameterQueue.DELIVERY_MODE)); + } // Create a new hashmap and registry it in rListener @@ -143,11 +147,13 @@ // Add the correlation ID and create a replyTo property - BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).type(serializerType).build(); + BasicProperties props = new BasicProperties.Builder().appId(uid).correlationId(corrId).replyTo(replyQueueName).type(serializerType) + .deliveryMode(deliveryMode).build(); // Publish the message byte[] bytesRequest = serializer.serialize(serializerType, request); broker.getChannel().basicPublish(exchange, routingkey, props, bytesRequest); - logger.debug("Proxymq: " + uid + " invokes " + request.getMethod() + ", corrID" + corrId + ", exchange: " + exchange + ", replyQueue: " - + replyQueueName + ", serializerType: " + serializerType + ", multi call: " + request.isMulti() + ", async call: " + request.isAsync()); + logger.debug("Proxymq: " + uid + " invokes '" + request.getMethod() + "' , corrID: " + corrId + ", exchange: " + exchange + ", replyQueue: " + + replyQueueName + ", serializerType: " + serializerType + ", multi call: " + request.isMulti() + ", async call: " + request.isAsync() + + ", delivery mode: " + deliveryMode); } Index: trunk/src/main/java/omq/common/util/OmqConnectionFactory.java =================================================================== --- trunk/src/main/java/omq/common/util/OmqConnectionFactory.java (revision 76) +++ trunk/src/main/java/omq/common/util/OmqConnectionFactory.java (revision 77) @@ -56,6 +56,6 @@ // Get host info of rabbimq (where it is) - String host = env.getProperty(ParameterQueue.SERVER_HOST); - int port = Integer.parseInt(env.getProperty(ParameterQueue.SERVER_PORT)); + String host = env.getProperty(ParameterQueue.RABBIT_HOST); + int port = Integer.parseInt(env.getProperty(ParameterQueue.RABBIT_PORT)); boolean ssl = Boolean.parseBoolean(env.getProperty(ParameterQueue.ENABLE_SSL)); Index: trunk/src/main/java/omq/common/util/ParameterQueue.java =================================================================== --- trunk/src/main/java/omq/common/util/ParameterQueue.java (revision 76) +++ trunk/src/main/java/omq/common/util/ParameterQueue.java (revision 77) @@ -12,20 +12,23 @@ */ - public static String SERIALIZER_NAME = "omq.serializer"; + /** + * Set the proxy's serializer method + */ + public static String PROXY_SERIALIZER = "omq.serializer"; /** * Set whether the messages must be compressed or not */ - public static String ENABLECOMPRESSION = "omq.compression"; + public static String ENABLE_COMPRESSION = "omq.compression"; /** * Set the ip where the rabbitmq server is. */ - public static String SERVER_HOST = "omq.host"; + public static String RABBIT_HOST = "omq.host"; /** * Set the port that rabbitmq uses. */ - public static String SERVER_PORT = "omq.port"; + public static String RABBIT_PORT = "omq.port"; /** @@ -51,10 +54,4 @@ /** - * Set the clients event queue. Every client must have a different queue - * name. - */ - public static String EVENT_REPLY_QUEUE = "omq.reply_queue_event"; - - /** * Set if the queues must be durable. The queues won't be lost when rabbitmq * crashes if DURABLE_QUEUES is set trues. @@ -72,22 +69,44 @@ public static String MESSAGE_TTL_IN_QUEUES = "omq.message_ttl_queue"; - // TODO persisten messages? the messages will be saved in the disk if this + // TODO persistent messages? the messages will be saved in the disk if this // flag is set true + /** + * Set if the system will use ssl + */ public static String ENABLE_SSL = "omq.enable_ssl"; - public static String DEBUGFILE = "omq.debug_file"; + /** + * Set how many time we have to wait to retry the connection with the server + * when this goes down + */ public static String RETRY_TIME_CONNECTION = "omq.retry_connection"; - /* - * Values + /** + * Set how many threads will be created to invoke remote methods */ - - // Change this!!! - public static String RPC_TYPE = "direct"; - public static String NUM_THREADS = "omq.num_threads"; - public static String REGISTRY_NAME = "REGISTRY"; + /** + * Set the specific name of a multi queue in a specific object + */ + public static String MULTI_QUEUE_NAME = "omq.multi_queue_name"; + + /** + * Set if server will delete a queue when is no longer in use + */ + public static String AUTO_DELETE_QUEUE = "omq.auto_delete"; + + /** + * Set if we are declaring an exclusive queue (restricted to this + * connection) + */ + public static String EXCLUSIVE_QUEUE = "omq.exclusive_queue"; + + /** + * Set 1 to indicate the message will be nonpersistent and 2 to indicate it + * will be persistent + */ + public static String DELIVERY_MODE = "omq.delivery_mode"; /** Index: trunk/src/main/java/omq/common/util/Serializer.java =================================================================== --- trunk/src/main/java/omq/common/util/Serializer.java (revision 76) +++ trunk/src/main/java/omq/common/util/Serializer.java (revision 77) @@ -40,5 +40,5 @@ private Boolean getEnableCompression() { - return Boolean.valueOf(env.getProperty(ParameterQueue.ENABLECOMPRESSION, "false")); + return Boolean.valueOf(env.getProperty(ParameterQueue.ENABLE_COMPRESSION, "false")); } @@ -46,5 +46,5 @@ if (serializer == null) { try { - String className = env.getProperty(ParameterQueue.SERIALIZER_NAME, Serializer.JAVA); + String className = env.getProperty(ParameterQueue.PROXY_SERIALIZER, Serializer.JAVA); if (className == null || className.isEmpty()) { @@ -98,5 +98,4 @@ } - // TODO: remove this function and think about the event serialization public byte[] serialize(Object obj) throws SerializerException { ISerializer instance = getInstance(); Index: trunk/src/main/java/omq/server/RemoteObject.java =================================================================== --- trunk/src/main/java/omq/server/RemoteObject.java (revision 76) +++ trunk/src/main/java/omq/server/RemoteObject.java (revision 77) @@ -218,8 +218,14 @@ String queue = UID; String routingKey = UID; + // Multi info String multiExchange = multi + UID; + if (env.getProperty(ParameterQueue.MULTI_QUEUE_NAME) != null) { + multiQueue = env.getProperty(ParameterQueue.MULTI_QUEUE_NAME); + } boolean durable = Boolean.parseBoolean(env.getProperty(ParameterQueue.DURABLE_QUEUES, "false")); + boolean exclusive = Boolean.parseBoolean(env.getProperty(ParameterQueue.EXCLUSIVE_QUEUE, "false")); + boolean autoDelete = Boolean.parseBoolean(env.getProperty(ParameterQueue.AUTO_DELETE_QUEUE, "false")); // Start channel @@ -227,12 +233,14 @@ // Declares and bindings - logger.info("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable); + logger.info("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable + ", Exclusive: " + + exclusive + ", AutoDelete: " + autoDelete); channel.exchangeDeclare(exchange, "direct"); - channel.queueDeclare(queue, durable, false, false, null); + channel.queueDeclare(queue, durable, exclusive, autoDelete, null); channel.queueBind(queue, exchange, routingKey); - logger.info("RemoteObject: " + UID + " declaring fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + durable); + logger.info("RemoteObject: " + UID + " declaring fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + durable + + ", Exclusive: " + exclusive + ", AutoDelete: " + autoDelete); channel.exchangeDeclare(multiExchange, "fanout"); - channel.queueDeclare(multiQueue, durable, false, false, null); + channel.queueDeclare(multiQueue, durable, exclusive, autoDelete, null); channel.queueBind(multiQueue, multiExchange, "");