From bcec586b8261ee99285973fd2640a6cc4c25451b Mon Sep 17 00:00:00 2001
From: Zlatko Murgoski <zlatko.murgoski@nokia.com>
Date: Mon, 31 Dec 2018 11:29:52 +0100
Subject: Fix bug with processing event list

DMaapEventPublisher processing of eventList

Change-Id: I3aa31dd55aefdc5130fcc36712f20bd15c3f7cce
Issue-ID: DCAEGEN2-1035
Signed-off-by: Zlatko Murgoski <zlatko.murgoski@nokia.com>
---
 .../java/org/onap/dcae/commonFunction/EventProcessor.java     | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

(limited to 'src/main')

diff --git a/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java b/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java
index 7d27399d..0d150c51 100644
--- a/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java
+++ b/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java
@@ -50,7 +50,6 @@ public class EventProcessor implements Runnable {
     private static final String COMMON_EVENT_HEADER = "commonEventHeader";
     private final SimpleDateFormat dateFormat = new SimpleDateFormat("EEE, MM dd yyyy hh:mm:ss z");
 
-    public JSONObject event;
     private EventPublisher eventPublisher;
     private Map<String, String[]> streamidHash;
     private ApplicationSettings properties;
@@ -66,7 +65,7 @@ public class EventProcessor implements Runnable {
     public void run() {
         try {
             while (true) {
-                event = VesApplication.fProcessingInputQueue.take();
+                JSONObject event = VesApplication.fProcessingInputQueue.take();
                 // As long as the producer is running we remove elements from
                 // the queue.
                 log.info("QueueSize:" + VesApplication.fProcessingInputQueue.size() + "\tEventProcessor\tRemoving element: " + event);
@@ -81,7 +80,7 @@ public class EventProcessor implements Runnable {
                         .onEmpty(() -> {
                             log.error("No StreamID defined for publish - Message dropped" + event);
                         }).forEach(streamIds -> {
-                    sendEventsToStreams(streamIds);
+                    sendEventsToStreams(event, streamIds);
                 });
                 log.debug("Message published" + event);
             }
@@ -91,7 +90,7 @@ public class EventProcessor implements Runnable {
         }
     }
 
-    public void overrideEvent() {
+    public void overrideEvent(JSONObject event) {
         // Set collector timestamp in event payload before publish
         addCurrentTimeToEvent(event);
 
@@ -112,10 +111,10 @@ public class EventProcessor implements Runnable {
         log.debug("Modified event:" + event);
     }
 
-    private void sendEventsToStreams(String[] streamIdList) {
+    private void sendEventsToStreams(JSONObject event, String[] streamIdList) {
         for (String aStreamIdList : streamIdList) {
             log.info("Invoking publisher for streamId:" + aStreamIdList);
-            this.overrideEvent();
+            overrideEvent(event);
             eventPublisher.sendEvent(event, aStreamIdList);
         }
     }
-- 
cgit 1.2.3-korg