Merge branch 'array_aggregation' into master

vasile-baluta · web-flow · commit 07370f8da5e2 · 2018-06-26T12:52:02.000+02:00
diff --git a/.travis.yml b/.travis.yml
@@ -21,6 +21,6 @@ script:
 #
 # It is also importan that the first test to be run is a Spring Boot Test since spring boot also downloads
 # another version of embedded mongo db so all Spring Boot tests will end up in a race condition also
-  -  mvn -DsomeModule.test.includes="**/QueryServiceTest.java" test
-  -  mvn -DsomeModule.test.excludes="**/QueryServiceTest.java" test
+  -  mvn -DsomeModule.test.includes="**/QueryServiceTest.java" test jacoco:report coveralls:report
+  -  mvn -DsomeModule.test.excludes="**/QueryServiceTest.java" test jacoco:report coveralls:report
  
diff --git a/README.md b/README.md
@@ -15,13 +15,19 @@
    limitations under the License.
 --->
 
+[![Build Status](https://travis-ci.org/Ericsson/eiffel-intelligence.svg?branch=master)](https://travis-ci.org/Ericsson/eiffel-intelligence)
+[![Coverage Status](https://coveralls.io/repos/github/Ericsson/eiffel-intelligence/badge.svg?branch=master)](https://coveralls.io/github/Ericsson/eiffel-intelligence?branch=master)
+[![](https://jitpack.io/v/Ericsson/eiffel-intelligence.svg)](https://jitpack.io/#Ericsson/eiffel-intelligence)
+
 # Eiffel Intelligence
 Eiffel Intelligence is a real time data aggregation and analysis solution for Eiffel events. While Eiffel represents an event based architecture, Eiffel Intelligence addresses the need of stakeholders to view the current state of the system by bridging the divide from immutable events to mutable state representation.
 
 This repository contains a software implementation based on the Eiffel protocol. For more information on Eiffel, its vocabulary, descriptions, guides and schemas, please see [the Eiffel repository](https://github.com/Ericsson/eiffel). For news, discussions and questions, please visit the [Eiffel Community Google group](https://groups.google.com/forum/#!forum/eiffel-community).
 
 Eiffel Intelligence is licensed under the [Apache License 2.0](./LICENSE).
 
+## Detailed documentation is provided at the following link: http://ericsson.github.io/eiffel-intelligence
+
 ## How to Propose Changes
 
 Anyone is welcome to propose changes to this repository by creating a new [Issue](https://github.com/Ericsson/eiffel-intelligence/issues) ticket in GitHub. These requests may concern anything contained in the repo: changes to documentation, changes to interfaces, changes to implementations, additional tests et cetera.
diff --git a/docs/_config.yml b/docs/_config.yml
@@ -0,0 +1 @@
+theme: jekyll-theme-slate
diff --git a/docs/index.md b/docs/index.md
@@ -0,0 +1,12 @@
+
+# _Introduction_
+
+Eiffel Intelligence is a service that aggregates informations from different events of a flow with the purpose of notifying subscribers when content of interest has been collected from desired events. The aggregation is stored in a JSON bject. By flow we mean a chain of Eiffel events that are linked together directly or indirectly. 
+
+Eiffel Intelligence uses a set of rules to define what information will be extracted from an Eiffel event in the flow and at what location to store this information in the aggregated object. Today only one rule set can be run in each instance and the reason is that in an Eiffel domain with million of events flowing we will have multiple extractions and checks for each rule set and multiple rule sets will require large machines and scalability problems.
+
+Eiffel intelligence uses subscriptions to notify interested parties when several evens have occured with desired content in each event aggregated in an aggregated object. Every time an aggregated object is updated we check whether any subscription is fulfilled by the curent state of the aggregated object. 
+
+[Rules](rules.md)
+
+[Subscriptions](subscription.md)
diff --git a/docs/rules.md b/docs/rules.md
@@ -0,0 +1 @@
+# Rules in Eiffel Intelligence
diff --git a/docs/subscriptions.md b/docs/subscriptions.md
@@ -0,0 +1 @@
+# Subscriptions in Eiffel Intelligence
diff --git a/pom.xml b/pom.xml
@@ -98,7 +98,7 @@
             <artifactId>spring-security-test</artifactId>
             <scope>test</scope>
         </dependency>
-
+      
         <dependency>
             <groupId>org.springframework.session</groupId>
             <artifactId>spring-session-data-mongodb</artifactId>
@@ -244,7 +244,7 @@
 
     </dependencies>
     <build>
-        <plugins>
+        <plugins>		          		
             <plugin>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
@@ -343,4 +343,4 @@
 
         </plugins>
     </build>
-</project>
+</project>
diff --git a/src/main/java/com/ericsson/ei/handlers/UpStreamEventsHandler.java b/src/main/java/com/ericsson/ei/handlers/UpStreamEventsHandler.java
@@ -19,6 +19,7 @@
 import com.ericsson.ei.rules.RulesHandler;
 import com.ericsson.ei.rules.RulesObject;
 import com.fasterxml.jackson.databind.JsonNode;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -107,15 +108,21 @@ private void traverseTree(final JsonNode jsonArray, final String aggregatedObjec
             final String pathInAggregatedObject) {
 
         final JsonNode parent = jsonArray.get(0);
-        RulesObject rules = rulesHandler.getRulesForEvent(parent.toString());
-
-        String np = historyExtractionHandler.runHistoryExtraction(aggregatedObjectId, rules, parent.toString(),
-                pathInAggregatedObject);
+        JsonNode parentId = parent.at("/meta/id");
+        String np = pathInAggregatedObject;
+        if (!aggregatedObjectId.equals(parentId.textValue())) {
+            // parent event is not the same as the starting event so we can
+            // start collecting history
+            RulesObject rules = rulesHandler.getRulesForEvent(parent.toString());
+
+            np = historyExtractionHandler.runHistoryExtraction(aggregatedObjectId, rules, parent.toString(),
+                    pathInAggregatedObject);
+        }
         String prevNp = null;
         for (int i = 1; i < jsonArray.size(); i++) {
             if (jsonArray.get(i).isObject()) {
                 String event = jsonArray.get(i).toString();
-                rules = rulesHandler.getRulesForEvent(event);
+                RulesObject rules = rulesHandler.getRulesForEvent(event);
                 prevNp = historyExtractionHandler.runHistoryExtraction(aggregatedObjectId, rules, event,
                         pathInAggregatedObject);
             } else {
diff --git a/src/main/java/com/ericsson/ei/jmespath/JmesPathInterface.java b/src/main/java/com/ericsson/ei/jmespath/JmesPathInterface.java
@@ -16,13 +16,13 @@
  */
 package com.ericsson.ei.jmespath;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
 import io.burt.jmespath.Expression;
 import io.burt.jmespath.JmesPath;
 import io.burt.jmespath.function.FunctionRegistry;
@@ -50,14 +50,16 @@ public JsonNode runRuleOnEvent(String rule, String input) {
         if (input == null) {
             input = "";
         }
-        Expression<JsonNode> expression = jmespath.compile(rule);
+
+        JsonNode result = null;
         ObjectMapper objectMapper = new ObjectMapper();
         try {
+            Expression<JsonNode> expression = jmespath.compile(rule);
             event = objectMapper.readValue(input, JsonNode.class);
+            result = expression.search(event);
         } catch (Exception e) {
             log.info(e.getMessage(), e);
         }
-        JsonNode result = expression.search(event);
 
         return result;
     }
diff --git a/src/main/java/com/ericsson/ei/jsonmerge/MergePrepare.java b/src/main/java/com/ericsson/ei/jsonmerge/MergePrepare.java
@@ -16,22 +16,27 @@
 */
 package com.ericsson.ei.jsonmerge;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.commons.lang3.StringUtils;
-import org.json.*;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
 import com.ericsson.ei.jmespath.JmesPathInterface;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.NullNode;
 import com.github.wnameless.json.flattener.JsonFlattener;
 
-import lombok.Setter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
 
-import java.util.*;
+import lombok.Setter;
 
 @Component
 public class MergePrepare {
@@ -241,10 +246,10 @@ public String getMergePath(String originObject, String mergeRule, boolean skipPa
             if (pos > 0)
                 ruleKey = ruleKey.substring(0, pos);
             try {
-                Object object = objectJSONObject.get(ruleKey);
-                if (object != null)
+                JsonNode jsonResult = jmesPathInterface.runRuleOnEvent(ruleKey, originObject);
+                if (!(jsonResult instanceof NullNode))
                     mergePath = ruleKey;
-            } catch (JSONException e) {
+            } catch (Exception e) {
                 log.error(e.getMessage(), e);
             }
         } else {
@@ -320,13 +325,14 @@ public String getMergePath(String originObject, String mergeRule, boolean skipPa
                     Object value = parsedJson.at(mergePath);
                     if (value instanceof ArrayNode) {
                         int arraySize = ((ArrayNode) value).size();
-                        mergePath += "." + arraySize++ + "." + ruleKeyLast;
-                    } else {
+                        mergePath += "." + arraySize++;
+                    }
+                    if (!StringUtils.isAllBlank(ruleKeyLast)) {
                         mergePath += "." + ruleKeyLast;
                     }
+
                     mergePath = mergePath.replaceFirst("\\/", "");
                     mergePath = mergePath.replaceAll("\\/", "\\.");
-
                 } catch (Exception e) {
                     log.error(e.getMessage(), e);
                 }
diff --git a/src/main/java/com/ericsson/ei/rules/RulesObject.java b/src/main/java/com/ericsson/ei/rules/RulesObject.java
@@ -71,7 +71,7 @@ public String getDownstreamMergeRules() {
     }
 
     public String getHistoryExtractionRules() {
-        return rulesObject.get("HistoryExtractionRules").textValue();
+        return getTextValue("HistoryExtractionRules");
     }
 
     public String getHistoryPathRules() {
@@ -83,18 +83,18 @@ public String fetchProcessRules() {
     }
 
     public String getTextValue(String fieldName) {
-         JsonNode jsonNode = rulesObject.get(fieldName);
-         if (jsonNode != null)
-             return jsonNode.textValue();
-         return null;
+        JsonNode jsonNode = rulesObject.get(fieldName);
+        if (jsonNode != null)
+            return jsonNode.textValue();
+        return null;
     }
 
     public String getString(String fieldName) {
         JsonNode jsonNode = rulesObject.get(fieldName);
         if (jsonNode != null)
             return jsonNode.toString();
         return "";
-   }
+    }
 
     public boolean equals(Object other) {
         if (other instanceof RulesObject) {
diff --git a/src/test/java/com/ericsson/ei/flowtests/ArrayAggregationTest.java b/src/test/java/com/ericsson/ei/flowtests/ArrayAggregationTest.java
@@ -0,0 +1,103 @@
+package com.ericsson.ei.flowtests;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.when;
+
+import com.ericsson.ei.App;
+import com.ericsson.ei.erqueryservice.ERQueryService;
+import com.ericsson.ei.erqueryservice.SearchOption;
+import com.ericsson.ei.handlers.UpStreamEventsHandler;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.io.IOException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.test.context.TestExecutionListeners;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.support.DependencyInjectionTestExecutionListener;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@TestExecutionListeners(listeners = { DependencyInjectionTestExecutionListener.class, ArrayAggregationTest.class })
+@SpringBootTest(classes = App.class)
+public class ArrayAggregationTest extends FlowTestBase {
+
+    private static final String UPSTREAM_RESULT_FILE = "arrayAggregationUpstreamResult.json";
+    private static final String EVENTS_FILE_PATH = "src/test/resources/arrayAggregationEvents.json";
+    private static final String RULES_FILE_PATH = "src/test/resources/arrayAggregationRules.json";
+    private static final String AGGREGATED_OBJECT_FILE_PATH = "src/test/resources/arrayAggregatedObject.json";
+    private static final String AGGREGATED_OBJECT_ID = "175f08ff-1e4b-4265-a0d4-36e744297dc3";
+
+    @Autowired
+    private UpStreamEventsHandler upStreamEventsHandler;
+
+    @Mock
+    private ERQueryService erQueryService;
+
+    @Before
+    public void before() throws IOException {
+        MockitoAnnotations.initMocks(this);
+        upStreamEventsHandler.setEventRepositoryQueryService(erQueryService);
+
+        final URL upStreamResult = this.getClass().getClassLoader().getResource(UPSTREAM_RESULT_FILE);
+        ObjectMapper objectMapper = new ObjectMapper();
+        ObjectNode objectNode = objectMapper.createObjectNode();
+        objectNode.set("upstreamLinkObjects", objectMapper.readTree(upStreamResult));
+        objectNode.set("downstreamLinkObjects", objectMapper.createArrayNode());
+
+        when(erQueryService.getEventStreamDataById(anyString(), any(SearchOption.class), anyInt(), anyInt(),
+                anyBoolean())).thenReturn(new ResponseEntity<>(objectNode, HttpStatus.OK));
+    }
+
+    @Override
+    String getRulesFilePath() {
+        return RULES_FILE_PATH;
+    }
+
+    @Override
+    String getEventsFilePath() {
+        return EVENTS_FILE_PATH;
+    }
+
+    @Override
+    protected int extraEventsCount() {
+        // extra events from ER upstream
+        return 2;
+    }
+
+    @Override
+    List<String> getEventNamesToSend() {
+        List<String> eventNames = new ArrayList<>();
+        eventNames.add("EiffelArtifactPublishedEvent_1");
+        eventNames.add("EiffelArtifactPublishedEvent_2");
+        eventNames.add("EiffelCompositionDefinedEvent");
+        eventNames.add("EiffelArtifactPublishedEvent_3");
+        eventNames.add("EiffelArtifactPublishedEvent_4");
+        return eventNames;
+    }
+
+    @Override
+    Map<String, JsonNode> getCheckData() throws IOException {
+        JsonNode expectedJSON = getJSONFromFile(AGGREGATED_OBJECT_FILE_PATH);
+        Map<String, JsonNode> checkData = new HashMap<>();
+        checkData.put(AGGREGATED_OBJECT_ID, expectedJSON);
+        return checkData;
+    }
+
+}
diff --git a/src/test/java/com/ericsson/ei/flowtests/FlowTestBase.java b/src/test/java/com/ericsson/ei/flowtests/FlowTestBase.java
@@ -121,7 +121,7 @@ public void setFirstEventWaitTime(int value) {
 
     /**
      * Override this if you have more events that will be registered to event to
-     * object map but it is not visible in the test. For example form upstream
+     * object map but it is not visible in the test. For example from upstream
      * or downstream from event repository
      * 
      * @return
diff --git a/src/test/resources/arrayAggregatedObject.json b/src/test/resources/arrayAggregatedObject.json
diff --git a/src/test/resources/arrayAggregationEvents.json b/src/test/resources/arrayAggregationEvents.json
diff --git a/src/test/resources/arrayAggregationRules.json b/src/test/resources/arrayAggregationRules.json
diff --git a/src/test/resources/arrayAggregationUpstreamResult.json b/src/test/resources/arrayAggregationUpstreamResult.json

Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,6 @@ script:`
`21`	`21`	`#`
`22`	`22`	`# It is also importan that the first test to be run is a Spring Boot Test since spring boot also downloads`
`23`	`23`	`# another version of embedded mongo db so all Spring Boot tests will end up in a race condition also`
`24`		`- - mvn -DsomeModule.test.includes="**/QueryServiceTest.java" test`
`25`		`- - mvn -DsomeModule.test.excludes="**/QueryServiceTest.java" test`
	`24`	`+ - mvn -DsomeModule.test.includes="**/QueryServiceTest.java" test jacoco:report coveralls:report`
	`25`	`+ - mvn -DsomeModule.test.excludes="**/QueryServiceTest.java" test jacoco:report coveralls:report`
`26`	`26`