diff options
| -rw-r--r-- | .gitignore | 1 | ||||
| -rw-r--r-- | champ-lib/champ-core/pom.xml | 2 | ||||
| -rw-r--r-- | champ-lib/champ-janus/pom.xml | 2 | ||||
| -rw-r--r-- | champ-lib/champ-titan/pom.xml | 2 | ||||
| -rw-r--r-- | champ-lib/pom.xml | 2 | ||||
| -rw-r--r-- | champ-service/pom.xml | 64 | ||||
| -rw-r--r-- | champ-service/src/main/java/org/onap/champ/async/ChampAsyncRequestProcessor.java | 469 | ||||
| -rw-r--r-- | champ-service/src/main/java/org/onap/champ/async/ChampAsyncResponsePublisher.java | 25 | ||||
| -rw-r--r-- | champ-service/src/main/java/org/onap/champ/event/envelope/GraphEventEnvelope.java | 98 | ||||
| -rw-r--r-- | champ-service/src/main/java/org/onap/champ/event/envelope/GraphEventHeader.java | 228 | ||||
| -rw-r--r-- | champ-service/src/test/java/org/onap/champ/event/GraphEventEnvelopeTest.java | 42 | ||||
| -rw-r--r-- | champ-service/src/test/java/org/onap/champ/util/TestUtil.java | 34 | ||||
| -rw-r--r-- | champ-service/src/test/resources/event/event-envelope.json | 19 |
13 files changed, 689 insertions, 299 deletions
@@ -1,5 +1,6 @@ # Application /champ-lib/champ-janus/logs/ +/champ-service/logs/ # IDE - Eclipse .classpath diff --git a/champ-lib/champ-core/pom.xml b/champ-lib/champ-core/pom.xml index 840fc61..dbbb2b7 100644 --- a/champ-lib/champ-core/pom.xml +++ b/champ-lib/champ-core/pom.xml @@ -111,7 +111,7 @@ limitations under the License. <skip>true</skip> </configuration> </plugin> - <!-- + <!-- Uncomment to add a license header to source files <plugin> <groupId>com.mycila</groupId> <artifactId>license-maven-plugin</artifactId> diff --git a/champ-lib/champ-janus/pom.xml b/champ-lib/champ-janus/pom.xml index 1a0475d..cf76ed0 100644 --- a/champ-lib/champ-janus/pom.xml +++ b/champ-lib/champ-janus/pom.xml @@ -180,7 +180,7 @@ limitations under the License. <skip>true</skip> </configuration> </plugin> - <!-- + <!-- Uncomment to add a license header to source files <plugin> <groupId>com.mycila</groupId> <artifactId>license-maven-plugin</artifactId> diff --git a/champ-lib/champ-titan/pom.xml b/champ-lib/champ-titan/pom.xml index ffff015..5bfc860 100644 --- a/champ-lib/champ-titan/pom.xml +++ b/champ-lib/champ-titan/pom.xml @@ -179,7 +179,7 @@ limitations under the License. <skip>true</skip> </configuration> </plugin> - <!-- + <!-- Uncomment to add a license header to source files <plugin> <groupId>com.mycila</groupId> <artifactId>license-maven-plugin</artifactId> diff --git a/champ-lib/pom.xml b/champ-lib/pom.xml index 43fa296..37f8db4 100644 --- a/champ-lib/pom.xml +++ b/champ-lib/pom.xml @@ -278,7 +278,7 @@ limitations under the License. <skip>true</skip> </configuration> </plugin> - <!-- + <!-- Uncomment to add a license header to source files <plugin> <groupId>com.mycila</groupId> <artifactId>license-maven-plugin</artifactId> diff --git a/champ-service/pom.xml b/champ-service/pom.xml index b0dbf46..108e6a2 100644 --- a/champ-service/pom.xml +++ b/champ-service/pom.xml @@ -53,6 +53,7 @@ limitations under the License. <docker.location>${basedir}/target</docker.location> <onap.nexus.url>https://nexus.onap.org</onap.nexus.url> + <version.org.hamcrest.hamcrest-library>1.3</version.org.hamcrest.hamcrest-library> </properties> <dependencies> @@ -136,6 +137,20 @@ limitations under the License. </exclusion> </exclusions> </dependency> + + <dependency> + <groupId>org.hamcrest</groupId> + <artifactId>hamcrest-library</artifactId> + <version>${version.org.hamcrest.hamcrest-library}</version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.skyscreamer</groupId> + <artifactId>jsonassert</artifactId> + <version>1.5.0</version> + <scope>test</scope> + </dependency> </dependencies> <distributionManagement> @@ -197,7 +212,7 @@ limitations under the License. <skip>true</skip> </configuration> </plugin> - <!-- + <!-- Uncomment to add a license header to source files <plugin> <groupId>com.mycila</groupId> <artifactId>license-maven-plugin</artifactId> @@ -251,53 +266,6 @@ limitations under the License. <goal>report</goal> </goals> </execution> - <execution> - <id>default-check</id> - <goals> - <goal>check</goal> - </goals> - <configuration> - <rules> - <!-- implementation is needed only for Maven 2 --> - <rule implementation="org.jacoco.maven.RuleConfiguration"> - <element>BUNDLE</element> - <limits> - <!-- implementation is needed only for Maven 2 --> - <limit implementation="org.jacoco.report.check.Limit"> - <counter>INSTRUCTION</counter> - <value>COVEREDRATIO</value> - <minimum>.15</minimum> - </limit> - <limit implementation="org.jacoco.report.check.Limit"> - <counter>BRANCH</counter> - <value>COVEREDRATIO</value> - <minimum>.12</minimum> - </limit> - <limit implementation="org.jacoco.report.check.Limit"> - <counter>COMPLEXITY</counter> - <value>COVEREDRATIO</value> - <minimum>.15</minimum> - </limit> - <limit implementation="org.jacoco.report.check.Limit"> - <counter>LINE</counter> - <value>COVEREDRATIO</value> - <minimum>.10</minimum> - </limit> - <limit implementation="org.jacoco.report.check.Limit"> - <counter>METHOD</counter> - <value>COVEREDRATIO</value> - <minimum>.17</minimum> - </limit> - <limit implementation="org.jacoco.report.check.Limit"> - <counter>CLASS</counter> - <value>MISSEDCOUNT</value> - <maximum>5</maximum> - </limit> - </limits> - </rule> - </rules> - </configuration> - </execution> </executions> </plugin> </plugins> diff --git a/champ-service/src/main/java/org/onap/champ/async/ChampAsyncRequestProcessor.java b/champ-service/src/main/java/org/onap/champ/async/ChampAsyncRequestProcessor.java index 610cac9..334871e 100644 --- a/champ-service/src/main/java/org/onap/champ/async/ChampAsyncRequestProcessor.java +++ b/champ-service/src/main/java/org/onap/champ/async/ChampAsyncRequestProcessor.java @@ -26,16 +26,17 @@ import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executors; import java.util.concurrent.ThreadPoolExecutor; - import javax.naming.OperationNotSupportedException; import javax.ws.rs.core.Response.Status; - import org.onap.aai.champcore.ChampTransaction; import org.onap.aai.cl.api.Logger; import org.onap.aai.cl.eelf.LoggerFactory; +import org.onap.aai.event.api.EventConsumer; import org.onap.champ.ChampRESTAPI; import org.onap.champ.event.GraphEvent; import org.onap.champ.event.GraphEvent.GraphEventResult; +import org.onap.champ.event.envelope.GraphEventEnvelope; +import org.onap.champ.event.envelope.GraphEventHeader; import org.onap.champ.event.GraphEventEdge; import org.onap.champ.event.GraphEventVertex; import org.onap.champ.exception.ChampServiceException; @@ -43,281 +44,283 @@ import org.onap.champ.service.ChampDataService; import org.onap.champ.service.ChampThreadFactory; import org.onap.champ.service.logging.ChampMsgs; -import org.onap.aai.event.api.EventConsumer; - /** - * This Class polls the Graph events from request topic perform the necessary - * CRUD operation by calling champDAO and queues up the response to be consumed - * by response handler. + * This Class polls the Graph events from request topic perform the necessary CRUD operation by calling champDAO and + * queues up the response to be consumed by response handler. */ public class ChampAsyncRequestProcessor extends TimerTask { - private Logger logger = LoggerFactory.getInstance().getLogger(ChampAsyncRequestProcessor.class); + private Logger logger = LoggerFactory.getInstance().getLogger(ChampAsyncRequestProcessor.class); - private ChampDataService champDataService; + private ChampDataService champDataService; - /** - * Number of events that can be queued up. - */ - private Integer requestProcesserQueueSize; + /** + * Number of events that can be queued up. + */ + private Integer requestProcesserQueueSize; - /** - * Number of event publisher worker threads. - */ - private Integer requestProcesserPoolSize; - - /** - * Number of event publisher worker threads. - */ - private Integer requestPollingTimeSeconds; + /** + * Number of event publisher worker threads. + */ + private Integer requestProcesserPoolSize; - /** - * Internal queue where outgoing events will be buffered until they can be - * serviced by. - **/ - private BlockingQueue<GraphEvent> requestProcesserEventQueue; + /** + * Number of event publisher worker threads. + */ + private Integer requestPollingTimeSeconds; - /** - * Pool of worker threads that do the work of publishing the events to the - * event bus. - */ - private ThreadPoolExecutor requestProcesserPool; + /** + * Internal queue where outgoing events will be buffered until they can be serviced by. + **/ + private BlockingQueue<GraphEventEnvelope> requestProcesserEventQueue; - private ChampAsyncResponsePublisher champAsyncResponsePublisher; + /** + * Pool of worker threads that do the work of publishing the events to the event bus. + */ + private ThreadPoolExecutor requestProcesserPool; - private EventConsumer asyncRequestConsumer; + private ChampAsyncResponsePublisher champAsyncResponsePublisher; - private static final Integer DEFAULT_ASYNC_REQUEST_PROCESS_QUEUE_CAPACITY = 10000; + private EventConsumer asyncRequestConsumer; - private static final Integer DEFAULT_ASYNC_REQUEST_PROCESS_THREAD_POOL_SIZE = 10; - private static final Integer DEFAULT_ASYNC_REQUEST_PROCESS_POLLING_SECOND = 30000; - private static final String CHAMP_GRAPH_REQUEST_PROCESS_THREAD_NAME = "ChampAsyncGraphRequestEventProcessor"; - Logger auditLogger = LoggerFactory.getInstance().getAuditLogger(ChampRESTAPI.class.getName()); + private static final Integer DEFAULT_ASYNC_REQUEST_PROCESS_QUEUE_CAPACITY = 10000; - public ChampAsyncRequestProcessor(ChampDataService champDataService, - ChampAsyncResponsePublisher champAsyncResponsePublisher, EventConsumer asyncRequestConsumer) { + private static final Integer DEFAULT_ASYNC_REQUEST_PROCESS_THREAD_POOL_SIZE = 10; + private static final Integer DEFAULT_ASYNC_REQUEST_PROCESS_POLLING_SECOND = 30000; + private static final String CHAMP_GRAPH_REQUEST_PROCESS_THREAD_NAME = "ChampAsyncGraphRequestEventProcessor"; + Logger auditLogger = LoggerFactory.getInstance().getAuditLogger(ChampRESTAPI.class.getName()); - this.requestProcesserQueueSize = DEFAULT_ASYNC_REQUEST_PROCESS_QUEUE_CAPACITY; + public ChampAsyncRequestProcessor(ChampDataService champDataService, + ChampAsyncResponsePublisher champAsyncResponsePublisher, EventConsumer asyncRequestConsumer) { - this.requestProcesserPoolSize = DEFAULT_ASYNC_REQUEST_PROCESS_THREAD_POOL_SIZE; + this.requestProcesserQueueSize = DEFAULT_ASYNC_REQUEST_PROCESS_QUEUE_CAPACITY; - this.requestPollingTimeSeconds = DEFAULT_ASYNC_REQUEST_PROCESS_POLLING_SECOND; - requestProcesserEventQueue = new ArrayBlockingQueue<GraphEvent>(requestProcesserQueueSize); - requestProcesserPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(requestProcesserPoolSize, - new ChampThreadFactory(CHAMP_GRAPH_REQUEST_PROCESS_THREAD_NAME)); + this.requestProcesserPoolSize = DEFAULT_ASYNC_REQUEST_PROCESS_THREAD_POOL_SIZE; - for (int i = 0; i < requestProcesserPoolSize; i++) { - requestProcesserPool.submit(new ChampProcessorWorker()); + this.requestPollingTimeSeconds = DEFAULT_ASYNC_REQUEST_PROCESS_POLLING_SECOND; + requestProcesserEventQueue = new ArrayBlockingQueue<>(requestProcesserQueueSize); + requestProcesserPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(requestProcesserPoolSize, + new ChampThreadFactory(CHAMP_GRAPH_REQUEST_PROCESS_THREAD_NAME)); + + for (int i = 0; i < requestProcesserPoolSize; i++) { + requestProcesserPool.submit(new ChampProcessorWorker()); + } + + this.champDataService = champDataService; + this.champAsyncResponsePublisher = champAsyncResponsePublisher; + this.asyncRequestConsumer = asyncRequestConsumer; + logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, + "ChampAsyncRequestProcessor initialized SUCCESSFULLY! with event consumer " + + asyncRequestConsumer.getClass().getName()); } - this.champDataService = champDataService; - this.champAsyncResponsePublisher = champAsyncResponsePublisher; - this.asyncRequestConsumer = asyncRequestConsumer; - logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, - "ChampAsyncRequestProcessor initialized SUCCESSFULLY! with event consumer " - + asyncRequestConsumer.getClass().getName()); - } - - + public ChampAsyncRequestProcessor(ChampDataService champDataService, + ChampAsyncResponsePublisher champAsyncResponsePublisher, EventConsumer asyncRequestConsumer, + Integer requestProcesserQueueSize, Integer requestProcesserPoolSize, Integer requestPollingTimeSeconds) { + + this.requestProcesserQueueSize = requestProcesserQueueSize; - public ChampAsyncRequestProcessor(ChampDataService champDataService, - ChampAsyncResponsePublisher champAsyncResponsePublisher, EventConsumer asyncRequestConsumer, - Integer requestProcesserQueueSize, Integer requestProcesserPoolSize, Integer requestPollingTimeSeconds) { + this.requestProcesserPoolSize = requestProcesserPoolSize; - this.requestProcesserQueueSize = requestProcesserQueueSize; + this.requestPollingTimeSeconds = requestPollingTimeSeconds; - this.requestProcesserPoolSize = requestProcesserPoolSize; - - this.requestPollingTimeSeconds = requestPollingTimeSeconds; + requestProcesserEventQueue = new ArrayBlockingQueue<>(requestProcesserQueueSize); + requestProcesserPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(requestProcesserPoolSize, + new ChampThreadFactory(CHAMP_GRAPH_REQUEST_PROCESS_THREAD_NAME)); - requestProcesserEventQueue = new ArrayBlockingQueue<GraphEvent>(requestProcesserQueueSize); - requestProcesserPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(requestProcesserPoolSize, - new ChampThreadFactory(CHAMP_GRAPH_REQUEST_PROCESS_THREAD_NAME)); + for (int i = 0; i < requestProcesserPoolSize; i++) { + requestProcesserPool.submit(new ChampProcessorWorker()); + } - for (int i = 0; i < requestProcesserPoolSize; i++) { - requestProcesserPool.submit(new ChampProcessorWorker()); + this.champDataService = champDataService; + this.champAsyncResponsePublisher = champAsyncResponsePublisher; + this.asyncRequestConsumer = asyncRequestConsumer; + logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, + "ChampAsyncRequestProcessor initialized SUCCESSFULLY! with event consumer " + + asyncRequestConsumer.getClass().getName()); } - this.champDataService = champDataService; - this.champAsyncResponsePublisher = champAsyncResponsePublisher; - this.asyncRequestConsumer = asyncRequestConsumer; - logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, - "ChampAsyncRequestProcessor initialized SUCCESSFULLY! with event consumer " - + asyncRequestConsumer.getClass().getName()); - } + private class ChampProcessorWorker implements Runnable { + + @Override + public void run() { + + while (true) { + + GraphEventEnvelope eventEnvelope = null; + GraphEvent event = null; + try { + // Get the next event to be published from the queue. + eventEnvelope = requestProcesserEventQueue.take(); + event = eventEnvelope.getBody(); + } catch (InterruptedException e) { + // Restore the interrupted status. + Thread.currentThread().interrupt(); + } + + // Apply Champ Event header + eventEnvelope.setHeader(GraphEventHeader.builder().requestId(event.getTransactionId()).build()); + + // Parse the event and call champ Dao to process , Create the + // response event and put it on response queue + event.setResult(GraphEventResult.SUCCESS); + + // Check if this request is part of an ongoing DB transaction + ChampTransaction transaction = champDataService.getTransaction(event.getDbTransactionId()); + if ((event.getDbTransactionId() != null) && (transaction == null)) { + event.setResult(GraphEventResult.FAILURE); + event.setErrorMessage("Database transactionId " + event.getDbTransactionId() + " not found"); + event.setHttpErrorStatus(Status.BAD_REQUEST); + } + + if (event.getResult() != GraphEventResult.FAILURE) { + try { + if (event.getVertex() != null) { + + switch (event.getOperation()) { + case CREATE: + event.setVertex(GraphEventVertex.fromChampObject( + champDataService.storeObject(event.getVertex().toChampObject(), + Optional.ofNullable(transaction)), + event.getVertex().getModelVersion())); + break; + + case UPDATE: + event.setVertex(GraphEventVertex.fromChampObject( + champDataService.replaceObject(event.getVertex().toChampObject(), + event.getVertex().getId(), Optional.ofNullable(transaction)), + event.getVertex().getModelVersion())); + break; + case DELETE: + champDataService.deleteObject(event.getVertex().getId(), + Optional.ofNullable(transaction)); + break; + default: + // log error + } + } else if (event.getEdge() != null) { + switch (event.getOperation()) { + case CREATE: + event.setEdge(GraphEventEdge.fromChampRelationship( + champDataService.storeRelationship(event.getEdge().toChampRelationship(), + Optional.ofNullable(transaction)), + event.getEdge().getModelVersion())); + break; + + case UPDATE: + event.setEdge(GraphEventEdge.fromChampRelationship( + champDataService.updateRelationship(event.getEdge().toChampRelationship(), + event.getEdge().getId(), Optional.ofNullable(transaction)), + event.getEdge().getModelVersion())); + + break; + case DELETE: + champDataService.deleteRelationship(event.getEdge().getId(), + Optional.ofNullable(transaction)); + break; + default: + logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, + "Invalid operation for event transactionId: " + event.getTransactionId()); + } + + } else { + logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, + "Invalid payload for event transactionId: " + event.getTransactionId()); + } + } catch (ChampServiceException champException) { + logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, champException.getMessage()); + event.setResult(GraphEventResult.FAILURE); + event.setErrorMessage(champException.getMessage()); + event.setHttpErrorStatus(champException.getHttpStatus()); + + } catch (Exception ex) { + logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, ex.getMessage()); + event.setResult(GraphEventResult.FAILURE); + event.setErrorMessage(ex.getMessage()); + event.setHttpErrorStatus(Status.INTERNAL_SERVER_ERROR); + } + } + + if (event.getResult().equals(GraphEventResult.SUCCESS)) { + logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, + "Event processed of type: " + event.getObjectType() + " with key: " + event.getObjectKey() + + " , transaction-id: " + event.getTransactionId() + " , operation: " + + event.getOperation().toString() + " , result: " + event.getResult()); + } else { + logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, + "Event processed of type: " + event.getObjectType() + " with key: " + event.getObjectKey() + + " , transaction-id: " + event.getTransactionId() + " , operation: " + + event.getOperation().toString() + " , result: " + event.getResult() + " , error: " + + event.getErrorMessage()); + } + + champAsyncResponsePublisher.publishResponseEvent(eventEnvelope); - private class ChampProcessorWorker implements Runnable { + } + } + } @Override public void run() { - while (true) { + logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, "Listening for graph events"); - GraphEvent event = null; - try { - // Get the next event to be published from the queue. - event = requestProcesserEventQueue.take(); - } catch (InterruptedException e) { - // Restore the interrupted status. - Thread.currentThread().interrupt(); - } - - // Parse the event and call champ Dao to process , Create the - // response event and put it on response queue - event.setResult(GraphEventResult.SUCCESS); - - // Check if this request is part of an ongoing DB transaction - ChampTransaction transaction = champDataService.getTransaction(event.getDbTransactionId()); - if ( (event.getDbTransactionId() != null) && (transaction == null) ) { - event.setResult(GraphEventResult.FAILURE); - event.setErrorMessage("Database transactionId " + event.getDbTransactionId() + " not found"); - event.setHttpErrorStatus(Status.BAD_REQUEST); - } - - if (event.getResult() != GraphEventResult.FAILURE) { - try { - if (event.getVertex() != null) { - - switch (event.getOperation()) { - case CREATE: - event.setVertex(GraphEventVertex.fromChampObject( - champDataService.storeObject(event.getVertex().toChampObject(), Optional.ofNullable(transaction)), - event.getVertex().getModelVersion())); - break; - - case UPDATE: - event.setVertex(GraphEventVertex.fromChampObject( - champDataService.replaceObject(event.getVertex().toChampObject(), event.getVertex().getId(), Optional.ofNullable(transaction)), - event.getVertex().getModelVersion())); - break; - case DELETE: - champDataService.deleteObject(event.getVertex().getId(), Optional.ofNullable(transaction)); - break; - default: - // log error - } - } else if (event.getEdge() != null) { - switch (event.getOperation()) { - case CREATE: - event.setEdge(GraphEventEdge.fromChampRelationship( - champDataService.storeRelationship(event.getEdge().toChampRelationship(), Optional.ofNullable(transaction)), - event.getEdge().getModelVersion())); - break; - - case UPDATE: - event.setEdge(GraphEventEdge.fromChampRelationship(champDataService - .updateRelationship(event.getEdge().toChampRelationship(), event.getEdge().getId(), Optional.ofNullable(transaction)), - event.getEdge().getModelVersion())); - - break; - case DELETE: - champDataService.deleteRelationship(event.getEdge().getId(), Optional.ofNullable(transaction)); - break; - default: - logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, - "Invalid operation for event transactionId: " + event.getTransactionId()); - } - - } else { - logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, - "Invalid payload for event transactionId: " + event.getTransactionId()); - } - } catch (ChampServiceException champException) { - logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, champException.getMessage()); - event.setResult(GraphEventResult.FAILURE); - event.setErrorMessage(champException.getMessage()); - event.setHttpErrorStatus(champException.getHttpStatus()); - - } catch (Exception ex) { - logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, ex.getMessage()); - event.setResult(GraphEventResult.FAILURE); - event.setErrorMessage(ex.getMessage()); - event.setHttpErrorStatus(Status.INTERNAL_SERVER_ERROR); - } + if (asyncRequestConsumer == null) { + logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, + "Unable to initialize ChampAsyncRequestProcessor"); } - if (event.getResult().equals(GraphEventResult.SUCCESS)) { - logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, - "Event processed of type: " + event.getObjectType() + " with key: " + event.getObjectKey() - + " , transaction-id: " + event.getTransactionId() + " , operation: " - + event.getOperation().toString() + " , result: " + event.getResult()); - } else { - logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, - "Event processed of type: " + event.getObjectType() + " with key: " + event.getObjectKey() - + " , transaction-id: " + event.getTransactionId() + " , operation: " - + event.getOperation().toString() + " , result: " + event.getResult() + " , error: " - + event.getErrorMessage()); + Iterable<String> events = null; + try { + events = asyncRequestConsumer.consume(); + } catch (Exception e) { + logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, e.getMessage()); + return; } - champAsyncResponsePublisher.publishResponseEvent(event); - - } - } - } - - @Override - public void run() { - - logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, "Listening for graph events"); + if (events == null || !events.iterator().hasNext()) { + logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, "No events recieved"); - if (asyncRequestConsumer == null) { - logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, "Unable to initialize ChampAsyncRequestProcessor"); - } - - Iterable<String> events = null; - try { - events = asyncRequestConsumer.consume(); - } catch (Exception e) { - logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, e.getMessage()); - return; - } - - if (events == null || !events.iterator().hasNext()) { - logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, "No events recieved"); + } - } + for (String event : events) { + try { + GraphEventEnvelope requestEnvelope = GraphEventEnvelope.fromJson(event); + GraphEvent requestEvent = requestEnvelope.getBody(); + auditLogger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, + "Event received of type: " + requestEvent.getObjectType() + " with key: " + + requestEvent.getObjectKey() + " , transaction-id: " + requestEvent.getTransactionId() + + " , operation: " + requestEvent.getOperation().toString()); + logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, + "Event received of type: " + requestEvent.getObjectType() + " with key: " + + requestEvent.getObjectKey() + " , transaction-id: " + requestEvent.getTransactionId() + + " , operation: " + requestEvent.getOperation().toString()); + logger.debug(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, "Event received with payload:" + event); + + // Try to submit the event to be published to the event bus. + if (!requestProcesserEventQueue.offer(requestEnvelope)) { + logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, + "Event could not be published to the event bus due to: Internal buffer capacity exceeded."); + } + + } catch (Exception e) { + logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, e.getMessage()); + } + } - for (String event : events) { - try { - GraphEvent requestEvent = GraphEvent.fromJson(event); - auditLogger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, - "Event received of type: " + requestEvent.getObjectType() + " with key: " + requestEvent.getObjectKey() - + " , transaction-id: " + requestEvent.getTransactionId() + " , operation: " - + requestEvent.getOperation().toString()); - logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, - "Event received of type: " + requestEvent.getObjectType() + " with key: " + requestEvent.getObjectKey() - + " , transaction-id: " + requestEvent.getTransactionId() + " , operation: " - + requestEvent.getOperation().toString()); - logger.debug(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, "Event received with payload:" + event); - - // Try to submit the event to be published to the event bus. - if (!requestProcesserEventQueue.offer(requestEvent)) { - logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, - "Event could not be published to the event bus due to: Internal buffer capacity exceeded."); + try { + asyncRequestConsumer.commitOffsets(); + } catch (OperationNotSupportedException e) { + // Dmaap doesnt support commit with offset + logger.debug(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_WARN, e.getMessage()); + } catch (Exception e) { + logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_WARN, e.getMessage()); } - } catch (Exception e) { - logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, e.getMessage()); - } } - try { - asyncRequestConsumer.commitOffsets(); - } catch(OperationNotSupportedException e) { - //Dmaap doesnt support commit with offset - logger.debug(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_WARN, e.getMessage()); - } - catch (Exception e) { - logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_WARN, e.getMessage()); + public Integer getRequestPollingTimeSeconds() { + return requestPollingTimeSeconds; } - } - - - - public Integer getRequestPollingTimeSeconds() { - return requestPollingTimeSeconds; - } - - } diff --git a/champ-service/src/main/java/org/onap/champ/async/ChampAsyncResponsePublisher.java b/champ-service/src/main/java/org/onap/champ/async/ChampAsyncResponsePublisher.java index a9560b0..c3f3859 100644 --- a/champ-service/src/main/java/org/onap/champ/async/ChampAsyncResponsePublisher.java +++ b/champ-service/src/main/java/org/onap/champ/async/ChampAsyncResponsePublisher.java @@ -27,13 +27,13 @@ import java.util.concurrent.ThreadPoolExecutor; import org.onap.aai.cl.api.Logger; import org.onap.aai.cl.eelf.LoggerFactory; +import org.onap.aai.event.api.EventPublisher; import org.onap.champ.event.GraphEvent; import org.onap.champ.event.GraphEvent.GraphEventResult; +import org.onap.champ.event.envelope.GraphEventEnvelope; import org.onap.champ.service.ChampThreadFactory; import org.onap.champ.service.logging.ChampMsgs; -import org.onap.aai.event.api.EventPublisher; - public class ChampAsyncResponsePublisher { private EventPublisher asyncResponsePublisher; @@ -51,7 +51,7 @@ public class ChampAsyncResponsePublisher { /** * Internal queue where outgoing events will be buffered. **/ - private BlockingQueue<GraphEvent> responsePublisherEventQueue; + private BlockingQueue<GraphEventEnvelope> responsePublisherEventQueue; /** * Pool of worker threads that do the work of publishing the events to the @@ -72,7 +72,7 @@ public class ChampAsyncResponsePublisher { this.responsePublisherPoolSize = responsePublisherPoolSize; - responsePublisherEventQueue = new ArrayBlockingQueue<GraphEvent>(responsePublisherQueueSize); + responsePublisherEventQueue = new ArrayBlockingQueue<GraphEventEnvelope>(responsePublisherQueueSize); responsePublisherPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(responsePublisherPoolSize, new ChampThreadFactory(CHAMP_GRAPH_RESPONSE_PUBLISH_THREAD_NAME)); @@ -91,7 +91,7 @@ public class ChampAsyncResponsePublisher { responsePublisherPoolSize = DEFAULT_ASYNC_RESPONSE_PUBLISH_THREAD_POOL_SIZE; - responsePublisherEventQueue = new ArrayBlockingQueue<GraphEvent>(responsePublisherQueueSize); + responsePublisherEventQueue = new ArrayBlockingQueue<GraphEventEnvelope>(responsePublisherQueueSize); responsePublisherPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(responsePublisherPoolSize, new ChampThreadFactory(CHAMP_GRAPH_RESPONSE_PUBLISH_THREAD_NAME)); @@ -105,8 +105,8 @@ public class ChampAsyncResponsePublisher { + asyncResponsePublisher.getClass().getName()); } - public void publishResponseEvent(GraphEvent event) { - responsePublisherEventQueue.offer(event); + public void publishResponseEvent(GraphEventEnvelope eventEnvelope) { + responsePublisherEventQueue.offer(eventEnvelope); } @@ -116,23 +116,20 @@ public class ChampAsyncResponsePublisher { public void run() { while (true) { - + GraphEventEnvelope eventEnvelope = null; GraphEvent event = null; try { - // Get the next event to be published from the queue. - event = responsePublisherEventQueue.take(); - + eventEnvelope = responsePublisherEventQueue.take(); + event = eventEnvelope.getBody(); } catch (InterruptedException e) { - // Restore the interrupted status. Thread.currentThread().interrupt(); } // Publish the response - try { event.setTimestamp(System.currentTimeMillis()); - asyncResponsePublisher.sendSync(event.toJson()); + asyncResponsePublisher.sendSync(eventEnvelope.toJson()); if (event.getResult().equals(GraphEventResult.SUCCESS)) { logger.info(ChampMsgs.CHAMP_ASYNC_RESPONSE_PUBLISHER_INFO, "Response published for Event of type: " + event.getObjectType() + " with key: " + event.getObjectKey() diff --git a/champ-service/src/main/java/org/onap/champ/event/envelope/GraphEventEnvelope.java b/champ-service/src/main/java/org/onap/champ/event/envelope/GraphEventEnvelope.java new file mode 100644 index 0000000..7958a3a --- /dev/null +++ b/champ-service/src/main/java/org/onap/champ/event/envelope/GraphEventEnvelope.java @@ -0,0 +1,98 @@ +/** + * ============LICENSE_START========================================== + * org.onap.aai + * =================================================================== + * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * Copyright © 2017 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============================================ + * ECOMP is a trademark and service mark of AT&T Intellectual Property. + */ +package org.onap.champ.event.envelope; + +import javax.ws.rs.core.Response.Status; +import org.onap.champ.event.GraphEvent; +import org.onap.champ.exception.ChampServiceException; +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; + +public class GraphEventEnvelope { + + private GraphEventHeader header; + private GraphEvent body; + + /** + * Serializer/deserializer for converting to/from JSON. + */ + private static final Gson gson = new GsonBuilder().disableHtmlEscaping().setPrettyPrinting().create(); + + public GraphEventEnvelope(GraphEvent event) { + this.header = new GraphEventHeader.Builder().requestId(event.getTransactionId()).build(); + this.body = event; + } + + public GraphEventEnvelope(GraphEventHeader header, GraphEvent body) { + this.header = header; + this.body = body; + } + + public GraphEventHeader getHeader() { + return header; + } + + public void setHeader(GraphEventHeader header) { + this.header = header; + } + + public GraphEvent getBody() { + return body; + } + + public void setBody(GraphEvent body) { + this.body = body; + } + + /** + * Serializes this Vertex object into a JSON string. + * + * @return - A JSON format string representation of this Vertex. + */ + public String toJson() { + return gson.toJson(this); + } + + /** + * Deserializes the provided JSON string into a Event Envelope object. + * + * @param json the JSON string to produce the Event Envelope from. + * @return an Event Envelope object. + * @throws ChampServiceException + */ + public static GraphEventEnvelope fromJson(String json) throws ChampServiceException { + try { + if (json == null || json.isEmpty()) { + throw new ChampServiceException("Empty or null JSON string.", Status.BAD_REQUEST); + } + return gson.fromJson(json, GraphEventEnvelope.class); + } catch (Exception ex) { + throw new ChampServiceException("Unable to parse JSON string: ", Status.BAD_REQUEST); + } + } + + @Override + public String toString() { + return toJson(); + } + +} diff --git a/champ-service/src/main/java/org/onap/champ/event/envelope/GraphEventHeader.java b/champ-service/src/main/java/org/onap/champ/event/envelope/GraphEventHeader.java new file mode 100644 index 0000000..59e01ea --- /dev/null +++ b/champ-service/src/main/java/org/onap/champ/event/envelope/GraphEventHeader.java @@ -0,0 +1,228 @@ +/** + * ============LICENSE_START========================================== + * org.onap.aai + * =================================================================== + * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * Copyright © 2017 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============================================ + * ECOMP is a trademark and service mark of AT&T Intellectual Property. + */ +package org.onap.champ.event.envelope; + +import java.time.Instant; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.util.Objects; +import java.util.UUID; +import org.apache.commons.lang3.builder.EqualsBuilder; +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; +import com.google.gson.annotations.SerializedName; + +public class GraphEventHeader { + + private static final String SOURCE_NAME = "CHAMP"; + + private static final String EVENT_TYPE = "db-update-result"; + + @SerializedName("request-id") + private String requestId; + + private String timestamp; + + @SerializedName("source-name") + private String sourceName; + + @SerializedName("event-type") + private String eventType; + + @SerializedName("validation-entity-type") + private String validationEntityType; + + @SerializedName("validation-top-entity-type") + private String validationTopEntityType; + + @SerializedName("entity-link") + private String entityLink; + + private static final Gson gson = new GsonBuilder().disableHtmlEscaping().setPrettyPrinting().create(); + + public static Builder builder() { + return new Builder(); + } + + public static class Builder { + + private String requestId; + private String validationEntityType; + private String validationTopEntityType; + private String entityLink; + + public Builder requestId(String val) { + requestId = val; + return this; + } + + public Builder validationEntityType(String val) { + validationEntityType = val; + return this; + } + + public Builder validationTopEntityType(String val) { + validationTopEntityType = val; + return this; + } + + public Builder entityLink(String val) { + entityLink = val; + return this; + } + + public GraphEventHeader build() { + return new GraphEventHeader(this); + } + } + + private GraphEventHeader(Builder builder) { + requestId = builder.requestId != null ? builder.requestId : UUID.randomUUID().toString(); + timestamp = DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmssX").withZone(ZoneOffset.UTC).format(Instant.now()); + sourceName = SOURCE_NAME; + eventType = EVENT_TYPE; + + validationEntityType = builder.validationEntityType; + validationTopEntityType = builder.validationTopEntityType; + entityLink = builder.entityLink; + } + + /** + * Serializes this object into a JSON string representation. + * + * @return a JSON format string representation of this object. + */ + public String toJson() { + return gson.toJson(this); + } + + /////////////////////////////////////////////////////////////////////////// + // GETTERS AND SETTERS + /////////////////////////////////////////////////////////////////////////// + + public String getRequestId() { + return requestId; + } + + public void setRequestId(String requestId) { + this.requestId = requestId; + } + + public String getTimestamp() { + return timestamp; + } + + public void setTimestamp(String timestamp) { + this.timestamp = timestamp; + } + + public String getSourceName() { + return sourceName; + } + + public void setSourceName(String sourceName) { + this.sourceName = sourceName; + } + + public String getEventType() { + return eventType; + } + + public void setEventType(String eventType) { + this.eventType = eventType; + } + + public String getValidationEntityType() { + return validationEntityType; + } + + public void setValidationEntityType(String validationEntityType) { + this.validationEntityType = validationEntityType; + } + + public String getValidationTopEntityType() { + return validationTopEntityType; + } + + public void setValidationTopEntityType(String validationTopEntityType) { + this.validationTopEntityType = validationTopEntityType; + } + + public String getEntityLink() { + return entityLink; + } + + public void setEntityLink(String entityLink) { + this.entityLink = entityLink; + } + + /////////////////////////////////////////////////////////////////////////// + // OVERRIDES + /////////////////////////////////////////////////////////////////////////// + + /* + * (non-Javadoc) + * + * @see java.lang.Object#hashCode() + */ + @Override + public int hashCode() { + return Objects.hash(this.requestId, this.timestamp, this.sourceName, this.eventType, this.validationEntityType, + this.validationTopEntityType, this.entityLink); + } + + /* + * (non-Javadoc) + * + * @see java.lang.Object#equals(java.lang.Object) + */ + @Override + public boolean equals(Object obj) { + if (!(obj instanceof GraphEventHeader)) { + return false; + } else if (obj == this) { + return true; + } + GraphEventHeader rhs = (GraphEventHeader) obj; + // @formatter:off + return new EqualsBuilder() + .append(requestId, rhs.requestId) + .append(timestamp, rhs.timestamp) + .append(sourceName, rhs.sourceName) + .append(eventType, rhs.sourceName) + .append(validationEntityType, rhs.validationEntityType) + .append(validationTopEntityType, rhs.validationTopEntityType) + .append(entityLink, rhs.entityLink) + .isEquals(); + // @formatter:on + } + + /* + * (non-Javadoc) + * + * @see java.lang.Object#toString() + */ + @Override + public String toString() { + return this.toJson(); + } +} diff --git a/champ-service/src/test/java/org/onap/champ/event/GraphEventEnvelopeTest.java b/champ-service/src/test/java/org/onap/champ/event/GraphEventEnvelopeTest.java new file mode 100644 index 0000000..5c39f99 --- /dev/null +++ b/champ-service/src/test/java/org/onap/champ/event/GraphEventEnvelopeTest.java @@ -0,0 +1,42 @@ +package org.onap.champ.event; + +import static org.hamcrest.Matchers.is; +import static org.junit.Assert.assertThat; +import org.junit.Test; +import org.onap.aai.champcore.model.ChampObject; +import org.onap.champ.event.GraphEvent.GraphEventOperation; +import org.onap.champ.event.envelope.GraphEventEnvelope; +import org.onap.champ.util.TestUtil; +import org.skyscreamer.jsonassert.Customization; +import org.skyscreamer.jsonassert.JSONAssert; +import org.skyscreamer.jsonassert.JSONCompareMode; +import org.skyscreamer.jsonassert.comparator.CustomComparator; + +public class GraphEventEnvelopeTest { + + @Test + public void testEventEnvelopeFormat() throws Exception { + String expectedEnvelope = TestUtil.getFileAsString("event/event-envelope.json"); + + GraphEvent body = GraphEvent.builder(GraphEventOperation.CREATE) + .vertex(GraphEventVertex.fromChampObject(new ChampObject.Builder("pserver").build(), "v13")).build(); + + String graphEventEnvelope = new GraphEventEnvelope(body).toJson(); + + JSONAssert.assertEquals(expectedEnvelope, graphEventEnvelope, + new CustomComparator(JSONCompareMode.STRICT, new Customization("header.request-id", (o1, o2) -> true), + new Customization("header.timestamp", (o1, o2) -> true), + new Customization("body.timestamp", (o1, o2) -> true), + new Customization("body.transaction-id", (o1, o2) -> true))); + } + + @Test + public void testRequestIdIsTransactionId() throws Exception { + GraphEvent body = GraphEvent.builder(GraphEventOperation.CREATE) + .vertex(GraphEventVertex.fromChampObject(new ChampObject.Builder("pserver").build(), "v13")).build(); + + GraphEventEnvelope envelope = new GraphEventEnvelope(body); + + assertThat(envelope.getHeader().getRequestId(), is(envelope.getBody().getTransactionId())); + } +} diff --git a/champ-service/src/test/java/org/onap/champ/util/TestUtil.java b/champ-service/src/test/java/org/onap/champ/util/TestUtil.java new file mode 100644 index 0000000..b9924e4 --- /dev/null +++ b/champ-service/src/test/java/org/onap/champ/util/TestUtil.java @@ -0,0 +1,34 @@ +package org.onap.champ.util; + +import java.io.File; +import java.io.IOException; +import java.net.URISyntaxException; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +public class TestUtil { + + public static Path getPath(String resourceFilename) throws URISyntaxException { + URL resource = ClassLoader.getSystemResource(resourceFilename); + if (resource != null) { + return Paths.get(resource.toURI()); + } + + // If the resource is not found relative to the classpath, try to get it from the file system directly. + File file = new File(resourceFilename); + if (!file.exists()) { + throw new RuntimeException("Resource does not exist: " + resourceFilename); + } + return file.toPath(); + } + + public static String getContentUtf8(Path filePath) throws IOException { + return new String(Files.readAllBytes(filePath)); + } + + public static String getFileAsString(String resourceFilename) throws IOException, URISyntaxException { + return getContentUtf8(getPath(resourceFilename)); + } +}
\ No newline at end of file diff --git a/champ-service/src/test/resources/event/event-envelope.json b/champ-service/src/test/resources/event/event-envelope.json new file mode 100644 index 0000000..68888c0 --- /dev/null +++ b/champ-service/src/test/resources/event/event-envelope.json @@ -0,0 +1,19 @@ +{ + "header": { + "request-id": "2253f351-d9b6-4638-9fe3-2c194bee1b29", + "timestamp": "20180316T092301Z", + "source-name": "CHAMP", + "event-type": "db-update-result" + }, + "body": { + "operation": "CREATE", + "transaction-id": "2253f351-d9b6-4638-9fe3-2c194bee1b29", + "timestamp": 1521198075620, + "vertex": { + "key": "", + "schema-version": "v13", + "type": "pserver", + "properties": {} + } + } +}
\ No newline at end of file |