}
// 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);
*/
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
--- /dev/null
+/**
+ * ============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());
+ }
+ }
+}
--- /dev/null
+/**
+ * ============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
Code Review / aai / gizmo.git / commitdiff