This is an automated email from the ASF dual-hosted git repository. gomm pushed a commit to branch edge-extensions in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit bcc1134226b39eb87437ebcf01ccfd92c21c716e Author: Daniel Gomm <[email protected]> AuthorDate: Fri Feb 19 18:44:07 2021 +0100 Fix issues from merge with dev --- .../controller/container/management/node/NodeManager.java | 1 - .../manager/matching/InvocationGraphBuilder.java | 14 ++++++-------- .../org/apache/streampipes/rest/impl/PipelineResource.java | 6 +----- 3 files changed, 7 insertions(+), 14 deletions(-) diff --git a/streampipes-node-controller-container/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeManager.java b/streampipes-node-controller-container/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeManager.java index a597a03..d9ec40d 100644 --- a/streampipes-node-controller-container/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeManager.java +++ b/streampipes-node-controller-container/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeManager.java @@ -159,7 +159,6 @@ public class NodeManager { + NodeControllerConfig.INSTANCE.getBackendHost() + COLON + NodeControllerConfig.INSTANCE.getBackendPort() - + SLASH + BACKEND_BASE_ROUTE + NODE_REGISTRATION_ROUTE; } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/InvocationGraphBuilder.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/InvocationGraphBuilder.java index e5d9ae5..f7cae4f 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/InvocationGraphBuilder.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/InvocationGraphBuilder.java @@ -175,23 +175,21 @@ public class InvocationGraphBuilder { } private void connectSourceToTarget(DataProcessorInvocation source, InvocableStreamPipesEntity target, EventGrounding inputGrounding, boolean edgeToEdgeRelay){ + //connect a source entity with a target entity through an appropriate relay EventGrounding relayGrounding = edgeToEdgeRelay ? generateRelayGrounding(inputGrounding, target, true) : generateRelayGrounding(inputGrounding, false); - if (!eventRelayExists(source, target)) { - modifyTargetInputStream(source, target, relayGrounding, edgeToEdgeRelay); - // add initial relay grounding to source processor - source.addOutputStreamRelay(new SpDataStreamRelay(relayGrounding)); - } else { + if (eventRelayExists(source, target)) { removeExistingStreamRelay(source, target); - modifyTargetInputStream(source, target, relayGrounding, edgeToEdgeRelay); - source.addOutputStreamRelay(new SpDataStreamRelay(relayGrounding)); } + modifyTargetInputStream(source, target, relayGrounding, edgeToEdgeRelay); + source.addOutputStreamRelay(new SpDataStreamRelay(relayGrounding)); } private void modifyTargetInputStream(NamedStreamPipesEntity s, InvocableStreamPipesEntity t, EventGrounding grounding, boolean edgeToEdgeRelay) { + //modify the grounding of the InputStream of the target entity so that it matches the relay grounding SpDataStream inputStream = t.getInputStreams() - .get(getIndex(s.getDOM(),t)); //Is this correct? + .get(getIndex(s.getDOM(),t)); if(edgeToEdgeRelay){ inputStream.setEventGrounding(grounding); }else{ diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java index 74b71cf..28c0dee 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java @@ -274,11 +274,7 @@ public class PipelineResource extends AbstractAuthGuardedRestResource { @PathParam("pipelineId") String pipelineId, Pipeline pipelineNew) { try { - Pipeline pipelineOld = getPipelineStorage() - .getPipeline(pipelineId); - PipelineOperationStatus status = Operations.updatePipelineDeploymentPartial(pipelineOld, pipelineNew, - true,true, true); - Operations.overwritePipeline(pipelineNew); + PipelineOperationStatus status = Operations.migratePipelineProcessors(pipelineNew, true, true, true); return ok(status); } catch (Exception e) { e.printStackTrace();
