Remove connect-container-worker module and move code to streampipes-c…

…ontainer (#569)
apache · Jan 1, 2023 · b7e7446 · b7e7446
1 parent dec2a31
commit b7e7446
Show file tree

Hide file tree

Showing 29 changed files with 64 additions and 334 deletions.
diff --git a/pom.xml b/pom.xml
@@ -1367,7 +1367,6 @@ IoT data streams.
 			<module>streampipes-connect</module>
 			<module>streampipes-connect-api</module>
 			<module>streampipes-connect-container-master</module>
-			<module>streampipes-connect-container-worker</module>
 			<module>streampipes-container</module>
 			<module>streampipes-container-extensions</module>
 			<module>streampipes-data-explorer</module>

diff --git a/streampipes-connect-container-worker/pom.xml b/streampipes-connect-container-worker/pom.xml
diff --git a/...n/java/org/apache/streampipes/connect/container/worker/init/ConnectWorkerTagProvider.java b/...n/java/org/apache/streampipes/connect/container/worker/init/ConnectWorkerTagProvider.java
diff --git a/streampipes-connect-container-worker/src/main/resources/banner.txt b/streampipes-connect-container-worker/src/main/resources/banner.txt
diff --git a/streampipes-connect/pom.xml b/streampipes-connect/pom.xml
@@ -40,11 +40,6 @@
             <artifactId>streampipes-connect-api</artifactId>
             <version>0.91.0-SNAPSHOT</version>
         </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-container</artifactId>
-            <version>0.91.0-SNAPSHOT</version>
-        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-dataformat-json</artifactId>

diff --git a/...s/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java b/...s/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java
@@ -17,7 +17,7 @@
  */
 package org.apache.streampipes.container.extensions;
 
-import org.apache.streampipes.connect.container.worker.init.ConnectWorkerRegistrationService;
+import org.apache.streampipes.container.extensions.connect.ConnectWorkerRegistrationService;
 import org.apache.streampipes.container.extensions.function.StreamPipesFunctionHandler;
 import org.apache.streampipes.container.init.DeclarersSingleton;
 import org.apache.streampipes.container.model.SpServiceDefinition;

diff --git a/...ensions/src/main/java/org/apache/streampipes/container/extensions/ServiceTagProvider.java b/...ensions/src/main/java/org/apache/streampipes/container/extensions/ServiceTagProvider.java
@@ -18,7 +18,8 @@
 
 package org.apache.streampipes.container.extensions;
 
-import org.apache.streampipes.connect.container.worker.init.ConnectWorkerTagProvider;
+import org.apache.streampipes.connect.api.IAdapter;
+import org.apache.streampipes.connect.api.IProtocol;
 import org.apache.streampipes.container.declarer.Declarer;
 import org.apache.streampipes.container.init.DeclarersSingleton;
 import org.apache.streampipes.container.util.ServiceDefinitionUtil;
@@ -48,6 +49,14 @@ private List<SpServiceTag> extractPipelineElementServiceTags() {
   }
 
   private List<SpServiceTag> extractAdapterServiceTags() {
-    return new ConnectWorkerTagProvider().extractServiceTags();
+    var tags = new ArrayList<SpServiceTag>();
+    Collection<IAdapter> adapters = DeclarersSingleton.getInstance().getAllAdapters();
+    Collection<IProtocol> protocols = DeclarersSingleton.getInstance().getAllProtocols();
+    tags.addAll(ServiceDefinitionUtil.extractAppIdsFromAdapters(adapters));
+    tags.addAll(ServiceDefinitionUtil.extractAppIdsFromProtocols(protocols));
+    tags.add(DefaultSpServiceTags.CONNECT_WORKER);
+
+    return tags;
+
   }
 }
diff --git a/...r/worker/management/MasterRestClient.java → ...extensions/connect/ConnectRestClient.java b/...r/worker/management/MasterRestClient.java → ...extensions/connect/ConnectRestClient.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.connect.container.worker.management;
+package org.apache.streampipes.container.extensions.connect;
 
 import org.apache.streampipes.client.StreamPipesClient;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
@@ -27,9 +27,9 @@
 
 import java.util.List;
 
