This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch STREAMPIPES-494
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/STREAMPIPES-494 by this push:
new f42c57e [STREAMPIPES-494] Support less-strict validation and
moficiation of pipelines
f42c57e is described below
commit f42c57e55e4a7930e6a6add2258a0bfd74a51f8e
Author: Dominik Riemer <[email protected]>
AuthorDate: Wed Jan 5 11:51:47 2022 +0100
[STREAMPIPES-494] Support less-strict validation and moficiation of
pipelines
---
.../client/matching/MatchingResultMessage.java | 5 +
.../model/message/EdgeValidationStatus.java | 52 ++++++++++
.../model/message/EdgeValidationStatusType.java | 7 ++
.../model/message/PipelineEdgeValidation.java | 76 ++++++++++++++
.../model/message/PipelineModificationMessage.java | 30 +++++-
...ion.java => PipelineElementValidationInfo.java} | 37 ++++---
.../model/pipeline/PipelineModification.java | 21 +++-
...orDescription.java => ValidationInfoLevel.java} | 24 +----
.../DefaultStaticPropertyVisitor.java | 43 ++++++++
.../streampipes/manager/data/PipelineGraph.java | 4 +-
.../matching/PipelineModificationGenerator.java | 102 +++++++++++++++++++
.../matching/PipelineVerificationHandler.java | 12 ---
.../matching/PipelineVerificationHandlerV2.java | 33 +++---
.../manager/matching/v2/PropertyMatch.java | 1 -
.../pipeline/AbstractPipelineValidationStep.java | 45 ++++++++
.../matching/v2/pipeline/ApplyGroundingStep.java | 76 ++++++++++++++
.../v2/pipeline/CheckCompletedStep.java} | 24 +++--
.../v2/pipeline/CheckCompletedVisitor.java | 113 +++++++++++++++++++++
.../matching/v2/pipeline/ComputeOutputStep.java | 68 +++++++++++++
.../v2/pipeline/PipelineValidationSteps.java} | 22 ++--
.../v2/pipeline/PipelineValidator.java} | 25 ++++-
.../manager/matching/v2/pipeline/PrepareStep.java | 50 +++++++++
.../matching/v2/pipeline/SchemaValidationStep.java | 65 ++++++++++++
.../v2/pipeline/SpValidationException.java | 30 +++---
.../v2/pipeline/UpdateOutputStrategiesStep.java | 67 ++++++++++++
.../v2/pipeline/UpdateStaticPropertiesStep.java} | 22 ++--
.../v2/pipeline/UpdateStaticPropertiesVisitor.java | 108 ++++++++++++++++++++
.../streampipes/manager/operations/Operations.java | 17 ++--
ui/package.json | 2 +-
ui/src/app/core-model/gen/streampipes-model.ts | 71 +++++++++++--
.../pipeline-element-options.component.html | 8 +-
.../components/pipeline/pipeline.component.ts | 93 ++++++++++-------
.../matching-error/matching-error.component.html | 18 ++--
.../matching-error/matching-error.component.ts | 16 +--
.../app/editor/services/jsplumb-bridge.service.ts | 5 +-
35 files changed, 1206 insertions(+), 186 deletions(-)
diff --git
a/streampipes-model-client/src/main/java/org/apache/streampipes/model/client/matching/MatchingResultMessage.java
b/streampipes-model-client/src/main/java/org/apache/streampipes/model/client/matching/MatchingResultMessage.java
index c16ab9c..17591a5 100644
---
a/streampipes-model-client/src/main/java/org/apache/streampipes/model/client/matching/MatchingResultMessage.java
+++
b/streampipes-model-client/src/main/java/org/apache/streampipes/model/client/matching/MatchingResultMessage.java
@@ -85,5 +85,10 @@ public class MatchingResultMessage {
public void setDescription(String description) {
this.description = description;
}
+
+ @Override
+ public String toString() {
+ return title + " - " + description + "\n" + "(required: "
+requirementSubject + ")";
+ }
}
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/message/EdgeValidationStatus.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/EdgeValidationStatus.java
new file mode 100644
index 0000000..26448aa
--- /dev/null
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/EdgeValidationStatus.java
@@ -0,0 +1,52 @@
+/*
+ * 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.model.message;
+
+import java.util.List;
+
+public class EdgeValidationStatus {
+
+ private EdgeValidationStatusType validationStatusType;
+ private List<Notification> notifications;
+
+ public EdgeValidationStatus(EdgeValidationStatusType validationStatusType) {
+ this.validationStatusType = validationStatusType;
+ }
+
+ public EdgeValidationStatus(EdgeValidationStatusType validationStatusType,
+ List<Notification> notifications) {
+ this.validationStatusType = validationStatusType;
+ this.notifications = notifications;
+ }
+
+ public EdgeValidationStatusType getValidationStatusType() {
+ return validationStatusType;
+ }
+
+ public void setValidationStatusType(EdgeValidationStatusType
validationStatusType) {
+ this.validationStatusType = validationStatusType;
+ }
+
+ public List<Notification> getNotifications() {
+ return notifications;
+ }
+
+ public void setNotifications(List<Notification> notifications) {
+ this.notifications = notifications;
+ }
+}
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/message/EdgeValidationStatusType.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/EdgeValidationStatusType.java
new file mode 100644
index 0000000..58aa46f
--- /dev/null
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/EdgeValidationStatusType.java
@@ -0,0 +1,7 @@
+package org.apache.streampipes.model.message;
+
+public enum EdgeValidationStatusType {
+ COMPLETE,
+ INCOMPLETE,
+ INVALID
+}
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineEdgeValidation.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineEdgeValidation.java
new file mode 100644
index 0000000..d485e08
--- /dev/null
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineEdgeValidation.java
@@ -0,0 +1,76 @@
+/*
+ * 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.model.message;
+
+import java.util.List;
+
+public class PipelineEdgeValidation {
+
+ private String sourceId;
+ private String targetId;
+
+ private EdgeValidationStatus status;
+
+ public static PipelineEdgeValidation complete(String sourceId,
+ String targetId) {
+ EdgeValidationStatus status = new
EdgeValidationStatus(EdgeValidationStatusType.COMPLETE);
+ return new PipelineEdgeValidation(sourceId, targetId, status);
+ }
+
+ public static PipelineEdgeValidation invalid(String sourceId,
+ String targetId,
+ List<Notification>
notifications) {
+ EdgeValidationStatus status = new
EdgeValidationStatus(EdgeValidationStatusType.INVALID, notifications);
+ return new PipelineEdgeValidation(sourceId, targetId, status);
+ }
+
+ public PipelineEdgeValidation() {
+ }
+
+ public PipelineEdgeValidation(String sourceId,
+ String targetId,
+ EdgeValidationStatus edgeValidationStatus) {
+ this.sourceId = sourceId;
+ this.targetId = targetId;
+ this.status = edgeValidationStatus;
+ }
+
+ public String getSourceId() {
+ return sourceId;
+ }
+
+ public void setSourceId(String sourceId) {
+ this.sourceId = sourceId;
+ }
+
+ public String getTargetId() {
+ return targetId;
+ }
+
+ public void setTargetId(String targetId) {
+ this.targetId = targetId;
+ }
+
+ public EdgeValidationStatus getStatus() {
+ return status;
+ }
+
+ public void setStatus(EdgeValidationStatus status) {
+ this.status = status;
+ }
+}
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineModificationMessage.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineModificationMessage.java
index cd26edb..631ef1d 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineModificationMessage.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineModificationMessage.java
@@ -28,15 +28,18 @@ import java.util.List;
public class PipelineModificationMessage extends Message {
/**
- * Class that represents PipelineModification messages. Modifications
are used to update a SEPA within an already created pipeline
+ * Class that represents PipelineModification messages. Modifications
are used to update a pipeline element within an already created pipeline
*/
private List<PipelineModification> pipelineModifications;
+ private boolean pipelineValid;
+ private List<PipelineEdgeValidation> edgeValidations;
public PipelineModificationMessage(
List<PipelineModification> pipelineModifications) {
super(true);
this.pipelineModifications = pipelineModifications;
+ this.edgeValidations = new ArrayList<>();
}
public PipelineModificationMessage() {
@@ -52,11 +55,30 @@ public class PipelineModificationMessage extends Message {
List<PipelineModification> pipelineModifications) {
this.pipelineModifications = pipelineModifications;
}
-
- public void addPipelineModification(PipelineModification
pipelineModification)
- {
+
+ public boolean isPipelineValid() {
+ return pipelineValid;
+ }
+
+ public void setPipelineValid(boolean pipelineValid) {
+ this.pipelineValid = pipelineValid;
+ }
+
+ public List<PipelineEdgeValidation> getEdgeValidations() {
+ return edgeValidations;
+ }
+
+ public void setEdgeValidations(List<PipelineEdgeValidation>
edgeValidations) {
+ this.edgeValidations = edgeValidations;
+ }
+
+ public void addPipelineModification(PipelineModification
pipelineModification) {
pipelineModifications.add(pipelineModification);
}
+
+ public void addEdgeValidation(PipelineEdgeValidation
pipelineEdgeValidation) {
+ edgeValidations.add(pipelineEdgeValidation);
+ }
public boolean existsModification(String domId)
{
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementValidationInfo.java
similarity index 51%
copy from
streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
copy to
streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementValidationInfo.java
index 4a987a7..f88c98c 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementValidationInfo.java
@@ -18,25 +18,38 @@
package org.apache.streampipes.model.pipeline;
-public class PipelineModificationErrorDescription {
+public class PipelineElementValidationInfo {
- private String errorMessage;
-
+ private ValidationInfoLevel level;
+ private String message;
+
+ public static PipelineElementValidationInfo info(String message) {
+ return new
PipelineElementValidationInfo(ValidationInfoLevel.INFO, message);
+ }
+
+ public static PipelineElementValidationInfo error(String message) {
+ return new
PipelineElementValidationInfo(ValidationInfoLevel.ERROR, message);
+ }
- public PipelineModificationErrorDescription(String errorMessage)
- {
- this.errorMessage = errorMessage;
+ public PipelineElementValidationInfo(ValidationInfoLevel level,
+
String
message) {
+ this.level = level;
+ this.message = message;
}
+ public ValidationInfoLevel getLevel() {
+ return level;
+ }
- public String getErrorMessage() {
- return errorMessage;
+ public void setLevel(ValidationInfoLevel level) {
+ this.level = level;
}
+ public String getMessage() {
+ return message;
+ }
- public void setErrorMessage(String errorMessage) {
- this.errorMessage = errorMessage;
+ public void setMessage(String message) {
+ this.message = message;
}
-
-
}
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModification.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModification.java
index 898d372..1546e81 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModification.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModification.java
@@ -29,8 +29,9 @@ public class PipelineModification {
private String domId;
private String elementId;
+ private boolean pipelineElementValid;
- private List<PipelineModificationErrorDescription> errorDescriptions;
+ private List<PipelineElementValidationInfo> validationInfos;
private List<StaticProperty> staticProperties;
private List<OutputStrategy> outputStrategies;
private List<SpDataStream> inputStreams;
@@ -42,6 +43,8 @@ public class PipelineModification {
this.elementId = elementId;
this.staticProperties = staticProperties;
this.inputStreams = new ArrayList<>();
+ this.outputStrategies = new ArrayList<>();
+ this.validationInfos = new ArrayList<>();
}
public PipelineModification() {
@@ -91,4 +94,20 @@ public class PipelineModification {
public void addInputStream(SpDataStream inputStream) {
this.inputStreams.add(inputStream);
}
+
+ public boolean isPipelineElementValid() {
+ return pipelineElementValid;
+ }
+
+ public void setPipelineElementValid(boolean pipelineElementValid) {
+ this.pipelineElementValid = pipelineElementValid;
+ }
+
+ public List<PipelineElementValidationInfo> getValidationInfos() {
+ return validationInfos;
+ }
+
+ public void setValidationInfos(List<PipelineElementValidationInfo>
validationInfos) {
+ this.validationInfos = validationInfos;
+ }
}
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/ValidationInfoLevel.java
similarity index 70%
copy from
streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
copy to
streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/ValidationInfoLevel.java
index 4a987a7..f6b840c 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/ValidationInfoLevel.java
@@ -18,25 +18,7 @@
package org.apache.streampipes.model.pipeline;
-public class PipelineModificationErrorDescription {
-
- private String errorMessage;
-
-
- public PipelineModificationErrorDescription(String errorMessage)
- {
- this.errorMessage = errorMessage;
- }
-
-
- public String getErrorMessage() {
- return errorMessage;
- }
-
-
- public void setErrorMessage(String errorMessage) {
- this.errorMessage = errorMessage;
- }
-
-
+public enum ValidationInfoLevel {
+ INFO,
+ ERROR;
}
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/staticproperty/DefaultStaticPropertyVisitor.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/staticproperty/DefaultStaticPropertyVisitor.java
new file mode 100644
index 0000000..a300e55
--- /dev/null
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/staticproperty/DefaultStaticPropertyVisitor.java
@@ -0,0 +1,43 @@
+/*
+ * 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.model.staticproperty;
+
+public abstract class DefaultStaticPropertyVisitor implements
StaticPropertyVisitor {
+
+ @Override
+ public void visit(CollectionStaticProperty collectionStaticProperty) {
+ collectionStaticProperty.getStaticPropertyTemplate().accept(this);
+ collectionStaticProperty.getMembers().forEach(sp -> sp.accept(this));
+ }
+
+ @Override
+ public void visit(StaticPropertyAlternative staticPropertyAlternative) {
+ staticPropertyAlternative.getStaticProperty().accept(this);
+ }
+
+ @Override
+ public void visit(StaticPropertyAlternatives staticPropertyAlternatives) {
+ staticPropertyAlternatives.getAlternatives().forEach(alternative ->
alternative.accept(this));
+ }
+
+ @Override
+ public void visit(StaticPropertyGroup staticPropertyGroup) {
+ staticPropertyGroup.getStaticProperties().forEach(sp -> sp.accept(this));
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
index eb1a837..d873905 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
@@ -18,10 +18,10 @@
package org.apache.streampipes.manager.data;
-import org.jgrapht.graph.SimpleDirectedGraph;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.jgrapht.graph.DirectedMultigraph;
-public class PipelineGraph extends SimpleDirectedGraph<NamedStreamPipesEntity,
String> {
+public class PipelineGraph extends DirectedMultigraph<NamedStreamPipesEntity,
String> {
public PipelineGraph() {
super(String.class);
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineModificationGenerator.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineModificationGenerator.java
new file mode 100644
index 0000000..41f720c
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineModificationGenerator.java
@@ -0,0 +1,102 @@
+package org.apache.streampipes.manager.matching;
+
+import org.apache.streampipes.manager.data.PipelineGraph;
+import org.apache.streampipes.manager.data.PipelineGraphHelpers;
+import org.apache.streampipes.manager.matching.v2.pipeline.PipelineValidator;
+import
org.apache.streampipes.manager.matching.v2.pipeline.SpValidationException;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.client.matching.MatchingResultMessage;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.message.Notification;
+import org.apache.streampipes.model.message.PipelineEdgeValidation;
+import org.apache.streampipes.model.message.PipelineModificationMessage;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+import org.apache.streampipes.model.pipeline.PipelineModification;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+public class PipelineModificationGenerator {
+
+ private final PipelineGraph pipelineGraph;
+ private final Map<String, PipelineModification> pipelineModifications;
+ private List<PipelineEdgeValidation> edgeValidations;
+ private final PipelineValidator pipelineValidator;
+
+ public PipelineModificationGenerator(PipelineGraph pipelineGraph) {
+ this.pipelineGraph = pipelineGraph;
+ this.pipelineModifications = new HashMap<>();
+ this.pipelineValidator = new PipelineValidator();
+ this.edgeValidations = new ArrayList<>();
+ }
+
+ public PipelineModificationMessage buildPipelineModificationMessage() {
+ List<SpDataStream> streams =
PipelineGraphHelpers.findStreams(pipelineGraph);
+
+ for (SpDataStream stream : streams) {
+ Set<InvocableStreamPipesEntity> connectedElements =
getConnections(stream);
+ addModification(stream, connectedElements);
+ }
+
+ PipelineModificationMessage message = new PipelineModificationMessage();
+ message.setPipelineModifications(getModifications());
+ message.setEdgeValidations(this.edgeValidations);
+ return message;
+ }
+
+ private void addModification(NamedStreamPipesEntity source,
+ Set<InvocableStreamPipesEntity> targets) {
+
+ targets.forEach(t -> {
+ PipelineModification modification = new PipelineModification();
+ List<PipelineElementValidationInfo> validationInfos = new ArrayList<>();
+ modification.setDomId(t.getDOM());
+ modification.setElementId(t.getElementId());
+ try {
+ pipelineValidator.apply(source, t, targets, validationInfos);
+ buildModification(modification, t);
+ edgeValidations.add(PipelineEdgeValidation.complete(source.getDOM(),
t.getDOM()));
+ } catch (SpValidationException e) {
+ //e.getErrorLog().forEach(log ->
validationInfos.add(PipelineElementValidationInfo.error(log.toString())));
+ edgeValidations.add(PipelineEdgeValidation.invalid(source.getDOM(),
t.getDOM(), toNotifications(e.getErrorLog())));
+ modification.setPipelineElementValid(false);
+ }
+ modification.setValidationInfos(validationInfos);
+ this.pipelineModifications.put(t.getDOM(), modification);
+
+ addModification(t, getConnections(t));
+ });
+ }
+
+ private void buildModification(PipelineModification modification,
+ InvocableStreamPipesEntity t) {
+ if (t instanceof DataProcessorInvocation) {
+ modification.setOutputStrategies(((DataProcessorInvocation)
t).getOutputStrategies());
+ }
+ modification.setInputStreams(t.getInputStreams());
+ modification.setStaticProperties(t.getStaticProperties());
+ modification.setPipelineElementValid(true);
+ }
+
+ private Set<InvocableStreamPipesEntity>
getConnections(NamedStreamPipesEntity source) {
+ Set<String> outgoingEdges = pipelineGraph.outgoingEdgesOf(source);
+ return outgoingEdges
+ .stream()
+ .map(pipelineGraph::getEdgeTarget)
+ .map(g -> (InvocableStreamPipesEntity) g)
+ .collect(Collectors.toSet());
+ }
+
+ private List<PipelineModification> getModifications() {
+ return new ArrayList<>(this.pipelineModifications.values());
+ }
+
+ private List<Notification> toNotifications(List<MatchingResultMessage>
matchingResultMessages) {
+ return matchingResultMessages
+ .stream()
+ .map(m -> new Notification(m.getTitle(), m.toString()))
+ .collect(Collectors.toList());
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandler.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandler.java
index c6a2bb2..7e63678 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandler.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandler.java
@@ -191,21 +191,9 @@ public class PipelineVerificationHandler {
return pipelineModificationMessage;
}
- public List<InvocableStreamPipesEntity> getInvocationGraphs() {
- return this.invocationGraphs;
- }
-
public List<InvocableStreamPipesEntity> makeInvocationGraphs() {
PipelineGraph pipelineGraph = new
PipelineGraphBuilder(pipeline).buildGraph();
return new InvocationGraphBuilder(pipelineGraph, null).buildGraphs();
}
- private boolean onlyStreamAncestorsPresentInPipeline() {
- return rootPipelineElement
- .getConnectedTo()
- .stream()
- .map(connectedTo -> TreeUtils.findSEPAElement(connectedTo,
pipeline.getSepas(), pipeline.getStreams()))
- .allMatch(pe -> pe instanceof SpDataStream);
- }
-
}
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandlerV2.java
similarity index 54%
copy from
streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
copy to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandlerV2.java
index 4a987a7..0152dca 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/PipelineVerificationHandlerV2.java
@@ -15,28 +15,23 @@
* limitations under the License.
*
*/
+package org.apache.streampipes.manager.matching;
-package org.apache.streampipes.model.pipeline;
+import org.apache.streampipes.manager.data.PipelineGraph;
+import org.apache.streampipes.manager.data.PipelineGraphBuilder;
+import org.apache.streampipes.model.message.PipelineModificationMessage;
+import org.apache.streampipes.model.pipeline.Pipeline;
-public class PipelineModificationErrorDescription {
+public class PipelineVerificationHandlerV2 {
- private String errorMessage;
-
-
- public PipelineModificationErrorDescription(String errorMessage)
- {
- this.errorMessage = errorMessage;
- }
+ private Pipeline pipeline;
+ public PipelineVerificationHandlerV2(Pipeline pipeline) {
+ this.pipeline = pipeline;
+ }
- public String getErrorMessage() {
- return errorMessage;
- }
-
-
- public void setErrorMessage(String errorMessage) {
- this.errorMessage = errorMessage;
- }
-
-
+ public PipelineModificationMessage verifyPipeline() {
+ PipelineGraph graph = new PipelineGraphBuilder(pipeline).buildGraph();
+ return new
PipelineModificationGenerator(graph).buildPipelineModificationMessage();
+ }
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/PropertyMatch.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/PropertyMatch.java
index 95259e9..6786fce 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/PropertyMatch.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/PropertyMatch.java
@@ -31,7 +31,6 @@ public class PropertyMatch extends
AbstractMatcher<EventProperty, EventProperty>
public PropertyMatch() {
super(MatchingResultType.PROPERTY_MATCH);
- // TODO Auto-generated constructor stub
}
@Override
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/AbstractPipelineValidationStep.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/AbstractPipelineValidationStep.java
new file mode 100644
index 0000000..a248a19
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/AbstractPipelineValidationStep.java
@@ -0,0 +1,45 @@
+/*
+ * 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.manager.matching.v2.pipeline;
+
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.client.matching.MatchingResultMessage;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+
+import java.util.*;
+
+public abstract class AbstractPipelineValidationStep {
+
+ protected final Map<String, Integer> visitorHistory = new HashMap<>();
+
+ public abstract void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo>
validationInfos) throws SpValidationException;
+
+ public List<MatchingResultMessage> getNewErrorLog() {
+ return new ArrayList<>();
+ }
+
+ protected int getIndex(InvocableStreamPipesEntity target) {
+ return this.visitorHistory.containsKey(target.getDOM()) ? 1 : 0;
+ }
+
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/ApplyGroundingStep.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/ApplyGroundingStep.java
new file mode 100644
index 0000000..cc5e54e
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/ApplyGroundingStep.java
@@ -0,0 +1,76 @@
+/*
+ * 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.manager.matching.v2.pipeline;
+
+import org.apache.streampipes.manager.matching.GroundingBuilder;
+import org.apache.streampipes.manager.matching.v2.GroundingMatch;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.client.matching.MatchingResultMessage;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.grounding.EventGrounding;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+
+import java.util.List;
+import java.util.Set;
+
+public class ApplyGroundingStep extends AbstractPipelineValidationStep {
+
+ @Override
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos)
throws SpValidationException {
+
+ List<MatchingResultMessage> errorLog = getNewErrorLog();
+ boolean match = new GroundingMatch().match(
+ getSourceGrounding(source),
+ target.getSupportedGrounding(),
+ errorLog
+ );
+
+ if (!match) {
+ throw new SpValidationException(errorLog);
+ } else {
+ EventGrounding selectedGrounding = new GroundingBuilder(source,
allTargets).getEventGrounding();
+
+ if (source instanceof DataProcessorInvocation) {
+ ((DataProcessorInvocation) source)
+ .getOutputStream()
+ .setEventGrounding(selectedGrounding);
+ }
+
+ target
+ .getInputStreams()
+ .get(getIndex(target))
+ .setEventGrounding(selectedGrounding);
+
+ this.visitorHistory.put(target.getDOM(), 1);
+ }
+ }
+
+ private EventGrounding getSourceGrounding(NamedStreamPipesEntity source) {
+ if (source instanceof SpDataStream) {
+ return ((SpDataStream) source).getEventGrounding();
+ } else {
+ return ((DataProcessorInvocation)
source).getOutputStream().getEventGrounding();
+ }
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CheckCompletedStep.java
similarity index 51%
copy from
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
copy to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CheckCompletedStep.java
index eb1a837..de6a0e2 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CheckCompletedStep.java
@@ -16,14 +16,26 @@
*
*/
-package org.apache.streampipes.manager.data;
+package org.apache.streampipes.manager.matching.v2.pipeline;
-import org.jgrapht.graph.SimpleDirectedGraph;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
-public class PipelineGraph extends SimpleDirectedGraph<NamedStreamPipesEntity,
String> {
+import java.util.List;
+import java.util.Set;
- public PipelineGraph() {
- super(String.class);
- }
+public class CheckCompletedStep extends AbstractPipelineValidationStep {
+
+ @Override
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos)
throws SpValidationException {
+
+ CheckCompletedVisitor visitor = new CheckCompletedVisitor();
+ target.getStaticProperties().forEach(sp -> sp.accept(visitor));
+
+ validationInfos.addAll(visitor.getValidationInfos());
+ }
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CheckCompletedVisitor.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CheckCompletedVisitor.java
new file mode 100644
index 0000000..885ab77
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/CheckCompletedVisitor.java
@@ -0,0 +1,113 @@
+/*
+ * 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.manager.matching.v2.pipeline;
+
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+import org.apache.streampipes.model.staticproperty.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class CheckCompletedVisitor extends DefaultStaticPropertyVisitor {
+
+ private List<PipelineElementValidationInfo> validationInfos;
+
+ public CheckCompletedVisitor() {
+ this.validationInfos = new ArrayList<>();
+ }
+
+ @Override
+ public void visit(AnyStaticProperty property) {
+
+ }
+
+ @Override
+ public void visit(CodeInputStaticProperty codeInputStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(ColorPickerStaticProperty colorPickerStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(DomainStaticProperty domainStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(FileStaticProperty fileStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(FreeTextStaticProperty freeTextStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(MappingPropertyNary mappingPropertyNary) {
+ if (mappingPropertyNary
+ .getSelectedProperties()
+ .stream()
+ .noneMatch((p ->
mappingPropertyNary.getMapsFromOptions().contains(p)))) {
+ mappingPropertyNary.setSelectedProperties(mappingPropertyNary
+ .getSelectedProperties()
+ .stream()
+ .filter(p ->
mappingPropertyNary.getMapsFromOptions().contains(p))
+ .collect(Collectors.toList()));
+ validationInfos.add(PipelineElementValidationInfo.info("Error in mapping
property nary - removed invalid selection"));
+ }
+ }
+
+ @Override
+ public void visit(MappingPropertyUnary mappingPropertyUnary) {
+ if
(!(mappingPropertyUnary.getMapsFromOptions().contains(mappingPropertyUnary.getSelectedProperty())))
{
+ // TODO only if element touched
+ mappingPropertyUnary.setSelectedProperty("");
+ validationInfos.add(PipelineElementValidationInfo.info("Error in mapping
property unary - removed invalid selection"));
+ }
+ }
+
+ @Override
+ public void visit(MatchingStaticProperty matchingStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(OneOfStaticProperty oneOfStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(SecretStaticProperty secretStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(RemoteOneOfStaticProperty remoteOneOfStaticProperty) {
+
+ }
+
+ public List<PipelineElementValidationInfo> getValidationInfos() {
+ return this.validationInfos;
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/ComputeOutputStep.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/ComputeOutputStep.java
new file mode 100644
index 0000000..d213fe0
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/ComputeOutputStep.java
@@ -0,0 +1,68 @@
+/*
+ * 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.manager.matching.v2.pipeline;
+
+import org.apache.streampipes.manager.matching.output.OutputSchemaFactory;
+import org.apache.streampipes.manager.matching.output.OutputSchemaGenerator;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.output.OutputStrategy;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.sdk.helpers.Tuple2;
+
+import java.util.*;
+
+public class ComputeOutputStep extends AbstractPipelineValidationStep {
+
+ private final Map<String, DataProcessorInvocation> relatedPes = new
HashMap<>();
+
+ @Override
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos)
throws SpValidationException {
+
+ if (target instanceof DataProcessorInvocation) {
+ DataProcessorInvocation pe = (DataProcessorInvocation) target;
+ Tuple2<EventSchema, ? extends OutputStrategy> outputSettings;
+ OutputSchemaGenerator<?> schemaGenerator = new OutputSchemaFactory(pe)
+ .getOuputSchemaGenerator();
+
+ if (target.getInputStreams().size() == 1) {
+ outputSettings = schemaGenerator.buildFromOneStream(
+ pe.getInputStreams()
+ .get(0));
+ } else if (relatedPes.containsKey(pe.getDOM())) {
+ DataProcessorInvocation existingInvocation =
relatedPes.get(pe.getDOM());
+
+ outputSettings = schemaGenerator.buildFromTwoStreams(existingInvocation
+ .getInputStreams().get(0), pe.getInputStreams().get(1));
+ } else {
+ relatedPes.put(target.getDOM(), pe);
+ outputSettings = new Tuple2<>(new EventSchema(), pe
+ .getOutputStrategies().get(0));
+ }
+
+ pe.setOutputStrategies(Collections.singletonList(outputSettings.b));
+ ((DataProcessorInvocation)
target).getOutputStream().setEventSchema(outputSettings.a);
+ }
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidationSteps.java
similarity index 60%
copy from
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
copy to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidationSteps.java
index eb1a837..a54bb66 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidationSteps.java
@@ -16,14 +16,22 @@
*
*/
-package org.apache.streampipes.manager.data;
+package org.apache.streampipes.manager.matching.v2.pipeline;
-import org.jgrapht.graph.SimpleDirectedGraph;
-import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import java.util.Arrays;
+import java.util.List;
-public class PipelineGraph extends SimpleDirectedGraph<NamedStreamPipesEntity,
String> {
+public class PipelineValidationSteps {
- public PipelineGraph() {
- super(String.class);
- }
+ public List<AbstractPipelineValidationStep> collect() {
+ return Arrays.asList(
+ new PrepareStep(),
+ new ApplyGroundingStep(),
+ new SchemaValidationStep(),
+ new UpdateStaticPropertiesStep(),
+ new UpdateOutputStrategiesStep(),
+ new ComputeOutputStep(),
+ new CheckCompletedStep()
+ );
+ }
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidator.java
similarity index 51%
copy from
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
copy to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidator.java
index eb1a837..6648082 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PipelineValidator.java
@@ -16,14 +16,29 @@
*
*/
-package org.apache.streampipes.manager.data;
+package org.apache.streampipes.manager.matching.v2.pipeline;
-import org.jgrapht.graph.SimpleDirectedGraph;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
-public class PipelineGraph extends SimpleDirectedGraph<NamedStreamPipesEntity,
String> {
+import java.util.List;
+import java.util.Set;
- public PipelineGraph() {
- super(String.class);
+public class PipelineValidator {
+
+ private final List<AbstractPipelineValidationStep> steps;
+
+ public PipelineValidator() {
+ this.steps = new PipelineValidationSteps().collect();
+ }
+
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos)
throws SpValidationException {
+ for (AbstractPipelineValidationStep step : steps) {
+ step.apply(source, target, allTargets, validationInfos);
}
+ }
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PrepareStep.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PrepareStep.java
new file mode 100644
index 0000000..6b7a0bf
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/PrepareStep.java
@@ -0,0 +1,50 @@
+/*
+ * 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.manager.matching.v2.pipeline;
+
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+public class PrepareStep extends AbstractPipelineValidationStep {
+
+ @Override
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos)
throws SpValidationException {
+ if (target instanceof DataProcessorInvocation) {
+ if (target.getInputStreams() == null) {
+ target.setInputStreams(new ArrayList<>());
+ for (int i = 0; i < target.getStreamRequirements().size(); i++) {
+ target.getInputStreams().add(new SpDataStream());
+ }
+ }
+ if (((DataProcessorInvocation) target).getOutputStream() == null) {
+ ((DataProcessorInvocation) target).setOutputStream(new SpDataStream());
+ }
+ }
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/SchemaValidationStep.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/SchemaValidationStep.java
new file mode 100644
index 0000000..a306383
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/SchemaValidationStep.java
@@ -0,0 +1,65 @@
+/*
+ * 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.manager.matching.v2.pipeline;
+
+import org.apache.streampipes.manager.matching.v2.SchemaMatch;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.client.matching.MatchingResultMessage;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+import org.apache.streampipes.model.schema.EventSchema;
+
+import java.util.List;
+import java.util.Set;
+
+public class SchemaValidationStep extends AbstractPipelineValidationStep {
+
+ @Override
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos)
throws SpValidationException {
+
+ List<MatchingResultMessage> errorLog = getNewErrorLog();
+ boolean matches = new SchemaMatch().match(getSourceSchema(source),
getTargetRequirement(target), errorLog);
+
+ if (!matches) {
+ throw new SpValidationException(errorLog);
+ } else {
+
target.getInputStreams().get(getIndex(target)).setEventSchema(getSourceSchema(source));
+ }
+
+ this.visitorHistory.put(target.getDOM(), 1);
+ }
+
+ private EventSchema getSourceSchema(NamedStreamPipesEntity source) {
+ if (source instanceof SpDataStream) {
+ return ((SpDataStream) source).getEventSchema();
+ } else {
+ return ((DataProcessorInvocation)
source).getOutputStream().getEventSchema();
+ }
+ }
+
+ private EventSchema getTargetRequirement(InvocableStreamPipesEntity target) {
+ return
target.getStreamRequirements().get(getIndex(target)).getEventSchema();
+ }
+
+}
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/SpValidationException.java
similarity index 62%
rename from
streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
rename to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/SpValidationException.java
index 4a987a7..14e1002 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineModificationErrorDescription.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/SpValidationException.java
@@ -16,27 +16,23 @@
*
*/
-package org.apache.streampipes.model.pipeline;
+package org.apache.streampipes.manager.matching.v2.pipeline;
-public class PipelineModificationErrorDescription {
+import org.apache.streampipes.model.client.matching.MatchingResultMessage;
- private String errorMessage;
-
-
- public PipelineModificationErrorDescription(String errorMessage)
- {
- this.errorMessage = errorMessage;
- }
+import java.util.List;
+public class SpValidationException extends Exception {
- public String getErrorMessage() {
- return errorMessage;
- }
+ private List<MatchingResultMessage> errorLog;
+ public SpValidationException(List<MatchingResultMessage> errorLog) {
+ this.errorLog = errorLog;
+ }
- public void setErrorMessage(String errorMessage) {
- this.errorMessage = errorMessage;
- }
-
-
+ public SpValidationException() {}
+
+ public List<MatchingResultMessage> getErrorLog() {
+ return errorLog;
+ }
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/UpdateOutputStrategiesStep.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/UpdateOutputStrategiesStep.java
new file mode 100644
index 0000000..2040732
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/UpdateOutputStrategiesStep.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.manager.matching.v2.pipeline;
+
+import org.apache.streampipes.manager.selector.PropertySelectorGenerator;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
+
+import java.util.List;
+import java.util.Set;
+
+public class UpdateOutputStrategiesStep extends AbstractPipelineValidationStep
{
+
+ @Override
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos)
throws SpValidationException {
+ List<SpDataStream> inputStreams = target.getInputStreams();
+ if (target instanceof DataProcessorInvocation) {
+ ((DataProcessorInvocation) target).getOutputStrategies()
+ .stream()
+ .filter(strategy -> strategy instanceof CustomOutputStrategy)
+ .map(strategy -> (CustomOutputStrategy) strategy)
+ .forEach(strategy -> {
+ PropertySelectorGenerator generator = getGenerator(inputStreams,
strategy);
+ strategy.setAvailablePropertyKeys(generator.generateSelectors());
+ });
+ }
+ }
+
+ private PropertySelectorGenerator getGenerator(List<SpDataStream>
inputStreams,
+ CustomOutputStrategy
strategy) {
+ if (inputStreams.size() == 1 || (inputStreams.size() > 1 &&
!strategy.isOutputRight())) {
+ return new PropertySelectorGenerator(
+ inputStreams.get(0).getEventSchema(),
+ false
+ );
+ } else {
+ return new PropertySelectorGenerator(
+ inputStreams.get(0).getEventSchema(),
+ inputStreams.get(1).getEventSchema(),
+ false
+ );
+ }
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/UpdateStaticPropertiesStep.java
similarity index 52%
copy from
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
copy to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/UpdateStaticPropertiesStep.java
index eb1a837..c6549b0 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraph.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/UpdateStaticPropertiesStep.java
@@ -16,14 +16,24 @@
*
*/
-package org.apache.streampipes.manager.data;
+package org.apache.streampipes.manager.matching.v2.pipeline;
-import org.jgrapht.graph.SimpleDirectedGraph;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
-public class PipelineGraph extends SimpleDirectedGraph<NamedStreamPipesEntity,
String> {
+import java.util.List;
+import java.util.Set;
- public PipelineGraph() {
- super(String.class);
- }
+public class UpdateStaticPropertiesStep extends AbstractPipelineValidationStep
{
+
+ @Override
+ public void apply(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ Set<InvocableStreamPipesEntity> allTargets,
+ List<PipelineElementValidationInfo> validationInfos)
throws SpValidationException {
+
+ UpdateStaticPropertiesVisitor visitor = new
UpdateStaticPropertiesVisitor(target);
+ target.getStaticProperties().forEach(t -> t.accept(visitor));
+ }
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/UpdateStaticPropertiesVisitor.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/UpdateStaticPropertiesVisitor.java
new file mode 100644
index 0000000..18cb7a2
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/pipeline/UpdateStaticPropertiesVisitor.java
@@ -0,0 +1,108 @@
+/*
+ * 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.manager.matching.v2.pipeline;
+
+import
org.apache.streampipes.manager.matching.mapping.AbstractRequirementsSelectorGenerator;
+import
org.apache.streampipes.manager.matching.mapping.RequirementsSelectorGeneratorFactory;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.staticproperty.*;
+
+import java.util.List;
+
+public class UpdateStaticPropertiesVisitor extends
DefaultStaticPropertyVisitor {
+
+ private final List<SpDataStream> inputStreams;
+ private final InvocableStreamPipesEntity pipelineElement;
+
+ public UpdateStaticPropertiesVisitor(InvocableStreamPipesEntity
pipelineElement) {
+ this.inputStreams = pipelineElement.getInputStreams();
+ this.pipelineElement = pipelineElement;
+ }
+
+ @Override
+ public void visit(AnyStaticProperty property) {
+ // Do nothing
+ }
+
+ @Override
+ public void visit(CodeInputStaticProperty codeInputStaticProperty) {
+ // Do nothing
+ }
+
+ @Override
+ public void visit(ColorPickerStaticProperty colorPickerStaticProperty) {
+ // Do nothing
+ }
+
+ @Override
+ public void visit(DomainStaticProperty domainStaticProperty) {
+ // Do nothing
+ }
+
+ @Override
+ public void visit(FileStaticProperty fileStaticProperty) {
+ // Do nothing
+ }
+
+ @Override
+ public void visit(FreeTextStaticProperty freeTextStaticProperty) {
+ // Do nothing
+ }
+
+ @Override
+ public void visit(MappingPropertyNary mappingPropertyNary) {
+ updateMappingProperty(mappingPropertyNary);
+ }
+
+ @Override
+ public void visit(MappingPropertyUnary mappingPropertyUnary) {
+ updateMappingProperty(mappingPropertyUnary);
+ }
+
+ @Override
+ public void visit(MatchingStaticProperty matchingStaticProperty) {
+ // compute
+ }
+
+ @Override
+ public void visit(OneOfStaticProperty oneOfStaticProperty) {
+ // do nothing
+ }
+
+ @Override
+ public void visit(SecretStaticProperty secretStaticProperty) {
+ // do nothing
+ }
+
+ @Override
+ public void visit(RemoteOneOfStaticProperty remoteOneOfStaticProperty) {
+
+ }
+
+ private void updateMappingProperty(MappingProperty mappingProperty) {
+ AbstractRequirementsSelectorGenerator generator =
RequirementsSelectorGeneratorFactory
+ .getRequirementsSelector(
+ mappingProperty,
+ inputStreams,
+ pipelineElement
+ );
+ mappingProperty.setMapsFromOptions(generator.generateSelectors());
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
index 59ac301..1afd5cf 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
@@ -25,7 +25,7 @@ import
org.apache.streampipes.manager.endpoint.EndpointItemFetcher;
import org.apache.streampipes.manager.execution.http.PipelineExecutor;
import org.apache.streampipes.manager.execution.http.PipelineStorageService;
import org.apache.streampipes.manager.matching.DataSetGroundingSelector;
-import org.apache.streampipes.manager.matching.PipelineVerificationHandler;
+import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
import org.apache.streampipes.manager.recommender.ElementRecommender;
import org.apache.streampipes.manager.remote.ContainerProvidedOptionsHandler;
import
org.apache.streampipes.manager.runtime.PipelineElementRuntimeInfoFetcher;
@@ -78,13 +78,14 @@ public class Operations {
*/
public static PipelineModificationMessage validatePipeline(Pipeline
pipeline, boolean isPartial, String username)
throws Exception {
- PipelineVerificationHandler validator = new PipelineVerificationHandler(
- pipeline);
- return validator
- .validateConnection()
- .computeMappingProperties()
- .storeConnection()
- .getPipelineModificationMessage();
+ return new PipelineVerificationHandlerV2(pipeline).verifyPipeline();
+// PipelineVerificationHandler validator = new PipelineVerificationHandler(
+// pipeline);
+// return validator
+// .validateConnection()
+// .computeMappingProperties()
+// .storeConnection()
+// .getPipelineModificationMessage();
}
public static DataSetModificationMessage updateDataSet(SpDataSet dataSet) {
diff --git a/ui/package.json b/ui/package.json
index da51b0d..1f9fa78 100644
--- a/ui/package.json
+++ b/ui/package.json
@@ -54,7 +54,7 @@
"angular-plotly.js": "^4.0.0",
"@circlon/angular-tree-component": "11.0.4",
"angular-ui-tree": "2.9.0",
- "angular2-uuid": "1.1.1",
+ "angular2-uuid": "^1.1.1",
"bootstrap": "3.3.4",
"calendar-heatmap-graph": "0.0.14",
"console-browserify": "^1.2.0",
diff --git a/ui/src/app/core-model/gen/streampipes-model.ts
b/ui/src/app/core-model/gen/streampipes-model.ts
index 1f2b259..2f6062a 100644
--- a/ui/src/app/core-model/gen/streampipes-model.ts
+++ b/ui/src/app/core-model/gen/streampipes-model.ts
@@ -18,7 +18,7 @@
/* tslint:disable */
/* eslint-disable */
// @ts-nocheck
-// Generated using typescript-generator version 2.27.744 on 2021-11-08
19:04:54.
+// Generated using typescript-generator version 2.27.744 on 2022-01-05
10:45:36.
export class AbstractStreamPipesEntity {
"@class": "org.apache.streampipes.model.base.AbstractStreamPipesEntity" |
"org.apache.streampipes.model.base.NamedStreamPipesEntity" |
"org.apache.streampipes.model.connect.adapter.AdapterDescription" |
"org.apache.streampipes.model.connect.adapter.AdapterSetDescription" |
"org.apache.streampipes.model.connect.adapter.GenericAdapterSetDescription" |
"org.apache.streampipes.model.connect.adapter.SpecificAdapterSetDescription" |
"org.apache.streampipes.model.connect.adapter.AdapterStream [...]
@@ -151,8 +151,8 @@ export class NamedStreamPipesEntity extends
AbstractStreamPipesEntity {
instance.applicationLinks =
__getCopyArrayFn(ApplicationLink.fromData)(data.applicationLinks);
instance.internallyManaged = data.internallyManaged;
instance.connectedTo =
__getCopyArrayFn(__identity<string>())(data.connectedTo);
- instance.uri = data.uri;
instance.dom = data.dom;
+ instance.uri = data.uri;
instance._rev = data._rev;
return instance;
}
@@ -1179,6 +1179,21 @@ export class DomainStaticProperty extends StaticProperty
{
}
}
+export class EdgeValidationStatus {
+ notifications: Notification[];
+ validationStatusType: EdgeValidationStatusType;
+
+ static fromData(data: EdgeValidationStatus, target?: EdgeValidationStatus):
EdgeValidationStatus {
+ if (!data) {
+ return data;
+ }
+ const instance = target || new EdgeValidationStatus();
+ instance.validationStatusType = data.validationStatusType;
+ instance.notifications =
__getCopyArrayFn(Notification.fromData)(data.notifications);
+ return instance;
+ }
+}
+
export class ElementComposition {
description: string;
name: string;
@@ -1647,9 +1662,9 @@ export class GenericAdapterSetDescription extends
AdapterSetDescription implemen
}
const instance = target || new GenericAdapterSetDescription();
super.fromData(data, instance);
- instance.eventSchema = EventSchema.fromData(data.eventSchema);
instance.protocolDescription =
ProtocolDescription.fromData(data.protocolDescription);
instance.formatDescription =
FormatDescription.fromData(data.formatDescription);
+ instance.eventSchema = EventSchema.fromData(data.eventSchema);
return instance;
}
}
@@ -1666,9 +1681,9 @@ export class GenericAdapterStreamDescription extends
AdapterStreamDescription im
}
const instance = target || new GenericAdapterStreamDescription();
super.fromData(data, instance);
- instance.eventSchema = EventSchema.fromData(data.eventSchema);
instance.protocolDescription =
ProtocolDescription.fromData(data.protocolDescription);
instance.formatDescription =
FormatDescription.fromData(data.formatDescription);
+ instance.eventSchema = EventSchema.fromData(data.eventSchema);
return instance;
}
}
@@ -2179,6 +2194,23 @@ export class PipelineCategory {
}
}
+export class PipelineEdgeValidation {
+ sourceId: string;
+ status: EdgeValidationStatus;
+ targetId: string;
+
+ static fromData(data: PipelineEdgeValidation, target?:
PipelineEdgeValidation): PipelineEdgeValidation {
+ if (!data) {
+ return data;
+ }
+ const instance = target || new PipelineEdgeValidation();
+ instance.sourceId = data.sourceId;
+ instance.targetId = data.targetId;
+ instance.status = EdgeValidationStatus.fromData(data.status);
+ return instance;
+ }
+}
+
export class PipelineElementMetadata {
customName: string;
position: CanvasPosition;
@@ -2314,12 +2346,29 @@ export class PipelineElementTemplateConfig {
}
}
+export class PipelineElementValidationInfo {
+ level: ValidationInfoLevel;
+ message: string;
+
+ static fromData(data: PipelineElementValidationInfo, target?:
PipelineElementValidationInfo): PipelineElementValidationInfo {
+ if (!data) {
+ return data;
+ }
+ const instance = target || new PipelineElementValidationInfo();
+ instance.level = data.level;
+ instance.message = data.message;
+ return instance;
+ }
+}
+
export class PipelineModification {
domId: string;
elementId: string;
inputStreams: SpDataStreamUnion[];
outputStrategies: OutputStrategyUnion[];
+ pipelineElementValid: boolean;
staticProperties: StaticPropertyUnion[];
+ validationInfos: PipelineElementValidationInfo[];
static fromData(data: PipelineModification, target?: PipelineModification):
PipelineModification {
if (!data) {
@@ -2328,6 +2377,8 @@ export class PipelineModification {
const instance = target || new PipelineModification();
instance.domId = data.domId;
instance.elementId = data.elementId;
+ instance.pipelineElementValid = data.pipelineElementValid;
+ instance.validationInfos =
__getCopyArrayFn(PipelineElementValidationInfo.fromData)(data.validationInfos);
instance.staticProperties =
__getCopyArrayFn(StaticProperty.fromDataUnion)(data.staticProperties);
instance.outputStrategies =
__getCopyArrayFn(OutputStrategy.fromDataUnion)(data.outputStrategies);
instance.inputStreams =
__getCopyArrayFn(SpDataStream.fromDataUnion)(data.inputStreams);
@@ -2336,7 +2387,9 @@ export class PipelineModification {
}
export class PipelineModificationMessage extends Message {
+ edgeValidations: PipelineEdgeValidation[];
pipelineModifications: PipelineModification[];
+ pipelineValid: boolean;
static fromData(data: PipelineModificationMessage, target?:
PipelineModificationMessage): PipelineModificationMessage {
if (!data) {
@@ -2345,6 +2398,8 @@ export class PipelineModificationMessage extends Message {
const instance = target || new PipelineModificationMessage();
super.fromData(data, instance);
instance.pipelineModifications =
__getCopyArrayFn(PipelineModification.fromData)(data.pipelineModifications);
+ instance.pipelineValid = data.pipelineValid;
+ instance.edgeValidations =
__getCopyArrayFn(PipelineEdgeValidation.fromData)(data.edgeValidations);
return instance;
}
}
@@ -2437,9 +2492,9 @@ export class PipelineTemplateDescription extends
NamedStreamPipesEntity {
const instance = target || new PipelineTemplateDescription();
super.fromData(data, instance);
instance.boundTo =
__getCopyArrayFn(BoundPipelineElement.fromData)(data.boundTo);
- instance.pipelineTemplateDescription = data.pipelineTemplateDescription;
instance.pipelineTemplateId = data.pipelineTemplateId;
instance.pipelineTemplateName = data.pipelineTemplateName;
+ instance.pipelineTemplateDescription = data.pipelineTemplateDescription;
return instance;
}
}
@@ -2806,8 +2861,8 @@ export class SpDataSet extends SpDataStream {
instance.datasetInvocationId = data.datasetInvocationId;
instance.correspondingPipeline = data.correspondingPipeline;
instance.selectedEndpointUrl = data.selectedEndpointUrl;
- instance.actualTopicName = data.actualTopicName;
instance.brokerHostname = data.brokerHostname;
+ instance.actualTopicName = data.actualTopicName;
return instance;
}
}
@@ -3110,6 +3165,8 @@ export type AdapterSetDescriptionUnion =
GenericAdapterSetDescription | Specific
export type AdapterStreamDescriptionUnion = SpecificAdapterStreamDescription;
+export type EdgeValidationStatusType = "COMPLETE" | "INCOMPLETE" | "INVALID";
+
export type EventPropertyQualityDefinitionUnion = Accuracy | MeasurementRange
| Precision | Resolution;
export type EventPropertyUnion = EventPropertyList | EventPropertyNested |
EventPropertyPrimitive;
@@ -3142,6 +3199,8 @@ export type TransformationRuleDescriptionUnion =
AddTimestampRuleDescription | A
export type TransportProtocolUnion = JmsTransportProtocol |
KafkaTransportProtocol | MqttTransportProtocol;
+export type ValidationInfoLevel = "INFO" | "ERROR";
+
export type ValueSpecificationUnion = QuantitativeValue | Enumeration;
export type ValueTransformationRuleDescriptionUnion =
AddTimestampRuleDescription | AddValueTransformationRuleDescription |
TimestampTranfsformationRuleDescription | UnitTransformRuleDescription |
CorrectionValueTransformationRuleDescription;
diff --git
a/ui/src/app/editor/components/pipeline-element-options/pipeline-element-options.component.html
b/ui/src/app/editor/components/pipeline-element-options/pipeline-element-options.component.html
index cb5f15c..adb8d5f 100644
---
a/ui/src/app/editor/components/pipeline-element-options/pipeline-element-options.component.html
+++
b/ui/src/app/editor/components/pipeline-element-options/pipeline-element-options.component.html
@@ -28,19 +28,19 @@
*ngIf="pipelineElement.type=='stream' && isWildcardTopic()"
style="z-index:10">
<button mat-button mat-icon-button matTooltip="Configure Element"
[matTooltipPosition]="'above'"
- (click)="openCustomizeStreamDialog()"
[disabled]="!isRootElement()">
+ (click)="openCustomizeStreamDialog()">
<i class="material-icons">settings</i>
</button>
</span>
<span class="options-button delete-button" style="z-index:10">
<button mat-button mat-icon-button matTooltip="Delete Element"
[matTooltipPosition]="'above'"
- (click)="removeElement(pipelineElement)"
[disabled]="!isRootElement()">
+ (click)="removeElement(pipelineElement)">
<i class="material-icons">clear</i>
</button>
</span>
<span class="options-button possible-button"
*ngIf="pipelineElement.type!='action'" style="z-index:10">
<button mat-button mat-icon-button matTooltip="Compatible
Elements" [matTooltipPosition]="'below'"
- [disabled]="!possibleElements || possibleElements.length
== 0 || !isConfigured()"
+ [disabled]="!possibleElements || possibleElements.length
== 0"
(click)="openPossibleElementsDialog()"
[attr.data-cy]="'sp-possible-elements-' +
pipelineElement.payload.name.toLowerCase().replaceAll(' ', '_')"_>
<i class="material-icons">account_tree</i>
@@ -51,7 +51,7 @@
style="z-index:10">
<button mat-button mat-icon-button matTooltip="Recommended
Elements" [matTooltipPosition]="'below'"
(click)="showRecommendations($event)"
- [disabled]="!recommendationsAvailable || !isConfigured()">
+ [disabled]="!recommendationsAvailable">
<i class="material-icons">add</i>
</button>
</span>
diff --git a/ui/src/app/editor/components/pipeline/pipeline.component.ts
b/ui/src/app/editor/components/pipeline/pipeline.component.ts
index ee4a47e..62a5eca 100644
--- a/ui/src/app/editor/components/pipeline/pipeline.component.ts
+++ b/ui/src/app/editor/components/pipeline/pipeline.component.ts
@@ -26,9 +26,9 @@ import { InvocablePipelineElementUnion,
PipelineElementConfig, PipelineElementUn
import {
CustomOutputStrategy,
DataProcessorInvocation,
- DataSinkInvocation,
+ DataSinkInvocation, Notification,
Pipeline,
- PipelineCanvasMetadata,
+ PipelineCanvasMetadata, PipelineEdgeValidation, PipelineModificationMessage,
PipelinePreviewModel,
SpDataSet,
SpDataStream
@@ -110,7 +110,7 @@ export class PipelineComponent implements OnInit, OnDestroy
{
private pipelineValidationService: PipelineValidationService,
private dialogService: DialogService,
private dialog: MatDialog,
- private ngZone: NgZone, ) {
+ private ngZone: NgZone) {
this.plumbReady = false;
this.currentMouseOverElement = '';
this.currentPipelineModel = new Pipeline();
@@ -129,7 +129,7 @@ export class PipelineComponent implements OnInit, OnDestroy
{
setTimeout(() => {
this.ngZone.run(() => {
this.pipelineValid = this.pipelineValidationService
- .isValidPipeline(this.rawPipelineModel.filter(pe =>
!(pe.settings.disabled)), this.preview);
+ .isValidPipeline(this.rawPipelineModel.filter(pe =>
!(pe.settings.disabled)), this.preview);
});
});
}
@@ -150,16 +150,16 @@ export class PipelineComponent implements OnInit,
OnDestroy {
getElementCss(currentPipelineElementSettings) {
return 'position:absolute;'
- + (this.preview ? 'width:75px;' : 'width:110px;')
- + (this.preview ? 'height:75px;' : 'height:110px;')
- + 'left: ' + currentPipelineElementSettings.position.x + 'px; '
- + 'top: ' + currentPipelineElementSettings.position.y + 'px; ';
+ + (this.preview ? 'width:75px;' : 'width:110px;')
+ + (this.preview ? 'height:75px;' : 'height:110px;')
+ + 'left: ' + currentPipelineElementSettings.position.x + 'px; '
+ + 'top: ' + currentPipelineElementSettings.position.y + 'px; ';
}
getElementCssClasses(currentPipelineElement) {
return currentPipelineElement.type + ' ' +
(currentPipelineElement.settings.openCustomize ? '' : '')
- + currentPipelineElement.settings.connectable + ' '
- + currentPipelineElement.settings.displaySettings;
+ + currentPipelineElement.settings.connectable + ' '
+ + currentPipelineElement.settings.displaySettings;
}
isStreamInPipeline() {
@@ -199,12 +199,12 @@ export class PipelineComponent implements OnInit,
OnDestroy {
this.editorService.makePipelineAssemblyEmpty(false);
const newElementId = pipelineElement.elementId + ':' +
this.jsplumbService.makeId(5);
const pipelineElementConfig =
this.jsplumbService.createNewPipelineElementConfig(pipelineElement,
- this.pipelineEditorService.getCoordinates(ui,
this.currentZoomLevel),
- false,
- false,
- newElementId);
+ this.pipelineEditorService.getCoordinates(ui,
this.currentZoomLevel),
+ false,
+ false,
+ newElementId);
if ((this.isStreamInPipeline() && pipelineElementConfig.type ===
'set') ||
- this.isSetInPipeline() && pipelineElementConfig.type ===
'stream') {
+ this.isSetInPipeline() && pipelineElementConfig.type === 'stream')
{
this.showMixedStreamAlert();
} else {
this.rawPipelineModel.push(pipelineElementConfig);
@@ -256,16 +256,16 @@ export class PipelineComponent implements OnInit,
OnDestroy {
checkTopicModel(pipelineElementConfig: PipelineElementConfig) {
setTimeout(() => {
this.jsplumbService.dataStreamDropped(pipelineElementConfig.payload.dom,
- pipelineElementConfig.payload as SpDataStream,
- true,
- false);
+ pipelineElementConfig.payload as SpDataStream,
+ true,
+ false);
}, 10);
const streamDescription = pipelineElementConfig.payload as SpDataStream;
if (streamDescription
- .eventGrounding
- .transportProtocols[0]
- .topicDefinition['@class'] ===
'org.apache.streampipes.model.grounding.WildcardTopicDefinition') {
+ .eventGrounding
+ .transportProtocols[0]
+ .topicDefinition['@class'] ===
'org.apache.streampipes.model.grounding.WildcardTopicDefinition') {
// this.EditorDialogManager.showCustomizeStreamDialog(streamDescription);
}
}
@@ -330,8 +330,12 @@ export class PipelineComponent implements OnInit,
OnDestroy {
this.currentPipelineModel =
this.objectProvider.makePipeline(this.rawPipelineModel);
pe.settings.loadingStatus = true;
this.objectProvider.updatePipeline(this.currentPipelineModel)
- .subscribe(pipelineModificationMessage => {
- pe.settings.loadingStatus = false;
+ .subscribe(pipelineModificationMessage => {
+ pe.settings.loadingStatus = false;
+ const edgeValidations =
this.getTargetEdgeValidations(pipelineModificationMessage, info.target.id);
+ const currentConnectionValid = this.currentConnectionValid(pe,
edgeValidations);
+ this.processEdgeValidations(pipelineModificationMessage,
info.target.id);
+ if (currentConnectionValid) {
info.targetEndpoint.setType('token');
this.validatePipeline();
this.modifyPipeline(pipelineModificationMessage.pipelineModifications);
@@ -345,16 +349,17 @@ export class PipelineComponent implements OnInit,
OnDestroy {
this.announceConfiguredElement(pe);
}
}
- }, status => {
- pe.settings.loadingStatus = false;
+ } else {
this.JsplumbBridge.detach(info.connection);
- if (Array.isArray(status.error)) {
- const matchingResultMessage = (status.error as any[]).map(e =>
MatchingResultMessage.fromData(e as MatchingResultMessage));
- this.showMatchingErrorDialog(matchingResultMessage);
- } else {
- this.showErrorDialog(status.error.title,
status.error.description);
+ const invalidEdgeValidation = edgeValidations.find(e =>
e.sourceId === info.source.id);
+ console.log(invalidEdgeValidation);
+ if (invalidEdgeValidation) {
+
this.showMatchingErrorDialog(invalidEdgeValidation.status.notifications);
}
- });
+ }
+ }, status => {
+ this.showErrorDialog(status.error.title, status.error.description);
+ });
}
});
@@ -367,14 +372,34 @@ export class PipelineComponent implements OnInit,
OnDestroy {
}, 100);
}
+ currentConnectionValid(pe: PipelineElementConfig,
+ targetEdges: PipelineEdgeValidation[]) {
+ const entity = pe.payload as InvocablePipelineElementUnion;
+ return entity.streamRequirements.length === targetEdges.length &&
targetEdges.every(e => e.status.validationStatusType === 'COMPLETE');
+ }
+
+ getTargetEdgeValidations(pipelineModificationMessage:
PipelineModificationMessage,
+ targetDomId: string): PipelineEdgeValidation[] {
+ const edgeValidations = pipelineModificationMessage.edgeValidations;
+ return edgeValidations.filter(ev => ev.targetId === targetDomId);
+ }
+
+ processEdgeValidations(pipelineModificationMessage:
PipelineModificationMessage,
+ targetDomId: string) {
+
+ }
+
modifyPipeline(pipelineModifications) {
if (pipelineModifications) {
+ console.log(pipelineModifications);
pipelineModifications.forEach(modification => {
const id = modification.domId;
if (id !== 'undefined') {
const pe = this.objectProvider.findElement(id,
this.rawPipelineModel);
(pe.payload as InvocablePipelineElementUnion).staticProperties =
modification.staticProperties;
- (pe.payload as DataProcessorInvocation).outputStrategies =
modification.outputStrategies;
+ if (pe instanceof DataProcessorInvocation) {
+ (pe.payload as DataProcessorInvocation).outputStrategies =
modification.pipelineElement.outputStrategies;
+ }
(pe.payload as InvocablePipelineElementUnion).inputStreams =
modification.inputStreams;
}
});
@@ -419,12 +444,12 @@ export class PipelineComponent implements OnInit,
OnDestroy {
});
}
- showMatchingErrorDialog(matchingResultMessage: MatchingResultMessage[]) {
+ showMatchingErrorDialog(notifications: Notification[]) {
this.dialogService.open(MatchingErrorComponent, {
panelType: PanelType.STANDARD_PANEL,
title: 'Invalid Connection',
data: {
- 'matchingResultMessage': matchingResultMessage
+ 'notifications': notifications
}
});
}
diff --git
a/ui/src/app/editor/dialog/matching-error/matching-error.component.html
b/ui/src/app/editor/dialog/matching-error/matching-error.component.html
index 5303852..e0a7fa8 100644
--- a/ui/src/app/editor/dialog/matching-error/matching-error.component.html
+++ b/ui/src/app/editor/dialog/matching-error/matching-error.component.html
@@ -26,17 +26,17 @@
<div *ngIf="statusDetailsVisible">Hide Details</div>
</button>
<div *ngIf="statusDetailsVisible">
- <div *ngFor="let entry of matchingResultMessage">
+ <div *ngFor="let entry of notifications">
<div fxFlex="100" class="md-whiteframe-z1"
style="margin-bottom:10px;">
<div fxFlex fxLayout="column" class="md-padding">
- <div>
- <i class="material-icons"
*ngIf="entry.matchingResultMessage">done</i>
- <i class="material-icons"
*ngIf="entry.matchingSuccessful">warning</i>
- <b>{{entry.title}} </b>
- </div>
+<!-- <div>-->
+<!-- <i class="material-icons"
*ngIf="entry.matchingResultMessage">done</i>-->
+<!-- <i class="material-icons"
*ngIf="entry.matchingSuccessful">warning</i>-->
+<!-- <b>{{entry.title}} </b>-->
+<!-- </div>-->
<div fxFlex="column">
- {{entry.requirementSubject}}<br/>
- {{entry.reasonText}}
+ {{entry.title}}<br/>
+ {{entry.description}}
</div>
</div>
</div>
@@ -50,4 +50,4 @@
Close
</button>
</div>
-</div>
\ No newline at end of file
+</div>
diff --git
a/ui/src/app/editor/dialog/matching-error/matching-error.component.ts
b/ui/src/app/editor/dialog/matching-error/matching-error.component.ts
index ac21a03..e9b89f9 100644
--- a/ui/src/app/editor/dialog/matching-error/matching-error.component.ts
+++ b/ui/src/app/editor/dialog/matching-error/matching-error.component.ts
@@ -16,9 +16,9 @@
*
*/
-import {Component, Input} from "@angular/core";
-import {DialogRef} from "../../../core-ui/dialog/base-dialog/dialog-ref";
-import {MatchingResultMessage} from
"../../../core-model/gen/streampipes-model-client";
+import { Component, Input } from '@angular/core';
+import { DialogRef } from '../../../core-ui/dialog/base-dialog/dialog-ref';
+import { Notification } from '../../../core-model/gen/streampipes-model';
@Component({
selector: 'matching-error',
@@ -28,20 +28,20 @@ import {MatchingResultMessage} from
"../../../core-model/gen/streampipes-model-c
export class MatchingErrorComponent {
@Input()
- matchingResultMessage: MatchingResultMessage[];
+ notifications: Notification[];
msg: any;
statusDetailsVisible: any;
- constructor(private DialogRef: DialogRef<MatchingErrorComponent>) {
+ constructor(private dialogRef: DialogRef<MatchingErrorComponent>) {
}
close() {
- this.DialogRef.close();
- };
+ this.dialogRef.close();
+ }
toggleStatusDetailsVisible() {
this.statusDetailsVisible = !(this.statusDetailsVisible);
}
-}
\ No newline at end of file
+}
diff --git a/ui/src/app/editor/services/jsplumb-bridge.service.ts
b/ui/src/app/editor/services/jsplumb-bridge.service.ts
index 69883a4..cc607c0 100644
--- a/ui/src/app/editor/services/jsplumb-bridge.service.ts
+++ b/ui/src/app/editor/services/jsplumb-bridge.service.ts
@@ -53,9 +53,8 @@ export class JsplumbBridge {
}
deleteEveryEndpoint() {
- // TODO
- // this.jsPlumbInstance.destroy();
- // this.jsPlumbInstance.deleteEveryEndpoint();
+ this.jsPlumbInstance.selectEndpoints().deleteAll();
+ this.jsPlumbInstance.deleteEveryConnection();
}
setContainer(container) {
