This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch remove-consul in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 2e18ab0e59ab42343855710aed3082718dbf35d0 Author: Dominik Riemer <[email protected]> AuthorDate: Sun Aug 6 17:02:03 2023 +0200 Migrate config extractors --- .../apache/streampipes/client/api/IAdminApi.java | 2 + .../streampipes/client/api/AbstractClientApi.java | 17 +++++++ .../apache/streampipes/client/api/AdminApi.java | 9 +++- .../extensions/api/config/IConfigExtractor.java | 15 +++++- .../management/config/ConfigExtractor.java | 58 ++++++++++++++++++---- .../context/AdapterContextGenerator.java | 2 - .../context/SpAdapterGuessSchemaContext.java | 8 +-- .../context/SpAdapterRuntimeContext.java | 4 +- .../management/init/DeclarersSingleton.java | 6 +++ .../management/util/RuntimeContextUtils.java | 5 +- .../googlemaps/GoogleMapsGeocoderProcessor.java | 2 +- .../GoogleMapsStaticGeocoderProcessor.java | 2 +- .../simulator/simulator/VehicleDataSimulator.java | 6 +-- .../simulator/vehicle/streams/VehicleStream.java | 4 +- .../simulator/utils/WatertankDataSimulator.java | 6 +-- .../watertank/streams/FlowRate1Stream.java | 4 +- .../watertank/streams/FlowRate2Stream.java | 4 +- .../watertank/streams/PressureTankStream.java | 4 +- .../watertank/streams/WaterLevel1Stream.java | 4 +- .../watertank/streams/WaterLevel2Stream.java | 4 +- .../impl/admin/ServiceConfigurationResource.java | 8 +++ .../sources/AbstractAdapterIncludedStream.java | 2 +- .../sources/AbstractAlreadyExistingStream.java | 2 +- 23 files changed, 134 insertions(+), 44 deletions(-) diff --git a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java index 486d95f0b..ff0bc2d11 100644 --- a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java +++ b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java @@ -34,6 +34,8 @@ public interface IAdminApi { void registerServiceConfiguration(SpServiceConfiguration serviceConfiguration); + SpServiceConfiguration getServiceConfiguration(String serviceGroup); + void registerAdapters(List<AdapterDescription> adapters); void registerFunctions(List<FunctionDefinition> functions); diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java index 78c9d02bc..c4df217fe 100644 --- a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java +++ b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java @@ -27,8 +27,11 @@ import org.apache.streampipes.client.model.StreamPipesClientConfig; import org.apache.streampipes.client.serializer.ObjectSerializer; import org.apache.streampipes.client.serializer.Serializer; import org.apache.streampipes.client.util.StreamPipesApiPath; +import org.apache.streampipes.commons.exceptions.SpHttpErrorStatusCode; import org.apache.streampipes.commons.exceptions.SpRuntimeException; +import java.util.Optional; + public class AbstractClientApi { protected StreamPipesClientConfig clientConfig; @@ -79,4 +82,18 @@ public class AbstractClientApi { ObjectSerializer<Void, T> serializer = new ObjectSerializer<>(); return new GetRequest<>(clientConfig, apiPath, targetClass, serializer).executeRequest(); } + + protected <T> Optional<T> getSingleOpt(StreamPipesApiPath apiPath, + Class<T> targetClass) throws SpRuntimeException { + try { + ObjectSerializer<Void, T> serializer = new ObjectSerializer<>(); + return Optional.of(new GetRequest<>(clientConfig, apiPath, targetClass, serializer).executeRequest()); + } catch (SpHttpErrorStatusCode e) { + if (e.getHttpStatusCode() == 404) { + return Optional.empty(); + } else { + throw e; + } + } + } } diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java index 8bf8a3b6a..3f9403614 100644 --- a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java +++ b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java @@ -49,6 +49,14 @@ public class AdminApi extends AbstractClientApi implements IAdminApi { post(getExtensionsServiceConfigurationPath(), serviceConfiguration); } + @Override + public SpServiceConfiguration getServiceConfiguration(String serviceGroup) { + var opt = getSingleOpt( + getExtensionsServiceConfigurationPath().addToPath(serviceGroup), SpServiceConfiguration.class); + + return opt.orElseGet(SpServiceConfiguration::new); + } + @Override public void registerAdapters(List<AdapterDescription> adapters) { post(getConnectPath(), adapters); @@ -84,7 +92,6 @@ public class AdminApi extends AbstractClientApi implements IAdminApi { private StreamPipesApiPath getMessagingSettingsPath() { return StreamPipesApiPath .fromBaseApiPath() - .addToPath("consul") .addToPath("messaging"); } diff --git a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/config/IConfigExtractor.java b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/config/IConfigExtractor.java index f423d5eb7..e1a672748 100644 --- a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/config/IConfigExtractor.java +++ b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/config/IConfigExtractor.java @@ -18,10 +18,21 @@ package org.apache.streampipes.extensions.api.config; -import org.apache.streampipes.svcdiscovery.api.SpConfig; +import org.apache.streampipes.model.extensions.configuration.ConfigItem; import java.io.Serializable; public interface IConfigExtractor extends Serializable { - SpConfig getConfig(); + + boolean getBoolean(String key); + + int getInteger(String key); + + double getDouble(String key); + + String getString(String key); + + <T> T getObject(String key, Class<T> clazz, T defaultValue); + + ConfigItem getConfigItem(String key); } diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/config/ConfigExtractor.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/config/ConfigExtractor.java index 4b173b592..dd1fef5f6 100644 --- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/config/ConfigExtractor.java +++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/config/ConfigExtractor.java @@ -18,23 +18,63 @@ package org.apache.streampipes.extensions.management.config; import org.apache.streampipes.extensions.api.config.IConfigExtractor; -import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.SpConfig; +import org.apache.streampipes.extensions.management.client.StreamPipesClientResolver; +import org.apache.streampipes.model.extensions.configuration.ConfigItem; +import org.apache.streampipes.model.extensions.configuration.SpServiceConfiguration; public class ConfigExtractor implements IConfigExtractor { - private SpConfig config; + private final SpServiceConfiguration serviceConfig; - private ConfigExtractor(String serviceGroup) { - this.config = SpServiceDiscovery.getSpConfig(serviceGroup); + private ConfigExtractor(SpServiceConfiguration serviceConfig) { + this.serviceConfig = serviceConfig; } - public static ConfigExtractor from(String serviceId) { - return new ConfigExtractor(serviceId); + public static ConfigExtractor from(String serviceGroup) { + var client = new StreamPipesClientResolver().makeStreamPipesClientInstance(); + var serviceConfig = client.adminApi().getServiceConfiguration(serviceGroup); + return new ConfigExtractor(serviceConfig); + } + + public static ConfigExtractor from(SpServiceConfiguration serviceConfig) { + return new ConfigExtractor(serviceConfig); + } + + @Override + public boolean getBoolean(String key) { + return Boolean.parseBoolean(getString(key)); } @Override - public SpConfig getConfig() { - return this.config; + public int getInteger(String key) { + return Integer.parseInt(getString(key)); + } + + @Override + public double getDouble(String key) { + return Double.parseDouble(getString(key)); + } + + @Override + public String getString(String key) { + return getItem(key).getValue(); + } + + @Override + public <T> T getObject(String key, Class<T> clazz, T defaultValue) { + return null; + } + + @Override + public ConfigItem getConfigItem(String key) { + return getItem(key); + } + + private ConfigItem getItem(String key) { + return serviceConfig.getConfigs() + .stream() + .filter(c -> c.getKey().equals(key)) + .findFirst() + .orElseThrow(IllegalArgumentException::new); } } diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/AdapterContextGenerator.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/AdapterContextGenerator.java index 77e24c234..f774a4140 100644 --- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/AdapterContextGenerator.java +++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/AdapterContextGenerator.java @@ -37,6 +37,4 @@ public class AdapterContextGenerator { public IAdapterGuessSchemaContext makeGuessSchemaContext() { return new SpAdapterGuessSchemaContext(makeConfigExtractor(), makeStreamPipesClient()); } - - } diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/SpAdapterGuessSchemaContext.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/SpAdapterGuessSchemaContext.java index b746accb1..f80ae4cee 100644 --- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/SpAdapterGuessSchemaContext.java +++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/SpAdapterGuessSchemaContext.java @@ -19,22 +19,22 @@ package org.apache.streampipes.extensions.management.context; import org.apache.streampipes.client.StreamPipesClient; +import org.apache.streampipes.extensions.api.config.IConfigExtractor; import org.apache.streampipes.extensions.api.connect.context.IAdapterGuessSchemaContext; -import org.apache.streampipes.extensions.management.config.ConfigExtractor; public class SpAdapterGuessSchemaContext implements IAdapterGuessSchemaContext { - protected ConfigExtractor configExtractor; + protected IConfigExtractor configExtractor; protected StreamPipesClient streamPipesClient; - public SpAdapterGuessSchemaContext(ConfigExtractor configExtractor, + public SpAdapterGuessSchemaContext(IConfigExtractor configExtractor, StreamPipesClient streamPipesClient) { this.configExtractor = configExtractor; this.streamPipesClient = streamPipesClient; } @Override - public ConfigExtractor getConfigStore() { + public IConfigExtractor getConfigStore() { return configExtractor; } diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/SpAdapterRuntimeContext.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/SpAdapterRuntimeContext.java index 2fc6fd95e..2dea83542 100644 --- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/SpAdapterRuntimeContext.java +++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/context/SpAdapterRuntimeContext.java @@ -19,9 +19,9 @@ package org.apache.streampipes.extensions.management.context; import org.apache.streampipes.client.StreamPipesClient; +import org.apache.streampipes.extensions.api.config.IConfigExtractor; import org.apache.streampipes.extensions.api.connect.context.IAdapterRuntimeContext; import org.apache.streampipes.extensions.api.monitoring.IExtensionsLogger; -import org.apache.streampipes.extensions.management.config.ConfigExtractor; import java.io.Serializable; @@ -31,7 +31,7 @@ public class SpAdapterRuntimeContext extends SpAdapterGuessSchemaContext private final IExtensionsLogger extensionsLogger; public SpAdapterRuntimeContext(IExtensionsLogger extensionsLogger, - ConfigExtractor configExtractor, + IConfigExtractor configExtractor, StreamPipesClient streamPipesClient) { super(configExtractor, streamPipesClient); this.extensionsLogger = extensionsLogger; diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java index 204ab1e6c..8b98511b8 100644 --- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java +++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java @@ -72,6 +72,7 @@ public class DeclarersSingleton implements IDeclarersSingleton { private List<IStreamPipesRuntimeProvider> runtimeProviders; private String serviceId; + private String serviceGroup; private int port; private String route; @@ -105,6 +106,7 @@ public class DeclarersSingleton implements IDeclarersSingleton { this.setPort(port); this.addDeclarers(serviceDef.getDeclarers()); this.serviceId = serviceDef.getServiceId(); + this.serviceGroup = serviceDef.getServiceGroup(); this.registerProtocols(serviceDef.getProtocolDefinitionFactories()); this.registerDataFormats(serviceDef.getDataFormatFactories()); this.runtimeProviders = serviceDef.getRuntimeProviders(); @@ -281,6 +283,10 @@ public class DeclarersSingleton implements IDeclarersSingleton { return serviceDef; } + public String getServiceGroup() { + return serviceGroup; + } + public SpServiceDefinition getServiceDefinition() { return this.serviceDefinition; } diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/util/RuntimeContextUtils.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/util/RuntimeContextUtils.java index 1823557b7..c8223431a 100644 --- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/util/RuntimeContextUtils.java +++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/util/RuntimeContextUtils.java @@ -19,14 +19,15 @@ package org.apache.streampipes.extensions.management.util; import org.apache.streampipes.client.StreamPipesClient; +import org.apache.streampipes.extensions.api.config.IConfigExtractor; import org.apache.streampipes.extensions.management.client.StreamPipesClientResolver; import org.apache.streampipes.extensions.management.config.ConfigExtractor; import org.apache.streampipes.extensions.management.init.DeclarersSingleton; public class RuntimeContextUtils { - public static ConfigExtractor makeConfigExtractor() { - var serviceId = DeclarersSingleton.getInstance().getServiceId(); + public static IConfigExtractor makeConfigExtractor() { + var serviceId = DeclarersSingleton.getInstance().getServiceGroup(); return ConfigExtractor.from(serviceId); } diff --git a/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/latlong/processor/geocoder/googlemaps/GoogleMapsGeocoderProcessor.java b/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/latlong/processor/geocoder/googlemaps/GoogleMapsGeocoderProcessor.java index e07af4cf6..65cb65c96 100644 --- a/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/latlong/processor/geocoder/googlemaps/GoogleMapsGeocoderProcessor.java +++ b/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/latlong/processor/geocoder/googlemaps/GoogleMapsGeocoderProcessor.java @@ -79,7 +79,7 @@ public class GoogleMapsGeocoderProcessor extends StreamPipesDataProcessor { this.geocoderRequest = parameters.extractor().mappingPropertyValue(GEOCODER_REQUEST_KEY); - String googleMapsApiKey = runtimeContext.getConfigStore().getConfig().getString(ConfigKeys.GOOGLE_API_KEY); + String googleMapsApiKey = runtimeContext.getConfigStore().getString(ConfigKeys.GOOGLE_API_KEY); if (googleMapsApiKey == null || googleMapsApiKey.equals("")) { throw new SpRuntimeException("Could not start Geocoder. Did you forget to add a Google Maps API key?"); diff --git a/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/latlong/processor/geocoder/googlemapsstatic/GoogleMapsStaticGeocoderProcessor.java b/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/latlong/processor/geocoder/googlemapsstatic/GoogleMapsStaticGeocoderProcessor.java index 8f9f8cdcb..edf141427 100644 --- a/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/latlong/processor/geocoder/googlemapsstatic/GoogleMapsStaticGeocoderProcessor.java +++ b/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/latlong/processor/geocoder/googlemapsstatic/GoogleMapsStaticGeocoderProcessor.java @@ -80,7 +80,7 @@ public class GoogleMapsStaticGeocoderProcessor extends StreamPipesDataProcessor this.staticGeocoderRequest = parameters.extractor().singleValueParameter(STATIC_GEOCODER_REQUEST_KEY, String.class); - String googleMapsApiKey = runtimeContext.getConfigStore().getConfig().getString(ConfigKeys.GOOGLE_API_KEY); + String googleMapsApiKey = runtimeContext.getConfigStore().getString(ConfigKeys.GOOGLE_API_KEY); if (googleMapsApiKey == null || googleMapsApiKey.equals("")) { throw new SpRuntimeException("Could not start Geocoder. Did you forget to add a Google Maps" + " API key?"); diff --git a/streampipes-extensions/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/simulator/VehicleDataSimulator.java b/streampipes-extensions/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/simulator/VehicleDataSimulator.java index 63a036b6a..62fbad54f 100644 --- a/streampipes-extensions/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/simulator/VehicleDataSimulator.java +++ b/streampipes-extensions/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/simulator/VehicleDataSimulator.java @@ -47,7 +47,7 @@ public class VehicleDataSimulator implements Runnable { private void initSimulation() { try { ConfigExtractor configExtractor = - ConfigExtractor.from(DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup()); + ConfigExtractor.from(DeclarersSingleton.getInstance().getServiceGroup()); SimulationConfig config = buildSimulationConfig(); Map<String, TopicAwareWorkflow> workflows = buildSimWorkflows(config); String kafkaHost = getKafkaHost(configExtractor); @@ -78,12 +78,12 @@ public class VehicleDataSimulator implements Runnable { private String getKafkaHost(ConfigExtractor configExtractor) { return env.getSpDebug().getValueOrDefault() - ? "localhost" : configExtractor.getConfig().getString(ConfigKeys.KAFKA_HOST); + ? "localhost" : configExtractor.getString(ConfigKeys.KAFKA_HOST); } private Integer getKafkaPort(ConfigExtractor configExtractor) { return env.getSpDebug().getValueOrDefault() - ? 9094 : configExtractor.getConfig().getInteger(ConfigKeys.KAFKA_PORT); + ? 9094 : configExtractor.getInteger(ConfigKeys.KAFKA_PORT); } @Override diff --git a/streampipes-extensions/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/vehicle/streams/VehicleStream.java b/streampipes-extensions/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/vehicle/streams/VehicleStream.java index d29c2bf37..44e050439 100644 --- a/streampipes-extensions/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/vehicle/streams/VehicleStream.java +++ b/streampipes-extensions/streampipes-sources-vehicle-simulator/src/main/java/org/apache/streampipes/sources/vehicle/simulator/vehicle/streams/VehicleStream.java @@ -46,8 +46,8 @@ public class VehicleStream extends AbstractAdapterIncludedStream { + "value of the vehicle's position"), "longitude", Geo.LNG)) .format(Formats.jsonFormat()) .protocol(Protocols.kafka( - configExtractor().getConfig().getString(ConfigKeys.KAFKA_HOST), - configExtractor().getConfig().getInteger(ConfigKeys.KAFKA_PORT), + configExtractor().getString(ConfigKeys.KAFKA_HOST), + configExtractor().getInteger(ConfigKeys.KAFKA_PORT), "org.apache.streampipes.examples.sources.vehicle")) .build(); } diff --git a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/utils/WatertankDataSimulator.java b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/utils/WatertankDataSimulator.java index 35dd6f2e0..48d124c35 100644 --- a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/utils/WatertankDataSimulator.java +++ b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/utils/WatertankDataSimulator.java @@ -47,7 +47,7 @@ public class WatertankDataSimulator implements Runnable { private void initSimulation() { try { ConfigExtractor configExtractor = - ConfigExtractor.from(DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup()); + ConfigExtractor.from(DeclarersSingleton.getInstance().getServiceGroup()); SimulationConfig config = buildSimulationConfig(); Map<String, TopicAwareWorkflow> workflows = buildSimWorkflows(config); String kafkaHost = getKafkaHost(configExtractor); @@ -69,12 +69,12 @@ public class WatertankDataSimulator implements Runnable { private String getKafkaHost(ConfigExtractor configExtractor) { return env.getSpDebug().getValueOrDefault() - ? "localhost" : configExtractor.getConfig().getString(ConfigKeys.KAFKA_HOST); + ? "localhost" : configExtractor.getString(ConfigKeys.KAFKA_HOST); } private Integer getKafkaPort(ConfigExtractor configExtractor) { return env.getSpDebug().getValueOrDefault() - ? 9094 : configExtractor.getConfig().getInteger(ConfigKeys.KAFKA_PORT); + ? 9094 : configExtractor.getInteger(ConfigKeys.KAFKA_PORT); } private Map<String, TopicAwareWorkflow> buildSimWorkflows(SimulationConfig config) throws IOException { diff --git a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/FlowRate1Stream.java b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/FlowRate1Stream.java index af6b2f5a8..1005ba03d 100644 --- a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/FlowRate1Stream.java +++ b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/FlowRate1Stream.java @@ -68,8 +68,8 @@ public class FlowRate1Stream extends AbstractAdapterIncludedStream { .build()) .format(Formats.jsonFormat()) .protocol(Protocols.kafka( - configExtractor().getConfig().getString(ConfigKeys.KAFKA_HOST), - configExtractor().getConfig().getInteger(ConfigKeys.KAFKA_PORT), + configExtractor().getString(ConfigKeys.KAFKA_HOST), + configExtractor().getInteger(ConfigKeys.KAFKA_PORT), "org.apache.streampipes.examples.flowrate-1") ) .build(); diff --git a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/FlowRate2Stream.java b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/FlowRate2Stream.java index 023fdc8e7..cfd4638ac 100644 --- a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/FlowRate2Stream.java +++ b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/FlowRate2Stream.java @@ -85,8 +85,8 @@ public class FlowRate2Stream extends AbstractAlreadyExistingStream { .build()) .format(Formats.jsonFormat()) .protocol(Protocols.kafka( - configExtractor().getConfig().getString(ConfigKeys.KAFKA_HOST), - configExtractor().getConfig().getInteger(ConfigKeys.KAFKA_PORT), + configExtractor().getString(ConfigKeys.KAFKA_HOST), + configExtractor().getInteger(ConfigKeys.KAFKA_PORT), "org.apache.streampipes.examples.flowrate2")) .build(); } diff --git a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/PressureTankStream.java b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/PressureTankStream.java index fa7bfe71d..cbe4dbd0f 100644 --- a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/PressureTankStream.java +++ b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/PressureTankStream.java @@ -57,8 +57,8 @@ public class PressureTankStream extends AbstractAlreadyExistingStream { .build()) .format(Formats.jsonFormat()) .protocol(Protocols.kafka( - configExtractor().getConfig().getString(ConfigKeys.KAFKA_HOST), - configExtractor().getConfig().getInteger(ConfigKeys.KAFKA_PORT), + configExtractor().getString(ConfigKeys.KAFKA_HOST), + configExtractor().getInteger(ConfigKeys.KAFKA_PORT), "org.apache.streampipes.examples.pressuretank")) .build(); } diff --git a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/WaterLevel1Stream.java b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/WaterLevel1Stream.java index c516afc17..5e54e9429 100644 --- a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/WaterLevel1Stream.java +++ b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/WaterLevel1Stream.java @@ -71,8 +71,8 @@ public class WaterLevel1Stream extends AbstractAlreadyExistingStream { .build()) .format(Formats.jsonFormat()) .protocol(Protocols.kafka( - configExtractor().getConfig().getString(ConfigKeys.KAFKA_HOST), - configExtractor().getConfig().getInteger(ConfigKeys.KAFKA_PORT), + configExtractor().getString(ConfigKeys.KAFKA_HOST), + configExtractor().getInteger(ConfigKeys.KAFKA_PORT), "org.apache.streampipes.examples.waterlevel")) .build(); } diff --git a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/WaterLevel2Stream.java b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/WaterLevel2Stream.java index eece73b8a..685ec156d 100644 --- a/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/WaterLevel2Stream.java +++ b/streampipes-extensions/streampipes-sources-watertank-simulator/src/main/java/org/apache/streampipes/sources/watertank/simulator/watertank/streams/WaterLevel2Stream.java @@ -63,8 +63,8 @@ public class WaterLevel2Stream extends AbstractAlreadyExistingStream { .build()) .format(Formats.jsonFormat()) .protocol(Protocols.kafka( - configExtractor().getConfig().getString(ConfigKeys.KAFKA_HOST), - configExtractor().getConfig().getInteger(ConfigKeys.KAFKA_PORT), + configExtractor().getString(ConfigKeys.KAFKA_HOST), + configExtractor().getInteger(ConfigKeys.KAFKA_PORT), "org.apache.streampipes.examples.waterlevel2")) .build(); } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceConfigurationResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceConfigurationResource.java index df7437abf..8a89b41db 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceConfigurationResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceConfigurationResource.java @@ -51,6 +51,14 @@ public class ServiceConfigurationResource extends AbstractAuthGuardedRestResourc return ok(extensionsServicesConfigStorage.getAll()); } + @GET + @Path("{serviceGroup}") + @Produces(MediaType.APPLICATION_JSON) + public Response getServiceConfiguration(@PathParam("serviceGroup") String serviceGroup) { + var config = extensionsServicesConfigStorage.getElementById(serviceGroup); + return config != null ? ok(config) : notFound(); + } + @POST @Consumes(MediaType.APPLICATION_JSON) public Response registerServiceConfiguration(SpServiceConfiguration serviceConfiguration) { diff --git a/streampipes-sources/src/main/java/org/apache/streampipes/sources/AbstractAdapterIncludedStream.java b/streampipes-sources/src/main/java/org/apache/streampipes/sources/AbstractAdapterIncludedStream.java index 7f31edcda..5c1cba4a2 100644 --- a/streampipes-sources/src/main/java/org/apache/streampipes/sources/AbstractAdapterIncludedStream.java +++ b/streampipes-sources/src/main/java/org/apache/streampipes/sources/AbstractAdapterIncludedStream.java @@ -33,7 +33,7 @@ public abstract class AbstractAdapterIncludedStream implements IStreamPipesDataS } public ConfigExtractor configExtractor() { - return ConfigExtractor.from(DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup()); + return ConfigExtractor.from(DeclarersSingleton.getInstance().getServiceGroup()); } @Override diff --git a/streampipes-sources/src/main/java/org/apache/streampipes/sources/AbstractAlreadyExistingStream.java b/streampipes-sources/src/main/java/org/apache/streampipes/sources/AbstractAlreadyExistingStream.java index 42eb4705b..d08d435b0 100644 --- a/streampipes-sources/src/main/java/org/apache/streampipes/sources/AbstractAlreadyExistingStream.java +++ b/streampipes-sources/src/main/java/org/apache/streampipes/sources/AbstractAlreadyExistingStream.java @@ -39,7 +39,7 @@ public abstract class AbstractAlreadyExistingStream implements IStreamPipesDataS } public ConfigExtractor configExtractor() { - return ConfigExtractor.from(DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup()); + return ConfigExtractor.from(DeclarersSingleton.getInstance().getServiceGroup()); } @Override
