Improve internal composition handling

Improve aggregation of data from events prior a start event in aggregation.

Author: vasile-baluta

src/main/java/com/ericsson/ei/controller/SubscriptionControllerImpl.java

@@ -58,7 +58,7 @@ public class SubscriptionControllerImpl implements SubscriptionController {
     public ResponseEntity<SubscriptionResponse> createSubscription(@RequestBody List<Subscription> subscriptions) {
         ResponseEntity<SubscriptionResponse> subResponse = null;
         SubscriptionResponse subscriptionResponse = new SubscriptionResponse();
-        for (Subscription subscription :  subscriptions){
+        for (Subscription subscription : subscriptions) {
             subResponse = null;
             try {
                 subscription.setCreated(Instant.now().toEpochMilli());
@@ -69,7 +69,8 @@ public ResponseEntity<SubscriptionResponse> createSubscription(@RequestBody List
                 LOG.error(msg);
                 subscriptionResponse.setMsg(msg);
                 subscriptionResponse.setStatusCode(HttpStatus.PRECONDITION_FAILED.value());
-                subResponse = new ResponseEntity<SubscriptionResponse>(subscriptionResponse, HttpStatus.PRECONDITION_FAILED);
+                subResponse = new ResponseEntity<SubscriptionResponse>(subscriptionResponse,
+                        HttpStatus.PRECONDITION_FAILED);
             }
 
             if (!subscriptionService.doSubscriptionExist(subscription.getSubscriptionName())) {

src/main/java/com/ericsson/ei/handlers/HistoryExtractionHandler.java

@@ -14,17 +14,16 @@
 
 package com.ericsson.ei.handlers;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-
 import com.ericsson.ei.jmespath.JmesPathInterface;
 import com.ericsson.ei.jsonmerge.MergeHandler;
 import com.ericsson.ei.jsonmerge.MergePrepare;
 import com.ericsson.ei.rules.RulesObject;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.github.wnameless.json.flattener.JsonFlattener;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 // TODO: Auto-generated Javadoc
@@ -67,11 +66,23 @@ public String runHistoryExtraction(String aggregatedObjectId, RulesObject rules,
         String ruleString = ruleJson.toString();
 
         // if we need to add append to an array then array_path will not be
-        // empty so we use it instead passed pathInAggregatedObject
+        // empty so we use it instead of passed pathInAggregatedObject
         String aggregatedObject = mergeHandler.getAggregatedObject(aggregatedObjectId, false);
-        String array_path = getPathFromExtractedContent(aggregatedObject, ruleString);
+        String objAtPathStr = "";
+        String pathTrimmed = mergePrepare.trimLastInPath(pathInAggregatedObject, ".");
+        try {
+            pathTrimmed = mergePrepare.makeJmespathArrayIndexes(pathTrimmed);
+            JsonNode objAtPath = jmesPathInterface.runRuleOnEvent(pathTrimmed, aggregatedObject);
+            objAtPathStr = objAtPath.toString();
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+        }
+        String array_path = getPathFromExtractedContent(objAtPathStr, ruleString);
+
         if (!array_path.isEmpty()) {
-            pathInAggregatedObject = array_path;
+            // pathInAggregatedObject = array_path;
+            pathInAggregatedObject = pathTrimmed + "." + array_path;
+            pathInAggregatedObject = MergePrepare.destringify(pathInAggregatedObject);
         } else {
             String ruleKey = getRulePath(ruleString);
             if (pathInAggregatedObject.isEmpty()) {

src/main/java/com/ericsson/ei/handlers/ObjectHandler.java

@@ -43,11 +43,13 @@ public class ObjectHandler {
 
     static Logger log = (Logger) LoggerFactory.getLogger(ObjectHandler.class);
 
-    @Getter @Setter
+    @Getter
+    @Setter
     @Value("${aggregated.collection.name}")
     private String collectionName;
 
-    @Getter @Setter
+    @Getter
+    @Setter
     @Value("${database.name}")
     private String databaseName;
 
@@ -66,7 +68,7 @@ public class ObjectHandler {
     @Setter
     @Autowired
     private SubscriptionHandler subscriptionHandler;
-    
+
     @Getter
     @Value("${aggregated.collection.ttlValue}")
     private int ttlValue;
@@ -81,12 +83,11 @@ public boolean insertObject(String aggregatedObject, RulesObject rulesObject, St
         log.debug("ObjectHandler: Aggregated Object document to be inserted: " + document.toString());
 
         mongoDbHandler.createTTLIndex(databaseName, collectionName, "Time", ttlValue);
- 
 
         boolean result = mongoDbHandler.insertDocument(databaseName, collectionName, document.toString());
         if (result)
             eventToObjectMap.updateEventToObjectMapInMemoryDB(rulesObject, event, id);
-            subscriptionHandler.checkSubscriptionForObject(aggregatedObject);
+        subscriptionHandler.checkSubscriptionForObject(aggregatedObject);
         return result;
     }
 
@@ -95,12 +96,18 @@ public boolean insertObject(JsonNode aggregatedObject, RulesObject rulesObject,
     }
 
     /**
-     * This method uses previously locked in database aggregatedObject (lock was set in lockDocument method)
-     * and modifies this document with the new values and removes the lock in one query
-     * @param aggregatedObject String to insert in database
-     * @param rulesObject used for fetching id
-     * @param event String to fetch id if it was not specified
-     * @param id String
+     * This method uses previously locked in database aggregatedObject (lock was
+     * set in lockDocument method) and modifies this document with the new
+     * values and removes the lock in one query
+     * 
+     * @param aggregatedObject
+     *            String to insert in database
+     * @param rulesObject
+     *            used for fetching id
+     * @param event
+     *            String to fetch id if it was not specified
+     * @param id
+     *            String
      * @return true if operation succeed
      */
     public boolean updateObject(String aggregatedObject, RulesObject rulesObject, String event, String id) {
@@ -109,15 +116,15 @@ public boolean updateObject(String aggregatedObject, RulesObject rulesObject, St
             JsonNode idNode = jmespathInterface.runRuleOnEvent(idRules, event);
             id = idNode.textValue();
         }
-        log.debug("ObjectHandler: Updating Aggregated Object:\n" + aggregatedObject +
-        		"\nEvent:\n" + event);
+        log.debug("ObjectHandler: Updating Aggregated Object:\n" + aggregatedObject + "\nEvent:\n" + event);
         JsonNode document = prepareDocumentForInsertion(id, aggregatedObject);
         String condition = "{\"_id\" : \"" + id + "\"}";
         String documentStr = document.toString();
         boolean result = mongoDbHandler.updateDocument(databaseName, collectionName, condition, documentStr);
-        if (result)
+        if (result) {
             eventToObjectMap.updateEventToObjectMapInMemoryDB(rulesObject, event, id);
             subscriptionHandler.checkSubscriptionForObject(aggregatedObject);
+        }
         return result;
     }
 
@@ -132,9 +139,6 @@ public List<String> findObjectsByCondition(String condition) {
     public String findObjectById(String id) {
         String condition = "{\"_id\" : \"" + id + "\"}";
         String document = findObjectsByCondition(condition).get(0);
-//        JsonNode result = getAggregatedObject(document);
-//        if (result != null)
-//            return result.asText();
         return document;
     }
 
@@ -158,54 +162,59 @@ public JsonNode prepareDocumentForInsertion(String id, String object) {
 
             return jsonNode;
         } catch (Exception e) {
-            log.info(e.getMessage(),e);
+            log.error(e.getMessage(), e);
         }
         return null;
     }
 
     public JsonNode getAggregatedObject(String dbDocument) {
-         ObjectMapper mapper = new ObjectMapper();
-         try {
-             JsonNode documentJson = mapper.readValue(dbDocument, JsonNode.class);
-             JsonNode objectDoc = documentJson.get("aggregatedObject");
-             return objectDoc;
-         } catch (Exception e) {
-             log.info(e.getMessage(),e);
-         }
-         return null;
+        ObjectMapper mapper = new ObjectMapper();
+        try {
+            JsonNode documentJson = mapper.readValue(dbDocument, JsonNode.class);
+            JsonNode objectDoc = documentJson.get("aggregatedObject");
+            return objectDoc;
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+        }
+        return null;
     }
 
     public String extractObjectId(JsonNode aggregatedDbObject) {
         return aggregatedDbObject.get("_id").textValue();
     }
 
     /**
-     * Locks the document in database to achieve pessimistic locking. Method findAndModify is used to optimize
-     * the quantity of requests towards database.
-     * @param id String to search
+     * Locks the document in database to achieve pessimistic locking. Method
+     * findAndModify is used to optimize the quantity of requests towards
+     * database.
+     * 
+     * @param id
+     *            String to search
      * @return String aggregated document
      */
-    public String lockDocument(String id){
+    public String lockDocument(String id) {
         boolean documentLocked = true;
         String conditionId = "{\"_id\" : \"" + id + "\"}";
         String conditionLock = "[ { \"lock\" :  null } , { \"lock\" : \"0\"}]";
         String setLock = "{ \"$set\" : { \"lock\" : \"1\"}}";
         ObjectMapper mapper = new ObjectMapper();
-        while (documentLocked==true){
+        while (documentLocked == true) {
             try {
                 JsonNode documentJson = mapper.readValue(setLock, JsonNode.class);
                 JsonNode queryCondition = mapper.readValue(conditionId, JsonNode.class);
                 ((ObjectNode) queryCondition).set("$or", mapper.readValue(conditionLock, JsonNode.class));
-                Document result = mongoDbHandler.findAndModify(databaseName, collectionName, queryCondition.toString(), documentJson.toString());
+                Document result = mongoDbHandler.findAndModify(databaseName, collectionName, queryCondition.toString(),
+                        documentJson.toString());
                 if (result != null) {
-                    log.info("DB locked by " + Thread.currentThread().getId() + " thread");
+                    log.debug("DB locked by " + Thread.currentThread().getId() + " thread");
                     documentLocked = false;
                     return JSON.serialize(result);
                 }
                 // To Remove
-                log.info("Waiting by " + Thread.currentThread().getId() + " thread");
+                log.debug("Waiting by " + Thread.currentThread().getId() + " thread");
             } catch (Exception e) {
-                log.info(e.getMessage(),e); }
+                log.error(e.getMessage(), e);
+            }
         }
         return null;
     }

src/main/java/com/ericsson/ei/handlers/UpStreamEventsHandler.java

@@ -109,13 +109,14 @@ private void traverseTree(final JsonNode jsonArray, final String aggregatedObjec
         final JsonNode parent = jsonArray.get(0);
         RulesObject rules = rulesHandler.getRulesForEvent(parent.toString());
 
-        final String np = historyExtractionHandler.runHistoryExtraction(aggregatedObjectId, rules, parent.toString(),
+        String np = historyExtractionHandler.runHistoryExtraction(aggregatedObjectId, rules, parent.toString(),
                 pathInAggregatedObject);
         String prevNp = null;
         for (int i = 1; i < jsonArray.size(); i++) {
             if (jsonArray.get(i).isObject()) {
-                rules = rulesHandler.getRulesForEvent(jsonArray.get(i).toString());
-                historyExtractionHandler.runHistoryExtraction(aggregatedObjectId, rules, jsonArray.get(i).toString(),
+                String event = jsonArray.get(i).toString();
+                rules = rulesHandler.getRulesForEvent(event);
+                prevNp = historyExtractionHandler.runHistoryExtraction(aggregatedObjectId, rules, event,
                         pathInAggregatedObject);
             } else {
                 // if we have prevNp then we should use that because it is the