diff options
author | Bogumil Zebek <bogumil.zebek@nokia.com> | 2018-07-06 10:38:45 +0200 |
---|---|---|
committer | Bogumil Zebek <bogumil.zebek@nokia.com> | 2018-07-09 13:29:01 +0200 |
commit | a14d63079541c40f9471407b8753d56e1fcc0ea4 (patch) | |
tree | debac624152db6a3bc9adee5860bb1fd880c6bdc /src/main/java/org/onap/crud/service/CrudAsyncResponseConsumer.java | |
parent | 4b09e9e4fe77a4aa749f1a7ffa07545b64d3e2cb (diff) |
Fix code architecure in CrudAsyncResponseConsumer
- fix code complexity
- make code testable
- add missing tests
Change-Id: I542e7f603a5ffe956ae3fc3f58874b7dd2ce87a1
Issue-ID: AAI-1365
Signed-off-by: Bogumil Zebek <bogumil.zebek@nokia.com>
Diffstat (limited to 'src/main/java/org/onap/crud/service/CrudAsyncResponseConsumer.java')
-rw-r--r-- | src/main/java/org/onap/crud/service/CrudAsyncResponseConsumer.java | 109 |
1 files changed, 33 insertions, 76 deletions
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 |