refactor(#2938): replace ExtensionsModelSubmitter with StreamPipesExt…

…ensionsServiceBase (#2939) * refactor(#2938): Replace ExtensionsModelSubmitter with StreamPipesExtensionsServiceBase * refactor(#2938): Replace ExtensionsModelSubmitter in archetypes * refactor(#2938): Remove deprecated class ExtensionsModelSubmitter * refactor(#2938): Fix checkstyle
apache · Jun 18, 2024 · 974abf1 · 974abf1
1 parent 1a5073c
commit 974abf1
Show file tree

Hide file tree

Showing 10 changed files with 45 additions and 56 deletions.
diff --git a/...s-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java b/...s-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java
@@ -32,15 +32,15 @@
 import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
 import org.apache.streampipes.messaging.nats.SpNatsProtocolFactory;
 import org.apache.streampipes.messaging.pulsar.SpPulsarProtocolFactory;
-import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
+import org.apache.streampipes.service.extensions.StreamPipesExtensionsServiceBase;
 import org.apache.streampipes.wrapper.standalone.runtime.StandaloneStreamPipesRuntimeProvider;
 
 import ${package}.pe.${packageName}.${classNamePrefix}DataProcessor;
 import ${package}.pe.${packageName}.${classNamePrefix}DataSink;
 import ${package}.pe.${packageName}.${classNamePrefix}GenericAdapter;
 import ${package}.pe.${packageName}.${classNamePrefix}SpecificAdapter;
 
-public class Init extends ExtensionsModelSubmitter {
+public class Init extends StreamPipesExtensionsServiceBase {
 
   public static void main(String[] args) {
     new Init().init();

diff --git a/...hetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java b/...hetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java
@@ -36,9 +36,9 @@
 import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
 import org.apache.streampipes.messaging.pulsar.SpPulsarProtocolFactory;
 import org.apache.streampipes.messaging.nats.SpNatsProtocolFactory;
-import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
+import org.apache.streampipes.service.extensions.StreamPipesExtensionsServiceBase;
 
-public class Init extends ExtensionsModelSubmitter {
+public class Init extends StreamPipesExtensionsServiceBase {
 
   public static void main(String[] args) throws Exception {
 	  new Init().init();

diff --git a/...s-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java b/...s-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java
@@ -32,12 +32,12 @@
 import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
 import org.apache.streampipes.messaging.pulsar.SpPulsarProtocolFactory;
 import org.apache.streampipes.messaging.nats.SpNatsProtocolFactory;
-import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
+import org.apache.streampipes.service.extensions.StreamPipesExtensionsServiceBase;
 
 import ${package}.config.ConfigKeys;
 import ${package}.pe.sink.${packageName}.${classNamePrefix}Controller;
 
-public class Init extends ExtensionsModelSubmitter {
+public class Init extends StreamPipesExtensionsServiceBase {
 
   public static void main(String[] args) throws Exception {
     new Init().init();

diff --git a/...-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java b/...-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java
@@ -45,14 +45,14 @@
 import org.apache.streampipes.processors.filters.jvm.FilterExtensionModuleExport;
 import org.apache.streampipes.processors.siddhi.SiddhiFilterExtensionModuleExport;
 import org.apache.streampipes.processors.transformation.jvm.TransformationExtensionModuleExport;
-import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
+import org.apache.streampipes.service.extensions.StreamPipesExtensionsServiceBase;
 import org.apache.streampipes.sinks.brokers.jvm.BrokerSinksExtensionModuleExport;
 import org.apache.streampipes.sinks.databases.jvm.DatabaseSinksExtensionModuleExport;
 import org.apache.streampipes.sinks.internal.jvm.InternalSinksExtensionModuleExports;
 import org.apache.streampipes.sinks.notifications.jvm.NotificationsExtensionModuleExport;
 import org.apache.streampipes.wrapper.standalone.runtime.StandaloneStreamPipesRuntimeProvider;
 
-public class AllExtensionsIIoTInit extends ExtensionsModelSubmitter {
+public class AllExtensionsIIoTInit extends StreamPipesExtensionsServiceBase {
 
   public static void main(String[] args) {
     new AllExtensionsIIoTInit().init();

diff --git a/...ns-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java b/...ns-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
@@ -48,15 +48,15 @@
 import org.apache.streampipes.processors.siddhi.SiddhiFilterExtensionModuleExport;
 import org.apache.streampipes.processors.textmining.jvm.TextMiningExtensionModuleExport;
 import org.apache.streampipes.processors.transformation.jvm.TransformationExtensionModuleExport;
-import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
+import org.apache.streampipes.service.extensions.StreamPipesExtensionsServiceBase;
 import org.apache.streampipes.sinks.brokers.jvm.BrokerSinksExtensionModuleExport;
 import org.apache.streampipes.sinks.databases.jvm.DatabaseSinksExtensionModuleExport;
 import org.apache.streampipes.sinks.internal.jvm.InternalSinksExtensionModuleExports;
 import org.apache.streampipes.sinks.notifications.jvm.NotificationsExtensionModuleExport;
 import org.apache.streampipes.wrapper.standalone.runtime.StandaloneStreamPipesRuntimeProvider;
 
 
-public class AllExtensionsInit extends ExtensionsModelSubmitter {
+public class AllExtensionsInit extends StreamPipesExtensionsServiceBase {
 
   public static void main(String[] args) {
     new AllExtensionsInit().init();

diff --git a/...c/main/java/org/apache/streampipes/extensions/iiot/minimal/ExtensionsIIoTMinimalInit.java b/...c/main/java/org/apache/streampipes/extensions/iiot/minimal/ExtensionsIIoTMinimalInit.java
@@ -37,14 +37,14 @@
 import org.apache.streampipes.processors.filters.jvm.FilterExtensionModuleExport;
 import org.apache.streampipes.processors.siddhi.SiddhiFilterExtensionModuleExport;
 import org.apache.streampipes.processors.transformation.jvm.TransformationExtensionModuleExport;
-import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
+import org.apache.streampipes.service.extensions.StreamPipesExtensionsServiceBase;
 import org.apache.streampipes.sinks.brokers.jvm.BrokerSinksExtensionModuleExport;
 import org.apache.streampipes.sinks.databases.jvm.DatabaseSinksExtensionModuleExport;
 import org.apache.streampipes.sinks.internal.jvm.InternalSinksExtensionModuleExports;
 import org.apache.streampipes.sinks.notifications.jvm.NotificationsExtensionModuleExport;
 import org.apache.streampipes.wrapper.standalone.runtime.StandaloneStreamPipesRuntimeProvider;
 
-public class ExtensionsIIoTMinimalInit extends ExtensionsModelSubmitter {
+public class ExtensionsIIoTMinimalInit extends StreamPipesExtensionsServiceBase {
 
   public static void main(String[] args) {
     new ExtensionsIIoTMinimalInit().init();

diff --git a/...tal-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java b/...tal-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java
@@ -47,11 +47,11 @@
 import org.apache.streampipes.pe.flink.processor.urldereferencing.UrlDereferencingController;
 import org.apache.streampipes.pe.flink.processor.wordcount.WordCountController;
 import org.apache.streampipes.pe.flink.sink.elasticsearch.ElasticSearchController;
-import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
+import org.apache.streampipes.service.extensions.StreamPipesExtensionsServiceBase;
 import org.apache.streampipes.wrapper.flink.FlinkRuntimeProvider;
 
 
-public class AllFlinkPipelineElementsInit extends ExtensionsModelSubmitter {
+public class AllFlinkPipelineElementsInit extends StreamPipesExtensionsServiceBase {
 
   public static void main(String[] args) {
     new AllFlinkPipelineElementsInit().init();

diff --git a/streampipes-maven-plugin/README.md b/streampipes-maven-plugin/README.md
@@ -31,7 +31,7 @@ The streampipes-maven-plugin can either be started from the command line or embe
 
 ### Prerequisites
 
-The plugin must be started from a module which contains an `Init` class which inherits `ExtensionsModelSubmitter`.
+The plugin must be started from a module which contains an `Init` class which inherits `StreamPipesExtensionsServiceBase`.
 By default, the goal runs in the `package` phase.
 
 ### Command line

diff --git a/...pes-maven-plugin/src/main/java/org/apache/streampipes/smp/extractor/ExtensionsFinder.java b/...pes-maven-plugin/src/main/java/org/apache/streampipes/smp/extractor/ExtensionsFinder.java
@@ -23,7 +23,7 @@
 import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
 import org.apache.streampipes.extensions.api.pe.config.IPipelineElementConfiguration;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinition;
-import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
+import org.apache.streampipes.service.extensions.StreamPipesExtensionsServiceBase;
 import org.apache.streampipes.smp.constants.PeType;
 import org.apache.streampipes.smp.model.AssetModel;
 
@@ -39,17 +39,23 @@ public class ExtensionsFinder {
   private final ClassLoader loader;
   private final String initClass;
 
-  public ExtensionsFinder(ClassLoader loader,
-                          String initClass) {
+  public ExtensionsFinder(
+      ClassLoader loader,
+      String initClass
+  ) {
     this.loader = loader;
     this.initClass = initClass;
   }
 
   public List<AssetModel> findExtensions()
       throws MalformedURLException, DependencyResolutionRequiredException, ClassNotFoundException,
-      InstantiationException, IllegalAccessException {
+             InstantiationException, IllegalAccessException {
     var extensions = new ArrayList<AssetModel>();
-    var serviceDef = ((ExtensionsModelSubmitter) loader.loadClass(initClass).newInstance()).provideServiceDefinition();
+    var serviceDef = (
+        (StreamPipesExtensionsServiceBase) loader
+            .loadClass(initClass)
+            .newInstance()
+    ).provideServiceDefinition();
 
     extensions.addAll(findAdapters(serviceDef));
     extensions.addAll(findPipelineElements(serviceDef, IDataProcessorConfiguration.class, PeType.PROCESSOR));
@@ -58,20 +64,28 @@ public List<AssetModel> findExtensions()
     return extensions;
   }
 
-  private List<AssetModel> findPipelineElements(SpServiceDefinition serviceDef,
-                                                Class<? extends IPipelineElementConfiguration<?, ?>> configType,
-                                                PeType peType) {
+  private List<AssetModel> findPipelineElements(
+      SpServiceDefinition serviceDef,
+      Class<? extends IPipelineElementConfiguration<?, ?>> configType,
+      PeType peType
+  ) {
     return serviceDef.getDeclarers()
-        .stream()
-        .map(IStreamPipesPipelineElement::declareConfig)
-        .filter(configType::isInstance)
-        .map(config -> new AssetModel(config.getDescription().getAppId(), peType)).toList();
+                     .stream()
+                     .map(IStreamPipesPipelineElement::declareConfig)
+                     .filter(configType::isInstance)
+                     .map(config -> new AssetModel(config.getDescription()
+                                                         .getAppId(), peType))
+                     .toList();
   }
 
   private Collection<? extends AssetModel> findAdapters(SpServiceDefinition serviceDef) {
-    return serviceDef.getAdapters().stream().map(adapter -> {
-      var config = adapter.declareConfig();
-      return new AssetModel(config.getAdapterDescription().getAppId(), PeType.ADAPTER);
-    }).toList();
+    return serviceDef.getAdapters()
+                     .stream()
+                     .map(adapter -> {
+                       var config = adapter.declareConfig();
+                       return new AssetModel(config.getAdapterDescription()
+                                                   .getAppId(), PeType.ADAPTER);
+                     })
+                     .toList();
   }
 }
diff --git a/...ons/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java b/...ons/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java