refactor(#1960): Refactor transformation rules, add schema restore

streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/AdapterEventPreviewPipeline.java

@@ -41,7 +41,8 @@ public class AdapterEventPreviewPipeline implements IAdapterPipeline {
 
   public AdapterEventPreviewPipeline(AdapterEventPreview previewRequest) {
     this.objectMapper = new ObjectMapper();
-    this.pipelineElements = new AdapterPipelineGeneratorBase().makeAdapterPipelineElements(previewRequest.getRules());
+    this.pipelineElements = new AdapterPipelineGeneratorBase()
+        .makeAdapterPipelineElements(previewRequest.getRules(), false);
     this.event = previewRequest.getInputData();
   }
 

streampipes-connect-shared/pom.xml

@@ -46,11 +46,21 @@
             <version>0.93.0-SNAPSHOT</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-sdk</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/AdapterPipelineGeneratorBase.java

@@ -18,93 +18,30 @@
 
 package org.apache.streampipes.connect.shared;
 
-import org.apache.streampipes.connect.shared.preprocessing.elements.AddTimestampPipelineElement;
-import org.apache.streampipes.connect.shared.preprocessing.elements.AddValuePipelineElement;
-import org.apache.streampipes.connect.shared.preprocessing.elements.TransformSchemaAdapterPipelineElement;
-import org.apache.streampipes.connect.shared.preprocessing.elements.TransformValueAdapterPipelineElement;
+import org.apache.streampipes.connect.shared.preprocessing.elements.AdapterTransformationPipelineElement;
+import org.apache.streampipes.connect.shared.preprocessing.generator.StatefulTransformationRuleGeneratorVisitor;
+import org.apache.streampipes.connect.shared.preprocessing.generator.StatelessTransformationRuleGeneratorVisitor;
 import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
 import org.apache.streampipes.model.connect.rules.TransformationRuleDescription;
-import org.apache.streampipes.model.connect.rules.schema.SchemaTransformationRuleDescription;
-import org.apache.streampipes.model.connect.rules.stream.EventRateTransformationRuleDescription;
-import org.apache.streampipes.model.connect.rules.stream.RemoveDuplicatesTransformationRuleDescription;
-import org.apache.streampipes.model.connect.rules.value.AddTimestampRuleDescription;
-import org.apache.streampipes.model.connect.rules.value.AddValueTransformationRuleDescription;
-import org.apache.streampipes.model.connect.rules.value.ValueTransformationRuleDescription;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.stream.Collectors;
 
 public class AdapterPipelineGeneratorBase {
 
-  public List<IAdapterPipelineElement> makeAdapterPipelineElements(List<TransformationRuleDescription> rules) {
-    List<IAdapterPipelineElement> pipelineElements = new ArrayList<>();
-
-    // Must be before the schema transformations to ensure that user can move this event property
-    var timestampTransformationRuleDescription = getTimestampRule(rules);
-    if (timestampTransformationRuleDescription != null) {
-      pipelineElements.add(new AddTimestampPipelineElement(
-          timestampTransformationRuleDescription.getRuntimeKey()));
-    }
-
-    var valueTransformationRuleDescription = getAddValueRule(rules);
-    if (valueTransformationRuleDescription != null) {
-      pipelineElements.add(new AddValuePipelineElement(
-          valueTransformationRuleDescription.getRuntimeKey(),
-          valueTransformationRuleDescription.getStaticValue()));
-    }
-
-    // first transform schema before transforming vales
-    // value rules should use unique keys for of new schema
-    pipelineElements.add(new TransformSchemaAdapterPipelineElement(getSchemaRules(rules)));
-    pipelineElements.add(new TransformValueAdapterPipelineElement(getValueRules(rules)));
-
-    return pipelineElements;
-  }
-
-  protected RemoveDuplicatesTransformationRuleDescription getRemoveDuplicateRule(
-      List<TransformationRuleDescription> rules) {
-    return getRule(rules, RemoveDuplicatesTransformationRuleDescription.class);
-  }
-
-  protected EventRateTransformationRuleDescription getEventRateTransformationRule(
-      List<TransformationRuleDescription> rules) {
-    return getRule(rules, EventRateTransformationRuleDescription.class);
-  }
-
-  protected AddTimestampRuleDescription getTimestampRule(List<TransformationRuleDescription> rules) {
-    return getRule(rules, AddTimestampRuleDescription.class);
-  }
-
-  protected AddValueTransformationRuleDescription getAddValueRule(List<TransformationRuleDescription> rules) {
-    return getRule(rules, AddValueTransformationRuleDescription.class);
-  }
-
-  private <T extends TransformationRuleDescription> T getRule(List<TransformationRuleDescription> rules,
-                                                              Class<T> type) {
-
-    if (rules != null) {
-      for (TransformationRuleDescription tr : rules) {
-        if (type.isInstance(tr)) {
-          return type.cast(tr);
-        }
-      }
+  public List<IAdapterPipelineElement> makeAdapterPipelineElements(List<TransformationRuleDescription> rules,
+                                                                   boolean includeStateful) {
+    var elements = new ArrayList<IAdapterPipelineElement>();
+    elements.add(new AdapterTransformationPipelineElement(
+        rules,
+        new StatelessTransformationRuleGeneratorVisitor())
+    );
+    if (includeStateful) {
+      elements.add(new AdapterTransformationPipelineElement(
+          rules,
+          new StatefulTransformationRuleGeneratorVisitor())
+      );
     }
-
-    return null;
-  }
-
-  private List<TransformationRuleDescription> getValueRules(List<TransformationRuleDescription> rules) {
-    return rules
-        .stream()
-        .filter(r -> r instanceof ValueTransformationRuleDescription && !(r instanceof AddTimestampRuleDescription))
-        .collect(Collectors.toList());
-  }
-
-  private List<TransformationRuleDescription> getSchemaRules(List<TransformationRuleDescription> rules) {
-    return rules
-        .stream()
-        .filter(r -> r instanceof SchemaTransformationRuleDescription)
-        .collect(Collectors.toList());
+    return elements;
   }
 }

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/SupportsNestedTransformationRule.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ *
+ */
+
+package org.apache.streampipes.connect.shared.preprocessing;
+
+import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+
+import java.util.List;
+import java.util.Map;
+
+public abstract class SupportsNestedTransformationRule implements TransformationRule {
+
+  @Override
+  public Map<String, Object> apply(Map<String, Object> event) {
+    return applyNested(event, getEventKeys());
+  }
+
+  protected Map<String, Object> applyNested(Map<String, Object> event,
+                                          List<String> eventKey) {
+    if (eventKey.size() == 1) {
+      applyTransformation(event, eventKey);
+    } else {
+      String key = eventKey.get(0);
+      List<String> newKeysTmpList = eventKey.subList(1, eventKey.size());
+
+      Map<String, Object> newSubEvent =
+          applyNested((Map<String, Object>) event.get(eventKey.get(0)), newKeysTmpList);
+
+      event.remove(key);
+      event.put(key, newSubEvent);
+    }
+    return event;
+  }
+
+  protected abstract List<String> getEventKeys();
+
+  protected abstract void applyTransformation(Map<String, Object> event,
+                                              List<String> eventKey);
+
+
+}

streampipes-extensions/streampipes-connect-adapters/src/test/java/org/apache/streampipes/connect/adapters/generic/Mock.java → streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/convert/ProvidesConversionResult.java

@@ -16,11 +16,14 @@
  *
  */
 
-package org.apache.streampipes.connect.adapters.generic;
+package org.apache.streampipes.connect.shared.preprocessing.convert;
 
-public class Mock {
-  public static final int PORT = 8042;
+import org.apache.streampipes.model.connect.rules.ITransformationRuleVisitor;
+import org.apache.streampipes.model.schema.EventProperty;
 
-  public static final String HOST = "http://localhost:" + PORT;
+import java.util.List;
 
+public interface ProvidesConversionResult extends ITransformationRuleVisitor {
+
+  List<EventProperty> getTransformedProperties();
 }

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/convert/SchemaConverter.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ *
+ */
+
+package org.apache.streampipes.connect.shared.preprocessing.convert;
+
+import org.apache.streampipes.model.connect.rules.TransformationRuleDescription;
+import org.apache.streampipes.model.schema.EventSchema;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class SchemaConverter {
+
+  public EventSchema toOriginalSchema(EventSchema transformedSchema,
+                                      List<TransformationRuleDescription> transformationRules) {
+
+    var converter = new ToOriginalSchemaConverter(
+        transformedSchema.getEventProperties()
+    );
+
+    return transform(transformationRules, converter, true);
+  }
+
+  public EventSchema toTransformedSchema(EventSchema originalSchema,
+                                         List<TransformationRuleDescription> transformationRules) {
+
+    var converter = new ToTransformedSchemaConverter(
+        originalSchema.getEventProperties()
+    );
+
+    return transform(transformationRules, converter, false);
+  }
+
+  private EventSchema transform(List<TransformationRuleDescription> transformationRules,
+                                ProvidesConversionResult converter,
+                                boolean reverseRules) {
+    var rules = transformationRules
+        .stream()
+        .sorted(Comparator.comparingInt(TransformationRuleDescription::getRulePriority))
+        .collect(Collectors.toCollection(ArrayList::new));
+
+    if (reverseRules) {
+      Collections.reverse(rules);
+    }
+
+    rules.forEach(rule -> rule.accept(converter));
+    return new EventSchema(converter.getTransformedProperties());
+  }
+}