Policy 1707 commit to LF

Change-Id: Ibe6f01d92f9a434c040abb05d5386e89d675ae65
Signed-off-by: ITSERVICES\rb7147 <rb7147@att.com>

9 files changed, 878 insertions, 330 deletions

diff --git a/PolicyEngineUtils/pom.xml b/PolicyEngineUtils/pom.xml
index 50d98ebbb..98eae1e3c 100644
--- a/PolicyEngineUtils/pom.xml
+++ b/PolicyEngineUtils/pom.xml
@@ -17,7 +17,6 @@
   limitations under the License.
   ============LICENSE_END=========================================================
   -->
-
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
@@ -26,13 +25,32 @@
 		<artifactId>PolicyEngineSuite</artifactId>
 		<version>1.1.0-SNAPSHOT</version>
 	</parent>
-
 	<artifactId>PolicyEngineUtils</artifactId>
 	<dependencies>
 		<dependency>
-			<groupId>log4j</groupId>
-			<artifactId>log4j</artifactId>
-			<version>1.2.17</version>
+			<groupId>org.openecomp.policy.common</groupId>
+			<artifactId>integrity-monitor</artifactId>
+			<version>${common-modules.version}</version> 
+		</dependency>
+		<dependency>
+			<groupId>org.openecomp.policy.common</groupId>
+			<artifactId>integrity-audit</artifactId>
+			<version>${common-modules.version}</version> 
+		</dependency>
+		<dependency>
+			<groupId>org.openecomp.policy.common</groupId>
+			<artifactId>ECOMP-Logging</artifactId>
+			<version>${common-modules.version}</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.powermock</groupId>
+					<artifactId>powermock-module-junit4</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>org.powermock</groupId>
+					<artifactId>powermock-core</artifactId>
+				</exclusion>
+			</exclusions>
 		</dependency>
 		<dependency>
 			<groupId>org.eclipse.persistence</groupId>
@@ -45,11 +63,6 @@
 			<version>2.6.0</version>
 		</dependency>
 		<dependency>
-			<groupId>com.h2database</groupId>
-			<artifactId>h2</artifactId>
-			<version>[1.4.186,)</version>
-		</dependency>
-		<dependency>
 			<groupId>mysql</groupId>
 			<artifactId>mysql-connector-java</artifactId>
 			<version>5.1.30</version>
@@ -65,21 +78,57 @@
 			<version>1.9</version>
 		</dependency>
 		<dependency>
+			<groupId>org.apache.maven</groupId>
+			<artifactId>maven-model</artifactId>
+			<version>3.3.9</version>
+		</dependency>
+		<dependency>
 			<groupId>junit</groupId>
 			<artifactId>junit</artifactId>
 			<version>4.11</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.mariadb.jdbc</groupId>
+			<artifactId>mariadb-java-client</artifactId>
+			<version>1.2.3</version>
 		</dependency>
-		<!-- https://mvnrepository.com/artifact/com.att.cadi/cadi-aaf -->
-		<dependency>
-    		<groupId>com.att.cadi</groupId>
-    		<artifactId>cadi-aaf</artifactId>
-    		<version>1.3.0</version>
-    		<exclusions>
-    			<exclusion>
-    				<groupId>com.att.aft</groupId>
-					<artifactId>dme2</artifactId>
-    			</exclusion>
-    		</exclusions>
+		<dependency>
+			<groupId>org.drools</groupId>
+			<artifactId>drools-verifier</artifactId>
+			<version>6.3.0.Final</version>
+		</dependency>
+		<dependency>
+			<groupId>com.att.cadi</groupId>
+			<artifactId>cadi-aaf</artifactId>
+			<version>1.3.0</version>
+		</dependency>
+		<dependency>
+    		<groupId>com.att.aft</groupId>
+    		<artifactId>dme2</artifactId>
+    		<version>3.1.200</version>
+		</dependency>
+		<dependency>
+			<groupId>com.att.nsa</groupId>
+			<artifactId>cambriaClient</artifactId>
+			<version>0.0.1</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.slf4j</groupId>
+					<artifactId>slf4j-log4j12</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>com.att.nsa</groupId>
+			<artifactId>dmaapClient</artifactId>
+			<version>0.2.12</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.slf4j</groupId>
+					<artifactId>slf4j-log4j12</artifactId>
+				</exclusion>
+			</exclusions>
 		</dependency>
 	</dependencies>
 </project>
