fix: Properly update output streams in functions

apache · Sep 17, 2024 · 3adaec7 · 3adaec7
1 parent 878e68d
commit 3adaec7
Showing 1 changed file with 25 additions and 20 deletions.
diff --git a/...n/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java b/...n/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java
@@ -62,27 +62,32 @@ public void registerDataStreams(Collection<IStreamPipesFunctionDeclarer> functio
     var client = new StreamPipesClientResolver().makeStreamPipesClientInstance();
     functions.forEach(function -> {
       function.getFunctionConfig().getOutputDataStreams().values().forEach(ds -> {
-        DeclarersSingleton.getInstance().add(new IStreamPipesDataStream() {
-          @Override
-          public IDataStreamConfiguration declareConfig() {
-            return DataStreamConfiguration.create(
-                () -> this,
-                ds
-            );
+        if (!DeclarersSingleton.getInstance().getDataStreams().containsKey(ds.getAppId())) {
+          DeclarersSingleton.getInstance().add(new IStreamPipesDataStream() {
+            @Override
+            public IDataStreamConfiguration declareConfig() {
+              return DataStreamConfiguration.create(
+                  () -> this,
+                  ds
+              );
+            }
+
+            @Override
+            public void executeStream() {
+
+            }
+
+            @Override
+            public boolean isExecutable() {
+              return false;
+            }
+          });
+          if (client.streams().get(ds.getElementId()).isEmpty()) {
+            client.streams().create(ds);
+          } else {
+            client.streams().update(ds);
           }
-
-          @Override
-          public void executeStream() {
-
-          }
-
-          @Override
-          public boolean isExecutable() {
-            return false;
-          }
-        });
-
-        client.streams().create(ds);
+        }
       });
     });
   }