Merge "Fix code architecure in CrudAsyncResponseConsumer"

4 files changed, 220 insertions, 77 deletions

diff --git a/src/main/java/org/onap/crud/service/CrudAsyncGraphDataService.java b/src/main/java/org/onap/crud/service/CrudAsyncGraphDataService.java
index c2d2591..26f7427 100644
--- a/src/main/java/org/onap/crud/service/CrudAsyncGraphDataService.java
+++ b/src/main/java/org/onap/crud/service/CrudAsyncGraphDataService.java
@@ -120,7 +120,9 @@ public class CrudAsyncGraphDataService extends AbstractGraphDataService {
         }
 
         // Start the Response Consumer timer
-        CrudAsyncResponseConsumer crudAsyncResponseConsumer = new CrudAsyncResponseConsumer(asyncResponseConsumer);
+        CrudAsyncResponseConsumer crudAsyncResponseConsumer = new CrudAsyncResponseConsumer(
+            asyncResponseConsumer, new GraphEventUpdater()
+        );
         timer = new Timer("crudAsyncResponseConsumer-1");
         timer.schedule(crudAsyncResponseConsumer, responsePollInterval, responsePollInterval);
 
diff --git a/src/main/java/org/onap/crud/service/CrudAsyncResponseConsumer.java b/src/main/java/org/onap/crud/service/CrudAsyncResponseConsumer.java
index 94c1e1b..4c4ca2f 100644
--- a/src/main/java/org/onap/crud/service/CrudAsyncResponseConsumer.java
+++ b/src/main/java/org/onap/crud/service/CrudAsyncResponseConsumer.java
@@ -20,103 +20,60 @@
  */
 package org.onap.crud.service;
 
+import java.util.Objects;
 import java.util.TimerTask;
-
-import javax.naming.OperationNotSupportedException;
-
 import org.onap.aai.cl.api.Logger;
 import org.onap.aai.cl.eelf.LoggerFactory;
-import org.onap.crud.event.GraphEvent;
-import org.onap.crud.event.envelope.GraphEventEnvelope;
-import org.onap.crud.logging.CrudServiceMsgs;
-
 import org.onap.aai.event.api.EventConsumer;
+import org.onap.crud.logging.CrudServiceMsgs;
 
 public class CrudAsyncResponseConsumer extends TimerTask {
 
-  private static Logger logger = LoggerFactory.getInstance().getLogger(CrudAsyncResponseConsumer
-                                                                       .class.getName());
+    private static Logger logger = LoggerFactory.getInstance().getLogger(CrudAsyncResponseConsumer
+        .class.getName());
 
-  private static Logger auditLogger = LoggerFactory.getInstance()
-    .getAuditLogger(CrudAsyncResponseConsumer.class.getName());
+    private final EventConsumer asyncResponseConsumer;
+    private final GraphEventUpdater graphEventUpdater;
 
-  private EventConsumer asyncResponseConsumer;
 
- 
-  public CrudAsyncResponseConsumer(EventConsumer asyncResponseConsumer) {
-    this.asyncResponseConsumer = asyncResponseConsumer;
-    logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
-                "CrudAsyncResponseConsumer initialized SUCCESSFULLY! with event consumer "
+    public CrudAsyncResponseConsumer(EventConsumer asyncResponseConsumer, GraphEventUpdater graphEventUpdater) {
+        Objects.requireNonNull(asyncResponseConsumer);
+        Objects.requireNonNull(graphEventUpdater);
+        this.asyncResponseConsumer = asyncResponseConsumer;
+        this.graphEventUpdater = graphEventUpdater;
+        logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
+            "CrudAsyncResponseConsumer initialized SUCCESSFULLY! with event consumer "
                 + asyncResponseConsumer.getClass().getName());
-  }
-
-
-  @Override
-  public void run() {
-
-    logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "Listening for graph events");
-
-    if (asyncResponseConsumer == null) {
-      logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR,
-                   "Unable to initialize CrudAsyncRequestProcessor");
     }
 
