Replace nsaCore library with Spring

Change-Id: I2227939a67a2cbba2d392136d49ef4419600d186
Issue-ID: DCAEGEN2-602
Signed-off-by: PawelSzalapski <pawel.szalapski@nokia.com>

11 files changed, 695 insertions, 904 deletions

diff --git a/src/main/java/org/onap/dcae/commonFunction/AnyNode.java b/src/main/java/org/onap/dcae/commonFunction/AnyNode.java
index 97d73ddd..7be45b0c 100644
--- a/src/main/java/org/onap/dcae/commonFunction/AnyNode.java
+++ b/src/main/java/org/onap/dcae/commonFunction/AnyNode.java
@@ -19,16 +19,17 @@
  */
 package org.onap.dcae.commonFunction;
 
-import static io.vavr.API.Set;
-
 import io.vavr.collection.List;
 import io.vavr.collection.Set;
 import io.vavr.control.Option;
-import java.util.stream.StreamSupport;
 import org.json.JSONArray;
 import org.json.JSONException;
 import org.json.JSONObject;
 
+import java.util.stream.StreamSupport;
+
+import static io.vavr.API.Set;
+
 /**
  * This class is a wrapper for 2 most used entities of org.json lib: JSONArray and JSONObject and comprises utility
  * methods for fast access of json structures without need to explicitly coerce between them. While using this, bear in
@@ -109,5 +110,4 @@ public class AnyNode {
         return (JSONObject) this.obj;
     }
 
-
 }
diff --git a/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java b/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java
deleted file mode 100644
index 36713aa4..00000000
--- a/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java
+++ /dev/null
@@ -1,204 +0,0 @@
-/*-

- * ============LICENSE_START=======================================================

- * PROJECT

- * ================================================================================

- * Copyright (C) 2017-2018 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.onap.dcae.commonFunction;

-

-import com.att.nsa.apiServer.ApiServer;

-import com.att.nsa.apiServer.ApiServerConnector;

-import com.att.nsa.apiServer.endpoints.NsaBaseEndpoint;

-import com.att.nsa.drumlin.till.nv.rrNvReadable;

-import com.att.nsa.drumlin.till.nv.rrNvReadable.loadException;

-import com.fasterxml.jackson.core.JsonParseException;

-import com.fasterxml.jackson.databind.JsonNode;

-import com.github.fge.jackson.JsonLoader;

-import com.github.fge.jsonschema.core.exceptions.ProcessingException;

-import com.github.fge.jsonschema.core.report.ProcessingMessage;

-import com.github.fge.jsonschema.core.report.ProcessingReport;

-import com.github.fge.jsonschema.main.JsonSchema;

-import com.github.fge.jsonschema.main.JsonSchemaFactory;

-import org.apache.catalina.LifecycleException;

-import org.json.JSONArray;

-import org.json.JSONException;

-import org.json.JSONObject;

-import org.onap.dcae.ApplicationSettings;

-import org.onap.dcae.CLIUtils;

-import org.onap.dcae.commonFunction.event.publishing.DMaaPConfigurationParser;

-import org.onap.dcae.commonFunction.event.publishing.EventPublisher;

-import org.onap.dcae.restapi.RestfulCollectorServlet;

-import org.slf4j.Logger;

-import org.slf4j.LoggerFactory;

-

-import java.io.IOException;

-import java.nio.file.Files;

-import java.nio.file.Paths;

-import java.util.LinkedList;

-import java.util.List;

-import java.util.concurrent.ExecutorService;

-import java.util.concurrent.Executors;

-import java.util.concurrent.LinkedBlockingQueue;

-

-public class CommonStartup extends NsaBaseEndpoint implements Runnable {

-

-    private static final Logger metriclog = LoggerFactory.getLogger("com.att.ecomp.metrics");

-    public static final Logger inlog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.input");

-    static final Logger oplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.output");

-    public static final Logger eplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.error");

-

-    static int maxQueueEvent = 1024 * 4;

-    public static boolean schemaValidatorflag = false;

-    public static boolean authflag = false;

-    static boolean eventTransformFlag = true;

-    public static JSONObject schemaFileJson;

-    static String cambriaConfigFile;

-    public static io.vavr.collection.Map<String , String [] > streamID;

-

-    static LinkedBlockingQueue<JSONObject> fProcessingInputQueue;

-    private static ApiServer fTomcatServer = null;

-    private static final Logger log = LoggerFactory.getLogger(CommonStartup.class);

-

-    private CommonStartup(ApplicationSettings settings) throws loadException, IOException, rrNvReadable.missingReqdSetting {

-        final List<ApiServerConnector> connectors = new LinkedList<>();

-

-        if (!settings.authorizationEnabled()) {

-            connectors.add(new ApiServerConnector.Builder(settings.httpPort()).secure(false).build());

-        }

-

-        final int securePort = settings.httpsPort();

-        final String keystoreFile = settings.keystoreFileLocation();

-        final String keystorePasswordFile = settings.keystorePasswordFileLocation();

-        final String keyAlias = settings.keystoreAlias();

-

-        if (settings.authorizationEnabled()) {

-            String keystorePassword = readFile(keystorePasswordFile);

-            connectors.add(new ApiServerConnector.Builder(securePort).secure(true)

-                    .keystorePassword(keystorePassword).keystoreFile(keystoreFile).keyAlias(keyAlias).build());

-

-        }

-

-        schemaValidatorflag = settings.jsonSchemaValidationEnabled();

-        maxQueueEvent = settings.maximumAllowedQueuedEvents();

-        if (schemaValidatorflag) {

-            schemaFileJson = settings.jsonSchema();

-

-        }

-        authflag = settings.authorizationEnabled();

-        cambriaConfigFile = settings.cambriaConfigurationFileLocation();

-        streamID = settings.dMaaPStreamsMapping();

-        eventTransformFlag = settings.eventTransformingEnabled();

-

-        fTomcatServer = new ApiServer.Builder(connectors, new RestfulCollectorServlet(settings)).encodeSlashes(true)

-            .name("collector").build();

-    }

-

-    public static void main(String[] args) {

-        try {

-

-            fProcessingInputQueue = new LinkedBlockingQueue<>(CommonStartup.maxQueueEvent);

-

-            VESLogger.setUpEcompLogging();

-

-            CommonStartup cs = new CommonStartup(new ApplicationSettings(args, CLIUtils::processCmdLine));

-

-            Thread commonStartupThread = new Thread(cs);

-            commonStartupThread.start();

-

-            EventProcessor ep = new EventProcessor(EventPublisher.createPublisher(oplog,

-                                                                                  DMaaPConfigurationParser

-                                                                                      .parseToDomainMapping(Paths.get(cambriaConfigFile))

-                                                                                      .get()));

-            ExecutorService executor = Executors.newFixedThreadPool(20);

-            for (int i = 0; i < 20; ++i) {

-                executor.execute(ep);

-            }

-        } catch (Exception e) {

-            CommonStartup.eplog.error("Fatal error during application startup", e);

-            throw new RuntimeException(e);

-        }

-    }

-

-    public void run() {

-        try {

-            fTomcatServer.start();

-            fTomcatServer.await();

-        } catch (LifecycleException | IOException e) {

-            throw new RuntimeException(e);

-        }

-    }

-

-    public static class QueueFullException extends Exception {

-

-        private static final long serialVersionUID = 1L;

-    }

-

-    public static void handleEvents(JSONArray a) throws QueueFullException, JSONException {

-        CommonStartup.metriclog.info("EVENT_PUBLISH_START");

-        for (int i = 0; i < a.length(); i++) {

-            if (!fProcessingInputQueue.offer(a.getJSONObject(i))) {

-                throw new QueueFullException();

-            }

-        }

-        log.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");

-        CommonStartup.metriclog.info("EVENT_PUBLISH_END");

-    }

-

-    private static String readFile(String path) throws IOException {

-        byte[] encoded = Files.readAllBytes(Paths.get(path));

-        String pwd = new String(encoded);

-        return pwd.substring(0, pwd.length() - 1);

-    }

-

-    public static String validateAgainstSchema(String jsonData, String jsonSchema) {

-        ProcessingReport report;

-        String result = "false";

-

-        try {

-            log.trace("Schema validation for event:" + jsonData);

-            JsonNode schemaNode = JsonLoader.fromString(jsonSchema);

-            JsonNode data = JsonLoader.fromString(jsonData);

-            JsonSchemaFactory factory = JsonSchemaFactory.byDefault();

-            JsonSchema schema = factory.getJsonSchema(schemaNode);

-            report = schema.validate(data);

-        } catch (JsonParseException e) {

-            log.error("validateAgainstSchema:JsonParseException for event:" + jsonData);

-            return e.getMessage();

-        } catch (ProcessingException e) {

-            log.error("validateAgainstSchema:Processing exception for event:" + jsonData);

-            return e.getMessage();

-        } catch (IOException e) {

-            log.error(

-                "validateAgainstSchema:IO exception; something went wrong trying to read json data for event:" + jsonData);

-            return e.getMessage();

-        }

-        if (report != null) {

-            for (ProcessingMessage pm : report) {

-                log.trace("Processing Message: " + pm.getMessage());

-            }

-            result = String.valueOf(report.isSuccess());

-        }

-        try {

-            log.debug("Validation Result:" + result + " Validation report:" + report);

-        } catch (NullPointerException e) {

-            log.error("validateAgainstSchema:NullpointerException on report");

-        }

-        return result;

-    }

-

-

-}

diff --git a/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java b/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java
index a6de0fc8..e3d59098 100644
--- a/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java
+++ b/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java
@@ -8,9 +8,9 @@
  * 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.
@@ -21,483 +21,483 @@
 
 package org.onap.dcae.commonFunction;
 
-import java.text.DecimalFormat;
 import org.json.JSONArray;
 import org.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class ConfigProcessors {
-
-	private static final Logger log = LoggerFactory.getLogger(ConfigProcessors.class);
-	private static final String FIELD = "field";
-	private static final String OLD_FIELD = "oldField";
-	private static final String FILTER = "filter";
-	private static final String VALUE = "value";
-	private static final String REGEX = "\\[\\]";
-	private static final String OBJECT_NOT_FOUND = "ObjectNotFound";
-	private static final String FILTER_NOT_MET = "Filter not met";
-	private static final String COMP_FALSE = "==false";
-
-	private final JSONObject event;
-
-	public ConfigProcessors(JSONObject eventJson) {
-		event = eventJson;
-	}
-
-	public void getValue(JSONObject jsonObject) {
-
-		final String field = jsonObject.getString(FIELD);
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-
-		if (filter == null || isFilterMet(filter)) {
-			getEventObjectVal(field);
-		} else
-			log.info(FILTER_NOT_MET);
-	}
+import java.text.DecimalFormat;
 
+public class ConfigProcessors {
 
-	public void setValue(JSONObject jsonObject) {
-		final String field = jsonObject.getString(FIELD);
-		final String value = jsonObject.getString(VALUE);
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-		if (filter == null || isFilterMet(filter)) {
-			setEventObjectVal(field, value);
-		} else
-			log.info(FILTER_NOT_MET);
-	}
+    private static final Logger log = LoggerFactory.getLogger(ConfigProcessors.class);
+    private static final String FIELD = "field";
+    private static final String OLD_FIELD = "oldField";
+    private static final String FILTER = "filter";
+    private static final String VALUE = "value";
+    private static final String REGEX = "\\[\\]";
+    private static final String OBJECT_NOT_FOUND = "ObjectNotFound";
+    private static final String FILTER_NOT_MET = "Filter not met";
+    private static final String COMP_FALSE = "==false";
 
+    private final JSONObject event;
 
+    public ConfigProcessors(JSONObject eventJson) {
+        event = eventJson;
+    }
 
-	private String evaluate(String str) {
-		String value = str;
-		if (str.startsWith("$")) {
-			value = (String) getEventObjectVal(str.substring(1));
+    public void getValue(JSONObject jsonObject) {
 
-		}
-		return value;
-	}
+        final String field = jsonObject.getString(FIELD);
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
 
+        if (filter == null || isFilterMet(filter)) {
+            getEventObjectVal(field);
+        } else
+            log.info(FILTER_NOT_MET);
+    }
 
-	public void suppressEvent(JSONObject jsonObject) {
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
 
-		if (filter == null || isFilterMet(filter)) {
-			setEventObjectVal("suppressEvent", "true");
-		} else
-			log.info(FILTER_NOT_MET);
-	}
+    public void setValue(JSONObject jsonObject) {
+        final String field = jsonObject.getString(FIELD);
+        final String value = jsonObject.getString(VALUE);
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
+        if (filter == null || isFilterMet(filter)) {
+            setEventObjectVal(field, value);
+        } else
+            log.info(FILTER_NOT_MET);
+    }
 
 
-	public void addAttribute(JSONObject jsonObject) {
+    private String evaluate(String str) {
+        String value = str;
+        if (str.startsWith("$")) {
+            value = (String) getEventObjectVal(str.substring(1));
+
+        }
+        return value;
+    }
 
-		final String field = jsonObject.getString(FIELD);
-		final String value = evaluate(jsonObject.getString(VALUE));
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-		final String fieldType = jsonObject.optString("fieldType", "string").toLowerCase();
 
-		if (filter == null || isFilterMet(filter)) {
-			setEventObjectVal(field, value, fieldType);
-		} else
-			log.info(FILTER_NOT_MET);
-	}
+    public void suppressEvent(JSONObject jsonObject) {
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
 
+        if (filter == null || isFilterMet(filter)) {
+            setEventObjectVal("suppressEvent", "true");
+        } else
+            log.info(FILTER_NOT_MET);
+    }
 
-	public void updateAttribute(JSONObject jsonObject) {
 
-		final String field = jsonObject.getString(FIELD);
-		final String value = evaluate(jsonObject.getString(VALUE));
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-		if (filter == null || isFilterMet(filter)) {
-			setEventObjectVal(field, value);
-		} else
-			log.info(FILTER_NOT_MET);
-	}
+    public void addAttribute(JSONObject jsonObject) {
+
+        final String field = jsonObject.getString(FIELD);
+        final String value = evaluate(jsonObject.getString(VALUE));
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
+        final String fieldType = jsonObject.optString("fieldType", "string").toLowerCase();
+
+        if (filter == null || isFilterMet(filter)) {
+            setEventObjectVal(field, value, fieldType);
+        } else
+            log.info(FILTER_NOT_MET);
+    }
 
 
-	public void removeAttribute(JSONObject jsonObject) {
-
-		final String field = jsonObject.getString(FIELD);
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-
-		if (filter == null || isFilterMet(filter)) {
-			removeEventKey(field);
-		} else
-			log.info(FILTER_NOT_MET);
-	}
-
-
-	private void renameArrayInArray(JSONObject jsonObject) // map
-	{
-		log.info("renameArrayInArray");
-		final String field = jsonObject.getString(FIELD);
-		final String oldField = jsonObject.getString(OLD_FIELD);
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-
-		if (filter == null || isFilterMet(filter)) {
-
-			final String[] fsplit = field.split(REGEX, field.length());
-			final String[] oldfsplit = oldField.split(REGEX, oldField.length());
-
-			final String oldValue = getEventObjectVal(oldfsplit[0]).toString();
-			if (!oldValue.equals(OBJECT_NOT_FOUND)) {
-				final String oldArrayName = oldfsplit[1].substring(1);
-				final String newArrayName = fsplit[1].substring(1);
-				final String value = oldValue.replaceAll(oldArrayName, newArrayName);
-
-				log.info("oldValue ==" + oldValue);
-				log.info("value ==" + value);
-				JSONArray ja = new JSONArray(value);
-				removeEventKey(oldfsplit[0]);
-				setEventObjectVal(fsplit[0], ja);
-			}
-		} else
-			log.info(FILTER_NOT_MET);
-	}
-
-
-	public void map(JSONObject jsonObject) {
-
-		final String field = jsonObject.getString(FIELD);
-		if (field.contains("[]")) {
-			if (field.matches(".*\\[\\]\\..*\\[\\]"))
-				renameArrayInArray(jsonObject);
-			else
-				mapToJArray(jsonObject);
-		} else
-			mapAttribute(jsonObject);
-	}
-
-	private String performOperation(String operation, String value) {
-		log.info("performOperation");
-		if ("convertMBtoKB".equals(operation)) {
-			float kbValue = Float.parseFloat(value) * 1024;
-			value = String.valueOf(kbValue);
-		}
-		return value;
-	}
-
-
-	public void mapAttribute(JSONObject jsonObject) {
-
-		final String field = jsonObject.getString(FIELD);
-		final String oldField = jsonObject.getString(OLD_FIELD);
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-		final String operation = jsonObject.optString("operation");
-		String value;
-		if (filter == null || isFilterMet(filter)) {
-
-			value = getEventObjectVal(oldField).toString();
-			if (!value.equals(OBJECT_NOT_FOUND)) {
-				if (operation != null && !operation.isEmpty())
-					value = performOperation(operation, value);
-
-				setEventObjectVal(field, value);
-
-				removeEventKey(oldField);
-			}
-		} else
-			log.info(FILTER_NOT_MET);
-	}
-
-
-	private void mapToJArray(JSONObject jsonObject) {
-		log.info("mapToJArray");
-		String field = jsonObject.getString(FIELD);
-		String oldField = jsonObject.getString(OLD_FIELD);
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-		final JSONObject attrMap = jsonObject.optJSONObject("attrMap");
-		oldField = oldField.replaceAll(REGEX, "");
-		field = field.replaceAll(REGEX, "");
-
-		if (filter == null || isFilterMet(filter)) {
-
-			String value = getEventObjectVal(oldField).toString();
-			if (!value.equals(OBJECT_NOT_FOUND)) {
-				log.info("old value ==" + value);
-				// update old value based on attrMap
-				if (attrMap != null) {
-					// loop thru attrMap and update attribute name to new name
-					for (String key : attrMap.keySet()) {
-						value = value.replaceAll(key, attrMap.getString(key));
-					}
-				}
-
-				log.info("new value ==" + value);
-				char c = value.charAt(0);
-				if (c != '[') {
-					// oldfield is JsonObject
-					JSONObject valueJO = new JSONObject(value);
-					// if the array already exists
-					String existingValue = getEventObjectVal(field).toString();
-					if (!existingValue.equals(OBJECT_NOT_FOUND)) {
-						JSONArray ja = new JSONArray(existingValue);
-						JSONObject jo = ja.optJSONObject(0);
-						if (jo != null) {
-							for (String key : valueJO.keySet()) {
-								jo.put(key, valueJO.get(key));
-
-							}
-							ja.put(0, jo);
-
-							setEventObjectVal(field, ja);
-						}
-					} else // if new array
-						setEventObjectVal(field + "[0]", new JSONObject(value), "JArray");
-				} else // oldfield is jsonArray
-					setEventObjectVal(field, new JSONArray(value));
-
-				removeEventKey(oldField);
-			}
-		} else
-			log.info(FILTER_NOT_MET);
-	}
-
-	/**
-	 * example - { "functionName": "concatenateValue", "args":{ "filter":
-	 * {"event.commonEventHeader.event":"heartbeat"},
-	 * FIELD:"event.commonEventHeader.eventName", "concatenate":
-	 * ["event.commonEventHeader.domain","event.commonEventHeader.eventType","event.commonEventHeader.alarmCondition"],
-	 * "delimiter":"_" } }
-	 **/
-	public void concatenateValue(JSONObject jsonObject) {
-
-		final String field = jsonObject.getString(FIELD);
-		final String delimiter = jsonObject.getString("delimiter");
-		final JSONArray values = jsonObject.getJSONArray("concatenate");
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-		if (filter == null || isFilterMet(filter)) {
-			StringBuilder value = new StringBuilder();
-			for (int i = 0; i < values.length(); i++) {
-
-				String tempVal = evaluate(values.getString(i));
-				if (!tempVal.equals(OBJECT_NOT_FOUND)) {
-					if (i == 0)
-						value.append(tempVal);
-					else
-						value.append(delimiter).append(tempVal);
-				}
-			}
-
-			setEventObjectVal(field, value.toString());
-		} else
-			log.info(FILTER_NOT_MET);
-	}
-
-	public void subtractValue(JSONObject jsonObject) {
-
-		final String field = jsonObject.getString(FIELD);
-		final JSONArray values = jsonObject.getJSONArray("subtract");
-		final JSONObject filter = jsonObject.optJSONObject(FILTER);
-		if (filter == null || isFilterMet(filter)) {
-			float value = 0;
-			for (int i = 0; i < values.length(); i++) {
-				log.info(values.getString(i));
-				String tempVal = evaluate(values.getString(i));
-				log.info("tempVal==" + tempVal);
-				if (!tempVal.equals(OBJECT_NOT_FOUND)) {
-					if (i == 0)
-						value = value + Float.valueOf(tempVal);
-					else
-						value = value - Float.valueOf(tempVal);
-				}
-			}
-			log.info("value ==" + value);
-			setEventObjectVal(field, value, "number");
-		} else
-			log.info(FILTER_NOT_MET);
-	}
-
-
-	private void removeEventKey(String field) {
-		String[] keySet = field.split("\\.", field.length());
-		JSONObject keySeries = event;
-		for (int i = 0; i < (keySet.length - 1); i++) {
-
-			keySeries = keySeries.getJSONObject(keySet[i]);
-		}
-
-		keySeries.remove(keySet[keySet.length - 1]);
-	}
-
-
-	private boolean checkFilter(JSONObject jo, String key, String logicKey) {
-		String filterValue = jo.getString(key);
-		if (filterValue.contains(":")) {
-			String[] splitVal = filterValue.split(":");
-			if ("matches".equals(splitVal[0])) {
-				if ("not".equals(logicKey)) {
-					if (getEventObjectVal(key).toString().matches(splitVal[1])) {
-						log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
-						return false;
-					}
-				} else {
-					if (!(getEventObjectVal(key).toString().matches(splitVal[1]))) {
-						log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
-						return false;
-					}
-				}
-
-			}
-			if ("contains".equals(splitVal[0])) {
-				if ("not".equals(logicKey)) {
-					if (getEventObjectVal(key).toString().contains(splitVal[1])) {
-						log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
-						return false;
-					}
-				} else {
-					if (!(getEventObjectVal(key).toString().contains(splitVal[1]))) {
-						log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
-						return false;
-					}
-				}
-
-			}
-		} else {
-			if ("not".equals(logicKey)) {
-				if (getEventObjectVal(key).toString().equals(filterValue)) {
-					log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
-					return false;
-				}
-			} else {
-				if (!(getEventObjectVal(key).toString().equals(filterValue))) {
-					log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
-					return false;
-				}
-			}
-		}
-		return true;
-	}
-
-
-	public boolean isFilterMet(JSONObject jo) {
-		for (String key : jo.keySet()) {
-			if ("not".equals(key)) {
-				JSONObject njo = jo.getJSONObject(key);
-				for (String njoKey : njo.keySet()) {
-					if (!checkFilter(njo, njoKey, key))
-						return false;
-				}
-			} else {
-				if (!checkFilter(jo, key, key))
-					return false;
-			}
-		}
-		return true;
-	}
-
-	/**
-	 * returns a string or JSONObject or JSONArray
-	 **/
-	public Object getEventObjectVal(String keySeriesStr) {
-		keySeriesStr = keySeriesStr.replaceAll("\\[", ".");
-		keySeriesStr = keySeriesStr.replaceAll("\\]", ".");
-		if (keySeriesStr.contains("..")) {
-			keySeriesStr = keySeriesStr.replaceAll("\\.\\.", ".");
-		}
-
-		if (keySeriesStr.lastIndexOf(".") == keySeriesStr.length() - 1)
-			keySeriesStr = keySeriesStr.substring(0, keySeriesStr.length() - 1);
-		String[] keySet = keySeriesStr.split("\\.", keySeriesStr.length());
-		Object keySeriesObj = event;
-		for (String aKeySet : keySet) {
-			if (keySeriesObj != null) {
-				if (keySeriesObj instanceof String) {
-
-					log.info("STRING==" + keySeriesObj);
-				} else if (keySeriesObj instanceof JSONArray) {
-					keySeriesObj = ((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(aKeySet));
-
-				} else if (keySeriesObj instanceof JSONObject) {
-					keySeriesObj = ((JSONObject) keySeriesObj).opt(aKeySet);
-
-				} else {
-					log.info("unknown object==" + keySeriesObj);
-				}
-			}
-		}
-
-		if (keySeriesObj == null)
-			return OBJECT_NOT_FOUND;
-		return keySeriesObj;
-	}
-
-	public void setEventObjectVal(String keySeriesStr, Object value) {
-		setEventObjectVal(keySeriesStr, value, "string");
-	}
-
-	/**
-	 * returns a string or JSONObject or JSONArray
-	 **/
-	public void setEventObjectVal(String keySeriesStr, Object value, String fieldType) {
-		keySeriesStr = keySeriesStr.replaceAll("\\[", ".");
-		keySeriesStr = keySeriesStr.replaceAll("\\]", ".");
-		if (keySeriesStr.contains("..")) {
-			keySeriesStr = keySeriesStr.replaceAll("\\.\\.", ".");
-		}
-		log.info("fieldType==" + fieldType);
-
-		if (keySeriesStr.lastIndexOf(".") == keySeriesStr.length() - 1)
-			keySeriesStr = keySeriesStr.substring(0, keySeriesStr.length() - 1);
-		String[] keySet = keySeriesStr.split("\\.", keySeriesStr.length());
-		Object keySeriesObj = event;
-		for (int i = 0; i < (keySet.length - 1); i++) {
-
-			if (keySeriesObj instanceof JSONArray) {
-
-				if (((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(keySet[i])) == null) // if
-																									// the
-																									// object
-																									// is
-																									// not
-																									// there
-																									// then
-																									// add
-																									// it
-				{
-					log.info("Object is null, must add it");
-					if (keySet[i + 1].matches("[0-9]*")) // if index then array
-						((JSONArray) keySeriesObj).put(Integer.parseInt(keySet[i]), new JSONArray());
-					else
-						((JSONArray) keySeriesObj).put(Integer.parseInt(keySet[i]), new JSONObject());
-				}
-				keySeriesObj = ((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(keySet[i]));
-
-			} else if (keySeriesObj instanceof JSONObject) {
-				if (((JSONObject) keySeriesObj).opt(keySet[i]) == null) // if
-																		// the
-																		// object
-																		// is
-																		// not
-																		// there
-																		// then
-																		// add
-																		// it
-				{
-					if (keySet[i + 1].matches("[0-9]*")) // if index then array
-						((JSONObject) keySeriesObj).put(keySet[i], new JSONArray());
-					else
-						((JSONObject) keySeriesObj).put(keySet[i], new JSONObject());
-					log.info("Object is null, must add it");
-				}
-				keySeriesObj = ((JSONObject) keySeriesObj).opt(keySet[i]);
-			} else {
-				log.info("unknown object==" + keySeriesObj);
-			}
-		}
-		if ("number".equals(fieldType)) {
-			DecimalFormat df = new DecimalFormat("#.0");
-			if (value instanceof String)
-				((JSONObject) keySeriesObj).put(keySet[keySet.length - 1],
-						Float.valueOf(df.format(Float.valueOf((String) value))));
-			else
-				((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], Float.valueOf(df.format(value)));
-		} else if ("integer".equals(fieldType) && value instanceof String)
-			((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], Integer.valueOf((String) value));
-		else if ("JArray".equals(fieldType))
-			((JSONArray) keySeriesObj).put(value);
-		else
-			((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], value);
-
-	}
+    public void updateAttribute(JSONObject jsonObject) {
+
+        final String field = jsonObject.getString(FIELD);
+        final String value = evaluate(jsonObject.getString(VALUE));
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
+        if (filter == null || isFilterMet(filter)) {
+            setEventObjectVal(field, value);
+        } else
+            log.info(FILTER_NOT_MET);
+    }
+
+
+    public void removeAttribute(JSONObject jsonObject) {
+
+        final String field = jsonObject.getString(FIELD);
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
+
+        if (filter == null || isFilterMet(filter)) {
+            removeEventKey(field);
+        } else
+            log.info(FILTER_NOT_MET);
+    }
+
+
+    private void renameArrayInArray(JSONObject jsonObject) // map
+    {
+        log.info("renameArrayInArray");
+        final String field = jsonObject.getString(FIELD);
+        final String oldField = jsonObject.getString(OLD_FIELD);
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
+
+        if (filter == null || isFilterMet(filter)) {
+
+            final String[] fsplit = field.split(REGEX, field.length());
+            final String[] oldfsplit = oldField.split(REGEX, oldField.length());
+
+            final String oldValue = getEventObjectVal(oldfsplit[0]).toString();
+            if (!oldValue.equals(OBJECT_NOT_FOUND)) {
+                final String oldArrayName = oldfsplit[1].substring(1);
+                final String newArrayName = fsplit[1].substring(1);
+                final String value = oldValue.replaceAll(oldArrayName, newArrayName);
+
+                log.info("oldValue ==" + oldValue);
+                log.info("value ==" + value);
+                JSONArray ja = new JSONArray(value);
+                removeEventKey(oldfsplit[0]);
+                setEventObjectVal(fsplit[0], ja);
+            }
+        } else
+            log.info(FILTER_NOT_MET);
+    }
+
+
+    public void map(JSONObject jsonObject) {
+
+        final String field = jsonObject.getString(FIELD);
+        if (field.contains("[]")) {
+            if (field.matches(".*\\[\\]\\..*\\[\\]"))
+                renameArrayInArray(jsonObject);
+            else
+                mapToJArray(jsonObject);
+        } else
+            mapAttribute(jsonObject);
+    }
+
+    private String performOperation(String operation, String value) {
+        log.info("performOperation");
+        if ("convertMBtoKB".equals(operation)) {
+            float kbValue = Float.parseFloat(value) * 1024;
+            value = String.valueOf(kbValue);
+        }
+        return value;
+    }
+
+
+    public void mapAttribute(JSONObject jsonObject) {
+
+        final String field = jsonObject.getString(FIELD);
+        final String oldField = jsonObject.getString(OLD_FIELD);
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
+        final String operation = jsonObject.optString("operation");
+        String value;
+        if (filter == null || isFilterMet(filter)) {
+
+            value = getEventObjectVal(oldField).toString();
+            if (!value.equals(OBJECT_NOT_FOUND)) {
+                if (operation != null && !operation.isEmpty())
+                    value = performOperation(operation, value);
+
+                setEventObjectVal(field, value);
+
+                removeEventKey(oldField);
+            }
+        } else
+            log.info(FILTER_NOT_MET);
+    }
+
+
+    private void mapToJArray(JSONObject jsonObject) {
+        log.info("mapToJArray");
+        String field = jsonObject.getString(FIELD);
+        String oldField = jsonObject.getString(OLD_FIELD);
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
+        final JSONObject attrMap = jsonObject.optJSONObject("attrMap");
+        oldField = oldField.replaceAll(REGEX, "");
+        field = field.replaceAll(REGEX, "");
+
+        if (filter == null || isFilterMet(filter)) {
+
+            String value = getEventObjectVal(oldField).toString();
+            if (!value.equals(OBJECT_NOT_FOUND)) {
+                log.info("old value ==" + value);
+                // update old value based on attrMap
+                if (attrMap != null) {
+                    // loop thru attrMap and update attribute name to new name
+                    for (String key : attrMap.keySet()) {
+                        value = value.replaceAll(key, attrMap.getString(key));
+                    }
+                }
+
+                log.info("new value ==" + value);
+                char c = value.charAt(0);
+                if (c != '[') {
+                    // oldfield is JsonObject
+                    JSONObject valueJO = new JSONObject(value);
+                    // if the array already exists
+                    String existingValue = getEventObjectVal(field).toString();
+                    if (!existingValue.equals(OBJECT_NOT_FOUND)) {
+                        JSONArray ja = new JSONArray(existingValue);
+                        JSONObject jo = ja.optJSONObject(0);
+                        if (jo != null) {
+                            for (String key : valueJO.keySet()) {
+                                jo.put(key, valueJO.get(key));
+
+                            }
+                            ja.put(0, jo);
+
+                            setEventObjectVal(field, ja);
+                        }
+                    } else // if new array
+                        setEventObjectVal(field + "[0]", new JSONObject(value), "JArray");
+                } else // oldfield is jsonArray
+                    setEventObjectVal(field, new JSONArray(value));
+
+                removeEventKey(oldField);
+            }
+        } else
+            log.info(FILTER_NOT_MET);
+    }
+
+    /**
+     * example - { "functionName": "concatenateValue", "args":{ "filter":
+     * {"event.commonEventHeader.event":"heartbeat"},
+     * FIELD:"event.commonEventHeader.eventName", "concatenate":
+     * ["event.commonEventHeader.domain","event.commonEventHeader.eventType","event.commonEventHeader.alarmCondition"],
+     * "delimiter":"_" } }
+     **/
+    public void concatenateValue(JSONObject jsonObject) {
+
+        final String field = jsonObject.getString(FIELD);
+        final String delimiter = jsonObject.getString("delimiter");
+        final JSONArray values = jsonObject.getJSONArray("concatenate");
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
+        if (filter == null || isFilterMet(filter)) {
+            StringBuilder value = new StringBuilder();
+            for (int i = 0; i < values.length(); i++) {
+
+                String tempVal = evaluate(values.getString(i));
+                if (!tempVal.equals(OBJECT_NOT_FOUND)) {
+                    if (i == 0)
+                        value.append(tempVal);
+                    else
+                        value.append(delimiter).append(tempVal);
+                }
+            }
+
+            setEventObjectVal(field, value.toString());
+        } else
+            log.info(FILTER_NOT_MET);
+    }
+
+    public void subtractValue(JSONObject jsonObject) {
+
+        final String field = jsonObject.getString(FIELD);
+        final JSONArray values = jsonObject.getJSONArray("subtract");
+        final JSONObject filter = jsonObject.optJSONObject(FILTER);
+        if (filter == null || isFilterMet(filter)) {
+            float value = 0;
+            for (int i = 0; i < values.length(); i++) {
+                log.info(values.getString(i));
+                String tempVal = evaluate(values.getString(i));
+                log.info("tempVal==" + tempVal);
+                if (!tempVal.equals(OBJECT_NOT_FOUND)) {
+                    if (i == 0)
+                        value = value + Float.valueOf(tempVal);
+                    else
+                        value = value - Float.valueOf(tempVal);
+                }
+            }
+            log.info("value ==" + value);
+            setEventObjectVal(field, value, "number");
+        } else
+            log.info(FILTER_NOT_MET);
+    }
+
+
+    private void removeEventKey(String field) {
+        String[] keySet = field.split("\\.", field.length());
+        JSONObject keySeries = event;
+        for (int i = 0; i < (keySet.length - 1); i++) {
+
+            keySeries = keySeries.getJSONObject(keySet[i]);
+        }
+
+        keySeries.remove(keySet[keySet.length - 1]);
+    }
+
+
+    private boolean checkFilter(JSONObject jo, String key, String logicKey) {
+        String filterValue = jo.getString(key);
+        if (filterValue.contains(":")) {
+            String[] splitVal = filterValue.split(":");
+            if ("matches".equals(splitVal[0])) {
+                if ("not".equals(logicKey)) {
+                    if (getEventObjectVal(key).toString().matches(splitVal[1])) {
+                        log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
+                        return false;
+                    }
+                } else {
+                    if (!(getEventObjectVal(key).toString().matches(splitVal[1]))) {
+                        log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
+                        return false;
+                    }
+                }
+
+            }
+            if ("contains".equals(splitVal[0])) {
+                if ("not".equals(logicKey)) {
+                    if (getEventObjectVal(key).toString().contains(splitVal[1])) {
+                        log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
+                        return false;
+                    }
+                } else {
+                    if (!(getEventObjectVal(key).toString().contains(splitVal[1]))) {
+                        log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
+                        return false;
+                    }
+                }
+
+            }
+        } else {
+            if ("not".equals(logicKey)) {
+                if (getEventObjectVal(key).toString().equals(filterValue)) {
+                    log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
+                    return false;
+                }
+            } else {
+                if (!(getEventObjectVal(key).toString().equals(filterValue))) {
+                    log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE);
+                    return false;
+                }
+            }
+        }
+        return true;
+    }
+
+
+    public boolean isFilterMet(JSONObject jo) {
+        for (String key : jo.keySet()) {
+            if ("not".equals(key)) {
+                JSONObject njo = jo.getJSONObject(key);
+                for (String njoKey : njo.keySet()) {
+                    if (!checkFilter(njo, njoKey, key))
+                        return false;
+                }
+            } else {
+                if (!checkFilter(jo, key, key))
+                    return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * returns a string or JSONObject or JSONArray
+     **/
+    public Object getEventObjectVal(String keySeriesStr) {
+        keySeriesStr = keySeriesStr.replaceAll("\\[", ".");
+        keySeriesStr = keySeriesStr.replaceAll("\\]", ".");
+        if (keySeriesStr.contains("..")) {
+            keySeriesStr = keySeriesStr.replaceAll("\\.\\.", ".");
+        }
+
+        if (keySeriesStr.lastIndexOf(".") == keySeriesStr.length() - 1)
+            keySeriesStr = keySeriesStr.substring(0, keySeriesStr.length() - 1);
+        String[] keySet = keySeriesStr.split("\\.", keySeriesStr.length());
+        Object keySeriesObj = event;
+        for (String aKeySet : keySet) {
+            if (keySeriesObj != null) {
+                if (keySeriesObj instanceof String) {
+
+                    log.info("STRING==" + keySeriesObj);
+                } else if (keySeriesObj instanceof JSONArray) {
+                    keySeriesObj = ((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(aKeySet));
+
+                } else if (keySeriesObj instanceof JSONObject) {
+                    keySeriesObj = ((JSONObject) keySeriesObj).opt(aKeySet);
+
+                } else {
+                    log.info("unknown object==" + keySeriesObj);
+                }
+            }
+        }
+
+        if (keySeriesObj == null)
+            return OBJECT_NOT_FOUND;
+        return keySeriesObj;
+    }
+
+    public void setEventObjectVal(String keySeriesStr, Object value) {
+        setEventObjectVal(keySeriesStr, value, "string");
+    }
+
+    /**
+     * returns a string or JSONObject or JSONArray
+     **/
+    public void setEventObjectVal(String keySeriesStr, Object value, String fieldType) {
+        keySeriesStr = keySeriesStr.replaceAll("\\[", ".");
+        keySeriesStr = keySeriesStr.replaceAll("\\]", ".");
+        if (keySeriesStr.contains("..")) {
+            keySeriesStr = keySeriesStr.replaceAll("\\.\\.", ".");
+        }
+        log.info("fieldType==" + fieldType);
+
+        if (keySeriesStr.lastIndexOf(".") == keySeriesStr.length() - 1)
+            keySeriesStr = keySeriesStr.substring(0, keySeriesStr.length() - 1);
+        String[] keySet = keySeriesStr.split("\\.", keySeriesStr.length());
+        Object keySeriesObj = event;
+        for (int i = 0; i < (keySet.length - 1); i++) {
+
+            if (keySeriesObj instanceof JSONArray) {
+
+                if (((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(keySet[i])) == null) // if
+                // the
+                // object
+                // is
+                // not
+                // there
+                // then
+                // add
+                // it
+                {
+                    log.info("Object is null, must add it");
+                    if (keySet[i + 1].matches("[0-9]*")) // if index then array
+                        ((JSONArray) keySeriesObj).put(Integer.parseInt(keySet[i]), new JSONArray());
+                    else
+                        ((JSONArray) keySeriesObj).put(Integer.parseInt(keySet[i]), new JSONObject());
+                }
+                keySeriesObj = ((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(keySet[i]));
+
+            } else if (keySeriesObj instanceof JSONObject) {
+                if (((JSONObject) keySeriesObj).opt(keySet[i]) == null) // if
+                // the
+                // object
+                // is
+                // not
+                // there
+                // then
+                // add
+                // it
+                {
+                    if (keySet[i + 1].matches("[0-9]*")) // if index then array
+                        ((JSONObject) keySeriesObj).put(keySet[i], new JSONArray());
+                    else
+                        ((JSONObject) keySeriesObj).put(keySet[i], new JSONObject());
+                    log.info("Object is null, must add it");
+                }
+                keySeriesObj = ((JSONObject) keySeriesObj).opt(keySet[i]);
+            } else {
+                log.info("unknown object==" + keySeriesObj);
+            }
+        }
+        if ("number".equals(fieldType)) {
+            DecimalFormat df = new DecimalFormat("#.0");
+            if (value instanceof String)
+                ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1],
+                        Float.valueOf(df.format(Float.valueOf((String) value))));
+            else
+                ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], Float.valueOf(df.format(value)));
+        } else if ("integer".equals(fieldType) && value instanceof String)
+            ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], Integer.valueOf((String) value));
+        else if ("JArray".equals(fieldType))
+            ((JSONArray) keySeriesObj).put(value);
+        else
+            ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], value);
+
+    }
 }
