support SSL in rabbit mq in event client lib

Issue-ID: AAI-1929

Change-Id: I5912b0814e4b565942e32c5ed6d5094b45040c33
Signed-off-by: Bansal, Nitin (nb121v) <nitin.bansal@amdocs.com>

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);
+        }
+
+    }
+
+}