Merge pull request #287 from Cumulocity-IoT/feature/mapping-filter

Feature/mapping-filter

.gitignore

@@ -50,3 +50,5 @@ dynamic-mapping-ui/cypress/fixtures/mqttConnectionPostRequest.json
 dynamic-mapping-service/src/main/configuration/dynamic-mapping-service-logging.xml
 JSONata4Java/dependency-reduced-pom.xml
 dynamic-mapping-ui/node_modules_tmp/
+.vscode/launch.json
+.env

USERGUIDE.md

@@ -16,6 +16,7 @@
       - [Map Device Identifier](#map-device-identifier)
       - [Define templates and substitutions for source and target payload](#define-templates-and-substitutions-for-source-and-target-payload)
       - [Different type of substitutions](#different-type-of-substitutions)
+    - [Apply a filter for a mapping](#apply-a-filter-for-a-mapping)
     - [Test transformation from source to target format](#test-transformation-from-source-to-target-format)
     - [Send transformed test message to test device in Cumulocity](#send-transformed-test-message-to-test-device-in-cumulocity)
     - [Use snooped payloads in source templates](#use-snooped-payloads-in-source-templates)
@@ -395,6 +396,22 @@ then three requests are generated:
 
 ---
 
+### Apply a filter for a mapping
+
+You can apply a filter for a mapping, in case the mapping should only be processed if the payload meets a certain condition. This might be the case, if you want to turn a rather generic payload into a measurement, an event and an alarm depending on content of the payload.
+In this case you can apply a filter:
+<p align="center">
+<img src="resources/image/Dynamic_Mapper_Mapping_Table_Filter.png"  style="width: 70%;" />
+</p>
+
+by defining a `JSONata` expression that evaluates to `true` or `false`:
+
+<p align="center">
+<img src="resources/image/Dynamic_Mapper_Mapping_Table_Filter_Detail.png"  style="width: 70%;" />
+</p>
+
+---
+
 ### Test transformation from source to target format
 
 To test the defined transformation, press the button `Transform test message`. The result of the transformation and any errors are displayed.

dynamic-mapping-service/src/main/java/dynamic/mapping/core/MappingComponent.java

@@ -483,6 +483,32 @@ public Mapping setActivationMapping(String tenant, String mappingId, Boolean act
 		return mapping;
 	}
 
+	public Mapping setFilterMapping(String tenant, String mappingId, String filterMapping) throws Exception {
+		// step 1. update activation for mapping
+		log.debug("Tenant {} - Setting filterMapping: {} got mapping: {}", tenant, filterMapping, mappingId);
+		Mapping mapping = getMapping(tenant, mappingId);
+		mapping.setFilterMapping(filterMapping);
+		if (Direction.INBOUND.equals(mapping.direction)) {
+			// step 2. retrieve collected snoopedTemplates
+			mapping.setSnoopedTemplates(cacheMappingInbound.get(tenant).get(mappingId).getSnoopedTemplates());
+		}
+		// step 3. update mapping in inventory
+		// don't validate mapping when setting active = false, this allows to remove
+		// mappings that are not working
+		updateMapping(tenant, mapping, true, false);
+		// step 4. delete mapping from update cache
+		removeDirtyMapping(tenant, mapping);
+		// step 5. update caches
+		if (Direction.OUTBOUND.equals(mapping.direction)) {
+			rebuildMappingOutboundCache(tenant);
+		} else {
+			deleteFromCacheMappingInbound(tenant, mapping);
+			addToCacheMappingInbound(tenant, mapping);
+			cacheMappingInbound.get(tenant).put(mapping.id, mapping);
+		}
+		return mapping;
+	}
+
 	public void setDebugMapping(String tenant, String id, Boolean debug) throws Exception {
 		// step 1. update debug for mapping
 		log.info("Tenant {} - Setting debug: {} got mapping: {}", tenant, id, debug);

dynamic-mapping-service/src/main/java/dynamic/mapping/core/Operation.java

@@ -34,5 +34,6 @@ public enum Operation {
 	SNOOP_MAPPING,
 	SNOOP_RESET,
 	RESET_DEPLOYMENT_MAP,
-	CLEAR_CACHE
+	CLEAR_CACHE,
+	APPLY_MAPPING_FILTER
 }

dynamic-mapping-service/src/main/java/dynamic/mapping/model/Mapping.java

@@ -134,6 +134,12 @@ public class Mapping implements Serializable {
     @JsonSetter(nulls = Nulls.SKIP)
     public String filterOutbound;
 
+    // TODO filterOutbound has to be removed and ofr ountbound mappings as well JSONata expressions
+    // this has to be changed in MappingComponent.deleteFromMappingCache & MappingComponent.rebuildMappingOutboundCache
+    @NotNull
+    @JsonSetter(nulls = Nulls.SKIP)
+    public String filterMapping;
+
     @NotNull
     @JsonSetter(nulls = Nulls.SKIP)
     public Boolean autoAckOperation;

dynamic-mapping-service/src/main/java/dynamic/mapping/processor/extension/ExtensibleProcessorInbound.java

@@ -146,4 +146,9 @@ public void updateStatusExtension(String extName) {
         }
     }
 
+    @Override
+    public void applyFiler(ProcessingContext<byte[]> context) {
+        // do nothing
+    }
+
 }

dynamic-mapping-service/src/main/java/dynamic/mapping/processor/inbound/AsynchronousDispatcherInbound.java

@@ -122,11 +122,11 @@ public MappingInboundTask(ConfigurationRegistry configurationRegistry, List<Mapp
 					.tag("tenant", connectorMessage.getTenant()).description("Total number of inbound messages")
 					.tag("connector", connectorMessage.getConnectorIdent()).register(Metrics.globalRegistry);
 
-        }
+		}
 
 		@Override
 		public List<ProcessingContext<?>> call() throws Exception {
-			//long startTime = System.nanoTime();
+			// long startTime = System.nanoTime();
 			Timer.Sample timer = Timer.start(Metrics.globalRegistry);
 			String tenant = connectorMessage.getTenant();
 			String topic = connectorMessage.getTopic();
@@ -206,10 +206,13 @@ public List<ProcessingContext<?>> call() throws Exception {
 								}
 							} else {
 								processor.extractFromSource(context);
-								processor.substituteInTargetAndSend(context);
-								List<C8YRequest> resultRequests = context.getRequests();
-								if (context.hasError() || resultRequests.stream().anyMatch(r -> r.hasError())) {
-									mappingStatus.errors++;
+								processor.applyFiler(context);
+								if (!context.isIgnoreFurtherProcessing()) {
+									processor.substituteInTargetAndSend(context);
+									List<C8YRequest> resultRequests = context.getRequests();
+									if (context.hasError() || resultRequests.stream().anyMatch(r -> r.hasError())) {
+										mappingStatus.errors++;
+									}
 								}
 							}
 						} catch (Exception e) {