-    Iterable<String> events = null;
-    try {
-      events = asyncResponseConsumer.consume();
-    } catch (Exception e) {
-      logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
-      return;
-    }
-
-    if (events == null || !events.iterator().hasNext()) {
-      logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "No events recieved");
 
-    }
+    @Override
+    public void run() {
 
-    for (String event : events) {
-      try {
+        logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "Listening for graph events");
 
-        GraphEventEnvelope graphEventEnvelope = GraphEventEnvelope.fromJson(event);
-        GraphEvent graphEvent = graphEventEnvelope.getBody();
-        auditLogger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
-                           "Event received of type: " + graphEvent.getObjectType() + " with key: "
-                           + graphEvent.getObjectKey() + " , transaction-id: "
-                           + graphEvent.getTransactionId() + " , operation: "
-                           + graphEvent.getOperation().toString());
-          logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
-                      "Event received of type: " + graphEvent.getObjectType() + " with key: "
-                      + graphEvent.getObjectKey() + " , transaction-id: "
-                      + graphEvent.getTransactionId() + " , operation: "
-                      + graphEvent.getOperation().toString());
-          logger.debug(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
-                       "Event received with payload:" + event);
+        try {
+            Iterable<String> events = asyncResponseConsumer.consume();
+            processEvents(events);
+            asyncResponseConsumer.commitOffsets();
+        } catch (Exception e) {
+            logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e, e.getMessage());
+        }
+    }
 
-        if (CrudAsyncGraphEventCache.get(graphEvent.getTransactionId()) != null) {
-          CrudAsyncGraphEventCache.get(graphEvent.getTransactionId())
-            .populateGraphEventEnvelope(graphEventEnvelope);
+    private void processEvents(Iterable<String> events) {
+        if (areEventsAvailable(events)) {
+            for (String event : events) {
+                graphEventUpdater.update(event);
+            }
+            logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "No events recieved");
         } else {
-          logger.error(CrudServiceMsgs.ASYNC_DATA_SERVICE_ERROR,
-                       "Request timed out. Not sending response for transaction-id: "
-                       + graphEvent.getTransactionId());
+            logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "No events recieved");
         }
-
-      } catch (Exception e) {
-        logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
-      }
     }
 
-    try {
-      asyncResponseConsumer.commitOffsets();
-    }
-    catch(OperationNotSupportedException e) {
-        //Dmaap doesnt support commit with offset    
-        logger.debug(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
+    private boolean areEventsAvailable(Iterable<String> events) {
+        return !(events == null || !events.iterator().hasNext());
     }
-    catch (Exception e) {
-      logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
-    }
-
-  }
 
 }