diff --git a/PolicyEngineUtils/src/main/java/org/openecomp/policy/api/PEDependency.java b/PolicyEngineUtils/src/main/java/org/openecomp/policy/api/PEDependency.java
new file mode 100644
index 000000000..214174197
--- /dev/null
+++ b/PolicyEngineUtils/src/main/java/org/openecomp/policy/api/PEDependency.java
@@ -0,0 +1,90 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * PolicyEngineUtils
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.openecomp.policy.api;
+
+import org.apache.maven.model.Dependency;
+import org.apache.maven.model.Exclusion;
+
+public class PEDependency{
+    
+    private String groupId;
+    private String artifactId;
+    private String version;
+    private String classifier;
+	private String type;
+    private String scope;
+    private java.util.List<Exclusion> exclusions;
+    
+    public String getGroupId() {
+        return groupId;
+    }
+    public void setGroupId(String groupId) {
+        this.groupId = groupId;
+    }
+    public String getArtifactId() {
+        return artifactId;
+    }
+    public void setArtifactId(String artifactId) {
+        this.artifactId = artifactId;
+    }
+    public String getVersion() {
+        return version;
+    }
+    public void setVersion(String version) {
+        this.version = version;
+    }
+    public java.util.List<Exclusion> getExclusions() {
+        return exclusions;
+    }
+    public void setExclusions(java.util.List<Exclusion> exclusions) {
+        this.exclusions = exclusions;
+    }
+    public String getClassifier() {
+		return classifier;
+	}
+	public void setClassifier(String classifier) {
+		this.classifier = classifier;
+	}
+	public String getType() {
+		return type;
+	}
+	public void setType(String type) {
+		this.type = type;
+	}
+	public String getScope() {
+		return scope;
+	}
+	public void setScope(String scope) {
+		this.scope = scope;
+	}
+	
+    public Dependency getDependency(){
+        Dependency dependency = new Dependency();
+        dependency.setArtifactId(artifactId);
+        dependency.setGroupId(groupId);
+        dependency.setVersion(version);
+        dependency.setExclusions(exclusions);
+        dependency.setClassifier(classifier);
+        dependency.setScope(scope);
+        dependency.setType(type);
+        return dependency;
+    }
+}
diff --git a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/AAFPolicyClient.java b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/AAFPolicyClient.java
index 1565c18d0..0c8de2702 100644
--- a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/AAFPolicyClient.java
+++ b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/AAFPolicyClient.java
@@ -1,208 +1,30 @@
-/*-
- * ============LICENSE_START=======================================================
- * PolicyEngineUtils
- * ================================================================================
- * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * 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.openecomp.policy.utils;
 
-import java.io.IOException;
+import java.lang.reflect.Method;
 import java.util.Properties;
 
-import org.apache.log4j.Logger;
-
-import com.att.cadi.Access;
-import com.att.cadi.Access.Level;
-import com.att.cadi.CadiException;
-import com.att.cadi.aaf.AAFPermission;
-import com.att.cadi.aaf.v2_0.AAFAuthn;
-import com.att.cadi.aaf.v2_0.AAFCon;
-import com.att.cadi.aaf.v2_0.AAFConDME2;
-import com.att.cadi.aaf.v2_0.AAFLurPerm;
-import com.att.cadi.config.Config;
-
-
-/**
- * AAF Client: Generic AAF Client implementation to connect to AAF Resources to validate permissions and authorization. 
- * 
- */
-public class AAFPolicyClient {
-	  private static Logger LOGGER = Logger.getLogger(AAFPolicyClient.class.getName());
-
-		private static final String DEFAULT_AFT_LATITUDE = "32.780140";
-		private static final String DEFAULT_AFT_LONGITUDE = "-96.800451";
-		private static final String DEFAULT_AAF_USER_EXPIRES = Integer.toString(5*60000); 	// 5 minutes for found items to live in cache
-		private static final String DEFAULT_AAF_HIGH_COUNT = Integer.toString(400); 		// Maximum number of items in Cache
-		
-		private static AAFPolicyClient instance = null;
-		private static Properties props = new Properties();
-		private static AAFCon<?> aafCon = null;
-		private static AAFLurPerm aafLurPerm = null;
-		private static AAFAuthn<?> aafAuthn = null;
-		private static Access access = null;
-		
-		
-		private AAFPolicyClient(Properties properties) throws AAFPolicyException{
-			if(instance == null){
-				instance = this;
-			}
-			setup(properties);
-		}
-
-		/**
-		 * Gets the instance of the AAFClient instance. Needs Proper properties with CLIENT_ID, CLIENT_KEY and ENVIRONMENT
-		 * 
-		 * @param properties Properties with CLIENT_ID, CLIENT_KEY and ENVIRONMENT
-		 * @return AAFClient instance. 
-		 * @throws AAFPolicyException Exceptions. 
-		 */
-		public static synchronized AAFPolicyClient getInstance(Properties properties) throws AAFPolicyException{
-			if(instance == null) {
-			    LOGGER.info("Creating AAFClient Instance ");
-				instance = new AAFPolicyClient(properties);
-			}
+public interface AAFPolicyClient {
+	/*
+	 * Enumeration for the Resource Node Naming. Add here if required. 
+	 */
+	String AAF_DEFAULT_CLASS = "org.openecomp.policy.utils.AAFPolicyClientImpl";
+	public enum Environment{
+		DEVL,
+		TEST,
+		PROD
+	}
+	public boolean checkAuth(String userName, String pass);
+	public void updateProperties(Properties properties) throws AAFPolicyException;
+	public boolean checkAuthPerm(String mechID, String pass, String type, String instance, String action);
+	public boolean checkPerm(String userName, String pass, String type, String instance, String action);
+	public static AAFPolicyClient getInstance(Properties properties) throws AAFPolicyException{
+		try {
+			Class<?> aafPolicyClient = Class.forName(properties.getProperty("aafClient.impl.className", AAF_DEFAULT_CLASS));
+			Method method =  aafPolicyClient.getMethod("getInstance", Properties.class);
+			AAFPolicyClient instance = (AAFPolicyClient) method.invoke(null, properties);
 			return instance;
+		} catch (Exception e) {
+			throw new AAFPolicyException(e);
 		}
-		
-		// To set Property values && Connections. 
-		private void setup(Properties properties) throws AAFPolicyException {
-			/*if(properties!=null && !properties.isEmpty()){
-				props = System.getProperties();
-				props.setProperty("AFT_LATITUDE", properties.getProperty("AFT_LATITUDE", DEFAULT_AFT_LATITUDE));
-				props.setProperty("AFT_LONGITUDE", properties.getProperty("AFT_LONGITUDE", DEFAULT_AFT_LONGITUDE));
-				props.setProperty("aaf_id",properties.getProperty("aaf_id", "aafID"));
-				props.setProperty("aaf_password", properties.getProperty("aaf_password", "aafPass"));
-				if(properties.containsKey(Config.AAF_URL)){
-		        	// if given a value in properties file. 
-		        	props.setProperty(Config.AAF_URL, properties.getProperty(Config.AAF_URL));
-	        	}else{
-	        		LOGGER.error("Required Property value is missing : " + Config.AAF_URL);
-	 				throw new AAFPolicyException("Required Property value is missing : " + Config.AAF_URL);
-	        	}
-				
-				if(properties.containsKey("AFT_ENVIRONMENT")){
-					props.setProperty("AFT_ENVIRONMENT", properties.getProperty("AFT_ENVIRONMENT"));
-	        	}else{
-	        		LOGGER.error("Required Property value is missing : AFT_ENVIRONMENT");
-	 				throw new AAFPolicyException("Required Property value is missing : AFT_ENVIRONMENT");
-	        	}
-		        props.setProperty(Config.AAF_USER_EXPIRES, properties.getProperty(Config.AAF_USER_EXPIRES, DEFAULT_AAF_USER_EXPIRES));	
-		        props.setProperty(Config.AAF_HIGH_COUNT, properties.getProperty(Config.AAF_HIGH_COUNT, DEFAULT_AAF_HIGH_COUNT));
-			}else{
-			    LOGGER.error("Required Property value is missing ");
-				throw new AAFPolicyException("Required Property value is missing");
-			}
-			access = new PolicyAccess(props, Level.valueOf(properties.getProperty("AAF_LOG_LEVEL", Level.INFO.toString())));
-			setUpAAF();*/
-		}
-		
-		/**
-		 * Updates the Properties file in case if required. 
-		 * 
-		 * @param properties  Properties with CLIENT_ID, CLIENT_KEY and ENVIRONMENT
-		 * @throws AAFPolicyException exceptions if any.
-		 */
-		public void updateProperties(Properties properties) throws AAFPolicyException{
-			setup(properties);
-		}
-		
-		/**
-		 * Checks the Authentication and Permissions for the given values. 
-		 * 
-		 * @param pass Password pertaining to the loginId
-		 * @param type Permissions Type.
-		 * @param instance Permissions Instance. 
-		 * @param action Permissions Action. 
-		 * @return
-		 */
-		public boolean checkAuthPerm(String mechID, String pass, String type, String instance, String action){
-			if(checkAuth(mechID, pass) && checkPerm(mechID, pass, type, instance, action)){
-				return true;
-			}
-			return false;
-		}
-		
-		/**
-		 * Checks the Authentication of the UserName and Password Given. 
-		 * 
-		 * @param userName UserName 
-		 * @param pass Password.
-		 * @return True or False. 
-		 */
-		public boolean checkAuth(String userName, String pass){
-			/*if(aafAuthn!=null){
-				try {
-				    int i=0;
-				    do{
-				        if(aafAuthn.validate(userName, pass)==null){ 
-		                    return true; 
-		                }
-				        i++;
-				    }while(i<2);
-				} catch (Exception e) {
-				    LOGGER.error(e.getMessage());
-				}
-			}
-			LOGGER.debug("Authentication failed for : " + userName + " in " + props.getProperty("AAF_URL"));
-			return false;*/
-			return true;
-		}
-		
-		/**
-		 * Checks Permissions for the given UserName, Password and Type, Instance Action. 
-		 * 
-		 * @param userName UserName 
-		 * @param pass Password.
-		 * @param type Permissions Type. 
-		 * @param instance Permissions Instance. 
-		 * @param action Permissions Action. 
-		 * @return True or False. 
-		 */
-		public boolean checkPerm(String userName, String pass, String type, String instance, String action){
-			/*int i =0;
-			Boolean result= false;
-		    do{
-			    if(aafCon!=null && aafLurPerm !=null){
-		            try {
-		                aafCon.basicAuth(userName, pass);
-		                AAFPermission perm = new AAFPermission(type, instance, action);
-		                result = aafLurPerm.fish(userName, perm);
-		            } catch (CadiException e) {
-		                LOGGER.error(e.getMessage());
-		                aafLurPerm.destroy();
-		            }
-		        }
-			    LOGGER.debug("Permissions for : " + userName + " in " + props.getProperty("AAF_URL") + "for " + type  + "," + instance + "," + action + "\n Result is: " + result);
-		        i++;
-			}while(i<2 && !result); // Try once more to check if this can be passed. AAF has some issues. 
-	        return result;*/
-			return true;
-		}
-		
-		/*private boolean setUpAAF(){
-		    try {
-	            aafCon = new AAFConDME2(access);
-	            aafLurPerm = aafCon.newLur();//new AAFLurPerm(aafCon);
-	            aafAuthn = aafCon.newAuthn(aafLurPerm);//new AAFAuthn(aafCon, aafLurPerm);
-	            return true;
-	        } catch (Exception e) {
-	            LOGGER.error("Error while creating Connection " + e.getMessage());
-	            return false;
-	        }
-		}*/
+	}
 }
diff --git a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/AAFPolicyClientImpl.java b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/AAFPolicyClientImpl.java
new file mode 100644
index 000000000..bd640261f
--- /dev/null
+++ b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/AAFPolicyClientImpl.java
@@ -0,0 +1,217 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * PolicyEngineUtils
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.openecomp.policy.utils;
+
+import java.util.Properties;
+
+import org.apache.log4j.Logger;
+
+import com.att.cadi.Access;
+import com.att.cadi.Access.Level;
+import com.att.cadi.CadiException;
+import com.att.cadi.aaf.AAFPermission;
+import com.att.cadi.aaf.v2_0.AAFAuthn;
+import com.att.cadi.aaf.v2_0.AAFCon;
+import com.att.cadi.aaf.v2_0.AAFConDME2;
+import com.att.cadi.aaf.v2_0.AAFLurPerm;
+import com.att.cadi.config.Config;
+
+
+/**
+ * AAF Client: Generic AAF Client implementation to connect to AAF Resources to validate permissions and authorization. 
+ * 
+ */
+public class AAFPolicyClientImpl implements AAFPolicyClient{
+	private static Logger LOGGER = Logger.getLogger(AAFPolicyClientImpl.class.getName());
+
+	private static final String ENVIRONMENT = "ENVIRONMENT";
+	
+	// Warning Please don't Change these Values. Confirm with AAF team.  
+	private static final String DEVL_AAF_URL = "";
+	private static final String TEST_AAF_URL = "";
+	private static final String PROD_AAF_URL = "";
+	private static final String DEFAULT_AFT_LATITUDE = "32.780140";
+	private static final String DEFAULT_AFT_LONGITUDE = "-96.800451";
+	private static final String TEST_AFT_ENVIRONMENT = "AFTUAT";
+	private static final String PROD_AFT_ENVIRONMENT = "AFTPRD";
+	private static final String DEFAULT_AAF_USER_EXPIRES = Integer.toString(5*60000); 	// 5 minutes for found items to live in cache
+	private static final String DEFAULT_AAF_HIGH_COUNT = Integer.toString(400); 		// Maximum number of items in Cache
+
+	private static AAFPolicyClientImpl instance = null;
+
+	private static Properties props = new Properties();
+	private static AAFCon<?> aafCon = null;
+	private static AAFLurPerm aafLurPerm = null;
+	private static AAFAuthn<?> aafAuthn = null;
+	private static Access access = null;
+
+	private AAFPolicyClientImpl(Properties properties) throws AAFPolicyException{
+		if(instance == null){
+			instance = this;
+		}
+		setup(properties);
+	}
+
+	/**
+	 * Gets the instance of the AAFClient instance. Needs Proper properties with CLIENT_ID, CLIENT_KEY and ENVIRONMENT
+	 * 
+	 * @param properties Properties with CLIENT_ID, CLIENT_KEY and ENVIRONMENT
+	 * @return AAFClient instance. 
+	 * @throws AAFPolicyException Exceptions. 
+	 */
+	public static synchronized AAFPolicyClientImpl getInstance(Properties properties) throws AAFPolicyException{
+		if(instance == null) {
+			LOGGER.info("Creating AAFClient Instance ");
+			instance = new AAFPolicyClientImpl(properties);
+		}
+		return instance;
+	}
+
+	// To set Property values && Connections. 
+	private void setup(Properties properties) throws AAFPolicyException {
+		if(properties!=null && !properties.isEmpty()){
+			props = System.getProperties();
+			props.setProperty("AFT_LATITUDE", properties.getProperty("AFT_LATITUDE", DEFAULT_AFT_LATITUDE));
+			props.setProperty("AFT_LONGITUDE", properties.getProperty("AFT_LONGITUDE", DEFAULT_AFT_LONGITUDE));
+			String aftEnv = TEST_AFT_ENVIRONMENT;
+			//props.setProperty(Config.CADI_KEYFILE,"keyfile");
+			props.setProperty("aaf_id",properties.getProperty("aaf_id", "aafID"));
+			props.setProperty("aaf_password", properties.getProperty("aaf_password", "aafPass"));
+			if(properties.containsKey(Config.AAF_URL)){
+				// if given a value in properties file. 
+				props.setProperty(Config.AAF_URL, properties.getProperty(Config.AAF_URL));
+			}else{
+				// Set Default values. 
+				if(properties.getProperty(ENVIRONMENT, "DEVL").equalsIgnoreCase(Environment.TEST.toString())){
+					props.setProperty(Config.AAF_URL, TEST_AAF_URL);
+				}else if(properties.getProperty(ENVIRONMENT, "DEVL").equalsIgnoreCase(Environment.PROD.toString())){
+					props.setProperty(Config.AAF_URL, PROD_AAF_URL);
+					aftEnv = PROD_AFT_ENVIRONMENT;
+				}else{
+					props.setProperty(Config.AAF_URL, DEVL_AAF_URL);
+				}
+			}
+			props.setProperty("AFT_ENVIRONMENT", properties.getProperty("AFT_ENVIRONMENT", aftEnv));
+			props.setProperty(Config.AAF_USER_EXPIRES, properties.getProperty(Config.AAF_USER_EXPIRES, DEFAULT_AAF_USER_EXPIRES));	
+			props.setProperty(Config.AAF_HIGH_COUNT, properties.getProperty(Config.AAF_HIGH_COUNT, DEFAULT_AAF_HIGH_COUNT));
+		}else{
+			LOGGER.error("Required Property value is missing : " + ENVIRONMENT);
+			throw new AAFPolicyException("Required Property value is missing : " + ENVIRONMENT);
+		}
+		access = new PolicyAccess(props, Level.valueOf(properties.getProperty("AAF_LOG_LEVEL", Level.ERROR.toString())));
+		setUpAAF();
+	}
+
+	/**
+	 * Updates the Properties file in case if required. 
+	 * 
+	 * @param properties  Properties with CLIENT_ID, CLIENT_KEY and ENVIRONMENT
+	 * @throws AAFPolicyException exceptions if any.
+	 */
+	public void updateProperties(Properties properties) throws AAFPolicyException{
+		setup(properties);
+	}
+
+	/**
+	 * Checks the Authentication and Permissions for the given values. 
+	 * 
+	 * @param mechID MechID or ATT ID must be registered under the Name space. 
+	 * @param pass Password pertaining to the MechID or ATTID. 
+	 * @param type Permissions Type.
+	 * @param instance Permissions Instance. 
+	 * @param action Permissions Action. 
+	 * @return
+	 */
+	public boolean checkAuthPerm(String mechID, String pass, String type, String instance, String action){
+		if(checkAuth(mechID, pass) && checkPerm(mechID, pass, type, instance, action)){
+			return true;
+		}
+		return false;
+	}
+
+	/**
+	 * Checks the Authentication of the UserName and Password Given. 
+	 * 
+	 * @param userName UserName or MechID
+	 * @param pass Password.
+	 * @return True or False. 
+	 */
+	public boolean checkAuth(String userName, String pass){
+		if(aafAuthn!=null){
+			try {
+				int i=0;
+				do{
+					if(aafAuthn.validate(userName, pass)==null){ 
+						return true; 
+					}
+					i++;
+				}while(i<2);
+			} catch (Exception e) {
+				LOGGER.error(e.getMessage());
+			}
+		}
+		LOGGER.info("Authentication failed for : " + userName + " in " + props.getProperty(Config.AAF_URL));
+		return false;
+	}
+
+	/**
+	 * Checks Permissions for the given UserName, Password and Type, Instance Action. 
+	 * 
+	 * @param userName UserName or MechID
+	 * @param pass Password.
+	 * @param type Permissions Type. 
+	 * @param instance Permissions Instance. 
+	 * @param action Permissions Action. 
+	 * @return True or False. 
+	 */
+	public boolean checkPerm(String userName, String pass, String type, String instance, String action){
+		int i =0;
+		Boolean result= false;
+		do{
+			if(aafCon!=null && aafLurPerm !=null){
+				try {
+					aafCon.basicAuth(userName, pass);
+					AAFPermission perm = new AAFPermission(type, instance, action);
+					result = aafLurPerm.fish(userName, perm);
+				} catch (CadiException e) {
+					LOGGER.error(e.getMessage());
+					aafLurPerm.destroy();
+				}
+			}
+			LOGGER.info("Permissions for : " + userName + " in " + props.getProperty(Config.AAF_URL) + " for " + type  + "," + instance + "," + action + "\n Result is: " + result);
+			i++;
+		}while(i<2 && !result); // Try once more to check if this can be passed. AAF has some issues. 
+		return result;
+	}
+
+	private boolean setUpAAF(){
+		try {
+			aafCon = new AAFConDME2(access);
+			aafLurPerm = aafCon.newLur();//new AAFLurPerm(aafCon);
+			aafAuthn = aafCon.newAuthn(aafLurPerm);//new AAFAuthn(aafCon, aafLurPerm);
+			return true;
+		} catch (Exception e) {
+			LOGGER.error("Error while setting up AAF Connection " + e.getMessage());
+			return false;
+		}
+	}
+}
diff --git a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BackUpMonitor.java b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BackUpMonitor.java
index 1b35dcc57..f041d105c 100644
--- a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BackUpMonitor.java
+++ b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BackUpMonitor.java
@@ -21,6 +21,7 @@
 package org.openecomp.policy.utils;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.Properties;
@@ -50,9 +51,9 @@ import com.github.fge.jsonpatch.diff.JsonDiff;
  * 
  */
 public class BackUpMonitor {
-	private static final Logger logger = Logger.getLogger(BackUpMonitor.class.getName());
-	private static final int DEFAULT_PING = 60000; // Value is in milliseconds. 
-	
+	private static final Logger LOGGER = Logger.getLogger(BackUpMonitor.class.getName());
+	private static final int DEFAULT_PING = 15000; // Value is in milliseconds. 
+
 	private static BackUpMonitor instance = null;
 	private static String resourceName = null;
 	private static String resourceNodeName = null;
@@ -65,7 +66,7 @@ public class BackUpMonitor {
 	private static BackUpHandler handler= null;
 	private EntityManager em;
 	private EntityManagerFactory emf;
-	
+
 	/*
 	 * Enumeration for the Resource Node Naming. Add here if required. 
 	 */
@@ -73,7 +74,7 @@ public class BackUpMonitor {
 		BRMS,
 		ASTRA
 	}
-	
+
 	private BackUpMonitor(String resourceNodeName, String resourceName, Properties properties, BackUpHandler handler) throws Exception{
 		if(instance == null){
 			instance = this;
@@ -85,14 +86,14 @@ public class BackUpMonitor {
 		properties.setProperty(PersistenceUnitProperties.ECLIPSELINK_PERSISTENCE_XML, "META-INF/persistencePU.xml");
 		emf = Persistence.createEntityManagerFactory("PolicyEngineUtils", properties);
 		if(emf==null){
-			logger.error("Unable to create Entity Manger Factory ");
+			LOGGER.error("Unable to create Entity Manger Factory ");
 			throw new Exception("Unable to create Entity Manger Factory");
 		}
 		em = emf.createEntityManager();
-		
+
 		// Check Database if this is Master or Slave.
 		checkDataBase();
-		
+
 		// Start thread.
 		Thread t = new Thread(new BMonitor());
 		t.start();
@@ -108,103 +109,103 @@ public class BackUpMonitor {
 	 */
 	public static synchronized BackUpMonitor getInstance(String resourceNodeName, String resourceName, Properties properties, BackUpHandler handler) throws Exception {
 		if(resourceNodeName==null || resourceNodeName.trim().equals("") ||resourceName==null|| resourceName.trim().equals("") || properties == null || handler==null){
-			logger.error("Error while getting Instance. Please check resourceName and/or properties file");
+			LOGGER.error("Error while getting Instance. Please check resourceName and/or properties file");
 			return null;
 		}else if((resourceNodeName.equals(ResourceNode.ASTRA.toString()) || resourceNodeName.equals(ResourceNode.BRMS.toString())) && validate(properties) && instance==null){
-			logger.info("Creating Instance of BackUpMonitor");
+			LOGGER.info("Creating Instance of BackUpMonitor");
 			instance = new BackUpMonitor(resourceNodeName, resourceName, properties, handler);
 		}
 		return instance;
 	}
-	
+
 	// This is to validate given Properties with required values. 
 	private static Boolean validate(Properties properties){
 		if(properties.getProperty("javax.persistence.jdbc.driver")==null ||properties.getProperty("javax.persistence.jdbc.driver").trim().equals("")){
-			logger.error("javax.persistence.jdbc.driver property is empty");
+			LOGGER.error("javax.persistence.jdbc.driver property is empty");
 			return false;
 		}
 		if(properties.getProperty("javax.persistence.jdbc.url")==null || properties.getProperty("javax.persistence.jdbc.url").trim().equals("")){
-			logger.error("javax.persistence.jdbc.url property is empty");
+			LOGGER.error("javax.persistence.jdbc.url property is empty");
 			return false;
 		}
 		if(properties.getProperty("javax.persistence.jdbc.user")==null || properties.getProperty("javax.persistence.jdbc.user").trim().equals("")){
-			logger.error("javax.persistence.jdbc.user property is empty");
+			LOGGER.error("javax.persistence.jdbc.user property is empty");
 			return false;
 		}
 		if(properties.getProperty("javax.persistence.jdbc.password")==null || properties.getProperty("javax.persistence.jdbc.password").trim().equals("")){
-			logger.error("javax.persistence.jdbc.password property is empty");
+			LOGGER.error("javax.persistence.jdbc.password property is empty");
 			return false;
 		}
 		if(properties.getProperty("ping_interval")==null || properties.getProperty("ping_interval").trim().equals("")){
-			logger.info("ping_interval property not specified. Taking default value");
+			LOGGER.info("ping_interval property not specified. Taking default value");
 		}else{
 			try{
 				pingInterval = Integer.parseInt(properties.getProperty("ping_interval").trim());
 			}catch(NumberFormatException e){
-				logger.warn("Ignored invalid proeprty ping_interval. Taking default value.");
+				LOGGER.warn("Ignored invalid proeprty ping_interval. Taking default value: " + pingInterval);
 				pingInterval = DEFAULT_PING;
 			}
 		}
 		return true;
 	}
-	
+
 	// Sets the Flag for masterFlag to either True or False. 
 	private static void setFlag(Boolean flag){
 		synchronized (lock) {
 			masterFlag = flag;
 		}
 	}
-	
+
 	/**
 	 * Gets the Boolean value of Master(True) or Slave mode (False)
 	 * 
 	 * @return Boolean flag which if True means that the operation needs to be performed(Master mode) or if false the operation is in slave mode. 
 	 */
-	public synchronized Boolean getFlag(){
+	public Boolean getFlag(){
 		synchronized (lock) {
 			return masterFlag;
 		}
 	}
-	
+
 	// BackUpMonitor Thread 
 	private class BMonitor implements Runnable{
 		@Override
 		public void run() {
-			logger.info("Starting BackUpMonitor Thread.. ");
+			LOGGER.info("Starting BackUpMonitor Thread.. ");
 			while(true){
 				try {
 					TimeUnit.MILLISECONDS.sleep(pingInterval);
 					checkDataBase();
 				} catch (Exception e) {
-					logger.error("Error during Thread execution " + e.getMessage());
+					LOGGER.error("Error during Thread execution " + e.getMessage());
 				}
 			}
 		}
 	}
-	
+
 	// Set Master 
 	private static BackUpMonitorEntity setMaster(BackUpMonitorEntity bMEntity){
 		bMEntity.setFlag("MASTER");
 		setFlag(true);
 		return bMEntity;
 	}
-	
+
 	// Set Slave 
 	private static BackUpMonitorEntity setSlave(BackUpMonitorEntity bMEntity){
 		bMEntity.setFlag("SLAVE");
 		setFlag(false);
 		return bMEntity;
 	}
-	
+
 	// Check Database and set the Flag. 
 	private void checkDataBase() throws Exception {
 		EntityTransaction et = em.getTransaction();
 		notificationRecord = PolicyUtils.objectToJsonString(NotificationStore.getNotificationRecord());
 		// Clear Cache. 
-		logger.info("Clearing Cache");
+		LOGGER.info("Clearing Cache");
 		em.getEntityManagerFactory().getCache().evictAll();
 		try{
-			logger.info("Checking Datatbase for BackUpMonitor.. ");
+			LOGGER.info("Checking Datatbase for BackUpMonitor.. ");
 			et.begin();
 			Query query = em.createQuery("select b from BackUpMonitorEntity b where b.resourceNodeName = :nn");
 			if(resourceNodeName.equals(ResourceNode.ASTRA.toString())){
@@ -215,7 +216,7 @@ public class BackUpMonitor {
 			List<?> bMList = query.getResultList();
 			if(bMList.isEmpty()){
 				// This is New. create an entry as Master. 
-				logger.info("Adding resource " + resourceName + " to Database");
+				LOGGER.info("Adding resource " + resourceName + " to Database");
 				BackUpMonitorEntity bMEntity = new BackUpMonitorEntity();
 				bMEntity.setResoruceNodeName(resourceNodeName);
 				bMEntity.setResourceName(resourceName);
@@ -224,115 +225,154 @@ public class BackUpMonitor {
 				em.persist(bMEntity);
 				em.flush();
 			}else{
-				// Check if resourceName already exists and if there is a Master in Node.
-				Boolean masterFlag = false;
-				Boolean alreadyMaster = false;
-				Date currentTime;
-				BackUpMonitorEntity masterEntity = null;
-				BackUpMonitorEntity slaveEntity = null;
-				long timeDiff = 0;
+				// Check for other Master(s)
+				ArrayList<BackUpMonitorEntity> masterEntities = new ArrayList<BackUpMonitorEntity>();
+				// Check for self. 
+				BackUpMonitorEntity selfEntity = null;
+				// Check backup monitor entities.
 				for(int i=0; i< bMList.size(); i++){
 					BackUpMonitorEntity bMEntity = (BackUpMonitorEntity) bMList.get(i);
-					logger.info("Refreshing Entity. ");
+					LOGGER.info("Refreshing Entity. ");
 					em.refresh(bMEntity);
+					if(bMEntity.getFlag().equalsIgnoreCase("MASTER")){
+						masterEntities.add(bMEntity);
+					}
 					if(bMEntity.getResourceName().equals(resourceName)){
-						logger.info("Resource Name already Exists. " + resourceName);
-						if(bMEntity.getFlag().equalsIgnoreCase("MASTER")){
-							// Master mode
-							setFlag(true);
-							logger.info(resourceName + " is on Master Mode");
-							bMEntity.setTimeStamp(new Date());
-							bMEntity.setNotificationRecord(notificationRecord);
-							em.persist(bMEntity);
-							em.flush();
-							setLastNotification(null);
-							alreadyMaster = true;
-							break;
-						}else{
-							// Slave mode. 
-							setFlag(false);
-							slaveEntity = bMEntity;
-							logger.info(resourceName + " is on Slave Mode");
-						}
-					}else{
-						if(bMEntity.getFlag().equalsIgnoreCase("MASTER")){
-							// check if its time stamp is old.
-							currentTime = new Date();
-							timeDiff = currentTime.getTime()-bMEntity.getTimeStamp().getTime();
-							masterEntity = bMEntity;
-							masterFlag = true;
-						}
+						selfEntity = bMEntity;
 					}
 				}
-				// If there is master and no slave entry then add the slave entry to database. 
-				// If we are slave and there is a masterFlag then check timeStamp to find if master is down. 
-				if(!alreadyMaster){
-					BackUpMonitorEntity bMEntity;
-					if(slaveEntity==null){
-						bMEntity = new BackUpMonitorEntity();
+				if(selfEntity!=null){
+					LOGGER.info("Resource Name already Exists: " + resourceName);
+					if(selfEntity.getFlag().equalsIgnoreCase("MASTER")){
+						// Already Master Mode. 
+						setFlag(true);
+						LOGGER.info(resourceName + " is on Master Mode");
+						selfEntity.setTimeStamp(new Date());
+						selfEntity.setNotificationRecord(notificationRecord);
+						em.persist(selfEntity);
+						em.flush();
+						setLastNotification(null);
+						if(!masterEntities.contains(selfEntity)){
+							masterEntities.add(selfEntity);
+						}
 					}else{
-						bMEntity = slaveEntity;
+						// Already Slave Mode.
+						setFlag(false);
+						selfEntity.setTimeStamp(new Date());
+						selfEntity.setNotificationRecord(notificationRecord);
+						em.persist(selfEntity);
+						em.flush();
+						LOGGER.info(resourceName + " is on Slave Mode");
 					}
-					if(masterFlag && !getFlag()){
-						if(timeDiff > pingInterval){
-							// This is down or has an issue and we need to become Master while turning the Master to slave. 
-							masterEntity = setSlave(masterEntity);
-							String lastNotification = null;
-							if(masterEntity.getNotificationRecord()!=null){
-								lastNotification = calculatePatch(masterEntity.getNotificationRecord());
+				}else{
+					// Resource name is null -> No resource with same name. 
+					selfEntity = new BackUpMonitorEntity();
+					selfEntity.setResoruceNodeName(resourceNodeName);
+					selfEntity.setResourceName(resourceName);
+					selfEntity.setTimeStamp(new Date());
+					selfEntity = setSlave(selfEntity);
+					setLastNotification(null);
+					LOGGER.info("Creating: " + resourceName + " on Slave Mode");
+					em.persist(selfEntity);
+					em.flush();
+				}
+				// Correct the database if any errors and perform monitor checks.
+				if(masterEntities.size()!=1 || !getFlag()){
+					// We are either not master or there are more masters or no masters. 
+					if(masterEntities.size()==0){
+						// No Masters is a problem Convert ourselves to Master. 
+						selfEntity = setMaster(selfEntity);
+						selfEntity.setTimeStamp(new Date());
+						selfEntity.setNotificationRecord(notificationRecord);
+						LOGGER.info(resourceName + " changed to Master Mode - No Masters available.");
+						em.persist(selfEntity);
+						em.flush();
+					}else {
+						if(masterEntities.size()>1){
+							// More Masters is a problem, Fix the issue by looking for the latest one and make others Slave. 
+							BackUpMonitorEntity masterEntity = null;
+							for(BackUpMonitorEntity currentEntity: masterEntities){
+								if(currentEntity.getFlag().equalsIgnoreCase("MASTER")){
+									if(masterEntity==null){
+										masterEntity = currentEntity;
+									}else if(currentEntity.getTimeStamp().getTime() > masterEntity.getTimeStamp().getTime()){
+										// False Master, Update master to slave and take currentMaster as Master. 
+										masterEntity.setFlag("SLAVE");
+										masterEntity.setTimeStamp(new Date());
+										em.persist(masterEntity);
+										em.flush();
+										masterEntity = currentEntity;
+									}else{
+										currentEntity.setFlag("SLAVE");
+										currentEntity.setTimeStamp(new Date());
+										em.persist(currentEntity);
+										em.flush();
+									}
+								}
+							}
+							masterEntities = new ArrayList<BackUpMonitorEntity>();
+							masterEntities.add(masterEntity);
+						}
+						if(masterEntities.size()==1){
+							// Correct Size, Check if Master is Latest, if not Change Master to Slave and Slave to Master.   
+							BackUpMonitorEntity masterEntity = masterEntities.get(0);
+							if(!masterEntity.getResourceName().equals(selfEntity.getResourceName())){
+								Date currentTime = new Date();
+								long timeDiff = 0;
+								timeDiff = currentTime.getTime()-masterEntity.getTimeStamp().getTime();
+								if(timeDiff > (pingInterval+1500)){
+									// This is down or has an issue and we need to become Master while turning the Master to slave.
+									masterEntity.setFlag("SLAVE");
+									String lastNotification = null;
+									if(masterEntity.getNotificationRecord()!=null){
+										lastNotification = calculatePatch(masterEntity.getNotificationRecord());
+									}
+									setLastNotification(lastNotification);
+									em.persist(masterEntity);
+									em.flush();
+									// Lets Become Master. 
+									selfEntity = setMaster(selfEntity);
+									LOGGER.info("Changing "+ resourceName + " from slave to Master Mode");
+									selfEntity.setTimeStamp(new Date());
+									selfEntity.setNotificationRecord(notificationRecord);
+									em.persist(selfEntity);
+									em.flush();
+								}
 							}
-							setLastNotification(lastNotification);
-							em.persist(masterEntity);
-							em.flush();
-							bMEntity = setMaster(bMEntity);
-							logger.info(resourceName + " changed to Master Mode");
 						}else{
-							bMEntity = setSlave(bMEntity);
-							setLastNotification(null);
-							logger.info(resourceName + " is on Slave Mode");
+							LOGGER.error("Backup Monitor Issue, Masters out of sync, This will be fixed in next interval.");
 						}
-					}else{
-						// If there is no Master. we need to become Master. 
-						bMEntity = setMaster(bMEntity);
-						logger.info(resourceName + " is on Master Mode");
-						setLastNotification(null);
 					}
-					bMEntity.setNotificationRecord(notificationRecord);
-					bMEntity.setResoruceNodeName(resourceNodeName);
-					bMEntity.setResourceName(resourceName);
-					bMEntity.setTimeStamp(new Date());
-					em.persist(bMEntity);
-					em.flush();
 				}
 			}
 			et.commit();
 		}catch(Exception e){
-			logger.error("failed Database Operation " + e.getMessage());
+			LOGGER.error("failed Database Operation " + e.getMessage());
 			if(et.isActive()){
 				et.rollback();
 			}
 			throw new Exception(e);
 		}
 	}
-	
+
 	// Calculate Patch and return String JsonPatch of the notification Delta. 
 	private synchronized String calculatePatch(String oldNotificationRecord) {
 		try{
 			JsonNode notification = JsonLoader.fromString(notificationRecord);
 			JsonNode oldNotification = JsonLoader.fromString(oldNotificationRecord);
 			JsonNode patchNode = JsonDiff.asJson(oldNotification, notification);
-			logger.info("Generated JSON Patch is " + patchNode.toString());
+			LOGGER.info("Generated JSON Patch is " + patchNode.toString());
 			JsonPatch patch = JsonPatch.fromJson(patchNode);
 			try {
 				JsonNode patched = patch.apply(oldNotification);
-				logger.info("Generated New Notification is : " + patched.toString());
+				LOGGER.info("Generated New Notification is : " + patched.toString());
 				return patched.toString();
 			} catch (JsonPatchException e) {
-				logger.error("Error generating Patched "  +e.getMessage());
+				LOGGER.error("Error generating Patched "  +e.getMessage());
 				return null;
 			}
 		}catch(IOException e){
-			logger.error("Error generating Patched "  +e.getMessage());
+			LOGGER.error("Error generating Patched "  +e.getMessage());
 			return null;
 		}
 	}
@@ -346,31 +386,35 @@ public class BackUpMonitor {
 	public synchronized void updateNotification() throws Exception{
 		checkDataBase();
 	}
-	
+
 	// Take in string notification and send the record delta to Handler. 
 	private static void callHandler(String notification){
 		if(handler!=null){
 			try {
 				PDPNotification notificationObject = PolicyUtils.jsonStringToObject(notification, StdPDPNotification.class);
 				if(notificationObject.getNotificationType()!=null){
-					logger.info("Performing Patched notification ");
+					LOGGER.info("Performing Patched notification ");
 					try{
 						handler.runOnNotification(notificationObject);
+						notificationRecord = lastMasterNotification;
 					}catch (Exception e){
-						logger.error("Error in Clients Handler Object : " + e.getMessage());
+						LOGGER.error("Error in Clients Handler Object : " + e.getMessage());
 					}
 				}
 			} catch (IOException e) {
-				logger.info("Error while notification Conversion " + e.getMessage());
+				LOGGER.info("Error while notification Conversion " + e.getMessage());
 			}
 		}
 	}
-	
+
 	// Used to set LastMasterNotification Record. 
 	private static void setLastNotification(String notification){
 		synchronized(notificationLock){
 			lastMasterNotification = notification;
 			if(lastMasterNotification!=null && !lastMasterNotification.equals("\"notificationType\":null")){
+				if(lastMasterNotification.equals(notificationRecord)){
+					return;
+				}
 				callHandler(notification);
 			}
 		}
diff --git a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BusConsumer.java b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BusConsumer.java
new file mode 100644
index 000000000..9fcae39d7
--- /dev/null
+++ b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BusConsumer.java
@@ -0,0 +1,99 @@
+package org.openecomp.policy.utils;
+
+import java.util.List;
+import java.util.Properties;
+
+import com.att.nsa.mr.client.impl.MRConsumerImpl;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+public interface BusConsumer {
+	
+	/**
+	 * fetch messages
+	 * 
+	 * @return list of messages
+	 * @throws Exception when error encountered by underlying libraries
+	 */
+	public Iterable<String> fetch() throws Exception;
+	
+	/**
+	 * close underlying library consumer
+	 */
+	public void close();
+	
+	/**
+	 * MR based consumer
+	 */
+	public static class DmaapConsumerWrapper implements BusConsumer {
+		
+		/**
+		 * MR Consumer
+		 */
+		protected MRConsumerImpl consumer;
+		
+		/**
+		 * MR Consumer Wrapper
+		 * 
+		 * @param servers messaging bus hosts
+		 * @param topic topic
+		 * @param apiKey API Key
+		 * @param apiSecret API Secret
+		 * @param aafLogin AAF Login
+		 * @param aafPassword AAF Password
+		 * @param consumerGroup Consumer Group
+		 * @param consumerInstance Consumer Instance
+		 * @param fetchTimeout Fetch Timeout
+		 * @param fetchLimit Fetch Limit
+		 */
+		public DmaapConsumerWrapper(List<String> servers, String topic, 
+								String aafLogin, String aafPassword,
+								String consumerGroup, String consumerInstance,
+								int fetchTimeout, int fetchLimit) 
+		throws Exception {
+					
+			this.consumer = new MRConsumerImpl(servers, topic, 
+											   consumerGroup, consumerInstance, 
+											   fetchTimeout, fetchLimit, 
+									           null, aafLogin, aafPassword);
+			
+			this.consumer.setUsername(aafLogin);
+			this.consumer.setPassword(aafPassword);
+			
+			this.consumer.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+			
+			Properties props = new Properties();
+			props.setProperty("Protocol", "http");
+			this.consumer.setProps(props);
+			this.consumer.setHost(servers.get(0) + ":3904");
+		}
+		
+		/**
+		 * {@inheritDoc}
+		 */
+		public Iterable<String> fetch() throws Exception {
+			return this.consumer.fetch();
+		}
+		
+		/**
+		 * {@inheritDoc}
+		 */
+		public void close() {
+			this.consumer.close();
+		}
+		
+		@Override
+		public String toString() {
+			StringBuilder builder = new StringBuilder();
+			builder.
+			append("DmaapConsumerWrapper [").
+			append("consumer.getAuthDate()=").append(consumer.getAuthDate()).
+			append(", consumer.getAuthKey()=").append(consumer.getAuthKey()).
+			append(", consumer.getHost()=").append(consumer.getHost()).
+			append(", consumer.getProtocolFlag()=").append(consumer.getProtocolFlag()).
+			append(", consumer.getUsername()=").append(consumer.getUsername()).
+			append("]");
+			return builder.toString();
+		}
+	}
+
+}
diff --git a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BusPublisher.java b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BusPublisher.java
new file mode 100644
index 000000000..9c9e54b47
--- /dev/null
+++ b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/BusPublisher.java
@@ -0,0 +1,119 @@
+package org.openecomp.policy.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.openecomp.policy.common.logging.eelf.PolicyLogger;
+
+import com.att.nsa.mr.client.impl.MRSimplerBatchPublisher;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+public interface BusPublisher {
+	
+	/**
+	 * sends a message
+	 * 
+	 * @param partition id
+	 * @param message the message
+	 * @return true if success, false otherwise
+	 * @throws IllegalArgumentException if no message provided
+	 */
+	public boolean send(String partitionId, String message) throws IllegalArgumentException;
+	
+	/**
+	 * closes the publisher
+	 */
+	public void close();
+	
+	/**
+	 * DmaapClient library wrapper
+	 */
+	public static class DmaapPublisherWrapper implements BusPublisher {
+		/**
+		 * MR based Publisher
+		 */		
+		protected MRSimplerBatchPublisher publisher;
+		
+		public DmaapPublisherWrapper(List<String> servers, String topic,
+				                     String aafLogin,
+				                     String aafPassword) {
+			
+			ArrayList<String> dmaapServers = new ArrayList<String>();
+			for (String server: servers) {
+				dmaapServers.add(server + ":3904");
+			}
+					
+			this.publisher = 
+				new MRSimplerBatchPublisher.Builder().
+			                                againstUrls(dmaapServers).
+			                                onTopic(topic).
+			                                build();
+			
+			this.publisher.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+			
+			this.publisher.setUsername(aafLogin);
+			this.publisher.setPassword(aafPassword);  
+			
+			Properties props = new Properties();
+			props.setProperty("Protocol", "http");
+			props.setProperty("contenttype", "application/json");
+			
+			this.publisher.setProps(props);
+			
+			this.publisher.setHost(servers.get(0));
+			
+			if (PolicyLogger.isInfoEnabled())
+				PolicyLogger.info(DmaapPublisherWrapper.class.getName(), 
+						          "CREATION: " + this);
+		}
+
+		/**
+		 * {@inheritDoc}
+		 */
+		@Override
+		public void close() {
+			if (PolicyLogger.isInfoEnabled())
+				PolicyLogger.info(DmaapPublisherWrapper.class.getName(), 
+				                  "CREATION: " + this);
+			
+			try {
+				this.publisher.close(1, TimeUnit.SECONDS);
+			} catch (Exception e) {
+				PolicyLogger.warn(DmaapPublisherWrapper.class.getName(), 
+				                  "CLOSE: " + this + " because of " + 
+								  e.getMessage());
+			}
+		}
+		
+		/**
+		 * {@inheritDoc}
+		 */
+		@Override
+		public boolean send(String partitionId, String message) 
+				throws IllegalArgumentException {
+			if (message == null)
+				throw new IllegalArgumentException("No message provided");
+			
+			this.publisher.send(partitionId, message);
+			return true;
+			
+		}
+		
+		@Override
+		public String toString() {
+			StringBuilder builder = new StringBuilder();
+			builder.append("DmaapPublisherWrapper [").
+			append("publisher.getAuthDate()=").append(publisher.getAuthDate()).
+			append(", publisher.getAuthKey()=").append(publisher.getAuthKey()).
+			append(", publisher.getHost()=").append(publisher.getHost()).
+			append(", publisher.getProtocolFlag()=").append(publisher.getProtocolFlag()).
+			append(", publisher.getUsername()=").append(publisher.getUsername()).
+			append(", publisher.getPendingMessageCount()=").append(publisher.getPendingMessageCount()).
+			append("]");
+			return builder.toString();
+		}
+	}
+
+}
diff --git a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/PolicyAccess.java b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/PolicyAccess.java
index 6141a5f73..bbbffec0a 100644
--- a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/PolicyAccess.java
+++ b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/PolicyAccess.java
@@ -101,7 +101,6 @@ public class PolicyAccess implements Access {
 
 	@Override
 	public boolean willLog(Level arg0) {
-		// TODO Auto-generated method stub
 		return false;
 	}
 }
diff --git a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/PolicyUtils.java b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/PolicyUtils.java
index fd8102af3..75197932a 100644
--- a/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/PolicyUtils.java
+++ b/PolicyEngineUtils/src/main/java/org/openecomp/policy/utils/PolicyUtils.java
@@ -21,16 +21,34 @@
 package org.openecomp.policy.utils;
 
 import java.io.IOException;
+import java.io.StringReader;
+import java.io.UnsupportedEncodingException;
+import java.util.Arrays;
 import java.util.Base64;
+import java.util.List;
 import java.util.StringTokenizer;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.drools.core.io.impl.ReaderResource;
+import org.drools.verifier.Verifier;
+import org.drools.verifier.VerifierError;
+import org.drools.verifier.builder.VerifierBuilder;
+import org.drools.verifier.builder.VerifierBuilderFactory;
+import org.kie.api.io.ResourceType;
 
 import com.fasterxml.jackson.core.JsonParseException;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.CharMatcher;
 
 public class PolicyUtils {
 	
+	public static final String EMAIL_PATTERN =
+            "^[_A-Za-z0-9-\\+]+(\\.[_A-Za-z0-9-]+)*@"
+            + "[A-Za-z0-9-]+(\\.[A-Za-z0-9]+)*(\\.[A-Za-z]{2,})$";
+	
 	public static String objectToJsonString(Object o) throws JsonProcessingException{
 		ObjectMapper mapper = new ObjectMapper();
 		return mapper.writeValueAsString(o);
@@ -42,6 +60,14 @@ public class PolicyUtils {
 		return t;
 	}
 	
+	public static String decode(String encodedString) throws UnsupportedEncodingException { 
+		if(encodedString!=null && !encodedString.isEmpty()){ 
+			return new String(Base64.getDecoder().decode(encodedString) ,"UTF-8"); 
+		}else{ 
+			return null; 
+		} 
+	}
+	
 	public static String[] decodeBasicEncoding(String encodedValue) throws Exception{
 		if(encodedValue!=null && encodedValue.contains("Basic ")){
 			String encodedUserPassword = encodedValue.replaceFirst("Basic"	+ " ", "");
@@ -56,4 +82,87 @@ public class PolicyUtils {
 			return null;
 		}
 	}
+	
+	public static String  emptyPolicyValidator(String field){
+        String error = "success";
+        if (field.equals("") || field.contains(" ") || !field.matches("^[a-zA-Z0-9_]*$")) {
+            error = "The Value in Required Field will allow only '{0-9}, {a-z}, {A-Z}, _' following set of Combinations";
+            return error;
+        } else {
+            if(CharMatcher.ASCII.matchesAllOf((CharSequence) field)){
+                 error = "success";
+            }else{
+                error = "The Value Contains Non ASCII Characters";
+                return error;
+            }   
+        }
+        return error;   
+    } 
+    
+    public static String descriptionValidator(String field) {
+        String error = "success";
+        if (field.contains("@CreatedBy:") || field.contains("@ModifiedBy:")) {
+             error = "The value in the description shouldn't contain @CreatedBy: or @ModifiedBy:";
+             return error;
+        } else {
+            error = "success";
+        }
+        return error;   
+    }
+    
+    public static String validateEmailAddress(String emailAddressValue) {
+        String error = "success";
+        List<String> emailList = Arrays.asList(emailAddressValue.toString().split(","));
+        for(int i =0 ; i < emailList.size() ; i++){
+            Pattern pattern = Pattern.compile(EMAIL_PATTERN);
+            Matcher matcher = pattern.matcher(emailList.get(i).trim());
+            if(!matcher.matches()){
+                error = "Please check the Following Email Address is not Valid ....   " +emailList.get(i).toString();
+                return error;
+            }else{
+                error = "success";
+            }
+        }
+        return error;       
+    }
+    
+    /*
+     * Check for "[ERR" to see if there are any errors. 
+     */
+    public static String brmsRawValidate(String rule){
+        VerifierBuilder vBuilder = VerifierBuilderFactory.newVerifierBuilder();
+        Verifier verifier = vBuilder.newVerifier();
+        verifier.addResourcesToVerify(new ReaderResource(new StringReader(rule)), ResourceType.DRL);
+        // Check if there are any Errors in Verification. 
+        if(verifier.getErrors().size()!=0){
+            String message = "Not a Valid DRL rule"; 
+            for(VerifierError error: verifier.getErrors()){
+                // Ignore annotations Error Messages
+                if(!error.getMessage().contains("'@'")){
+                    message = message + "\n" + error.getMessage();
+                }
+            }
+            return message;
+        }
+        return "";
+    }
+    
+    /**
+	 * Given a version string consisting of integers with dots between them, convert it into an array of ints.
+	 * 
+	 * @param version
+	 * @return
+	 * @throws NumberFormatException
+	 */
+	public static int[] versionStringToArray(String version) throws NumberFormatException {
+		if (version == null || version.length() == 0) {
+			return new int[0];
+		}
+		String[] stringArray = version.split("\\.");
+		int[] resultArray = new int[stringArray.length];
+		for (int i = 0; i < stringArray.length; i++) {
+			resultArray[i] = Integer.parseInt(stringArray[i]);
+		}
+		return resultArray;
+	}
 }