This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch STREAMPIPES-319 in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
commit 3559fb4a0c802b3ebb37dd3567d18bd587ad3310 Author: Dominik Riemer <[email protected]> AuthorDate: Tue Jul 13 11:05:15 2021 +0200 [STREAMPIPES-319] Refactor config of vehicle simulator service --- .../vehicle/simulator/VehicleSimulatorInit.java | 19 ++++- .../vehicle/simulator/config/ConfigKeys.java | 8 +- .../simulator/config/VehicleSimulatorConfig.java | 85 ---------------------- .../simulator/simulator/VehicleDataSimulator.java | 9 ++- .../simulator/vehicle/streams/VehicleStream.java | 6 +- 5 files changed, 27 insertions(+), 100 deletions(-) diff --git a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/VehicleSimulatorInit.java b/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/VehicleSimulatorInit.java index 6daff2b..3320e06 100644 --- a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/VehicleSimulatorInit.java +++ b/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/VehicleSimulatorInit.java @@ -18,16 +18,27 @@ package org.apache.streampipes.sources.vehicle.simulator; -import org.apache.streampipes.container.init.DeclarersSingleton; +import org.apache.streampipes.container.model.SpServiceDefinition; +import org.apache.streampipes.container.model.SpServiceDefinitionBuilder; import org.apache.streampipes.container.standalone.init.StandaloneModelSubmitter; -import org.apache.streampipes.sources.vehicle.simulator.config.VehicleSimulatorConfig; +import org.apache.streampipes.sources.vehicle.simulator.config.ConfigKeys; import org.apache.streampipes.sources.vehicle.simulator.vehicle.streams.VehicleStream; public class VehicleSimulatorInit extends StandaloneModelSubmitter { public static void main(String[] args) { - DeclarersSingleton.getInstance().add(new VehicleStream()); + new VehicleSimulatorInit().init(); + } - new VehicleSimulatorInit().init(VehicleSimulatorConfig.INSTANCE); + @Override + public SpServiceDefinition provideServiceDefinition() { + return SpServiceDefinitionBuilder.create("org-apache-streampipes-sources-simulator-vehicle", + "Vehicle Simulator", + "", + 8090) + .registerPipelineElement(new VehicleStream()) + .addConfig(ConfigKeys.KAFKA_HOST, "kafka", "Host for kafka of the vehicle simulator") + .addConfig(ConfigKeys.KAFKA_PORT, 9092, "Port for kafka of the vehicle simulator") + .build(); } } diff --git a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/config/ConfigKeys.java b/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/config/ConfigKeys.java index 606eb61..4f2d718 100644 --- a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/config/ConfigKeys.java +++ b/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/config/ConfigKeys.java @@ -19,10 +19,6 @@ package org.apache.streampipes.sources.vehicle.simulator.config; public class ConfigKeys { - - final static String HOST = "SP_HOST"; - final static String PORT = "SP_PORT"; - final static String KAFKA_HOST = "SP_KAFKA_HOST"; - final static String KAFKA_PORT = "SP_KAFKA_PORT"; - final static String SERVICE_NAME = "SP_SERVICE_NAME"; + public final static String KAFKA_HOST = "SP_KAFKA_HOST"; + public final static String KAFKA_PORT = "SP_KAFKA_PORT"; } diff --git a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/config/VehicleSimulatorConfig.java b/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/config/VehicleSimulatorConfig.java deleted file mode 100644 index 35e4786..0000000 --- a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/config/VehicleSimulatorConfig.java +++ /dev/null @@ -1,85 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -package org.apache.streampipes.sources.vehicle.simulator.config; - -import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.SpConfig; -import org.apache.streampipes.container.model.PeConfig; - -public enum VehicleSimulatorConfig implements PeConfig { - INSTANCE; - - - private final static String service_id = "pe/org.apache.streampipes.sources.vehicle.simulator"; - private final static String service_name = "Sources Vehicle Simulator"; - private final static String service_container_name = "sources-vehicle-simulator"; - - private SpConfig config; - public static String serverUrl; - - VehicleSimulatorConfig() { - config = SpServiceDiscovery.getSpConfig(service_id); - - /* - TUTORIAL: - The second parameter is the default value for the configuration property. - This value is set in Consul when the parameter does not exist. - Important. Changes here are not effective if the configuration parameter is already set in consul. In - such cases the value has to be changed in consul directly. - */ - config.register(ConfigKeys.HOST, service_container_name, "Hostname for the examples-sources project"); - config.register(ConfigKeys.PORT, 8090, "Port of the sources project"); - config.register(ConfigKeys.KAFKA_HOST, "kafka", "Host for kafka of the pe demonstrator project"); - config.register(ConfigKeys.KAFKA_PORT, 9092, "Port for kafka of the pe demonstrator project"); - config.register(ConfigKeys.SERVICE_NAME, service_name, "StreamPipes example sources"); - } - - static { - serverUrl = VehicleSimulatorConfig.INSTANCE.getHost() + ":" + VehicleSimulatorConfig.INSTANCE.getPort(); - } - - @Override - public String getHost() { - return config.getString(ConfigKeys.HOST); - } - - @Override - public int getPort() { - return config.getInteger(ConfigKeys.PORT); - } - - public String getKafkaHost() { - return config.getString(ConfigKeys.KAFKA_HOST); - } - - public int getKafkaPort() { - return config.getInteger(ConfigKeys.KAFKA_PORT); - } - - @Override - public String getId() { - return service_id; - } - - @Override - public String getName() { - return config.getString(ConfigKeys.SERVICE_NAME); - } - -} diff --git a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/simulator/VehicleDataSimulator.java b/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/simulator/VehicleDataSimulator.java index 99bbbe3..f36b0b4 100644 --- a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/simulator/VehicleDataSimulator.java +++ b/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/simulator/VehicleDataSimulator.java @@ -21,9 +21,11 @@ package org.apache.streampipes.sources.vehicle.simulator.simulator; import net.acesinc.data.json.generator.config.JSONConfigReader; import net.acesinc.data.json.generator.config.SimulationConfig; import net.acesinc.data.json.generator.config.WorkflowConfig; +import org.apache.streampipes.container.config.ConfigExtractor; +import org.apache.streampipes.container.init.DeclarersSingleton; import org.apache.streampipes.pe.simulator.StreamPipesSimulationRunner; import org.apache.streampipes.pe.simulator.TopicAwareWorkflow; -import org.apache.streampipes.sources.vehicle.simulator.config.VehicleSimulatorConfig; +import org.apache.streampipes.sources.vehicle.simulator.config.ConfigKeys; import java.io.IOException; import java.util.HashMap; @@ -35,10 +37,11 @@ public class VehicleDataSimulator implements Runnable { private void initSimulation() { try { + ConfigExtractor configExtractor = ConfigExtractor.from(DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup()); SimulationConfig config = buildSimulationConfig(); Map<String, TopicAwareWorkflow> workflows = buildSimWorkflows(config); - String kafkaHost = VehicleSimulatorConfig.INSTANCE.getKafkaHost(); - Integer kafkaPort = VehicleSimulatorConfig.INSTANCE.getKafkaPort(); + String kafkaHost = configExtractor.getConfig().getString(ConfigKeys.KAFKA_HOST); + Integer kafkaPort = configExtractor.getConfig().getInteger(ConfigKeys.KAFKA_PORT); new StreamPipesSimulationRunner(config, workflows, kafkaHost, kafkaPort).startSimulation(); } catch (IOException e) { diff --git a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/vehicle/streams/VehicleStream.java b/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/vehicle/streams/VehicleStream.java index 30a93fd..63fc6d1 100644 --- a/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/vehicle/streams/VehicleStream.java +++ b/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/vehicle/streams/VehicleStream.java @@ -26,7 +26,7 @@ import org.apache.streampipes.sdk.helpers.Formats; import org.apache.streampipes.sdk.helpers.Labels; import org.apache.streampipes.sdk.helpers.Protocols; import org.apache.streampipes.sources.AbstractAdapterIncludedStream; -import org.apache.streampipes.sources.vehicle.simulator.config.VehicleSimulatorConfig; +import org.apache.streampipes.sources.vehicle.simulator.config.ConfigKeys; import org.apache.streampipes.sources.vehicle.simulator.simulator.VehicleDataSimulator; import org.apache.streampipes.vocabulary.Geo; @@ -45,7 +45,9 @@ public class VehicleStream extends AbstractAdapterIncludedStream { .property(EpProperties.doubleEp(Labels.from("longitude", "Longitude", "Denotes the longitude " + "value of the vehicle's position"), "longitude", Geo.lng)) .format(Formats.jsonFormat()) - .protocol(Protocols.kafka(VehicleSimulatorConfig.INSTANCE.getKafkaHost(), VehicleSimulatorConfig.INSTANCE.getKafkaPort(), + .protocol(Protocols.kafka( + configExtractor().getConfig().getString(ConfigKeys.KAFKA_HOST), + configExtractor().getConfig().getInteger(ConfigKeys.KAFKA_PORT), "org.apache.streampipes.examples.sources.vehicle")) .build(); }