\ No newline at end of file
diff --git a/src/main/java/org/onap/crud/service/GraphEventUpdater.java b/src/main/java/org/onap/crud/service/GraphEventUpdater.java
new file mode 100644
index 0000000..70fdd0d
--- /dev/null
+++ b/src/main/java/org/onap/crud/service/GraphEventUpdater.java
@@ -0,0 +1,69 @@
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2017-2018 Amdocs
+ * ================================================================================
+ * 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.crud.service;
+
+import org.onap.aai.cl.api.Logger;
+import org.onap.aai.cl.eelf.LoggerFactory;
+import org.onap.crud.event.GraphEvent;
+import org.onap.crud.event.envelope.GraphEventEnvelope;
+import org.onap.crud.logging.CrudServiceMsgs;
+
+
+public class GraphEventUpdater {
+
+    private static Logger logger = LoggerFactory.getInstance().getLogger(GraphEventUpdater
+        .class.getName());
+
+    private static Logger auditLogger = LoggerFactory.getInstance()
+        .getAuditLogger(GraphEventUpdater.class.getName());
+
+    public void update(String eventAsJson) {
+        try {
+
+            GraphEventEnvelope graphEventEnvelope = GraphEventEnvelope.fromJson(eventAsJson);
+            GraphEvent graphEvent = graphEventEnvelope.getBody();
+            auditLogger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
+                "Event received of type: " + graphEvent.getObjectType() + " with key: "
+                    + graphEvent.getObjectKey() + " , transaction-id: "
+                    + graphEvent.getTransactionId() + " , operation: "
+                    + graphEvent.getOperation().toString());
+            logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
+                "Event received of type: " + graphEvent.getObjectType() + " with key: "
+                    + graphEvent.getObjectKey() + " , transaction-id: "
+                    + graphEvent.getTransactionId() + " , operation: "
+                    + graphEvent.getOperation().toString());
+            logger.debug(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
+                "Event received with payload:" + eventAsJson);
+
+            if (CrudAsyncGraphEventCache.get(graphEvent.getTransactionId()) != null) {
+                CrudAsyncGraphEventCache.get(graphEvent.getTransactionId())
+                    .populateGraphEventEnvelope(graphEventEnvelope);
+            } else {
+                logger.error(CrudServiceMsgs.ASYNC_DATA_SERVICE_ERROR,
+                    "Request timed out. Not sending response for transaction-id: "
+                        + graphEvent.getTransactionId());
+            }
+
+        } catch (Exception e) {
+            logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
+        }
+    }
+}
diff --git a/src/test/java/org/onap/crud/service/CrudAsyncResponseConsumerTest.java b/src/test/java/org/onap/crud/service/CrudAsyncResponseConsumerTest.java
new file mode 100644
index 0000000..6e7cb3b
--- /dev/null
+++ b/src/test/java/org/onap/crud/service/CrudAsyncResponseConsumerTest.java
@@ -0,0 +1,115 @@
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2018 Nokia
+ * ================================================================================
+ * 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.crud.service;
+
+import static junit.framework.TestCase.fail;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.Lists;
+import java.util.ArrayList;
+import javax.naming.OperationNotSupportedException;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.onap.aai.event.api.EventConsumer;
+
+
+@RunWith(MockitoJUnitRunner.class)
+public class CrudAsyncResponseConsumerTest {
+
+    private static final ArrayList<String> EVENTS = Lists.newArrayList("event_json1", "event_json2");
+
+    @Mock
+    private EventConsumer eventConsumer;
+    @Mock
+    private GraphEventUpdater graphEventUpdater;
+
+    private CrudAsyncResponseConsumer crudAsyncResponseConsumer;
+
+    @Before
+    public void setUp() {
+        crudAsyncResponseConsumer = new CrudAsyncResponseConsumer(eventConsumer, graphEventUpdater);
+    }
+
+    @Test
+    public void shouldCommitOnlyOffsetsWhenEventsCollectionIsEmpty() throws Exception {
+        // given
+        when(eventConsumer.consume()).thenReturn(new ArrayList<>());
+
+        // when
+        crudAsyncResponseConsumer.run();
+
+        // then
+        verify(graphEventUpdater, never()).update(anyString());
+        verify(eventConsumer, times(1)).commitOffsets();
+    }
+
+    @Test
+    public void shouldCommitOnlyOffsetsWhenThereIsNoEventsToProcess() throws Exception {
+        // given
+        when(eventConsumer.consume()).thenReturn(null);
+
+        // when
+        crudAsyncResponseConsumer.run();
+
+        // then
+        verify(graphEventUpdater, never()).update(anyString());
+        verify(eventConsumer, times(1)).commitOffsets();
+    }
+
+    @Test
+    public void shouldProcessEventsWhenConsumerProvidesListOfEvents() throws Exception {
+        // given
+        when(eventConsumer.consume()).thenReturn(EVENTS);
+
+        // when
+        crudAsyncResponseConsumer.run();
+
+        // then
+        verify(graphEventUpdater, times(2)).update(anyString());
+        verify(eventConsumer, times(1)).commitOffsets();
+    }
+
+    @Test
+    public void shouldHandleAnyErrorCaseDuringCommitOffsets() throws Exception {
+        // given
+        when(eventConsumer.consume()).thenReturn(EVENTS);
+        doThrow(OperationNotSupportedException.class).when(eventConsumer).commitOffsets();
+
+        // when
+        try {
+            crudAsyncResponseConsumer.run();
+        } catch (Exception e) {
+            fail("Any error reported by run method is wrong!");
+        }
+
+        // then
+        verify(graphEventUpdater, times(2)).update(anyString());
+
+    }
+}
\ No newline at end of file