-public class MasterRestClient {
+public class ConnectRestClient {
 
-  private static final Logger LOG = LoggerFactory.getLogger(MasterRestClient.class);
+  private static final Logger LOG = LoggerFactory.getLogger(ConnectRestClient.class);
 
   public static boolean register(List<AdapterDescription> allAvailableAdapters) {
 

diff --git a/...nit/ConnectWorkerRegistrationService.java → ...ect/ConnectWorkerRegistrationService.java b/...nit/ConnectWorkerRegistrationService.java → ...ect/ConnectWorkerRegistrationService.java
@@ -15,9 +15,9 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.connect.container.worker.init;
+package org.apache.streampipes.container.extensions.connect;
 
-import org.apache.streampipes.connect.container.worker.management.MasterRestClient;
+import org.apache.streampipes.container.connect.ConnectWorkerDescriptionProvider;
 import org.apache.streampipes.container.model.SpServiceDefinition;
 
 import org.slf4j.Logger;
@@ -32,7 +32,7 @@ public void registerWorker(SpServiceDefinition serviceDef) {
 
     while (!connected) {
 
-      connected = MasterRestClient.register(
+      connected = ConnectRestClient.register(
           new ConnectWorkerDescriptionProvider().getContainerDescription(serviceDef.getServiceGroup()));
 
       if (connected) {

diff --git a/streampipes-container/pom.xml b/streampipes-container/pom.xml
@@ -36,6 +36,16 @@
             <artifactId>streampipes-connect-api</artifactId>
             <version>0.91.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-connect</artifactId>
+            <version>0.91.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-client</artifactId>
+            <version>0.91.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-dataformat</artifactId>
@@ -69,11 +79,31 @@
         </dependency>
 
         <!-- Test dependencies -->
+        <dependency>
+            <groupId>com.github.tomakehurst</groupId>
+            <artifactId>wiremock</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-server</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.jayway.restassured</groupId>
+            <artifactId>rest-assured</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-module-junit4</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

diff --git a/.../container/worker/utils/AdapterUtils.java → ...pipes/container/connect/AdapterUtils.java b/.../container/worker/utils/AdapterUtils.java → ...pipes/container/connect/AdapterUtils.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.connect.container.worker.utils;
+package org.apache.streampipes.container.connect;
 
 import org.apache.streampipes.connect.adapter.model.generic.GenericAdapter;
 import org.apache.streampipes.connect.adapter.model.generic.GenericDataSetAdapter;

diff --git a/...r/management/AdapterWorkerManagement.java → ...iner/connect/AdapterWorkerManagement.java b/...r/management/AdapterWorkerManagement.java → ...iner/connect/AdapterWorkerManagement.java
@@ -16,12 +16,11 @@
  *
  */
 
-package org.apache.streampipes.connect.container.worker.management;
+package org.apache.streampipes.container.connect;
 
 import org.apache.streampipes.connect.RunningAdapterInstances;
 import org.apache.streampipes.connect.api.IAdapter;
 import org.apache.streampipes.connect.api.exception.AdapterException;
-import org.apache.streampipes.connect.container.worker.utils.AdapterUtils;
 import org.apache.streampipes.container.monitoring.SpMonitoringManager;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.connect.adapter.AdapterSetDescription;

diff --git a/...nit/ConnectWorkerDescriptionProvider.java → ...ect/ConnectWorkerDescriptionProvider.java b/...nit/ConnectWorkerDescriptionProvider.java → ...ect/ConnectWorkerDescriptionProvider.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.connect.container.worker.init;
+package org.apache.streampipes.container.connect;
 
 import org.apache.streampipes.connect.api.IAdapter;
 import org.apache.streampipes.connect.api.IProtocol;

diff --git a/...er/worker/management/GuessManagement.java → ...es/container/connect/GuessManagement.java b/...er/worker/management/GuessManagement.java → ...es/container/connect/GuessManagement.java
@@ -16,12 +16,11 @@
  *
  */
 
-package org.apache.streampipes.connect.container.worker.management;
+package org.apache.streampipes.container.connect;
 
 import org.apache.streampipes.connect.api.IAdapter;
 import org.apache.streampipes.connect.api.exception.AdapterException;
 import org.apache.streampipes.connect.api.exception.ParseException;
-import org.apache.streampipes.connect.container.worker.utils.AdapterUtils;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
 

diff --git a/...nagement/HttpServerAdapterManagement.java → .../connect/HttpServerAdapterManagement.java b/...nagement/HttpServerAdapterManagement.java → .../connect/HttpServerAdapterManagement.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.connect.container.worker.management;
+package org.apache.streampipes.container.connect;
 
 import org.apache.streampipes.messaging.InternalEventProcessor;
 

diff --git a/...r/worker/management/RuntimeResovable.java → ...s/container/connect/RuntimeResovable.java b/...r/worker/management/RuntimeResovable.java → ...s/container/connect/RuntimeResovable.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.connect.container.worker.management;
+package org.apache.streampipes.container.connect;
 
 import org.apache.streampipes.connect.adapter.AdapterRegistry;
 import org.apache.streampipes.connect.api.Connector;