diff --git a/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java b/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java
index a57ea3f0..7d27399d 100644
--- a/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java
+++ b/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java
@@ -25,7 +25,10 @@ import com.att.nsa.logging.LoggingContext;
 import com.att.nsa.logging.log4j.EcompFields;

 import com.google.common.reflect.TypeToken;

 import com.google.gson.Gson;

+import io.vavr.collection.Map;

 import org.json.JSONObject;

+import org.onap.dcae.ApplicationSettings;

+import org.onap.dcae.VesApplication;

 import org.onap.dcae.commonFunction.event.publishing.EventPublisher;

 import org.slf4j.Logger;

 import org.slf4j.LoggerFactory;

@@ -35,37 +38,38 @@ import java.io.IOException;
 import java.lang.reflect.Method;

 import java.lang.reflect.Type;

 import java.text.SimpleDateFormat;

-import java.util.Arrays;

 import java.util.Date;

-import java.util.HashMap;

 import java.util.List;

-import java.util.Map;

 

-class EventProcessor implements Runnable {

+public class EventProcessor implements Runnable {

 

+    static final Type EVENT_LIST_TYPE = new TypeToken<List<Event>>() {

+    }.getType();

     private static final Logger log = LoggerFactory.getLogger(EventProcessor.class);

     private static final String EVENT_LITERAL = "event";

     private static final String COMMON_EVENT_HEADER = "commonEventHeader";

-    static final Type EVENT_LIST_TYPE = new TypeToken<List<Event>>() {}.getType();

     private final SimpleDateFormat dateFormat = new SimpleDateFormat("EEE, MM dd yyyy hh:mm:ss z");

 

-    static Map<String, String[]> streamidHash = new HashMap<>();

     public JSONObject event;

     private EventPublisher eventPublisher;

+    private Map<String, String[]> streamidHash;

+    private ApplicationSettings properties;

 

-    public EventProcessor(EventPublisher eventPublisher) {

+

+    public EventProcessor(EventPublisher eventPublisher, ApplicationSettings properties) {

         this.eventPublisher = eventPublisher;

-        streamidHash = CommonStartup.streamID.toJavaMap();

+        this.properties = properties;

+        this.streamidHash = properties.dMaaPStreamsMapping();

     }

 

     @Override

     public void run() {

         try {

             while (true) {

-                event = CommonStartup.fProcessingInputQueue.take();

+                event = VesApplication.fProcessingInputQueue.take();

                 // As long as the producer is running we remove elements from

                 // the queue.

-                log.info("QueueSize:" + CommonStartup.fProcessingInputQueue.size() + "\tEventProcessor\tRemoving element: " + event);

+                log.info("QueueSize:" + VesApplication.fProcessingInputQueue.size() + "\tEventProcessor\tRemoving element: " + event);

 

                 String uuid = event.get("VESuniqueId").toString();

                 LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid);

@@ -73,14 +77,12 @@ class EventProcessor implements Runnable {
 

                 String domain = event.getJSONObject(EVENT_LITERAL).getJSONObject(COMMON_EVENT_HEADER).getString("domain");

                 log.debug("event.VESuniqueId" + event.get("VESuniqueId") + "event.commonEventHeader.domain:" + domain);

-                String[] streamIdList = streamidHash.get(domain);

-                log.debug("streamIdList:" + Arrays.toString(streamIdList));

-

-                if (streamIdList.length == 0) {

-                    log.error("No StreamID defined for publish - Message dropped" + event);

-                } else {

-                    sendEventsToStreams(streamIdList);

-                }

+                streamidHash.get(domain)

+                        .onEmpty(() -> {

+                            log.error("No StreamID defined for publish - Message dropped" + event);

+                        }).forEach(streamIds -> {

+                    sendEventsToStreams(streamIds);

+                });

                 log.debug("Message published" + event);

             }

         } catch (InterruptedException e) {

@@ -93,7 +95,7 @@ class EventProcessor implements Runnable {
         // Set collector timestamp in event payload before publish

         addCurrentTimeToEvent(event);

 

-        if (CommonStartup.eventTransformFlag) {

+        if (properties.eventTransformingEnabled()) {

             // read the mapping json file

             try (FileReader fr = new FileReader("./etc/eventTransform.json")) {

                 log.info("parse eventTransform.json");

@@ -168,5 +170,4 @@ class EventProcessor implements Runnable {
             method.invoke(configProcessors, parameter);

         }

     }

-}

-

+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/commonFunction/VESLogger.java b/src/main/java/org/onap/dcae/commonFunction/VESLogger.java
index a967327e..2a392e81 100644
--- a/src/main/java/org/onap/dcae/commonFunction/VESLogger.java
+++ b/src/main/java/org/onap/dcae/commonFunction/VESLogger.java
@@ -8,9 +8,9 @@
  * 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.
@@ -35,128 +35,126 @@ import java.util.UUID;
 
 public class VESLogger {
 
-	public static final String VES_AGENT = "VES_AGENT";
-	public static final String REQUEST_ID = "requestId";
-	private static final String IP_ADDRESS = "127.0.0.1";
-	private static final String HOST_NAME = "localhost";
-
-	public static Logger auditLog;
-	public static Logger metricsLog;
-	public static Logger errorLog;
-	public static Logger debugLog;
-
-	// Common LoggingContext
-	private static LoggingContext commonLC;
-	// Thread-specific LoggingContext
-	private static LoggingContext threadLC;
-	public LoggingContext lc;
-
-	/**
-	 * Returns the common LoggingContext instance that is the base context for
-	 * all subsequent instances.
-	 *
-	 * @return the common LoggingContext
-	 */
-	public static LoggingContext getCommonLoggingContext() {
-		if (commonLC == null) {
-			commonLC = new LoggingContextFactory.Builder().build();
-			final UUID uuid = UUID.randomUUID();
-
-			commonLC.put(REQUEST_ID, uuid.toString());
-		}
-		return commonLC;
-	}
-
-	/**
-	 * Get a logging context for the current thread that's based on the common
-	 * logging context. Populate the context with context-specific values.
-	 *
-	 * @param aUuid
-	 *            uuid for request id
-	 * @return a LoggingContext for the current thread
-	 */
-	public static LoggingContext getLoggingContextForThread(UUID aUuid) {
-		// note that this operation requires everything from the common context
-		// to be (re)copied into the target context. That seems slow, but it
-		// actually
-		// helps prevent the thread from overwriting supposedly common data. It
-		// also
-		// should be fairly quick compared with the overhead of handling the
-		// actual
-		// service call.
-
-		threadLC = new LoggingContextFactory.Builder().withBaseContext(getCommonLoggingContext()).build();
-		// Establish the request-specific UUID, as long as we are here...
-		threadLC.put(REQUEST_ID, aUuid.toString());
-		threadLC.put(EcompFields.kEndTimestamp, SaClock.now());
-
-		return threadLC;
-	}
-
-	/**
-	 * Get a logging context for the current thread that's based on the common
-	 * logging context. Populate the context with context-specific values.
-	 *
-	 * @param aUuid
-	 *            uuid for request id
-	 * @return a LoggingContext for the current thread
-	 */
-	public static LoggingContext getLoggingContextForThread(String aUuid) {
-		// note that this operation requires everything from the common context
-		// to be (re)copied into the target context. That seems slow, but it
-		// actually
-		// helps prevent the thread from overwriting supposedly common data. It
-		// also
-		// should be fairly quick compared with the overhead of handling the
-		// actual
-		// service call.
-
-		threadLC = new LoggingContextFactory.Builder().withBaseContext(getCommonLoggingContext()).build();
-		// Establish the request-specific UUID, as long as we are here...
-		threadLC.put(REQUEST_ID, aUuid);
-		threadLC.put("statusCode", "COMPLETE");
-		threadLC.put(EcompFields.kEndTimestamp, SaClock.now());
-		return threadLC;
-	}
-
-	public static void setUpEcompLogging() {
-
-		// Create ECOMP Logger instances
-		auditLog = LoggerFactory.getLogger("com.att.ecomp.audit");
-		metricsLog = LoggerFactory.getLogger("com.att.ecomp.metrics");
-		debugLog = LoggerFactory.getLogger("com.att.ecomp.debug");
-		errorLog = LoggerFactory.getLogger("com.att.ecomp.error");
-
-		final LoggingContext lc = getCommonLoggingContext();
-
-		String ipAddr = IP_ADDRESS;
-		String hostname = HOST_NAME;
-		try {
-			final InetAddress ip = InetAddress.getLocalHost();
-			hostname = ip.getCanonicalHostName();
-			ipAddr = ip.getHostAddress();
-		} catch (UnknownHostException x) {
-			Log.debug(x.getMessage());
-		}
-
-		lc.put("serverName", hostname);
-		lc.put("serviceName", "VESCollecor");
-		lc.put("statusCode", "RUNNING");
-		lc.put("targetEntity", "NULL");
-		lc.put("targetServiceName", "NULL");
-		lc.put("server", hostname);
-		lc.put("serverIpAddress", ipAddr);
-
-		// instance UUID is meaningless here, so we just create a new one each
-		// time the
-		// server starts. One could argue each new instantiation of the service
-		// should
-		// have a new instance ID.
-		lc.put("instanceUuid", "");
-		lc.put("severity", "");
-		lc.put(EcompFields.kEndTimestamp, SaClock.now());
-		lc.put("EndTimestamp", SaClock.now());
-		lc.put("partnerName", "NA");
-	}
+    public static final String VES_AGENT = "VES_AGENT";
+    public static final String REQUEST_ID = "requestId";
+    private static final String IP_ADDRESS = "127.0.0.1";
+    private static final String HOST_NAME = "localhost";
+
+    public static Logger auditLog;
+    public static Logger metricsLog;
+    public static Logger errorLog;
+    public static Logger debugLog;
+
+    // Common LoggingContext
+    private static LoggingContext commonLC;
+    // Thread-specific LoggingContext
+    private static LoggingContext threadLC;
+    public LoggingContext lc;
+
+    /**
+     * Returns the common LoggingContext instance that is the base context for
+     * all subsequent instances.
+     *
+     * @return the common LoggingContext
+     */
+    public static LoggingContext getCommonLoggingContext() {
+        if (commonLC == null) {
+            commonLC = new LoggingContextFactory.Builder().build();
+            final UUID uuid = UUID.randomUUID();
+
+            commonLC.put(REQUEST_ID, uuid.toString());
+        }
+        return commonLC;
+    }
+
+    /**
+     * Get a logging context for the current thread that's based on the common
+     * logging context. Populate the context with context-specific values.
+     *
+     * @param aUuid uuid for request id
+     * @return a LoggingContext for the current thread
+     */
+    public static LoggingContext getLoggingContextForThread(UUID aUuid) {
+        // note that this operation requires everything from the common context
+        // to be (re)copied into the target context. That seems slow, but it
+        // actually
+        // helps prevent the thread from overwriting supposedly common data. It
+        // also
+        // should be fairly quick compared with the overhead of handling the
+        // actual
+        // service call.
+
+        threadLC = new LoggingContextFactory.Builder().withBaseContext(getCommonLoggingContext()).build();
+        // Establish the request-specific UUID, as long as we are here...
+        threadLC.put(REQUEST_ID, aUuid.toString());
+        threadLC.put(EcompFields.kEndTimestamp, SaClock.now());
+
+        return threadLC;
+    }
+
+    /**
+     * Get a logging context for the current thread that's based on the common
+     * logging context. Populate the context with context-specific values.
+     *
+     * @param aUuid uuid for request id
+     * @return a LoggingContext for the current thread
+     */
+    public static LoggingContext getLoggingContextForThread(String aUuid) {
+        // note that this operation requires everything from the common context
+        // to be (re)copied into the target context. That seems slow, but it
+        // actually
+        // helps prevent the thread from overwriting supposedly common data. It
+        // also
+        // should be fairly quick compared with the overhead of handling the
+        // actual
+        // service call.
+
+        threadLC = new LoggingContextFactory.Builder().withBaseContext(getCommonLoggingContext()).build();
+        // Establish the request-specific UUID, as long as we are here...
+        threadLC.put(REQUEST_ID, aUuid);
+        threadLC.put("statusCode", "COMPLETE");
+        threadLC.put(EcompFields.kEndTimestamp, SaClock.now());
+        return threadLC;
+    }
+
+    public static void setUpEcompLogging() {
+
+        // Create ECOMP Logger instances
+        auditLog = LoggerFactory.getLogger("com.att.ecomp.audit");
+        metricsLog = LoggerFactory.getLogger("com.att.ecomp.metrics");
+        debugLog = LoggerFactory.getLogger("com.att.ecomp.debug");
+        errorLog = LoggerFactory.getLogger("com.att.ecomp.error");
+
+        final LoggingContext lc = getCommonLoggingContext();
+
+        String ipAddr = IP_ADDRESS;
+        String hostname = HOST_NAME;
+        try {
+            final InetAddress ip = InetAddress.getLocalHost();
+            hostname = ip.getCanonicalHostName();
+            ipAddr = ip.getHostAddress();
+        } catch (UnknownHostException x) {
+            Log.debug(x.getMessage());
+        }
+
+        lc.put("serverName", hostname);
+        lc.put("serviceName", "VESCollecor");
+        lc.put("statusCode", "RUNNING");
+        lc.put("targetEntity", "NULL");
+        lc.put("targetServiceName", "NULL");
+        lc.put("server", hostname);
+        lc.put("serverIpAddress", ipAddr);
+
+        // instance UUID is meaningless here, so we just create a new one each
+        // time the
+        // server starts. One could argue each new instantiation of the service
+        // should
+        // have a new instance ID.
+        lc.put("instanceUuid", "");
+        lc.put("severity", "");
+        lc.put(EcompFields.kEndTimestamp, SaClock.now());
+        lc.put("EndTimestamp", SaClock.now());
+        lc.put("partnerName", "NA");
+    }
 
 }
diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java
index 5865b12c..179e8826 100644
--- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java
+++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java
@@ -19,21 +19,19 @@
  */
 package org.onap.dcae.commonFunction.event.publishing;
 
-import static io.vavr.API.List;
-import static io.vavr.API.Try;
-import static io.vavr.API.Tuple;
-import static io.vavr.API.unchecked;
-import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
-import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
-
 import io.vavr.collection.List;
 import io.vavr.collection.Map;
 import io.vavr.control.Option;
 import io.vavr.control.Try;
+import org.onap.dcae.commonFunction.AnyNode;
+
 import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import org.onap.dcae.commonFunction.AnyNode;
+
+import static io.vavr.API.*;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
 
 /**
  * @author Pawel Szalapski (pawel.szalapski@nokia.com)
@@ -43,23 +41,23 @@ public final class DMaaPConfigurationParser {
 
     public static Try<Map<String, PublisherConfig>> parseToDomainMapping(Path configLocation) {
         return readFromFile(configLocation)
-            .flatMap(DMaaPConfigurationParser::toJSON)
-            .flatMap(DMaaPConfigurationParser::toConfigMap);
+                .flatMap(DMaaPConfigurationParser::toJSON)
+                .flatMap(DMaaPConfigurationParser::toConfigMap);
     }
 
     private static Try<String> readFromFile(Path configLocation) {
         return Try(() -> new String(Files.readAllBytes(configLocation)))
-            .mapFailure(enhanceError(f("Could not read DMaaP configuration from location: '%s'", configLocation)));
+                .mapFailure(enhanceError(f("Could not read DMaaP configuration from location: '%s'", configLocation)));
     }
 
     private static Try<AnyNode> toJSON(String config) {
         return Try(() -> AnyNode.fromString(config))
-            .mapFailure(enhanceError(f("DMaaP configuration '%s' is not a valid JSON document", config)));
+                .mapFailure(enhanceError(f("DMaaP configuration '%s' is not a valid JSON document", config)));
     }
 
     private static Try<Map<String, PublisherConfig>> toConfigMap(AnyNode config) {
         return Try(() -> usesLegacyFormat(config) ? parseLegacyFormat(config) : parseNewFormat(config))
-            .mapFailure(enhanceError(f("Parsing DMaaP configuration: '%s' failed, probably it is in unexpected format", config)));
+                .mapFailure(enhanceError(f("Parsing DMaaP configuration: '%s' failed, probably it is in unexpected format", config)));
     }
 
     private static boolean usesLegacyFormat(AnyNode dMaaPConfig) {
@@ -68,40 +66,40 @@ public final class DMaaPConfigurationParser {
 
     private static Map<String, PublisherConfig> parseLegacyFormat(AnyNode root) {
         return root.get("channels").toList().toMap(
-            channel -> channel.get("name").toString(),
-            channel -> {
-                String destinationsStr = channel.getAsOption("cambria.url")
-                    .getOrElse(channel.getAsOption("cambria.hosts").get())
-                    .toString();
-                String topic = channel.get("cambria.topic").toString();
-                Option<String> maybeUser = channel.getAsOption("basicAuthUsername").map(AnyNode::toString);
-                Option<String> maybePassword = channel.getAsOption("basicAuthPassword").map(AnyNode::toString);
-                List<String> destinations = List(destinationsStr.split(","));
-                return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations);
-            });
+                channel -> channel.get("name").toString(),
+                channel -> {
+                    String destinationsStr = channel.getAsOption("cambria.url")
+                            .getOrElse(channel.getAsOption("cambria.hosts").get())
+                            .toString();
+                    String topic = channel.get("cambria.topic").toString();
+                    Option<String> maybeUser = channel.getAsOption("basicAuthUsername").map(AnyNode::toString);
+                    Option<String> maybePassword = channel.getAsOption("basicAuthPassword").map(AnyNode::toString);
+                    List<String> destinations = List(destinationsStr.split(","));
+                    return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations);
+                });
     }
 
     private static Map<String, PublisherConfig> parseNewFormat(AnyNode root) {
         return root.keys().toMap(
-            channelName -> channelName,
-            channelName -> {
-                AnyNode channelConfig = root.get(channelName);
-                Option<String> maybeUser = channelConfig.getAsOption("aaf_username").map(AnyNode::toString);
-                Option<String> maybePassword = channelConfig.getAsOption("aaf_password").map(AnyNode::toString);
-                URL topicURL = unchecked(
-                    () -> new URL(channelConfig.get("dmaap_info").get("topic_url").toString())).apply();
-                String[] pathSegments = topicURL.getPath().substring(1).split("/");
-                String topic = pathSegments[1];
-                String destination = "events".equals(pathSegments[0]) ? topicURL.getAuthority() : topicURL.getHost();
-                List<String> destinations = List(destination);
-                return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations);
-            });
+                channelName -> channelName,
+                channelName -> {
+                    AnyNode channelConfig = root.get(channelName);
+                    Option<String> maybeUser = channelConfig.getAsOption("aaf_username").map(AnyNode::toString);
+                    Option<String> maybePassword = channelConfig.getAsOption("aaf_password").map(AnyNode::toString);
+                    URL topicURL = unchecked(
+                            () -> new URL(channelConfig.get("dmaap_info").get("topic_url").toString())).apply();
+                    String[] pathSegments = topicURL.getPath().substring(1).split("/");
+                    String topic = pathSegments[1];
+                    String destination = "events".equals(pathSegments[0]) ? topicURL.getAuthority() : topicURL.getHost();
+                    List<String> destinations = List(destination);
+                    return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations);
+                });
     }
 
     private static PublisherConfig buildBasedOnAuth(Option<String> maybeUser, Option<String> maybePassword,
                                                     String topic, List<String> destinations) {
         return maybeUser.flatMap(user -> maybePassword.map(password -> Tuple(user, password)))
-            .map(credentials -> new PublisherConfig(destinations, topic, credentials._1, credentials._2))
-            .getOrElse(new PublisherConfig(destinations, topic));
+                .map(credentials -> new PublisherConfig(destinations, topic, credentials._1, credentials._2))
+                .getOrElse(new PublisherConfig(destinations, topic));
     }
 }
diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPEventPublisher.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPEventPublisher.java
index fd9b3ae1..a0ee3bfb 100644
--- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPEventPublisher.java
+++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPEventPublisher.java
@@ -21,20 +21,21 @@
 
 package org.onap.dcae.commonFunction.event.publishing;
 
-import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
-
 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
 import com.att.nsa.clock.SaClock;
 import com.att.nsa.logging.LoggingContext;
 import com.att.nsa.logging.log4j.EcompFields;
 import io.vavr.collection.Map;
 import io.vavr.control.Try;
-import java.io.IOException;
 import org.json.JSONObject;
 import org.onap.dcae.commonFunction.VESLogger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+
 /**
  * @author Pawel Szalapski (pawel.szalapski@nokia.com)
  */
@@ -55,9 +56,9 @@ class DMaaPEventPublisher implements EventPublisher {
     public void sendEvent(JSONObject event, String domain) {
         clearVesUniqueIdFromEvent(event);
         publishersCache.getPublisher(domain)
-            .onEmpty(() ->
-                log.warn(f("Could not find event publisher for domain: '%s', dropping message: '%s'", domain, event)))
-            .forEach(publisher -> sendEvent(event, domain, publisher));
+                .onEmpty(() ->
+                        log.warn(f("Could not find event publisher for domain: '%s', dropping message: '%s'", domain, event)))
+                .forEach(publisher -> sendEvent(event, domain, publisher));
     }
 
     @Override
@@ -67,11 +68,11 @@ class DMaaPEventPublisher implements EventPublisher {
 
     private void sendEvent(JSONObject event, String domain, CambriaBatchingPublisher publisher) {
         Try.run(() -> uncheckedSendEvent(event, domain, publisher))
-            .onFailure(exc -> closePublisher(event, domain, exc));
+                .onFailure(exc -> closePublisher(event, domain, exc));
     }
 
     private void uncheckedSendEvent(JSONObject event, String domain, CambriaBatchingPublisher publisher)
-        throws IOException {
+            throws IOException {
         int pendingMsgs = publisher.send("MyPartitionKey", event.toString());
         if (pendingMsgs > PENDING_MESSAGE_LOG_THRESHOLD) {
             log.info("Pending messages count: " + pendingMsgs);
@@ -83,7 +84,7 @@ class DMaaPEventPublisher implements EventPublisher {
 
     private void closePublisher(JSONObject event, String domain, Throwable e) {
         log.error(f("Unable to schedule event: '%s' on domain: '%s'. Closing publisher and dropping message.",
-            event, domain), e);
+                event, domain), e);
         publishersCache.closePublisherFor(domain);
     }
 
diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java
index a7865a45..489fcbf0 100644
--- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java
+++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java
@@ -19,15 +19,15 @@
  */
 package org.onap.dcae.commonFunction.event.publishing;
 
-import static io.vavr.API.Try;
-import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
-import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
-
 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
 import com.att.nsa.cambria.client.CambriaClientBuilders;
 import com.att.nsa.cambria.client.CambriaClientBuilders.PublisherBuilder;
 import io.vavr.control.Try;
 
+import static io.vavr.API.Try;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
+
 /**
  * @author Pawel Szalapski (pawel.szalapski@nokia.com)
  */
@@ -36,7 +36,7 @@ final class DMaaPPublishersBuilder {
     @SuppressWarnings("mapFailure takes a generic varargs, unchecked because of Javas type system limitation, actually safe to do")
     static Try<CambriaBatchingPublisher> buildPublisher(PublisherConfig config) {
         return Try(() -> builder(config).build())
-            .mapFailure(enhanceError(f("DMaaP client builder throws exception for this configuration: '%s'", config)));
+                .mapFailure(enhanceError(f("DMaaP client builder throws exception for this configuration: '%s'", config)));
     }
 
     private static PublisherBuilder builder(PublisherConfig config) {
@@ -49,14 +49,14 @@ final class DMaaPPublishersBuilder {
 
     private static PublisherBuilder authenticatedBuilder(PublisherConfig config) {
         return unAuthenticatedBuilder(config)
-            .usingHttps()
-            .authenticatedByHttp(config.userName().get(), config.password().get());
+                .usingHttps()
+                .authenticatedByHttp(config.userName().get(), config.password().get());
     }
 
     private static PublisherBuilder unAuthenticatedBuilder(PublisherConfig config) {
         return new CambriaClientBuilders.PublisherBuilder()
-            .usingHosts(config.destinations().mkString(","))
-            .onTopic(config.topic())
-            .logSendFailuresAfter(5);
+                .usingHosts(config.destinations().mkString(","))
+                .onTopic(config.topic())
+                .logSendFailuresAfter(5);
     }
 }
diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersCache.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersCache.java
index 102d2774..4cdf92da 100644
--- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersCache.java
+++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersCache.java
@@ -20,24 +20,20 @@
  */
 package org.onap.dcae.commonFunction.event.publishing;
 
-import static io.vavr.API.Option;
-import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
-
 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.CacheLoader;
-import com.google.common.cache.LoadingCache;
-import com.google.common.cache.RemovalListener;
-import com.google.common.cache.RemovalNotification;
+import com.google.common.cache.*;
 import io.vavr.collection.Map;
 import io.vavr.control.Option;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
-import javax.annotation.Nonnull;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
+import static io.vavr.API.Option;
+import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f;
 
 /**
  * @author Pawel Szalapski (pawel.szalapski@nokia.com)
@@ -51,8 +47,8 @@ class DMaaPPublishersCache {
     DMaaPPublishersCache(Map<String, PublisherConfig> dMaaPConfiguration) {
         this.dMaaPConfiguration = new AtomicReference<>(dMaaPConfiguration);
         this.publishersCache = CacheBuilder.newBuilder()
-            .removalListener(new OnPublisherRemovalListener())
-            .build(new CambriaPublishersCacheLoader());
+                .removalListener(new OnPublisherRemovalListener())
+                .build(new CambriaPublishersCacheLoader());
     }
 
     DMaaPPublishersCache(CambriaPublishersCacheLoader dMaaPPublishersCacheLoader,
@@ -60,8 +56,8 @@ class DMaaPPublishersCache {
                          Map<String, PublisherConfig> dMaaPConfiguration) {
         this.dMaaPConfiguration = new AtomicReference<>(dMaaPConfiguration);
         this.publishersCache = CacheBuilder.newBuilder()
-            .removalListener(onPublisherRemovalListener)
-            .build(dMaaPPublishersCacheLoader);
+                .removalListener(onPublisherRemovalListener)
+                .build(dMaaPPublishersCacheLoader);
     }
 
     Option<CambriaBatchingPublisher> getPublisher(String streamID) {
@@ -80,9 +76,9 @@ class DMaaPPublishersCache {
     synchronized void reconfigure(Map<String, PublisherConfig> newConfig) {
         Map<String, PublisherConfig> currentConfig = dMaaPConfiguration.get();
         Map<String, PublisherConfig> removedConfigurations = currentConfig
-            .filterKeys(domain -> !newConfig.containsKey(domain));
+                .filterKeys(domain -> !newConfig.containsKey(domain));
         Map<String, PublisherConfig> changedConfigurations = newConfig
-            .filterKeys(e -> currentConfig.containsKey(e) && !currentConfig.get(e).equals(newConfig.get(e)));
+                .filterKeys(e -> currentConfig.containsKey(e) && !currentConfig.get(e).equals(newConfig.get(e)));
         dMaaPConfiguration.set(newConfig);
         removedConfigurations.merge(changedConfigurations).forEach(e -> publishersCache.invalidate(e._1));
     }
@@ -99,7 +95,7 @@ class DMaaPPublishersCache {
                     java.util.List<?> stuck = publisher.close(timeout, unit);
                     if (!stuck.isEmpty()) {
                         log.error(f("Publisher got stuck and did not manage to close in '%s' '%s', "
-                            + "%s messages were dropped", stuck.size(), timeout, unit));
+                                + "%s messages were dropped", stuck.size(), timeout, unit));
                     }
                 } catch (InterruptedException | IOException e) {
                     log.error("Could not close Cambria publisher, some messages might have been dropped", e);
@@ -113,11 +109,11 @@ class DMaaPPublishersCache {
         @Override
         public CambriaBatchingPublisher load(@Nonnull String domain) {
             return dMaaPConfiguration.get()
-                .get(domain)
-                .toTry(() -> new RuntimeException(
-                    f("DMaaP configuration contains no configuration for domain: '%s'", domain)))
-                .flatMap(DMaaPPublishersBuilder::buildPublisher)
-                .get();
+                    .get(domain)
+                    .toTry(() -> new RuntimeException(
+                            f("DMaaP configuration contains no configuration for domain: '%s'", domain)))
+                    .flatMap(DMaaPPublishersBuilder::buildPublisher)
+                    .get();
         }
     }
 
diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/PublisherConfig.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/PublisherConfig.java
index 4a056778..f1cbb8e5 100644
--- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/PublisherConfig.java
+++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/PublisherConfig.java
@@ -21,6 +21,7 @@ package org.onap.dcae.commonFunction.event.publishing;
 
 import io.vavr.collection.List;
 import io.vavr.control.Option;
+
 import java.util.Objects;
 
 /**
@@ -76,9 +77,9 @@ public final class PublisherConfig {
         }
         PublisherConfig that = (PublisherConfig) o;
         return Objects.equals(destinations, that.destinations) &&
-            Objects.equals(topic, that.topic) &&
-            Objects.equals(userName, that.userName) &&
-            Objects.equals(password, that.password);
+                Objects.equals(topic, that.topic) &&
+                Objects.equals(userName, that.userName) &&
+                Objects.equals(password, that.password);
     }
 
     @Override
@@ -89,10 +90,10 @@ public final class PublisherConfig {
     @Override
     public String toString() {
         return "PublisherConfig{" +
-            "destinations=" + destinations +
-            ", topic='" + topic + '\'' +
-            ", userName='" + userName + '\'' +
-            ", password='" + password + '\'' +
-            '}';
+                "destinations=" + destinations +
+                ", topic='" + topic + '\'' +
+                ", userName='" + userName + '\'' +
+                ", password='" + password + '\'' +
+                '}';
     }
 }
diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java
index 9bf3ef8c..78f34ff4 100644
--- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java
+++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java
@@ -19,11 +19,11 @@
  */
 package org.onap.dcae.commonFunction.event.publishing;
 
-import static io.vavr.API.$;
-
 import io.vavr.API;
 import io.vavr.API.Match.Case;
 
+import static io.vavr.API.$;
+
 /**
  * @author Pawel Szalapski (pawel.szalapski@nokia.com)
  */