aboutsummaryrefslogtreecommitdiffstats
path: root/event-client-rabbitmq/src/main/java
diff options
context:
space:
mode:
Diffstat (limited to 'event-client-rabbitmq/src/main/java')
-rw-r--r--event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqClientConfig.java198
-rw-r--r--event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventConsumer.java191
-rw-r--r--event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventPublisher.java287
-rw-r--r--event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqUtils.java127
4 files changed, 416 insertions, 387 deletions
diff --git a/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqClientConfig.java b/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqClientConfig.java
new file mode 100644
index 0000000..d177989
--- /dev/null
+++ b/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqClientConfig.java
@@ -0,0 +1,198 @@
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2018 European Software Marketing Ltd.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.aai.event.client;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+
+public class RabbitMqClientConfig {
+
+ private static String BINDING_CONSUME_ALL = "#";
+
+ private String hosts;
+ private String username;
+ private String password;
+ private String exchangeName;
+ private String exchangeType;
+ private String routingKey;
+ private String virtualHost;
+ private String queue;
+ private String bindingKey;
+ private Map<String, Object> queueArguments;
+ private Map<String, Object> exchangeArguments;
+ private String sslKeyStoreFile;
+ private String sslTrustStoreFile;
+ private String sslKeyStorePassword;
+ private String sslTrustStorePassword;
+ private String connectionTimeout;
+ private String enableSsl;
+
+ public String getUsername() {
+ return username;
+ }
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ public String getHosts() {
+ return hosts;
+ }
+ public void setHosts(String hosts) {
+ this.hosts = hosts;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public String getExchangeName() {
+ return exchangeName;
+ }
+ public void setExchangeName(String exchangeName) {
+ this.exchangeName = exchangeName;
+ }
+
+ public BuiltinExchangeType getExchangeType() {
+ if (exchangeType == null) {
+ return BuiltinExchangeType.TOPIC;
+ }
+ return BuiltinExchangeType.valueOf(exchangeType);
+ }
+ public void setExchangeType(String exchangeType) {
+ this.exchangeType = exchangeType;
+ }
+
+ public String getRoutingKey() {
+ if (routingKey == null) {
+ return "";
+ }
+ return routingKey;
+ }
+ public void setRoutingKey(String routingKey) {
+ this.routingKey = routingKey;
+ }
+
+ public String getVirtualHost() {
+ return virtualHost;
+ }
+ public void setVirtualHost(String virtualHost) {
+ this.virtualHost = virtualHost;
+ }
+
+ public String getQueue() {
+ return queue;
+ }
+ public void setQueue(String queue) {
+ this.queue = queue;
+ }
+
+ public Map<String, Object> getQueueArguments() {
+ if (queueArguments == null) {
+ return new HashMap<String, Object>();
+ }
+ return queueArguments;
+ }
+ public void setQueueArguments(Map<String, Object> queueArguments) {
+ this.queueArguments = queueArguments;
+ }
+
+ public Map<String, Object> getExchangeArguments() {
+ if (exchangeArguments == null) {
+ return new HashMap<String, Object>();
+ }
+ return exchangeArguments;
+ }
+ public void setExchangeArguments(Map<String, Object> exchangeArguments) {
+ this.exchangeArguments = exchangeArguments;
+ }
+
+ public String getBindingKey() {
+ if (bindingKey == null) {
+ return BINDING_CONSUME_ALL;
+ }
+ return bindingKey;
+ }
+
+ public void setBindingKey(String bindingKey) {
+ this.bindingKey = bindingKey;
+ }
+
+ public String getSslKeyStoreFile() {
+ return sslKeyStoreFile;
+ }
+
+ public void setSslKeyStoreFile(String sslKeyStoreFile) {
+ this.sslKeyStoreFile = sslKeyStoreFile;
+ }
+
+ public String getSslTrustStoreFile() {
+ return sslTrustStoreFile;
+ }
+
+ public void setSslTrustStoreFile(String sslTrustStoreFile) {
+ this.sslTrustStoreFile = sslTrustStoreFile;
+ }
+
+ public String getSslKeyStorePassword() {
+ return sslKeyStorePassword;
+ }
+
+ public void setSslKeyStorePassword(String sslKeyStorePassword) {
+ this.sslKeyStorePassword = sslKeyStorePassword;
+ }
+
+ public String getSslTrustStorePassword() {
+ return sslTrustStorePassword;
+ }
+
+ public void setSslTrustStorePassword(String sslTrustStorePassword) {
+ this.sslTrustStorePassword = sslTrustStorePassword;
+ }
+
+ public String getConnectionTimeout() {
+ return connectionTimeout;
+ }
+
+ public void setConnectionTimeout(String connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ }
+
+ public Boolean getEnableSsl() {
+ if (enableSsl != null) {
+ return Boolean.valueOf(enableSsl);
+ } else {
+ return false;
+ }
+ }
+
+ public void setEnableSsl(String enableSsl) {
+ this.enableSsl = enableSsl;
+ }
+
+
+
+ }
diff --git a/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventConsumer.java b/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventConsumer.java
index c564198..c10b46c 100644
--- a/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventConsumer.java
+++ b/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventConsumer.java
@@ -22,28 +22,23 @@ package org.onap.aai.event.client;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
+import javax.naming.ConfigurationException;
+
import org.onap.aai.cl.api.Logger;
import org.onap.aai.cl.eelf.LoggerFactory;
import org.onap.aai.event.api.EventConsumer;
import org.onap.aai.event.api.MessageWithOffset;
-import com.rabbitmq.client.Address;
-import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.AMQP.Exchange.DeclareOk;
/**
* Event bus client consumer wrapper for RabbitMQ.
@@ -57,27 +52,16 @@ import com.rabbitmq.client.AMQP.Exchange.DeclareOk;
*/
public class RabbitMqEventConsumer implements EventConsumer {
private static Logger log = LoggerFactory.getInstance().getLogger(RabbitMqEventConsumer.class);
- private static ConnectionFactory factory = new ConnectionFactory();
- private BlockingQueue<MessageWithOffset> messageQueue;
- private static String BINDING_CONSUME_ALL = "#";
+ private BlockingQueue<MessageWithOffset> messageQueue;
private final Connection connection;
private final Channel channel;
- private final String queueName;
- private DeclareOk exchangeInfo;
private Long lastDeliveryTag;
private long timeout = 5000;
- private com.rabbitmq.client.AMQP.Queue.DeclareOk queueInfo;
+ private RabbitMqClientConfig config;
+
- /**
- * (intended for testing prupose only)
- * @param connFactory
- */
- static void setConnectionFactory(ConnectionFactory connFactory) {
- factory = connFactory;
- }
-
/**
* (intended for testing prupose only)
* @param messageQueue
@@ -86,148 +70,29 @@ public class RabbitMqEventConsumer implements EventConsumer {
this.messageQueue = messageQueue;
}
- /**
- * Constructor to open a consumer on single host port for a topic exchange with specific queue name which will
- * consume all messages from topic
- * @param host
- * @param port
- * @param userName
- * @param password
- * @param exchangeName
- * @param queueName
- * @throws Exception
- */
- public RabbitMqEventConsumer(String host, int port, String userName, String password, String exchangeName, String queueName) throws Exception {
- this(host, port, userName, password, BuiltinExchangeType.TOPIC.name(), exchangeName, queueName, BINDING_CONSUME_ALL);
- }
-
- /**
- * Constructor to open a consumer on single host port for a exchange with specific queue name which will
- * consume all messages from topic
- * @param host
- * @param port
- * @param userName
- * @param password
- * @param exchangeType - Supported values - DIRECT, FANOUT, TOPIC, HEADERS
- * @param exchangeName
- * @param queueName
- * @throws Exception
- */
- public RabbitMqEventConsumer(String host, int port, String userName, String password, String exchangeType, String exchangeName, String queueName) throws Exception {
- this(host, port, userName, password, exchangeType, exchangeName, queueName, BINDING_CONSUME_ALL);
- }
-
- /**
- * Constructor to open a consumer on single host port for a exchange with specific queue name
- * @param host
- * @param port
- * @param userName
- * @param password
- * @param exchangeType
- * @param exchangeName
- * @param queueName
- * @param bindingKey - Bind the queue to specific messages only
- * @throws Exception
- */
- public RabbitMqEventConsumer(String host, int port, String userName, String password, String exchangeType, String exchangeName, String queueName, String bindingKey) throws Exception {
- this(buildMap(host, port), userName, password, BuiltinExchangeType.valueOf(exchangeType), exchangeName, queueName, bindingKey, new HashMap<String, Object>());
- }
-
- /**
- * Constructor to open a consumer on multiple host port for a topic exchange with specific queue name which will
- * consume all messages from topic
- * @param hostPortMap
- * @param userName
- * @param password
- * @param exchangeName
- * @param queueName
- * @throws Exception
- */
- public RabbitMqEventConsumer(Map<String, Integer> hostPortMap, String userName, String password, String exchangeName, String queueName) throws Exception {
- this(hostPortMap, userName, password, BuiltinExchangeType.TOPIC.name(), exchangeName, queueName, BINDING_CONSUME_ALL);
- }
-
- /**
- * Constructor to open a consumer on multiple host port for a exchange with specific queue name which will
- * consume all messages from topic
- * @param hostPortMap
- * @param userName
- * @param password
- * @param exchangeType - Supported values - DIRECT, FANOUT, TOPIC, HEADERS
- * @param exchangeName
- * @param queueName
- * @throws Exception
- */
- public RabbitMqEventConsumer(Map<String, Integer> hostPortMap, String userName, String password, String exchangeType, String exchangeName, String queueName) throws Exception {
- this(hostPortMap, userName, password, exchangeType, exchangeName, queueName, BINDING_CONSUME_ALL);
- }
-
- /**
- * Constructor to open a consumer on multiple host port for a exchange with specific queue name
- * @param hostPortMap
- * @param userName
- * @param password
- * @param exchangeType
- * @param exchangeName
- * @param queueName
- * @param bindingKey - Bind the queue to specific messages only
- * @throws Exception
- */
- public RabbitMqEventConsumer(Map<String, Integer> hostPortMap, String userName, String password, String exchangeType, String exchangeName, String queueName, String bindingKey) throws Exception {
- this(hostPortMap, userName, password, BuiltinExchangeType.valueOf(exchangeType), exchangeName, queueName, bindingKey, new HashMap<String, Object>());
- }
-
- public RabbitMqEventConsumer(Map<String, Integer> hostPortMap, String userName, String password, BuiltinExchangeType exchangeType, String exchangeName, String queueName, String bindingKey, Map<String, Object> exchangeArguments) throws Exception {
- messageQueue = new ArrayBlockingQueue<>(1000);
- List<Address> addresses = new ArrayList<Address>();
- Iterator<String> iter = hostPortMap.keySet().iterator();
- while (iter.hasNext())
- {
- String host = iter.next();
- int port = hostPortMap.get(host);
- Address add = new Address(host,port);
- addresses.add(add);
- }
- factory.setUsername(userName);
- factory.setPassword(password);
- connection = factory.newConnection(addresses);
- channel = connection.createChannel();
- exchangeInfo = channel.exchangeDeclare(exchangeName, exchangeType, true, false, exchangeArguments);
- this.queueName = queueName;
- channel.queueDeclare(queueName, true, false, false, null);
- channel.queueBind(queueName, exchangeName, bindingKey);
- String consumerTag = channel.basicConsume(queueName, false, new CallBackConsumer(channel)); //AutoAck is false
- }
-
- public RabbitMqEventConsumer(String host, int port, String userName, String password, String queue) throws Exception {
- this(buildMap(host, port), userName, password, queue, new HashMap<String, Object>());
- }
-
- public RabbitMqEventConsumer(Map<String, Integer> hostPortMap, String userName, String password, String queue, Map<String, Object> queueArguments) throws Exception {
- messageQueue = new ArrayBlockingQueue<>(1000);
- List<Address> addresses = new ArrayList<Address>();
- Iterator<String> iter = hostPortMap.keySet().iterator();
- while (iter.hasNext())
- {
- String host = iter.next();
- int port = hostPortMap.get(host);
- Address add = new Address(host,port);
- addresses.add(add);
- }
- factory.setUsername(userName);
- factory.setPassword(password);
- connection = factory.newConnection(addresses);
- channel = connection.createChannel();
- this.queueName = queue;
- queueInfo = channel.queueDeclare(queueName, true, false, false, queueArguments);
- String consumerTag = channel.basicConsume(queueName, false, new CallBackConsumer(channel)); //AutoAck is false
- }
-
- private static Map<String, Integer> buildMap(String host, Integer port) {
- Map<String, Integer> hostPortMap = new HashMap<String, Integer>();
- hostPortMap.put(host, port);
- return hostPortMap;
- }
+ public RabbitMqEventConsumer(RabbitMqClientConfig config) throws Exception {
+ this.config = config;
+ this.messageQueue = new ArrayBlockingQueue<>(1000);
+
+ if (config.getQueue() == null) {
+ throw new ConfigurationException("Mandatory config param queue not set");
+ }
+
+ this.connection = RabbitMqUtils.createConnection(config);
+ this.channel = connection.createChannel();
+
+ if (config.getExchangeName() != null) {
+ channel.exchangeDeclare(config.getExchangeName(), config.getExchangeType(), true, false, config.getExchangeArguments());
+ channel.queueDeclare(config.getQueue(), true, false, false, config.getQueueArguments());
+ channel.queueBind(config.getQueue(), config.getExchangeName(), config.getBindingKey());
+ }
+ else {
+ channel.queueDeclare(config.getQueue(), true, false, false, config.getQueueArguments());
+ }
+
+ channel.basicConsume(config.getQueue(), false, new CallBackConsumer(channel)); //AutoAck is false
+ }
+
@Override
public Iterable<String> consumeAndCommit() throws Exception {
diff --git a/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventPublisher.java b/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventPublisher.java
index a7e0ee3..614ae54 100644
--- a/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventPublisher.java
+++ b/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqEventPublisher.java
@@ -21,23 +21,16 @@
package org.onap.aai.event.client;
-import java.util.ArrayList;
import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
+
+import javax.naming.ConfigurationException;
import org.onap.aai.cl.api.Logger;
import org.onap.aai.cl.eelf.LoggerFactory;
import org.onap.aai.event.api.EventPublisher;
-import com.rabbitmq.client.AMQP.Exchange.DeclareOk;
-import com.rabbitmq.client.Address;
-import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
/**
* Event bus client publisher wrapper for RabbitMQ
@@ -46,236 +39,82 @@ import com.rabbitmq.client.ConnectionFactory;
*/
public class RabbitMqEventPublisher implements EventPublisher {
- private static Logger log = LoggerFactory.getInstance().getLogger(RabbitMqEventPublisher.class);
- private static ConnectionFactory factory = new ConnectionFactory();
+ private static Logger log = LoggerFactory.getInstance().getLogger(RabbitMqEventPublisher.class);
- private final Connection connection;
- private final Channel channel;
- private final String exchangeName;
- private final String queueName;
- private final String routingKey;
- private DeclareOk exchangeInfo;
- private com.rabbitmq.client.AMQP.Queue.DeclareOk queueInfo;
+ private final Connection connection;
+ private final Channel channel;
- static void setConnectionFactory(ConnectionFactory connFactory) {
- factory = connFactory;
- }
- /**
- * Constructor for single host port for a topic exchange
- * @param host
- * @param port
- * @param userName
- * @param password
- * @param exchangeName
- * @throws Exception
- */
- public RabbitMqEventPublisher(String host, int port, String userName, String password, String exchangeName) throws Exception {
- this(host, port, userName, password, BuiltinExchangeType.TOPIC.name(), exchangeName,"");
- }
-
- /**
- * Constructor for single host port for a specific exchange type
- * @param host
- * @param port
- * @param userName
- * @param password
- * @param exchangeType - Supported values - DIRECT, FANOUT, TOPIC, HEADERS
- * @param exchangeName
- * @throws Exception
- */
- public RabbitMqEventPublisher(String host, int port, String userName, String password, String exchangeType, String exchangeName) throws Exception {
- this(host,port, userName, password, exchangeType, exchangeName,"");
- }
-
- /**
- * Constructor for single host port for a specific exchange type and specific routing key
- * @param host
- * @param port
- * @param userName
- * @param password
- * @param exchangeType
- * @param exchangeName
- * @param routingKey
- * @throws Exception
- */
- public RabbitMqEventPublisher(String host, int port, String userName, String password, String exchangeType, String exchangeName, String routingKey) throws Exception {
- this(buildMap(host,port), userName, password, BuiltinExchangeType.valueOf(exchangeType), exchangeName,routingKey, new HashMap<String, Object>());
- }
-
- /**
- * Constructor for multiple host and port for a topic exchange
- * @param hostPortMap
- * @param userName
- * @param password
- * @param exchangeName
- * @throws Exception
- */
- public RabbitMqEventPublisher(Map<String, Integer> hostPortMap, String userName, String password, String exchangeName) throws Exception {
- this(hostPortMap, userName, password, BuiltinExchangeType.TOPIC.name(), exchangeName,"");
- }
-
- /**
- * Constructor for multiple host port for a specific exchange type
- * @param hostPortMap
- * @param userName
- * @param password
- * @param exchangeType - Supported values - DIRECT, FANOUT, TOPIC, HEADERS
- * @param exchangeName
- * @throws Exception
- */
- public RabbitMqEventPublisher(Map<String, Integer> hostPortMap, String userName, String password, String exchangeType, String exchangeName) throws Exception {
- this(hostPortMap, userName, password, exchangeType , exchangeName,"");
- }
-
- /**
- * Constructor for multiple host port for a specific exchange type and specific routing key
- * @param hostPortMap
- * @param userName
- * @param password
- * @param exchangeType
- * @param exchangeName
- * @param routingKey
- * @throws Exception
- */
- public RabbitMqEventPublisher(Map<String, Integer> hostPortMap, String userName, String password, String exchangeType, String exchangeName, String routingKey) throws Exception {
- this(hostPortMap, userName, password, BuiltinExchangeType.valueOf(exchangeType), exchangeName,routingKey, new HashMap<String, Object>());
- }
+ private RabbitMqClientConfig config;
+
+ public RabbitMqEventPublisher(RabbitMqClientConfig config) throws Exception {
+ this.config = config;
+
+ if (config.getExchangeName() == null) {
+ throw new ConfigurationException("Mandatory config param exchangeName not set");
+ }
+
+ this.connection = RabbitMqUtils.createConnection(config);
+ this.channel = connection.createChannel();
- public RabbitMqEventPublisher(Map<String, Integer> hostPortMap, String userName, String password, BuiltinExchangeType exchangeType, String exchangeName, String routingKey, Map<String, Object> exchangeArguments) throws Exception {
- List<Address> addresses = new ArrayList<Address>();
- Iterator<String> iter = hostPortMap.keySet().iterator();
- while (iter.hasNext())
- {
- String host = iter.next();
- int port = hostPortMap.get(host);
- Address add = new Address(host,port);
- addresses.add(add);
- }
- factory.setUsername(userName);
- factory.setPassword(password);
- connection = factory.newConnection(addresses);
- channel = connection.createChannel();
- this.exchangeName = exchangeName;
- exchangeInfo = channel.exchangeDeclare(exchangeName, exchangeType, true, false, exchangeArguments); //Durable exchange and non delete
- queueName = null;
- this.routingKey=routingKey;
- }
-
- /**
- * Constructor for single host port for a queue connection
- * @param host
- * @param port
- * @param userName
- * @param password
- * @param queue
- * @param queueArguments
- * @throws Exception
- */
- public RabbitMqEventPublisher(String host, int port, String userName, String password, String queue, Map<String, Object> queueArguments) throws Exception {
- this(buildMap(host, port), userName, password, queue, queueArguments);
- }
-
- /**
- * Constructor for multiple host port for a queue connection
- * @param hostPortMap
- * @param userName
- * @param password
- * @param queue
- * @param queueArguments
- * @throws Exception
- */
- public RabbitMqEventPublisher(Map<String, Integer> hostPortMap, String userName, String password, String queue, Map<String, Object> queueArguments) throws Exception {
- List<Address> addresses = new ArrayList<Address>();
- Iterator<String> iter = hostPortMap.keySet().iterator();
- while (iter.hasNext())
- {
- String host = iter.next();
- int port = hostPortMap.get(host);
- Address add = new Address(host,port);
- addresses.add(add);
- }
- factory.setUsername(userName);
- factory.setPassword(password);
- connection = factory.newConnection(addresses);
- channel = connection.createChannel();
- exchangeName = "";
- routingKey=queue;
- this.queueName = queue;
- queueInfo = channel.queueDeclare(queueName, true, false, false, queueArguments); //Durable, non exclusive and non auto delete queue
- }
-
- private static Map<String, Integer> buildMap(String host, Integer port) {
- Map<String, Integer> hostPortMap = new HashMap<String, Integer>();
- hostPortMap.put(host, port);
- return hostPortMap;
- }
+ //Durable exchange and non delete
+ channel.exchangeDeclare(config.getExchangeName(), config.getExchangeType(), true, false, config.getExchangeArguments());
+ }
- @Override
- public void close() throws Exception {
- channel.close();
- connection.close();
- }
+ @Override
+ public void close() throws Exception {
+ channel.close();
+ connection.close();
+ }
- @Override
- public void sendAsync(String message) throws Exception {
- sendSync(message);
- }
+ @Override
+ public void sendAsync(String message) throws Exception {
+ sendSync(message);
+ }
- @Override
- public void sendAsync(Collection<String> messages) throws Exception {
- sendSync(messages);
- }
+ @Override
+ public void sendAsync(Collection<String> messages) throws Exception {
+ sendSync(messages);
+ }
- @Override
- public void sendAsync(String routingParam, String message) throws Exception {
- sendSync(routingParam, message);
- }
+ @Override
+ public void sendAsync(String routingParam, String message) throws Exception {
+ sendSync(routingParam, message);
+ }
- @Override
- public void sendAsync(String routingParam, Collection<String> messages) throws Exception {
- sendSync(routingParam, messages);
- }
+ @Override
+ public void sendAsync(String routingParam, Collection<String> messages) throws Exception {
+ sendSync(routingParam, messages);
+ }
- @Override
- public int sendSync(String message) throws Exception {
- channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
- log.debug(" [x] Sent '" + message + "'");
- return 1;
- }
+ @Override
+ public int sendSync(String message) throws Exception {
+ channel.basicPublish(config.getExchangeName(), config.getRoutingKey(), null, message.getBytes());
+ log.debug(" [x] Sent '" + message + "'");
+ return 1;
+ }
- @Override
- public int sendSync(Collection<String> messages) throws Exception {
- log.debug("Publishing" + messages.size() + " messages ");
+ @Override
+ public int sendSync(Collection<String> messages) throws Exception {
+ log.debug("Publishing" + messages.size() + " messages ");
for (String message : messages) {
sendSync(message);
}
return messages.size();
- }
+ }
- @Override
- public int sendSync(String routingParam, String message) throws Exception {
- //Can only route if exchange is setup
- if(queueName == null) {
- channel.basicPublish(exchangeName, routingParam, null, message.getBytes());
- return 1;
- }
- else
- throw new UnsupportedOperationException("Routing without exchange");
-
- }
+ @Override
+ public int sendSync(String routingParam, String message) throws Exception {
+ channel.basicPublish(config.getExchangeName(), routingParam, null, message.getBytes());
+ return 1;
+ }
- @Override
- public int sendSync(String routingParam, Collection<String> messages) throws Exception {
- log.debug("Publishing" + messages.size() + " messages ");
- //Can only route if exchange is setup
- if(queueName == null) {
- for (String message : messages) {
- sendSync(routingParam, message);
- }
- return messages.size();
- }
- else
- throw new UnsupportedOperationException("Routing without exchange");
- }
+ @Override
+ public int sendSync(String routingParam, Collection<String> messages) throws Exception {
+ log.debug("Publishing" + messages.size() + " messages ");
+ for (String message : messages) {
+ sendSync(routingParam, message);
+ }
+ return messages.size();
+ }
}
diff --git a/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqUtils.java b/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqUtils.java
new file mode 100644
index 0000000..0c5177a
--- /dev/null
+++ b/event-client-rabbitmq/src/main/java/org/onap/aai/event/client/RabbitMqUtils.java
@@ -0,0 +1,127 @@
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2018 European Software Marketing Ltd.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.aai.event.client;
+
+import java.io.FileInputStream;
+import java.security.KeyStore;
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.naming.ConfigurationException;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+
+import com.rabbitmq.client.Address;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.DefaultSaslConfig;
+
+public class RabbitMqUtils {
+ private static ConnectionFactory factory = new ConnectionFactory();
+
+ /**
+ * (intended for testing purpose only)
+ * @param connFactory
+ */
+ static void setConnectionFactory(ConnectionFactory connFactory) {
+ factory = connFactory;
+ }
+
+ public static Connection createConnection(RabbitMqClientConfig config) throws Exception {
+ if (config.getHosts() == null) {
+ throw new ConfigurationException("Mandatory config param hosts not set");
+ }
+
+ List<Address> addresses = new ArrayList<Address>();
+ String[] hosts = config.getHosts().split(",");
+ for (String host : hosts) {
+ String[] parts = host.split(":");
+ if (parts.length != 2) {
+ throw new ConfigurationException("Hosts must be specified in 'host:port' format");
+ }
+
+ int port = Integer.parseInt(parts[1]);
+ Address add = new Address(parts[0], port);
+ addresses.add(add);
+ }
+
+ if (config.getUsername() == null && !config.getEnableSsl()) {
+ throw new ConfigurationException("Mandatory config param username not set");
+ }
+
+ factory.setUsername(config.getUsername());
+
+ if (config.getPassword() == null && !config.getEnableSsl()) {
+ throw new ConfigurationException("Mandatory config param password not set");
+ }
+
+ factory.setPassword(config.getPassword());
+
+ if (config.getVirtualHost() != null) {
+ factory.setVirtualHost(config.getVirtualHost());
+ }
+
+ checkSSL(factory,config);
+ factory.setConnectionTimeout(120000);
+ if (config.getConnectionTimeout() != null) {
+ try {
+ int timeout = Integer.parseInt(config.getConnectionTimeout());
+ factory.setConnectionTimeout(timeout);
+ } catch (NumberFormatException ex) {
+ }
+ }
+
+ return factory.newConnection(addresses);
+ }
+
+ private static void checkSSL(ConnectionFactory factory, RabbitMqClientConfig config) throws Exception {
+
+ // Check if any of SSL params is configured
+ if (config.getEnableSsl()) {
+ if (config.getSslKeyStoreFile() == null || config.getSslKeyStorePassword() == null
+ || config.getSslTrustStoreFile() == null || config.getSslTrustStorePassword() == null) {
+ throw new ConfigurationException(
+ "Missing SSL configuration : sslKeyStoreFile , sslKeyStorePassword , sslTrustStorePassword or sslTrustStoreFile");
+ }
+ char[] keyPassphrase = config.getSslKeyStorePassword().toCharArray();
+ KeyStore ks = KeyStore.getInstance("PKCS12");
+ ks.load(new FileInputStream(config.getSslKeyStoreFile()), keyPassphrase);
+
+ KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509");
+ kmf.init(ks, keyPassphrase);
+
+ char[] trustPassphrase = config.getSslTrustStorePassword().toCharArray();
+ KeyStore tks = KeyStore.getInstance("JKS");
+ tks.load(new FileInputStream(config.getSslTrustStoreFile()), trustPassphrase);
+
+ TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
+ tmf.init(tks);
+ SSLContext c = SSLContext.getInstance("TLSv1.2");
+ c.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
+ factory.setSaslConfig(DefaultSaslConfig.EXTERNAL);
+ factory.useSslProtocol(c);
+ }
+
+ }
+
+}