You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by ch...@apache.org on 2014/10/15 21:37:20 UTC
git commit: adding the capability to register multiple consumers for
the same channel
Repository: airavata
Updated Branches:
refs/heads/master ccebc44de -> c39d0fce2
adding the capability to register multiple consumers for the same channel
Signed-off-by: Chathuri Wimalasena <ka...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/c39d0fce
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/c39d0fce
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/c39d0fce
Branch: refs/heads/master
Commit: c39d0fce2407ea4f610f8bcc23c9ec07db004b63
Parents: ccebc44
Author: supunkamburugamuva <su...@gmail.com>
Authored: Wed Oct 15 15:03:43 2014 -0400
Committer: Chathuri Wimalasena <ka...@gmail.com>
Committed: Wed Oct 15 15:35:59 2014 -0400
----------------------------------------------------------------------
.../airavata/messaging/core/Consumer.java | 10 ++-
.../airavata/messaging/core/MessageHandler.java | 4 +-
.../messaging/core/MessagingConstants.java | 7 ++
.../airavata/messaging/core/TestClient.java | 12 ++-
.../messaging/core/impl/RabbitMQConsumer.java | 83 +++++++++++++++++---
5 files changed, 100 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/airavata/blob/c39d0fce/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/Consumer.java
----------------------------------------------------------------------
diff --git a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/Consumer.java b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/Consumer.java
index ecd23bd..eb557ac 100644
--- a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/Consumer.java
+++ b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/Consumer.java
@@ -27,6 +27,14 @@ import org.apache.airavata.common.exception.AiravataException;
* This is the basic consumer
*/
public interface Consumer {
- public void listen(MessageHandler handler) throws AiravataException;
+ /**
+ * Start listening for messages, The binding properties are specified in the handler.
+ * Returns and unique id to this Consumer. This id can be used to stop the listening
+ * @param handler
+ * @return
+ * @throws AiravataException
+ */
+ public String listen(MessageHandler handler) throws AiravataException;
+ public void stopListen(final String id) throws AiravataException;
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/c39d0fce/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessageHandler.java
----------------------------------------------------------------------
diff --git a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessageHandler.java b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessageHandler.java
index 70c4cb6..8b897c5 100644
--- a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessageHandler.java
+++ b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessageHandler.java
@@ -1,7 +1,9 @@
package org.apache.airavata.messaging.core;
-import org.apache.airavata.model.messaging.event.Message;
+import java.util.Map;
public interface MessageHandler {
+ Map<String, String> getProperties();
+
void onMessage(MessageContext message);
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/c39d0fce/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessagingConstants.java
----------------------------------------------------------------------
diff --git a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessagingConstants.java b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessagingConstants.java
new file mode 100644
index 0000000..fa0946a
--- /dev/null
+++ b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/MessagingConstants.java
@@ -0,0 +1,7 @@
+package org.apache.airavata.messaging.core;
+
+public abstract class MessagingConstants {
+ public static final String RABBIT_ROUTING_KEY = "routingKey";
+ public static final String RABBIT_QUEUE= "queue";
+ public static final String RABBIT_CONSUMER_TAG = "consumerTag";
+}
http://git-wip-us.apache.org/repos/asf/airavata/blob/c39d0fce/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/TestClient.java
----------------------------------------------------------------------
diff --git a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/TestClient.java b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/TestClient.java
index 46e3193..4de9aba 100644
--- a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/TestClient.java
+++ b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/TestClient.java
@@ -29,6 +29,9 @@ import org.apache.airavata.model.messaging.event.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashMap;
+import java.util.Map;
+
public class TestClient {
public static final String RABBITMQ_BROKER_URL = "rabbitmq.broker.url";
@@ -41,9 +44,16 @@ public class TestClient {
AiravataUtils.setExecutionAsServer();
String brokerUrl = ServerSettings.getSetting(RABBITMQ_BROKER_URL);
String exchangeName = ServerSettings.getSetting(RABBITMQ_EXCHANGE_NAME);
- RabbitMQConsumer consumer = new RabbitMQConsumer(brokerUrl, exchangeName, experimentId);
+ RabbitMQConsumer consumer = new RabbitMQConsumer(brokerUrl, exchangeName);
consumer.listen(new MessageHandler() {
@Override
+ public Map<String, String> getProperties() {
+ Map<String, String> props = new HashMap<String, String>();
+ props.put(MessagingConstants.RABBIT_ROUTING_KEY, experimentId);
+ return props;
+ }
+
+ @Override
public void onMessage(MessageContext message) {
System.out.println(" Message Received with message id '" + message.getMessageId()
+ "' and with message type '" + message.getType());
http://git-wip-us.apache.org/repos/asf/airavata/blob/c39d0fce/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/impl/RabbitMQConsumer.java
----------------------------------------------------------------------
diff --git a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/impl/RabbitMQConsumer.java b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/impl/RabbitMQConsumer.java
index 5dbea03..f3c1942 100644
--- a/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/impl/RabbitMQConsumer.java
+++ b/modules/messaging/core/src/main/java/org/apache/airavata/messaging/core/impl/RabbitMQConsumer.java
@@ -28,6 +28,7 @@ import org.apache.airavata.common.utils.ThriftUtils;
import org.apache.airavata.messaging.core.Consumer;
import org.apache.airavata.messaging.core.MessageContext;
import org.apache.airavata.messaging.core.MessageHandler;
+import org.apache.airavata.messaging.core.MessagingConstants;
import org.apache.airavata.model.messaging.event.*;
import org.apache.thrift.TBase;
import org.apache.thrift.TException;
@@ -35,32 +36,54 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
public class RabbitMQConsumer implements Consumer {
+ private static Logger log = LoggerFactory.getLogger(RabbitMQConsumer.class);
+
private String exchangeName;
private String url;
private Connection connection;
private Channel channel;
- private String consumerTag = "hhh";
- private static Logger log = LoggerFactory.getLogger(RabbitMQConsumer.class);
- private String routingKey;
+ private Map<String, QueueDetails> queueDetailsMap = new HashMap<String, QueueDetails>();
- public RabbitMQConsumer(String brokerUrl, String exchangeName, String routingKey) {
+ public RabbitMQConsumer(String brokerUrl, String exchangeName) throws AiravataException {
this.exchangeName = exchangeName;
this.url = brokerUrl;
- this.routingKey = routingKey;
- }
- public void listen(final MessageHandler handler) throws AiravataException {
try {
connection = createConnection();
channel = connection.createChannel();
channel.exchangeDeclare(exchangeName, "direct", false);
- final String queueName = channel.queueDeclare().getQueue();
- channel.queueBind(queueName, exchangeName, routingKey);
+ } catch (Exception e) {
+ String msg = "could not open channel for exchange " + exchangeName;
+ log.error(msg);
+ throw new AiravataException(msg, e);
+ }
+ }
+ public String listen(final MessageHandler handler) throws AiravataException {
+ try {
+ Map<String, String> props = handler.getProperties();
+ final String routingKey = props.get(MessagingConstants.RABBIT_ROUTING_KEY);
+ if (routingKey == null) {
+ throw new IllegalArgumentException("The routing key must be present");
+ }
+ String queueName = props.get(MessagingConstants.RABBIT_QUEUE);
+ String consumerTag = props.get(MessagingConstants.RABBIT_CONSUMER_TAG);
+ if (queueName == null) {
+ queueName = channel.queueDeclare().getQueue();
+ } else {
+ channel.queueDeclare(queueName, true, false, false, null);
+ }
+ if (consumerTag == null) {
+ consumerTag = "default";
+ }
+ String id = routingKey + "." + queueName;
+ channel.queueBind(queueName, exchangeName, routingKey);
channel.basicConsume(queueName, true, consumerTag, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
@@ -101,21 +124,33 @@ public class RabbitMQConsumer implements Consumer {
MessageContext messageContext = new MessageContext(event, message.getMessageType(), message.getMessageId());
handler.onMessage(messageContext);
} catch (TException e) {
- String msg = "Failed to de-serialize the thrift message, exchange: " + exchangeName + " routingKey: " + routingKey + " queue: " + queueName;
+ String msg = "Failed to de-serialize the thrift message, exchange: " + exchangeName + " routingKey: " + routingKey;
log.warn(msg, e);
}
}
});
+ // save the name for deleting the queue
+ queueDetailsMap.put(id, new QueueDetails(queueName, routingKey));
+ return id;
} catch (Exception e) {
- reset();
String msg = "could not open channel for exchange " + exchangeName;
log.error(msg);
throw new AiravataException(msg, e);
}
}
- private void reset() {
- consumerTag = null;
+ public void stopListen(final String id) throws AiravataException {
+ QueueDetails details = queueDetailsMap.get(id);
+ if (details != null) {
+ try {
+ channel.queueUnbind(details.getQueueName(), exchangeName, details.getRoutingKey());
+ channel.queueDelete(details.getQueueName(), true, true);
+ } catch (IOException e) {
+ String msg = "could not un-bind queue: " + details.getQueueName() + " for exchange " + exchangeName;
+ log.error(msg);
+ throw new AiravataException(msg, e);
+ }
+ }
}
private Connection createConnection() throws IOException {
@@ -134,4 +169,26 @@ public class RabbitMQConsumer implements Consumer {
return null;
}
}
+
+ /**
+ * Private class for holding some information about the consumers registered
+ */
+ private class QueueDetails {
+ String queueName;
+
+ String routingKey;
+
+ private QueueDetails(String queueName, String routingKey) {
+ this.queueName = queueName;
+ this.routingKey = routingKey;
+ }
+
+ public String getQueueName() {
+ return queueName;
+ }
+
+ public String getRoutingKey() {
+ return routingKey;
+ }
+ }
}