Refactor properties and channel setup
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/AMQPSession.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/AMQPSession.java
index 92ebd02..3ff2da9 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/AMQPSession.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/AMQPSession.java
@@ -1,6 +1,5 @@
package com.googlesource.gerrit.plugins.rabbitmq;
-import com.google.gerrit.extensions.annotations.PluginName;
import com.google.inject.Inject;
import com.google.inject.Singleton;
@@ -22,18 +21,14 @@
public class AMQPSession implements ShutdownListener {
private static final Logger LOGGER = LoggerFactory.getLogger(AMQPSession.class);
- private static final String EXCHANGE_TYPE_DIRECT ="direct";
- private static final String EXCHANGE_TYPE_FANOUT ="fanout";
private final Properties properties;
- private final String pluginName;
private Connection connection;
private Channel publishChannel;
private String exchangeName;
@Inject
- public AMQPSession(@PluginName String pluginName, Properties properties) {
+ public AMQPSession(Properties properties) {
this.properties = properties;
- this.pluginName = pluginName;
}
public boolean isOpen() {
@@ -44,16 +39,16 @@
}
public void connect() {
- LOGGER.info("Connect to " + properties.getAMQPUri() + "...");
+ LOGGER.info("Connect to " + properties.getString(Keys.AMQP_URI) + "...");
ConnectionFactory factory = new ConnectionFactory();
try {
- if (StringUtils.isNotEmpty(properties.getAMQPUri())) {
- factory.setUri(properties.getAMQPUri());
- if (StringUtils.isNotEmpty(properties.getAMQPUsername())) {
- factory.setUsername(properties.getAMQPUsername());
+ if (StringUtils.isNotEmpty(properties.getString(Keys.AMQP_URI))) {
+ factory.setUri(properties.getString(Keys.AMQP_URI));
+ if (StringUtils.isNotEmpty(properties.getString(Keys.AMQP_USERNAME))) {
+ factory.setUsername(properties.getString(Keys.AMQP_USERNAME));
}
- if (StringUtils.isNotEmpty(properties.getAMQPPassword())) {
- factory.setPassword(properties.getAMQPPassword());
+ if (StringUtils.isNotEmpty(properties.getString(Keys.AMQP_PASSWORD))) {
+ factory.setPassword(properties.getString(Keys.AMQP_PASSWORD));
}
connection = factory.newConnection();
connection.addShutdownListener(this);
@@ -61,7 +56,7 @@
}
bind();
} catch (URISyntaxException ex) {
- LOGGER.error("URI syntax error: " + properties.getAMQPUri());
+ LOGGER.error("URI syntax error: " + properties.getString(Keys.AMQP_URI));
} catch (IOException ex) {
LOGGER.error("Connection cannot be opened.");
} catch (Exception ex) {
@@ -73,35 +68,34 @@
if (connection != null && publishChannel == null) {
try {
Channel ch = connection.createChannel();
- if (StringUtils.isNotEmpty(properties.getAMQPQueue())) {
- LOGGER.info("Queue mode");
- String exchangeType = EXCHANGE_TYPE_DIRECT;
- String routingKey = properties.getAMQPRoutingKey();
- if (routingKey.isEmpty()) {
- exchangeType = EXCHANGE_TYPE_FANOUT;
- routingKey = pluginName;
- }
- exchangeName = exchangeType + "-" + properties.getAMQPQueue();
- LOGGER.debug("Exchange type: " + exchangeType);
- LOGGER.debug("Declare exchange: " + exchangeName);
- ch.exchangeDeclare(exchangeName, exchangeType, true);
- LOGGER.debug("Declare queue: " + properties.getAMQPQueue());
- ch.queueDeclare(properties.getAMQPQueue(), true, false, false, null);
- LOGGER.debug("Bind exchange and queue with key: " + routingKey);
- ch.queueBind(properties.getAMQPQueue(), exchangeName, routingKey);
- publishChannel = ch;
- LOGGER.info("Channel for queue \"" + properties.getAMQPQueue() + "\" opened.");
- } else if (StringUtils.isNotEmpty(properties.getAMQPExchange())) {
- LOGGER.info("Exchange mode");
- exchangeName = properties.getAMQPExchange();
- LOGGER.debug("Declare exchange: " + exchangeName);
- ch.exchangeDeclarePassive(exchangeName);
- publishChannel = ch;
- LOGGER.info("Channel for exchange \"" + exchangeName + "\" opened.");
- } else {
- LOGGER.warn("Unrecognized bind mode.");
- throw new IOException();
+ LOGGER.info("Channel is opened.");
+ if (StringUtils.isNotEmpty(properties.getString(Keys.QUEUE_NAME))) {
+ LOGGER.info("Queue: " + properties.getString(Keys.QUEUE_NAME));
+ ch.queueDeclare(properties.getString(Keys.QUEUE_NAME),
+ properties.getBoolean(Keys.QUEUE_DURABLE),
+ properties.getBoolean(Keys.QUEUE_EXCLUSIVE),
+ properties.getBoolean(Keys.QUEUE_AUTODELETE), null);
+ exchangeName = "exchange-for-" + properties.getString(Keys.QUEUE_NAME);
}
+
+ if (StringUtils.isNotEmpty(properties.getString(Keys.EXCHANGE_NAME))) {
+ exchangeName = properties.getString(Keys.EXCHANGE_NAME);
+ }
+
+ LOGGER.info("Exchange: " + exchangeName);
+ ch.exchangeDeclare(exchangeName,
+ properties.getString(Keys.EXCHANGE_TYPE),
+ properties.getBoolean(Keys.EXCHANGE_DURABLE),
+ properties.getBoolean(Keys.EXCHANGE_AUTODELETE), null);
+
+ if (StringUtils.isNotEmpty(properties.getString(Keys.QUEUE_NAME))) {
+ LOGGER.debug("Bind exchange and queue with key: " + properties.getString(Keys.BIND_ROUTINGKEY));
+ ch.queueBind(properties.getString(Keys.QUEUE_NAME),
+ exchangeName, properties.getString(Keys.BIND_ROUTINGKEY));
+ }
+
+ publishChannel = ch;
+ LOGGER.info("Complete to setup channel.");
} catch (Exception ex) {
LOGGER.warn("#bind: " + ex.getClass().getName());
disconnect();
@@ -126,7 +120,7 @@
if (publishChannel != null && publishChannel.isOpen()) {
try {
LOGGER.debug("Send message.");
- publishChannel.basicPublish(exchangeName, properties.getAMQPRoutingKey(), properties.getBasicProperties(),
+ publishChannel.basicPublish(exchangeName, properties.getString(Keys.MESSAGE_ROUTINGKEY), properties.getBasicProperties(),
message.getBytes(CharEncoding.UTF_8));
} catch (Exception ex) {
LOGGER.warn("#sendMessage: " + ex.getClass().getName());
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/Properties.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/Properties.java
index 4bc20b9..8ff2aff 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/Properties.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/Properties.java
@@ -30,11 +30,6 @@
private final static String GERRIT = "gerrit";
private final static String CONTENT_TYPE_JSON = "application/json";
- private final static int DEFAULT_MESSAGE_DELIVERY_MODE = 1;
- private final static int DEFAULT_MESSAGE_PRIORITY = 0;
- private final static int DEFAULT_GERRIT_PORT = 29418;
- private final static String DEFAULT_GERRIT_SCHEME = "ssh";
- private final static int DEFAULT_CONNECTION_MONITOR_INTERVAL = 15000;
private final static int MINIMUM_CONNECTION_MONITOR_INTERVAL = 5000;
private final Config config;
@@ -70,73 +65,25 @@
return cfg;
}
- private String getConfigString(Keys key, String defaultValue) {
- String val = pluginConfig.getString(key.section, null, key.value);
+ public String getString(Keys key) {
+ String val = pluginConfig.getString(key.section, null, key.name);
if (val == null) {
- return defaultValue;
+ return key.defaultVal.toString();
} else {
return val;
}
}
- private int getConfigInt(Keys key, int defaultValue) {
- return pluginConfig.getInt(key.section, key.value, defaultValue);
+ public int getInt(Keys key) {
+ return pluginConfig.getInt(key.section, key.name, new Integer(key.defaultVal.toString()));
}
- private boolean getConfigBoolean(Keys key, boolean defaultValue) {
- return pluginConfig.getBoolean(key.section, key.value, defaultValue);
- }
-
- public String getAMQPUri() {
- return getConfigString(Keys.AMQP_URI, "amqp://localhost");
- }
-
- public String getAMQPUsername() {
- return getConfigString(Keys.AMQP_USERNAME, "");
- }
-
- public String getAMQPPassword() {
- return getConfigString(Keys.AMQP_PASSWORD, "");
- }
-
- public String getAMQPQueue() {
- return getConfigString(Keys.AMQP_QUEUE, "");
- }
-
- public String getAMQPExchange() {
- return getConfigString(Keys.AMQP_EXCHANGE, "");
- }
-
- public String getAMQPRoutingKey() {
- return getConfigString(Keys.AMQP_ROUTINGKEY, "");
- }
-
- public int getMessageDeliveryMode() {
- return getConfigInt(Keys.MESSAGE_DELIVERY_MODE, DEFAULT_MESSAGE_DELIVERY_MODE);
- }
-
- public int getMessagePriority() {
- return getConfigInt(Keys.MESSAGE_PRIORITY, DEFAULT_MESSAGE_PRIORITY);
- }
-
- public String getGerritName() {
- return getConfigString(Keys.GERRIT_NAME, "");
- }
-
- public String getGerritHostname() {
- return getConfigString(Keys.GERRIT_HOSTNAME, "");
- }
-
- public String getGerritScheme() {
- return getConfigString(Keys.GERRIT_SCHEME, DEFAULT_GERRIT_SCHEME);
- }
-
- public int getGerritPort() {
- return getConfigInt(Keys.GERRIT_PORT, DEFAULT_GERRIT_PORT);
+ public boolean getBoolean(Keys key) {
+ return pluginConfig.getBoolean(key.section, key.name, new Boolean(key.defaultVal.toString()));
}
public String getGerritFrontUrl() {
- return StringUtils.stripToEmpty(config.getString(GERRIT, null, Keys.GERRIT_FRONT_URL.value));
+ return StringUtils.stripToEmpty(config.getString(Keys.GERRIT_FRONT_URL.section, null, Keys.GERRIT_FRONT_URL.name));
}
public String getGerritVersion() {
@@ -144,7 +91,7 @@
}
public int getConnectionMonitorInterval() {
- int interval = getConfigInt(Keys.CONNECTION_MONITOR_INTERVAL, DEFAULT_CONNECTION_MONITOR_INTERVAL);
+ int interval = getInt(Keys.MONITOR_INTERVAL);
if (interval < MINIMUM_CONNECTION_MONITOR_INTERVAL) {
return MINIMUM_CONNECTION_MONITOR_INTERVAL;
}
@@ -154,19 +101,19 @@
public AMQP.BasicProperties getBasicProperties() {
if (properties == null) {
Map<String, Object> headers = new HashMap<String, Object>();
- headers.put(Keys.GERRIT_NAME.header, getGerritName());
- headers.put(Keys.GERRIT_HOSTNAME.header, getGerritHostname());
- headers.put(Keys.GERRIT_SCHEME.header, getGerritScheme());
- headers.put(Keys.GERRIT_PORT.header, String.valueOf(getGerritPort()));
- headers.put(Keys.GERRIT_FRONT_URL.header, getGerritFrontUrl());
- headers.put(Keys.GERRIT_VERSION.header, getGerritVersion());
+ headers.put(Keys.GERRIT_NAME.key, getString(Keys.GERRIT_NAME));
+ headers.put(Keys.GERRIT_HOSTNAME.key, getString(Keys.GERRIT_HOSTNAME));
+ headers.put(Keys.GERRIT_SCHEME.key, getString(Keys.GERRIT_SCHEME));
+ headers.put(Keys.GERRIT_PORT.key, String.valueOf(getInt(Keys.GERRIT_PORT)));
+ headers.put(Keys.GERRIT_FRONT_URL.key, getGerritFrontUrl());
+ headers.put(Keys.GERRIT_VERSION.key, getGerritVersion());
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.appId(GERRIT);
builder.contentEncoding(CharEncoding.UTF_8);
builder.contentType(CONTENT_TYPE_JSON);
- builder.deliveryMode(getMessageDeliveryMode());
- builder.priority(getMessagePriority());
+ builder.deliveryMode(getInt(Keys.MESSAGE_DELIVERY_MODE));
+ builder.priority(getInt(Keys.MESSAGE_PRIORITY));
builder.headers(headers);
properties = builder.build();
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/RabbitMQManager.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/RabbitMQManager.java
index a1ed156..f799871 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/RabbitMQManager.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/RabbitMQManager.java
@@ -40,7 +40,7 @@
session.connect();
}
}
- }, MONITOR_FIRSTTIME_DELAY, properties.getConnectionMonitorInterval());
+ }, MONITOR_FIRSTTIME_DELAY, properties.getInt(Keys.MONITOR_INTERVAL));
}
@Override
