1 files changed, 24 insertions, 70 deletions

diff --git a/src/main/java/org/onap/dcae/common/EventSender.java b/src/main/java/org/onap/dcae/common/EventSender.java
index 48268d6c..c1002af6 100644
--- a/src/main/java/org/onap/dcae/common/EventSender.java
+++ b/src/main/java/org/onap/dcae/common/EventSender.java
@@ -20,17 +20,12 @@
  */
 package org.onap.dcae.common;
 
-import com.google.common.reflect.TypeToken;
-import com.google.gson.Gson;
+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 java.io.FileReader;
-import java.io.IOException;
-import java.lang.reflect.Type;
-import java.text.SimpleDateFormat;
-import java.util.Date;
-import java.util.List;
+import org.json.JSONArray;
 import org.json.JSONObject;
-import org.onap.dcae.ApplicationException;
 import org.onap.dcae.ApplicationSettings;
 import org.onap.dcae.common.publishing.EventPublisher;
 import org.slf4j.Logger;
@@ -38,88 +33,47 @@ import org.slf4j.LoggerFactory;
 
 public class EventSender {
 
-  private static final String COULD_NOT_FIND_FILE = "Couldn't find file ./etc/eventTransform.json";
+  private static final Logger metriclog = LoggerFactory.getLogger("com.att.ecomp.metrics");
   private Map<String, String[]> streamidHash;
-  private ApplicationSettings properties;
   private EventPublisher eventPublisher;
-
-  private static final Type EVENT_LIST_TYPE = new TypeToken<List<Event>>() {}.getType();
+  private static final String VES_UNIQUE_ID = "VESuniqueId";
   private static final Logger log = LoggerFactory.getLogger(EventSender.class);
   private static final String EVENT_LITERAL = "event";
   private static final String COMMON_EVENT_HEADER = "commonEventHeader";
-  private final SimpleDateFormat dateFormat = new SimpleDateFormat("EEE, MM dd yyyy hh:mm:ss z");
 
   public EventSender( EventPublisher eventPublisher, ApplicationSettings properties) {
     this.eventPublisher = eventPublisher;
     this.streamidHash = properties.dMaaPStreamsMapping();
-    this.properties = properties;
-
   }
 
-  public void send(JSONObject event) {
-    streamidHash.get(getDomain(event))
-        .onEmpty(() -> log.error("No StreamID defined for publish - Message dropped" + event))
-        .forEach(streamIds -> sendEventsToStreams(event, streamIds));
+  public void send(JSONArray arrayOfEvents) {
+    for (int i = 0; i < arrayOfEvents.length(); i++) {
+      metriclog.info("EVENT_PUBLISH_START");
+      JSONObject object = (JSONObject) arrayOfEvents.get(i);
+      setLoggingContext(object);
+      streamidHash.get(getDomain(object))
+          .onEmpty(() -> log.error("No StreamID defined for publish - Message dropped" + object))
+          .forEach(streamIds -> sendEventsToStreams(object, streamIds));
+      log.debug("Message published" + object);
+    }
+    log.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");
+    metriclog.info("EVENT_PUBLISH_END");
   }
 
-  public static String getDomain(JSONObject event) {
+  private static String getDomain(JSONObject event) {
     return event.getJSONObject(EVENT_LITERAL).getJSONObject(COMMON_EVENT_HEADER).getString("domain");
   }
 
   private void sendEventsToStreams(JSONObject event, String[] streamIdList) {
     for (String aStreamIdList : streamIdList) {
       log.info("Invoking publisher for streamId:" + aStreamIdList);
-      eventPublisher.sendEvent(overrideEvent(event), aStreamIdList);
+      eventPublisher.sendEvent(event, aStreamIdList);
     }
   }
 
-  private JSONObject overrideEvent(JSONObject event) {
-    JSONObject jsonObject = addCurrentTimeToEvent(event);
-    if (properties.eventTransformingEnabled()) {
-      try (FileReader fr = new FileReader("./etc/eventTransform.json")) {
-        log.info("parse eventTransform.json");
-        List<Event> events = new Gson().fromJson(fr, EVENT_LIST_TYPE);
-        parseEventsJson(events, new ConfigProcessorAdapter(new ConfigProcessors(jsonObject)));
-      } catch (IOException e) {
-        log.error(COULD_NOT_FIND_FILE, e);
-        throw new ApplicationException(COULD_NOT_FIND_FILE, e);
-      }
-    }
-    if (jsonObject.has("VESversion"))
-      jsonObject.remove("VESversion");
-
-    log.debug("Modified event:" + jsonObject);
-    return jsonObject;
-  }
-
-  private JSONObject addCurrentTimeToEvent(JSONObject event) {
-    final Date currentTime = new Date();
-    JSONObject collectorTimeStamp = new JSONObject().put("collectorTimeStamp", dateFormat.format(currentTime));
-    JSONObject commonEventHeaderkey = event.getJSONObject(EVENT_LITERAL).getJSONObject(COMMON_EVENT_HEADER);
-    commonEventHeaderkey.put("internalHeaderFields", collectorTimeStamp);
-    event.getJSONObject(EVENT_LITERAL).put(COMMON_EVENT_HEADER, commonEventHeaderkey);
-    return event;
-  }
-
-  private void parseEventsJson(List<Event> eventsTransform, ConfigProcessorAdapter configProcessorAdapter) {
-    for (Event eventTransform : eventsTransform) {
-      JSONObject filterObj = new JSONObject(eventTransform.filter.toString());
-      if (configProcessorAdapter.isFilterMet(filterObj)) {
-        callProcessorsMethod(configProcessorAdapter, eventTransform.processors);
-      }
-    }
-  }
-
-  private void callProcessorsMethod(ConfigProcessorAdapter configProcessorAdapter, List<Processor> processors) {
-    for (Processor processor : processors) {
-      final String functionName = processor.functionName;
-      final JSONObject args = new JSONObject(processor.args.toString());
-      log.info(String.format("functionName==%s | args==%s", functionName, args));
-      try {
-        configProcessorAdapter.runConfigProcessorFunctionByName(functionName, args);
-      } catch (ReflectiveOperationException e) {
-        log.error("EventProcessor Exception" + e.getMessage() + e + e.getCause());
-      }
-    }
+  private void setLoggingContext(JSONObject event) {
+    LoggingContext localLC = VESLogger.getLoggingContextForThread(event.get(VES_UNIQUE_ID).toString());
+    localLC.put(EcompFields.kBeginTimestampMs, SaClock.now());
+    log.debug("event.VESuniqueId" + event.get(VES_UNIQUE_ID) + "event.commonEventHeader.domain:" + getDomain(event));
   }
 }