[incubator-streampipes] branch edge-extensions updated: Adapt local graph description after reconfiguration

[gomm]

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


The following commit(s) were added to refs/heads/edge-extensions by this push:
     new 37dfa9e  Adapt local graph description after reconfiguration
37dfa9e is described below

commit 37dfa9e112fcf883702ee7fa9649545b4a05de21
Author: Daniel Gomm <daniel.g...@outlook.de>
AuthorDate: Sun Apr 11 14:12:28 2021 +0200

    Adapt local graph description after reconfiguration
---
 .../management/pe/InvocableElementManager.java     | 26 +++++++++++++++++-----
 1 file changed, 20 insertions(+), 6 deletions(-)

diff --git 
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java
 
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java
index 2087587..5d1f833 100644
--- 
a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java
+++ 
b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java
@@ -34,6 +34,7 @@ import org.apache.streampipes.model.grounding.*;
 import org.apache.streampipes.model.node.NodeInfoDescription;
 import 
org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity;
 import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.StaticProperty;
 import 
org.apache.streampipes.node.controller.container.config.NodeControllerConfig;
 import 
org.apache.streampipes.node.controller.container.management.node.NodeManager;
 import org.apache.streampipes.serializers.json.JacksonSerializer;
@@ -46,10 +47,7 @@ import java.net.MalformedURLException;
 import java.net.Socket;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 public class InvocableElementManager implements PipelineElementLifeCycle {
 
@@ -158,13 +156,29 @@ public class InvocableElementManager implements 
PipelineElementLifeCycle {
         pub.publish(reconfigurationEvent);
         pub.disconnect();
 
-        // TODO: update RunningInvocableInstances with new reconfiguration 
value
-        // remove, then add
+        adaptPipelineDescription(graph, reconfigurationEntity);
 
         response.setSuccess(true);
         return response;
     }
 
+    private void adaptPipelineDescription(InvocableStreamPipesEntity graph, 
PipelineElementReconfigurationEntity reconfigurationEntity){
+        List<StaticProperty> staticProperties = new ArrayList<>();
+        graph.getStaticProperties().forEach(sp -> {
+            int ind = graph.getStaticProperties().indexOf(sp);
+            staticProperties.add(ind, sp);
+            reconfigurationEntity.getReconfiguredStaticProperties().forEach(rp 
-> {
+                if(sp.getInternalName().equals(rp.getInternalName())){
+                    staticProperties.remove(ind);
+                    staticProperties.add(ind, rp);
+                }
+            });
+        });
+        graph.setStaticProperties(staticProperties);
+        
RunningInvocableInstances.INSTANCE.remove(reconfigurationEntity.getDeploymentRunningInstanceId());
+        
RunningInvocableInstances.INSTANCE.add(reconfigurationEntity.getDeploymentRunningInstanceId(),
 graph);
+    }
+
     private boolean post(String endpoint, String body, Response response) {
         try {
             org.apache.http.client.fluent.Response httpResp = Request