+ 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);