Exception test Fix

Issue-ID: NONRTRIC-107
Change-Id: Ic4f43010b920e338d2c25660565afb25877296fb
Signed-off-by: Lathish <lathishbabu.ganesan@est.tech>

diff --git a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java
index 889dfaf..c132fa2 100644 (file)
--- a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java
+++ b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumer.java
@@ -49,11 +49,6 @@ public interface DmaapMessageConsumer {
      */
     public boolean isAlive();
 
-    /**
-     * To Stop the DMAAP Listener
-     */
-    public void stopConsumer();
-
     /**
      * It's a infinite loop run every configured seconds to fetch the message from DMAAP. This method can be stop by
      * setting the alive flag to false

diff --git a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumerImpl.java b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumerImpl.java
index c7dab58..bd3ba69 100644 (file)
--- a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumerImpl.java
+++ b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageConsumerImpl.java
@@ -20,8 +20,11 @@
 
 package org.oransc.policyagent.dmaap;
 
+import com.google.common.collect.Iterables;
 import java.io.IOException;
 import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import org.onap.dmaap.mr.client.MRClientFactory;
 import org.onap.dmaap.mr.client.MRConsumer;
 import org.onap.dmaap.mr.client.response.MRConsumerResponse;
@@ -42,16 +45,18 @@ public class DmaapMessageConsumerImpl implements DmaapMessageConsumer {
     private boolean alive = false;
     private final ApplicationConfig applicationConfig;
     protected MRConsumer consumer;
-    private MRConsumerResponse response = null;
     @Autowired
     private DmaapMessageHandler dmaapMessageHandler;
+    private final long FETCHTIMEOUT = 30000;
+
+    private CountDownLatch sleep = new CountDownLatch(1);
 
     @Autowired
     public DmaapMessageConsumerImpl(ApplicationConfig applicationConfig) {
         this.applicationConfig = applicationConfig;
     }
 
-    @Scheduled(fixedRate = 1000 * 10) // , initialDelay=60000)
+    @Scheduled(fixedRate = 1000 * 10)
     @Override
     public void run() {
         if (!alive) {
@@ -60,9 +65,11 @@ public class DmaapMessageConsumerImpl implements DmaapMessageConsumer {
         if (this.alive) {
             try {
                 Iterable<String> dmaapMsgs = fetchAllMessages();
-                logger.debug("Fetched all the messages from DMAAP and will start to process the messages");
-                for (String msg : dmaapMsgs) {
-                    processMsg(msg);
+                if (dmaapMsgs != null && Iterables.size(dmaapMsgs) > 0) {
+                    logger.debug("Fetched all the messages from DMAAP and will start to process the messages");
+                    for (String msg : dmaapMsgs) {
+                        processMsg(msg);
+                    }
                 }
             } catch (Exception e) {
                 logger.error("{}: cannot fetch because of ", this, e.getMessage(), e);
@@ -71,15 +78,17 @@ public class DmaapMessageConsumerImpl implements DmaapMessageConsumer {
     }
 
     private Iterable<String> fetchAllMessages() {
-        response = consumer.fetchWithReturnConsumerResponse();
-        if (response == null) {
+        MRConsumerResponse response = null;
+        try {
+            response = consumer.fetchWithReturnConsumerResponse();
+        } catch (Exception e) {
+            logger.error("Failed to get message from DMAAP", e);
+        }
+        if (response == null || !"200".equals(response.getResponseCode())) {
             logger.warn("{}: DMaaP NULL response received", this);
+            sleepAfterFailure();
         } else {
             logger.debug("DMaaP consumer received {} : {}", response.getResponseCode(), response.getResponseMessage());
-            if (!"200".equals(response.getResponseCode())) {
-                logger.error("DMaaP consumer received: {} : {}", response.getResponseCode(),
-                        response.getResponseMessage());
-            }
         }
         return response.getActualMessages();
     }
@@ -117,8 +126,12 @@ public class DmaapMessageConsumerImpl implements DmaapMessageConsumer {
         return alive;
     }
 
-    @Override
-    public void stopConsumer() {
-        alive = false;
+    private void sleepAfterFailure() {
+        logger.warn("DMAAP message Consumer is put to Sleep for {} milliseconds", FETCHTIMEOUT);
+        try {
+            sleep.await(FETCHTIMEOUT, TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            logger.error("Failed to put the thread to sleep: {}", e);
+        }
     }
 }

diff --git a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java
index 172fe98..61c1e1e 100644 (file)
--- a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java
+++ b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java
@@ -45,6 +45,7 @@ public class DmaapMessageHandler {
 
     private static final Logger logger = LoggerFactory.getLogger(DmaapMessageHandler.class);
 
+    private boolean initialize = false;
     @Autowired
     private ObjectMapper mapper;
     @Autowired
@@ -57,8 +58,9 @@ public class DmaapMessageHandler {
     // The publish properties is corrupted. It contains the subscribe property values.
     @Async("threadPoolTaskExecutor")
     public void handleDmaapMsg(String msg) {
-        logger.debug("Message  ---------->{}", msg);
-        init();
+        if (!initialize) {
+            init();
+        }
         DmaapRequestMessage dmaapRequestMessage = null;
         Optional<String> dmaapResponse = null;
         // Process the message
@@ -166,6 +168,7 @@ public class DmaapMessageHandler {
         topic = dmaapPublisherConfig.getProperty("topic");
         logger.debug("Read the topic & Service Name - {} , {}", host, topic);
         this.restClient = new AsyncRestClient("http://" + host + "/"); // get this value from application config
+        initialize = true;
 
     }
 }