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 124daea5934be3ed2a3db5445d1ca8c9274bc339 Author: Dominik Riemer <[email protected]> AuthorDate: Fri Aug 4 08:09:46 2023 +0200 Migrate configuration storage from Consul to CouchDB --- .../apache/streampipes/client/api/IAdminApi.java | 2 +- .../apache/streampipes/client/api/AdminApi.java | 2 +- streampipes-config/pom.xml | 5 + .../streampipes/config/backend/BackendConfig.java | 11 +- .../config/backend/DefaultMessagingSettings.java | 6 +- .../connect/management/util/GroundingUtils.java | 26 +-- .../management/util/TransportFormatGenerator.java | 6 +- .../streampipes/connect/management/util/Utils.java | 14 +- .../export/utils/EventGroundingProcessor.java | 31 ++- .../connect/adapter/AdapterPipelineGenerator.java | 12 +- .../apache/streampipes/mail/AbstractMailer.java | 6 +- .../org/apache/streampipes/mail/MailTester.java | 2 +- .../mail/config/MailConfigurationBuilder.java | 8 +- .../apache/streampipes/mail/utils/MailUtils.java | 13 +- streampipes-model/pom.xml | 5 + .../model/configuration}/EmailConfig.java | 2 +- .../model/configuration}/GeneralConfig.java | 2 +- .../model/configuration}/JwtSigningMode.java | 2 +- .../model/configuration}/LocalAuthConfig.java | 2 +- .../MessagingSettings.java | 2 +- .../model/configuration/SpCoreConfiguration.java | 219 +++++++++++++++++++++ .../{config => configuration}/SpDataFormat.java | 2 +- .../{config => configuration}/SpProtocol.java | 2 +- .../model/configuration}/TransportStrategy.java | 2 +- .../streampipes/manager/assets/AssetConstants.java | 9 +- .../streampipes/manager/file/FileConstants.java | 9 +- .../manager/matching/FormatSelector.java | 16 +- .../manager/matching/ProtocolSelector.java | 51 +++-- .../rest/core/base/impl/AbstractRestResource.java | 5 + .../streampipes/rest/impl/Authentication.java | 7 +- .../streampipes/rest/impl/EmailResource.java | 3 +- .../streampipes/rest/impl/admin/ConsulConfig.java | 9 +- .../impl/admin/EmailConfigurationResource.java | 10 +- .../impl/admin/GeneralConfigurationResource.java | 10 +- .../service/core/StreamPipesCoreApplication.java | 5 +- .../service/core/StreamPipesEnvChecker.java | 21 +- .../core/migrations/AvailableMigrations.java | 4 +- .../migrations/v093/ConsulConfigMigration.java | 71 +++++++ streampipes-service-discovery/pom.xml | 5 + .../svcdiscovery/SpServiceDiscovery.java | 4 +- .../streampipes/storage/api/INoSqlStorage.java | 2 + .../storage/api/ISpCoreConfigurationStorage.java | 18 +- .../storage/couchdb/CouchDbStorageManager.java | 7 + .../couchdb/impl/CoreConfigurationStorageImpl.java | 61 ++++++ .../streampipes/storage/couchdb/utils/Utils.java | 4 + .../user/management/jwt/JwtTokenProvider.java | 16 +- .../user/management/jwt/SpKeyResolver.java | 9 +- 47 files changed, 607 insertions(+), 133 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 14f3fb3b6..9af296e7a 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 @@ -18,7 +18,7 @@ package org.apache.streampipes.client.api; -import org.apache.streampipes.model.config.MessagingSettings; +import org.apache.streampipes.model.configuration.MessagingSettings; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.function.FunctionDefinition; import org.apache.streampipes.svcdiscovery.api.model.SpServiceConfiguration; 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 90f81f8f6..a94b70268 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 @@ -19,7 +19,7 @@ package org.apache.streampipes.client.api; import org.apache.streampipes.client.model.StreamPipesClientConfig; import org.apache.streampipes.client.util.StreamPipesApiPath; -import org.apache.streampipes.model.config.MessagingSettings; +import org.apache.streampipes.model.configuration.MessagingSettings; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.function.FunctionDefinition; import org.apache.streampipes.model.message.SuccessMessage; diff --git a/streampipes-config/pom.xml b/streampipes-config/pom.xml index 6f70e62e5..562dd7928 100644 --- a/streampipes-config/pom.xml +++ b/streampipes-config/pom.xml @@ -34,6 +34,11 @@ <artifactId>streampipes-model</artifactId> <version>0.93.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-model-client</artifactId> + <version>0.93.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-serializers-json</artifactId> diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java index d626021c3..b3abc1f00 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java +++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java @@ -22,10 +22,10 @@ package org.apache.streampipes.config.backend; import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.environment.Environments; import org.apache.streampipes.commons.random.TokenGenerator; -import org.apache.streampipes.config.backend.model.EmailConfig; -import org.apache.streampipes.config.backend.model.GeneralConfig; -import org.apache.streampipes.config.backend.model.LocalAuthConfig; -import org.apache.streampipes.model.config.MessagingSettings; +import org.apache.streampipes.model.configuration.EmailConfig; +import org.apache.streampipes.model.configuration.GeneralConfig; +import org.apache.streampipes.model.configuration.LocalAuthConfig; +import org.apache.streampipes.model.configuration.MessagingSettings; import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; import org.apache.streampipes.svcdiscovery.api.SpConfig; @@ -34,9 +34,6 @@ import java.io.File; public enum BackendConfig { INSTANCE; - private final char[] possibleCharacters = - ("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789~`!@#$%^&*()-_=+[{]}\\|;:'\",<.>/?") - .toCharArray(); private SpConfig config; BackendConfig() { diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/DefaultMessagingSettings.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/DefaultMessagingSettings.java index fa4e5ed8f..8e2180382 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/DefaultMessagingSettings.java +++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/DefaultMessagingSettings.java @@ -18,9 +18,9 @@ package org.apache.streampipes.config.backend; -import org.apache.streampipes.model.config.MessagingSettings; -import org.apache.streampipes.model.config.SpDataFormat; -import org.apache.streampipes.model.config.SpProtocol; +import org.apache.streampipes.model.configuration.MessagingSettings; +import org.apache.streampipes.model.configuration.SpDataFormat; +import org.apache.streampipes.model.configuration.SpProtocol; import java.util.Arrays; import java.util.List; diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/GroundingUtils.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/GroundingUtils.java index 6d8dd7c30..8f51d12de 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/GroundingUtils.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/GroundingUtils.java @@ -18,8 +18,7 @@ package org.apache.streampipes.connect.management.util; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.model.config.SpProtocol; +import org.apache.streampipes.model.configuration.SpProtocol; import org.apache.streampipes.model.grounding.EventGrounding; import org.apache.streampipes.model.grounding.JmsTransportProtocol; import org.apache.streampipes.model.grounding.KafkaTransportProtocol; @@ -39,41 +38,44 @@ public class GroundingUtils { public static EventGrounding createEventGrounding() { EventGrounding eventGrounding = new EventGrounding(); + var cfg = Utils + .getCoreConfigStorage() + .get(); String topic = TOPIC_PREFIX + UUID.randomUUID().toString(); TopicDefinition topicDefinition = new SimpleTopicDefinition(topic); SpProtocol prioritizedProtocol = - BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedProtocols().get(0); + cfg.getMessagingSettings().getPrioritizedProtocols().get(0); if (isPrioritized(prioritizedProtocol, JmsTransportProtocol.class)) { eventGrounding.setTransportProtocol( makeJmsTransportProtocol( - BackendConfig.INSTANCE.getJmsHost(), - BackendConfig.INSTANCE.getJmsPort(), + cfg.getJmsHost(), + cfg.getJmsPort(), topicDefinition)); } else if (isPrioritized(prioritizedProtocol, KafkaTransportProtocol.class)) { eventGrounding.setTransportProtocol( makeKafkaTransportProtocol( - BackendConfig.INSTANCE.getKafkaHost(), - BackendConfig.INSTANCE.getKafkaPort(), + cfg.getKafkaHost(), + cfg.getKafkaPort(), topicDefinition)); } else if (isPrioritized(prioritizedProtocol, MqttTransportProtocol.class)) { eventGrounding.setTransportProtocol( makeMqttTransportProtocol( - BackendConfig.INSTANCE.getMqttHost(), - BackendConfig.INSTANCE.getMqttPort(), + cfg.getMqttHost(), + cfg.getMqttPort(), topicDefinition)); } else if (isPrioritized(prioritizedProtocol, NatsTransportProtocol.class)) { eventGrounding.setTransportProtocol( makeNatsTransportProtocol( - BackendConfig.INSTANCE.getNatsHost(), - BackendConfig.INSTANCE.getNatsPort(), + cfg.getNatsHost(), + cfg.getNatsPort(), topicDefinition)); } else if (isPrioritized(prioritizedProtocol, PulsarTransportProtocol.class)) { eventGrounding.setTransportProtocol( makePulsarTransportProtocol( - BackendConfig.INSTANCE.getPulsarUrl(), + cfg.getPulsarUrl(), topicDefinition ) ); diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/TransportFormatGenerator.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/TransportFormatGenerator.java index 5c9bf0b92..7038866af 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/TransportFormatGenerator.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/TransportFormatGenerator.java @@ -17,8 +17,7 @@ */ package org.apache.streampipes.connect.management.util; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.model.config.SpDataFormat; +import org.apache.streampipes.model.configuration.SpDataFormat; import org.apache.streampipes.model.grounding.TransportFormat; import org.apache.streampipes.sdk.helpers.SupportedFormats; import org.apache.streampipes.vocabulary.MessageFormat; @@ -29,8 +28,9 @@ import java.util.List; public class TransportFormatGenerator { public static TransportFormat getTransportFormat() { + var cfg = Utils.getCoreConfigStorage().get(); List<SpDataFormat> supportedFormats = - BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedFormats(); + cfg.getMessagingSettings().getPrioritizedFormats(); if (supportedFormats.size() > 0) { return new TransportFormat(supportedFormats.get(0).getMessageFormat()); diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/Utils.java similarity index 66% copy from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java copy to streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/Utils.java index eda79448f..030b839bf 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/Utils.java @@ -15,12 +15,18 @@ * limitations under the License. * */ -package org.apache.streampipes.manager.assets; -import org.apache.streampipes.config.backend.BackendConfig; +package org.apache.streampipes.connect.management.util; -public class AssetConstants { +import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage; +import org.apache.streampipes.storage.management.StorageDispatcher; - public static final String ASSET_BASE_DIR = BackendConfig.INSTANCE.getAssetDir(); +public class Utils { + public static ISpCoreConfigurationStorage getCoreConfigStorage() { + return StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getSpCoreConfigurationStorage(); + } } diff --git a/streampipes-data-export/src/main/java/org/apache/streampipes/export/utils/EventGroundingProcessor.java b/streampipes-data-export/src/main/java/org/apache/streampipes/export/utils/EventGroundingProcessor.java index 4ad57762a..72bd19e23 100644 --- a/streampipes-data-export/src/main/java/org/apache/streampipes/export/utils/EventGroundingProcessor.java +++ b/streampipes-data-export/src/main/java/org/apache/streampipes/export/utils/EventGroundingProcessor.java @@ -19,19 +19,30 @@ package org.apache.streampipes.export.utils; import org.apache.streampipes.commons.exceptions.SpRuntimeException; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.model.config.SpProtocol; +import org.apache.streampipes.model.configuration.SpCoreConfiguration; +import org.apache.streampipes.model.configuration.SpProtocol; import org.apache.streampipes.model.grounding.KafkaTransportProtocol; import org.apache.streampipes.model.grounding.MqttTransportProtocol; import org.apache.streampipes.model.grounding.NatsTransportProtocol; import org.apache.streampipes.model.grounding.PulsarTransportProtocol; import org.apache.streampipes.model.grounding.TransportProtocol; +import org.apache.streampipes.storage.management.StorageDispatcher; public class EventGroundingProcessor { + SpCoreConfiguration cfg; SpProtocol configuredProtocol; + public EventGroundingProcessor() { - this.configuredProtocol = BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedProtocols().get(0); + this.cfg = StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getSpCoreConfigurationStorage() + .get(); + + this.configuredProtocol = cfg + .getMessagingSettings() + .getPrioritizedProtocols().get(0); } public TransportProtocol applyOverride(TransportProtocol protocol) { @@ -66,28 +77,28 @@ public class EventGroundingProcessor { private KafkaTransportProtocol makeKafkaProtocol() { var protocol = new KafkaTransportProtocol(); - protocol.setBrokerHostname(BackendConfig.INSTANCE.getKafkaHost()); - protocol.setKafkaPort(BackendConfig.INSTANCE.getKafkaPort()); + protocol.setBrokerHostname(cfg.getKafkaHost()); + protocol.setKafkaPort(cfg.getKafkaPort()); return protocol; } private MqttTransportProtocol makeMqttProtocol() { var protocol = new MqttTransportProtocol(); - protocol.setBrokerHostname(BackendConfig.INSTANCE.getMqttHost()); - protocol.setPort(BackendConfig.INSTANCE.getMqttPort()); + protocol.setBrokerHostname(cfg.getMqttHost()); + protocol.setPort(cfg.getMqttPort()); return protocol; } private NatsTransportProtocol makeNatsProtocol() { var protocol = new NatsTransportProtocol(); - protocol.setBrokerHostname(BackendConfig.INSTANCE.getNatsHost()); - protocol.setPort(BackendConfig.INSTANCE.getNatsPort()); + protocol.setBrokerHostname(cfg.getNatsHost()); + protocol.setPort(cfg.getNatsPort()); return protocol; } private PulsarTransportProtocol makePulsarProtocol() { var protocol = new PulsarTransportProtocol(); - protocol.setBrokerHostname(BackendConfig.INSTANCE.getNatsHost()); + protocol.setBrokerHostname(cfg.getNatsHost()); return protocol; } diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java index 4042b4074..5f0ae1102 100644 --- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java +++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java @@ -18,10 +18,10 @@ package org.apache.streampipes.extensions.management.connect.adapter; -import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.connect.shared.AdapterPipelineGeneratorBase; import org.apache.streampipes.connect.shared.preprocessing.elements.TransformStreamAdapterElement; import org.apache.streampipes.connect.shared.preprocessing.transform.stream.DuplicateFilterPipelineElement; +import org.apache.streampipes.extensions.management.client.StreamPipesClientResolver; import org.apache.streampipes.extensions.management.connect.adapter.model.pipeline.AdapterPipeline; import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToBrokerAdapterSink; import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToJmsAdapterSink; @@ -29,7 +29,8 @@ import org.apache.streampipes.extensions.management.connect.adapter.preprocessin import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToMqttAdapterSink; import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToNatsAdapterSink; import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToPulsarAdapterSink; -import org.apache.streampipes.model.config.SpProtocol; +import org.apache.streampipes.model.configuration.MessagingSettings; +import org.apache.streampipes.model.configuration.SpProtocol; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.grounding.JmsTransportProtocol; import org.apache.streampipes.model.grounding.KafkaTransportProtocol; @@ -70,7 +71,7 @@ public class AdapterPipelineGenerator extends AdapterPipelineGeneratorBase { private SendToBrokerAdapterSink<?> getAdapterSink(AdapterDescription adapterDescription) { var prioritizedProtocol = - BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedProtocols().get(0); + getMessagingSettings().getPrioritizedProtocols().get(0); if (isPrioritized(prioritizedProtocol, JmsTransportProtocol.class)) { return new SendToJmsAdapterSink(adapterDescription); @@ -89,4 +90,9 @@ public class AdapterPipelineGenerator extends AdapterPipelineGeneratorBase { Class<?> protocolClass) { return prioritizedProtocol.getProtocolClass().equals(protocolClass.getCanonicalName()); } + + private MessagingSettings getMessagingSettings() { + var client = new StreamPipesClientResolver().makeStreamPipesClientInstance(); + return client.adminApi().getMessagingSettings(); + } } diff --git a/streampipes-mail/src/main/java/org/apache/streampipes/mail/AbstractMailer.java b/streampipes-mail/src/main/java/org/apache/streampipes/mail/AbstractMailer.java index 59d113b41..b974929c2 100644 --- a/streampipes-mail/src/main/java/org/apache/streampipes/mail/AbstractMailer.java +++ b/streampipes-mail/src/main/java/org/apache/streampipes/mail/AbstractMailer.java @@ -17,9 +17,9 @@ */ package org.apache.streampipes.mail; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.config.backend.model.EmailConfig; import org.apache.streampipes.mail.config.MailConfigurationBuilder; +import org.apache.streampipes.mail.utils.MailUtils; +import org.apache.streampipes.model.configuration.EmailConfig; import org.apache.streampipes.user.management.encryption.SecretEncryptionManager; import org.simplejavamail.api.email.Email; @@ -44,7 +44,7 @@ public class AbstractMailer { } protected EmailConfig getEmailConfig() { - EmailConfig config = BackendConfig.INSTANCE.getEmailConfig(); + EmailConfig config = MailUtils.getSpCoreConfiguration().getEmailConfig(); return getDecryptedEmailConfig(config); } diff --git a/streampipes-mail/src/main/java/org/apache/streampipes/mail/MailTester.java b/streampipes-mail/src/main/java/org/apache/streampipes/mail/MailTester.java index d39749a56..1224332ca 100644 --- a/streampipes-mail/src/main/java/org/apache/streampipes/mail/MailTester.java +++ b/streampipes-mail/src/main/java/org/apache/streampipes/mail/MailTester.java @@ -17,7 +17,7 @@ */ package org.apache.streampipes.mail; -import org.apache.streampipes.config.backend.model.EmailConfig; +import org.apache.streampipes.model.configuration.EmailConfig; import org.apache.streampipes.mail.template.TestMailTemplate; import org.apache.streampipes.mail.utils.MailUtils; diff --git a/streampipes-mail/src/main/java/org/apache/streampipes/mail/config/MailConfigurationBuilder.java b/streampipes-mail/src/main/java/org/apache/streampipes/mail/config/MailConfigurationBuilder.java index 5efcbdb24..cdddd5d1e 100644 --- a/streampipes-mail/src/main/java/org/apache/streampipes/mail/config/MailConfigurationBuilder.java +++ b/streampipes-mail/src/main/java/org/apache/streampipes/mail/config/MailConfigurationBuilder.java @@ -17,7 +17,7 @@ */ package org.apache.streampipes.mail.config; -import org.apache.streampipes.config.backend.model.EmailConfig; +import org.apache.streampipes.model.configuration.EmailConfig; import org.simplejavamail.api.mailer.Mailer; import org.simplejavamail.api.mailer.config.TransportStrategy; @@ -60,10 +60,10 @@ public class MailConfigurationBuilder { } private TransportStrategy toTransportStrategy( - org.apache.streampipes.config.backend.model.TransportStrategy strategy) { - if (strategy == org.apache.streampipes.config.backend.model.TransportStrategy.SMTP) { + org.apache.streampipes.model.configuration.TransportStrategy strategy) { + if (strategy == org.apache.streampipes.model.configuration.TransportStrategy.SMTP) { return TransportStrategy.SMTP; - } else if (strategy == org.apache.streampipes.config.backend.model.TransportStrategy.SMTPS) { + } else if (strategy == org.apache.streampipes.model.configuration.TransportStrategy.SMTPS) { return TransportStrategy.SMTPS; } else { return TransportStrategy.SMTP_TLS; diff --git a/streampipes-mail/src/main/java/org/apache/streampipes/mail/utils/MailUtils.java b/streampipes-mail/src/main/java/org/apache/streampipes/mail/utils/MailUtils.java index 317f53496..4a6619436 100644 --- a/streampipes-mail/src/main/java/org/apache/streampipes/mail/utils/MailUtils.java +++ b/streampipes-mail/src/main/java/org/apache/streampipes/mail/utils/MailUtils.java @@ -18,8 +18,9 @@ package org.apache.streampipes.mail.utils; import org.apache.streampipes.commons.resources.Resources; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.config.backend.model.GeneralConfig; +import org.apache.streampipes.model.configuration.GeneralConfig; +import org.apache.streampipes.model.configuration.SpCoreConfiguration; +import org.apache.streampipes.storage.management.StorageDispatcher; import java.io.IOException; import java.nio.charset.StandardCharsets; @@ -27,16 +28,20 @@ import java.nio.charset.StandardCharsets; public class MailUtils { public static String extractBaseUrl() { - GeneralConfig config = BackendConfig.INSTANCE.getGeneralConfig(); + GeneralConfig config = getSpCoreConfiguration().getGeneralConfig(); return config.getProtocol() + "://" + config.getHostname() + ":" + config.getPort(); } public static String extractAppName() { - return BackendConfig.INSTANCE.getGeneralConfig().getAppName(); + return getSpCoreConfiguration().getGeneralConfig().getAppName(); } public static String readResourceFileToString(String filename) throws IOException { return Resources.asString(filename, StandardCharsets.UTF_8); } + + public static SpCoreConfiguration getSpCoreConfiguration() { + return StorageDispatcher.INSTANCE.getNoSqlStore().getSpCoreConfigurationStorage().get(); + } } diff --git a/streampipes-model/pom.xml b/streampipes-model/pom.xml index 059059f41..f77db63a9 100644 --- a/streampipes-model/pom.xml +++ b/streampipes-model/pom.xml @@ -42,6 +42,11 @@ <artifactId>streampipes-logging</artifactId> <version>0.93.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-service-discovery-api</artifactId> + <version>0.93.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-vocabulary</artifactId> diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/EmailConfig.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/EmailConfig.java similarity index 98% rename from streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/EmailConfig.java rename to streampipes-model/src/main/java/org/apache/streampipes/model/configuration/EmailConfig.java index ca247c4c2..57e243fb1 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/EmailConfig.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/EmailConfig.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.config.backend.model; +package org.apache.streampipes.model.configuration; public class EmailConfig { diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/GeneralConfig.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/GeneralConfig.java similarity index 98% rename from streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/GeneralConfig.java rename to streampipes-model/src/main/java/org/apache/streampipes/model/configuration/GeneralConfig.java index e2d0dfca3..2f3968d96 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/GeneralConfig.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/GeneralConfig.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.config.backend.model; +package org.apache.streampipes.model.configuration; import java.util.List; diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/JwtSigningMode.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/JwtSigningMode.java similarity index 94% rename from streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/JwtSigningMode.java rename to streampipes-model/src/main/java/org/apache/streampipes/model/configuration/JwtSigningMode.java index 7bc74e183..d48fe6dee 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/JwtSigningMode.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/JwtSigningMode.java @@ -16,7 +16,7 @@ * */ -package org.apache.streampipes.config.backend.model; +package org.apache.streampipes.model.configuration; public enum JwtSigningMode { HMAC, diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/LocalAuthConfig.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/LocalAuthConfig.java similarity index 98% rename from streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/LocalAuthConfig.java rename to streampipes-model/src/main/java/org/apache/streampipes/model/configuration/LocalAuthConfig.java index 044eb0efc..3137f0d7b 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/LocalAuthConfig.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/LocalAuthConfig.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.config.backend.model; +package org.apache.streampipes.model.configuration; public class LocalAuthConfig { diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/config/MessagingSettings.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java similarity index 98% rename from streampipes-model/src/main/java/org/apache/streampipes/model/config/MessagingSettings.java rename to streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java index 277aa3bf3..6796db780 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/config/MessagingSettings.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.model.config; +package org.apache.streampipes.model.configuration; import java.util.List; diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfiguration.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfiguration.java new file mode 100644 index 000000000..cf192d567 --- /dev/null +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpCoreConfiguration.java @@ -0,0 +1,219 @@ +/* + * 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.model.configuration; + +import com.google.gson.annotations.SerializedName; + +public class SpCoreConfiguration { + + public static final String ID = "core"; + + protected @SerializedName("_rev") String rev; + private @SerializedName("_id") String id = ID; + + private MessagingSettings messagingSettings; + private LocalAuthConfig localAuthConfig; + private EmailConfig emailConfig; + private GeneralConfig generalConfig; + + private String jmsHost; + private int jmsPort; + + private String mqttHost; + private int mqttPort; + + private String natsHost; + private int natsPort; + + private String kafkaHost; + private int kafkaPort; + + private String pulsarUrl; + + private String zookeeperHost; + private int zookeeperPort; + + private boolean isConfigured; + + private String assetDir; + private String filesDir; + + public SpCoreConfiguration() { + } + + public String getRev() { + return rev; + } + + public void setRev(String rev) { + this.rev = rev; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public MessagingSettings getMessagingSettings() { + return messagingSettings; + } + + public void setMessagingSettings(MessagingSettings messagingSettings) { + this.messagingSettings = messagingSettings; + } + + public LocalAuthConfig getLocalAuthConfig() { + return localAuthConfig; + } + + public void setLocalAuthConfig(LocalAuthConfig localAuthConfig) { + this.localAuthConfig = localAuthConfig; + } + + public EmailConfig getEmailConfig() { + return emailConfig; + } + + public void setEmailConfig(EmailConfig emailConfig) { + this.emailConfig = emailConfig; + } + + public GeneralConfig getGeneralConfig() { + return generalConfig; + } + + public void setGeneralConfig(GeneralConfig generalConfig) { + this.generalConfig = generalConfig; + } + + public String getJmsHost() { + return jmsHost; + } + + public void setJmsHost(String jmsHost) { + this.jmsHost = jmsHost; + } + + public int getJmsPort() { + return jmsPort; + } + + public void setJmsPort(int jmsPort) { + this.jmsPort = jmsPort; + } + + public String getMqttHost() { + return mqttHost; + } + + public void setMqttHost(String mqttHost) { + this.mqttHost = mqttHost; + } + + public int getMqttPort() { + return mqttPort; + } + + public void setMqttPort(int mqttPort) { + this.mqttPort = mqttPort; + } + + public String getNatsHost() { + return natsHost; + } + + public void setNatsHost(String natsHost) { + this.natsHost = natsHost; + } + + public int getNatsPort() { + return natsPort; + } + + public void setNatsPort(int natsPort) { + this.natsPort = natsPort; + } + + public String getKafkaHost() { + return kafkaHost; + } + + public void setKafkaHost(String kafkaHost) { + this.kafkaHost = kafkaHost; + } + + public int getKafkaPort() { + return kafkaPort; + } + + public void setKafkaPort(int kafkaPort) { + this.kafkaPort = kafkaPort; + } + + public String getPulsarUrl() { + return pulsarUrl; + } + + public void setPulsarUrl(String pulsarUrl) { + this.pulsarUrl = pulsarUrl; + } + + public String getZookeeperHost() { + return zookeeperHost; + } + + public void setZookeeperHost(String zookeeperHost) { + this.zookeeperHost = zookeeperHost; + } + + public int getZookeeperPort() { + return zookeeperPort; + } + + public void setZookeeperPort(int zookeeperPort) { + this.zookeeperPort = zookeeperPort; + } + + public boolean isConfigured() { + return isConfigured; + } + + public void setConfigured(boolean configured) { + isConfigured = configured; + } + + public String getAssetDir() { + return assetDir; + } + + public void setAssetDir(String assetDir) { + this.assetDir = assetDir; + } + + public String getFilesDir() { + return filesDir; + } + + public void setFilesDir(String filesDir) { + this.filesDir = filesDir; + } +} diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/config/SpDataFormat.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpDataFormat.java similarity index 96% rename from streampipes-model/src/main/java/org/apache/streampipes/model/config/SpDataFormat.java rename to streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpDataFormat.java index c409eeaab..a2c5b6bf3 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/config/SpDataFormat.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpDataFormat.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.model.config; +package org.apache.streampipes.model.configuration; import org.apache.streampipes.vocabulary.MessageFormat; diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/config/SpProtocol.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpProtocol.java similarity index 96% rename from streampipes-model/src/main/java/org/apache/streampipes/model/config/SpProtocol.java rename to streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpProtocol.java index af7af7263..657ae968d 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/config/SpProtocol.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpProtocol.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.model.config; +package org.apache.streampipes.model.configuration; public enum SpProtocol { diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/TransportStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/TransportStrategy.java similarity index 94% rename from streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/TransportStrategy.java rename to streampipes-model/src/main/java/org/apache/streampipes/model/configuration/TransportStrategy.java index ffb701eee..4155f789d 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/model/TransportStrategy.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/TransportStrategy.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.config.backend.model; +package org.apache.streampipes.model.configuration; public enum TransportStrategy { SMTP, diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java index eda79448f..67ad1e198 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java @@ -17,10 +17,15 @@ */ package org.apache.streampipes.manager.assets; -import org.apache.streampipes.config.backend.BackendConfig; +import org.apache.streampipes.storage.management.StorageDispatcher; public class AssetConstants { - public static final String ASSET_BASE_DIR = BackendConfig.INSTANCE.getAssetDir(); + public static final String ASSET_BASE_DIR = StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getSpCoreConfigurationStorage() + .get() + .getAssetDir(); } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/file/FileConstants.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/file/FileConstants.java index 7ed491315..372ddda1d 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/file/FileConstants.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/file/FileConstants.java @@ -17,9 +17,14 @@ */ package org.apache.streampipes.manager.file; -import org.apache.streampipes.config.backend.BackendConfig; +import org.apache.streampipes.storage.management.StorageDispatcher; public class FileConstants { - public static final String FILES_BASE_DIR = BackendConfig.INSTANCE.getFilesDir(); + public static final String FILES_BASE_DIR = StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getSpCoreConfigurationStorage() + .get() + .getFilesDir(); } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/FormatSelector.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/FormatSelector.java index 128098f7c..7b3830117 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/FormatSelector.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/FormatSelector.java @@ -18,12 +18,12 @@ package org.apache.streampipes.manager.matching; -import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.model.SpDataStream; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; import org.apache.streampipes.model.base.NamedStreamPipesEntity; -import org.apache.streampipes.model.config.SpDataFormat; +import org.apache.streampipes.model.configuration.SpDataFormat; import org.apache.streampipes.model.grounding.TransportFormat; +import org.apache.streampipes.storage.management.StorageDispatcher; import org.apache.streampipes.vocabulary.MessageFormat; import java.net.URI; @@ -46,11 +46,17 @@ public class FormatSelector extends GroundingSelector { .get(0); } else { List<SpDataFormat> prioritizedFormats = - BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedFormats(); + StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getSpCoreConfigurationStorage() + .get() + .getMessagingSettings() + .getPrioritizedFormats(); List<SpDataFormat> supportedFormats = prioritizedFormats .stream() - .filter(pf -> supportsFormat(pf.getMessageFormat())).collect(Collectors.toList()); + .filter(pf -> supportsFormat(pf.getMessageFormat())).toList(); if (supportedFormats.size() > 0) { return new TransportFormat(supportedFormats.get(0).getMessageFormat()); @@ -60,7 +66,7 @@ public class FormatSelector extends GroundingSelector { } } - public <T extends TransportFormat> boolean supportsFormat(String format) { + public boolean supportsFormat(String format) { List<InvocableStreamPipesEntity> elements = buildInvocables(); return elements .stream() diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/ProtocolSelector.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/ProtocolSelector.java index 30a152343..ea6e0b649 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/ProtocolSelector.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/ProtocolSelector.java @@ -18,12 +18,12 @@ package org.apache.streampipes.manager.matching; -import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.manager.util.TopicGenerator; import org.apache.streampipes.model.SpDataStream; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; import org.apache.streampipes.model.base.NamedStreamPipesEntity; -import org.apache.streampipes.model.config.SpProtocol; +import org.apache.streampipes.model.configuration.SpCoreConfiguration; +import org.apache.streampipes.model.configuration.SpProtocol; import org.apache.streampipes.model.grounding.JmsTransportProtocol; import org.apache.streampipes.model.grounding.KafkaTransportProtocol; import org.apache.streampipes.model.grounding.MqttTransportProtocol; @@ -31,6 +31,7 @@ import org.apache.streampipes.model.grounding.NatsTransportProtocol; import org.apache.streampipes.model.grounding.PulsarTransportProtocol; import org.apache.streampipes.model.grounding.SimpleTopicDefinition; import org.apache.streampipes.model.grounding.TransportProtocol; +import org.apache.streampipes.storage.management.StorageDispatcher; import java.util.List; import java.util.Set; @@ -40,11 +41,19 @@ public class ProtocolSelector extends GroundingSelector { private final String outputTopic; private final List<SpProtocol> prioritizedProtocols; + private final SpCoreConfiguration cfg; + public ProtocolSelector(NamedStreamPipesEntity source, Set<InvocableStreamPipesEntity> targets) { super(source, targets); this.outputTopic = TopicGenerator.generateRandomTopic(); + this.cfg = StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getSpCoreConfigurationStorage() + .get(); + this.prioritizedProtocols = - BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedProtocols(); + cfg.getMessagingSettings().getPrioritizedProtocols(); } public TransportProtocol getPreferredProtocol() { @@ -68,7 +77,7 @@ public class ProtocolSelector extends GroundingSelector { return natsTopic(); } else if (prioritizedProtocol.getProtocolClass().equals(PulsarTransportProtocol.class.getCanonicalName()) && supportsProtocol(PulsarTransportProtocol.class)) { - return new PulsarTransportProtocol(BackendConfig.INSTANCE.getPulsarUrl(), + return new PulsarTransportProtocol(cfg.getPulsarUrl(), new SimpleTopicDefinition(outputTopic)); } } @@ -77,30 +86,37 @@ public class ProtocolSelector extends GroundingSelector { } private TransportProtocol mqttTopic() { - return new MqttTransportProtocol(BackendConfig.INSTANCE.getMqttHost(), - BackendConfig.INSTANCE.getMqttPort(), - outputTopic); + return new MqttTransportProtocol( + cfg.getMqttHost(), + cfg.getMqttPort(), + outputTopic + ); } private TransportProtocol jmsTopic() { - return new JmsTransportProtocol(BackendConfig.INSTANCE.getJmsHost(), - BackendConfig.INSTANCE.getJmsPort(), - outputTopic); + return new JmsTransportProtocol( + cfg.getJmsHost(), + cfg.getJmsPort(), + outputTopic + ); } private TransportProtocol natsTopic() { return new NatsTransportProtocol( - BackendConfig.INSTANCE.getNatsHost(), - BackendConfig.INSTANCE.getNatsPort(), - outputTopic); + cfg.getNatsHost(), + cfg.getNatsPort(), + outputTopic + ); } private TransportProtocol kafkaTopic() { - return new KafkaTransportProtocol(BackendConfig.INSTANCE.getKafkaHost(), - BackendConfig.INSTANCE.getKafkaPort(), + return new KafkaTransportProtocol( + cfg.getKafkaHost(), + cfg.getKafkaPort(), outputTopic, - BackendConfig.INSTANCE.getZookeeperHost(), - BackendConfig.INSTANCE.getZookeeperPort()); + cfg.getZookeeperHost(), + cfg.getZookeeperPort() + ); } @@ -114,6 +130,5 @@ public class ProtocolSelector extends GroundingSelector { .getTransportProtocols() .stream() .anyMatch(protocol::isInstance)); - } } diff --git a/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java b/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java index 10c74f619..d3edd0090 100644 --- a/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java +++ b/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java @@ -32,6 +32,7 @@ import org.apache.streampipes.storage.api.INotificationStorage; import org.apache.streampipes.storage.api.IPipelineElementDescriptionStorageCache; import org.apache.streampipes.storage.api.IPipelineElementTemplateStorage; import org.apache.streampipes.storage.api.IPipelineStorage; +import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage; import org.apache.streampipes.storage.api.IUserStorage; import org.apache.streampipes.storage.api.IVisualizationStorage; import org.apache.streampipes.storage.management.StorageDispatcher; @@ -51,6 +52,10 @@ import java.net.URLDecoder; public abstract class AbstractRestResource extends AbstractSharedRestInterface { + protected ISpCoreConfigurationStorage getSpCoreConfigurationStorage() { + return getNoSqlStorage().getSpCoreConfigurationStorage(); + } + protected IPipelineElementDescriptionStorageCache getPipelineElementRdfStorage() { return StorageManager.INSTANCE.getPipelineElementStorage(); } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Authentication.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Authentication.java index 5d7a5e151..ad0dfc6f1 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Authentication.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Authentication.java @@ -20,13 +20,12 @@ package org.apache.streampipes.rest.impl; import org.apache.streampipes.commons.exceptions.UserNotFoundException; import org.apache.streampipes.commons.exceptions.UsernameAlreadyTakenException; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.config.backend.model.GeneralConfig; import org.apache.streampipes.model.client.user.JwtAuthenticationResponse; import org.apache.streampipes.model.client.user.LoginRequest; import org.apache.streampipes.model.client.user.Principal; import org.apache.streampipes.model.client.user.RegistrationData; import org.apache.streampipes.model.client.user.UserAccount; +import org.apache.streampipes.model.configuration.GeneralConfig; import org.apache.streampipes.model.message.ErrorMessage; import org.apache.streampipes.model.message.NotificationType; import org.apache.streampipes.model.message.Notifications; @@ -97,7 +96,7 @@ public class Authentication extends AbstractRestResource { @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.APPLICATION_JSON) public Response doRegister(RegistrationData data) { - GeneralConfig config = BackendConfig.INSTANCE.getGeneralConfig(); + GeneralConfig config = getSpCoreConfigurationStorage().get().getGeneralConfig(); if (!config.isAllowSelfRegistration()) { throw new WebApplicationException(Response.Status.FORBIDDEN); } @@ -129,7 +128,7 @@ public class Authentication extends AbstractRestResource { @JacksonSerialized @Produces(MediaType.APPLICATION_JSON) public Response getAuthSettings() { - GeneralConfig config = BackendConfig.INSTANCE.getGeneralConfig(); + GeneralConfig config = getSpCoreConfigurationStorage().get().getGeneralConfig(); Map<String, Object> response = new HashMap<>(); response.put("allowSelfRegistration", config.isAllowSelfRegistration()); response.put("allowPasswordRecovery", config.isAllowPasswordRecovery()); diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/EmailResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/EmailResource.java index 1ebbbecce..84d80ce88 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/EmailResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/EmailResource.java @@ -17,7 +17,6 @@ */ package org.apache.streampipes.rest.impl; -import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.mail.MailSender; import org.apache.streampipes.model.mail.SpEmail; import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; @@ -35,7 +34,7 @@ public class EmailResource extends AbstractAuthGuardedRestResource { @POST @Consumes(MediaType.APPLICATION_JSON) public Response sendEmail(SpEmail email) { - if (BackendConfig.INSTANCE.getEmailConfig().isEmailConfigured()) { + if (getSpCoreConfigurationStorage().get().getEmailConfig().isEmailConfigured()) { new MailSender().sendEmail(email); return ok(); } else { diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ConsulConfig.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ConsulConfig.java index ab1acffac..26e650c03 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ConsulConfig.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ConsulConfig.java @@ -19,8 +19,7 @@ package org.apache.streampipes.rest.impl.admin; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.model.config.MessagingSettings; +import org.apache.streampipes.model.configuration.MessagingSettings; import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; import org.apache.streampipes.rest.security.AuthConstants; import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; @@ -165,7 +164,7 @@ public class ConsulConfig extends AbstractAuthGuardedRestResource { @Path("/messaging") @PreAuthorize(AuthConstants.IS_ADMIN_ROLE) public Response getMessagingSettings() { - return ok(BackendConfig.INSTANCE.getMessagingSettings()); + return ok(getSpCoreConfigurationStorage().get().getMessagingSettings()); } @POST @@ -175,7 +174,9 @@ public class ConsulConfig extends AbstractAuthGuardedRestResource { @Path("messaging") @PreAuthorize(AuthConstants.IS_ADMIN_ROLE) public Response updateMessagingSettings(MessagingSettings messagingSettings) { - BackendConfig.INSTANCE.setMessagingSettings(messagingSettings); + var cfg = getSpCoreConfigurationStorage().get(); + cfg.setMessagingSettings(messagingSettings); + getSpCoreConfigurationStorage().updateElement(cfg); return ok(); } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/EmailConfigurationResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/EmailConfigurationResource.java index a5c18094f..a14e8f7bc 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/EmailConfigurationResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/EmailConfigurationResource.java @@ -17,9 +17,8 @@ */ package org.apache.streampipes.rest.impl.admin; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.config.backend.model.EmailConfig; import org.apache.streampipes.mail.MailTester; +import org.apache.streampipes.model.configuration.EmailConfig; import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; import org.apache.streampipes.rest.security.AuthConstants; import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; @@ -49,7 +48,7 @@ public class EmailConfigurationResource extends AbstractAuthGuardedRestResource @Produces(MediaType.APPLICATION_JSON) @PreAuthorize(AuthConstants.IS_ADMIN_ROLE) public Response getMailConfiguration() { - return ok(BackendConfig.INSTANCE.getEmailConfig()); + return ok(getSpCoreConfigurationStorage().get().getEmailConfig()); } @PUT @@ -67,7 +66,10 @@ public class EmailConfigurationResource extends AbstractAuthGuardedRestResource config.setSmtpPassword(SecretEncryptionManager.encrypt(config.getSmtpPassword())); config.setSmtpPassEncrypted(true); } - BackendConfig.INSTANCE.updateEmailConfig(config); + var storage = getSpCoreConfigurationStorage(); + var cfg = storage.get(); + cfg.setEmailConfig(config); + storage.updateElement(cfg); return ok(); } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/GeneralConfigurationResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/GeneralConfigurationResource.java index 4b32fc147..df1a06a05 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/GeneralConfigurationResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/GeneralConfigurationResource.java @@ -17,8 +17,7 @@ */ package org.apache.streampipes.rest.impl.admin; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.config.backend.model.GeneralConfig; +import org.apache.streampipes.model.configuration.GeneralConfig; import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; import org.apache.streampipes.rest.security.AuthConstants; import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; @@ -51,7 +50,7 @@ public class GeneralConfigurationResource extends AbstractAuthGuardedRestResourc @Produces(MediaType.APPLICATION_JSON) @PreAuthorize(AuthConstants.IS_ADMIN_ROLE) public GeneralConfig getGeneralConfiguration() { - return BackendConfig.INSTANCE.getGeneralConfig(); + return getSpCoreConfigurationStorage().get().getGeneralConfig(); } @PUT @@ -60,7 +59,10 @@ public class GeneralConfigurationResource extends AbstractAuthGuardedRestResourc @PreAuthorize(AuthConstants.IS_ADMIN_ROLE) public Response updateGeneralConfiguration(GeneralConfig config) { config.setConfigured(true); - BackendConfig.INSTANCE.updateGeneralConfig(config); + var storage = getSpCoreConfigurationStorage(); + var cfg = storage.get(); + cfg.setGeneralConfig(config); + storage.updateElement(cfg); return ok(); } diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java index 969b760ea..7ad6db969 100644 --- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java +++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java @@ -128,8 +128,7 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { this.healthCheckExecutorService = Executors.newSingleThreadScheduledExecutor(); this.logCheckExecutorService = Executors.newSingleThreadScheduledExecutor(); - // TODO - //new StreamPipesEnvChecker().updateEnvironmentVariables(); + new StreamPipesEnvChecker().updateEnvironmentVariables(); new CouchDbViewGenerator().createGenericDatabaseIfNotExists(); if (!isConfigured()) { @@ -150,8 +149,6 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { LOG_FETCH_INTERVAL, LOG_FETCH_INTERVAL, LOG_FETCH_UNIT); - - } private boolean isConfigured() { diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesEnvChecker.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesEnvChecker.java index 64e187c78..f200276a5 100644 --- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesEnvChecker.java +++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesEnvChecker.java @@ -20,9 +20,11 @@ package org.apache.streampipes.service.core; import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.environment.Environments; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.config.backend.model.JwtSigningMode; -import org.apache.streampipes.config.backend.model.LocalAuthConfig; +import org.apache.streampipes.model.configuration.JwtSigningMode; +import org.apache.streampipes.model.configuration.LocalAuthConfig; +import org.apache.streampipes.model.configuration.SpCoreConfiguration; +import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage; +import org.apache.streampipes.storage.management.StorageDispatcher; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -35,7 +37,8 @@ public class StreamPipesEnvChecker { private static final Logger LOG = LoggerFactory.getLogger(StreamPipesEnvChecker.class); - BackendConfig coreConfig; + private ISpCoreConfigurationStorage configStorage; + private SpCoreConfiguration coreConfig; private Environment env; @@ -44,7 +47,12 @@ public class StreamPipesEnvChecker { } public void updateEnvironmentVariables() { - this.coreConfig = BackendConfig.INSTANCE; + this.configStorage = StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getSpCoreConfigurationStorage(); + + this.coreConfig = configStorage.get(); LOG.info("Checking and updating environment variables..."); updateJwtSettings(); @@ -92,7 +100,8 @@ public class StreamPipesEnvChecker { } if (!incompleteConfig) { LOG.info("Updating local auth config with signing mode {}", localAuthConfig.getJwtSigningMode().name()); - coreConfig.updateLocalAuthConfig(localAuthConfig); + coreConfig.setLocalAuthConfig(localAuthConfig); + configStorage.updateElement(coreConfig); } } diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/AvailableMigrations.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/AvailableMigrations.java index 6a8e81160..3336f462d 100644 --- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/AvailableMigrations.java +++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/AvailableMigrations.java @@ -24,6 +24,7 @@ import org.apache.streampipes.service.core.migrations.v070.CreateDefaultAssetMig import org.apache.streampipes.service.core.migrations.v070.CreateFileAssetTypeMigration; import org.apache.streampipes.service.core.migrations.v090.UpdateUsernameViewMigration; import org.apache.streampipes.service.core.migrations.v093.AdapterMigration; +import org.apache.streampipes.service.core.migrations.v093.ConsulConfigMigration; import java.util.Arrays; import java.util.List; @@ -36,7 +37,8 @@ public class AvailableMigrations { new CreateDefaultAssetMigration(), new CreateFileAssetTypeMigration(), new UpdateUsernameViewMigration(), - new AdapterMigration() + new AdapterMigration(), + new ConsulConfigMigration() ); } } diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java new file mode 100644 index 000000000..eebc1c356 --- /dev/null +++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/ConsulConfigMigration.java @@ -0,0 +1,71 @@ +/* + * 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.service.core.migrations.v093; + +import org.apache.streampipes.config.backend.BackendConfig; +import org.apache.streampipes.model.configuration.SpCoreConfiguration; +import org.apache.streampipes.service.core.migrations.Migration; +import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage; +import org.apache.streampipes.storage.management.StorageDispatcher; + +public class ConsulConfigMigration implements Migration { + + private final ISpCoreConfigurationStorage storage; + + public ConsulConfigMigration() { + this.storage = StorageDispatcher.INSTANCE.getNoSqlStore().getSpCoreConfigurationStorage(); + } + + @Override + public boolean shouldExecute() { + return storage.getAll().size() == 0; + } + + @Override + public void executeMigration() { + var currConf = BackendConfig.INSTANCE; + var newConf = new SpCoreConfiguration(); + + newConf.setMessagingSettings(currConf.getMessagingSettings()); + newConf.setLocalAuthConfig(currConf.getLocalAuthConfig()); + newConf.setEmailConfig(currConf.getEmailConfig()); + newConf.setGeneralConfig(currConf.getGeneralConfig()); + + newConf.setJmsHost(currConf.getJmsHost()); + newConf.setJmsPort(currConf.getJmsPort()); + newConf.setMqttHost(currConf.getMqttHost()); + newConf.setMqttPort(currConf.getMqttPort()); + newConf.setNatsHost(currConf.getNatsHost()); + newConf.setNatsPort(currConf.getNatsPort()); + newConf.setKafkaHost(currConf.getKafkaHost()); + newConf.setKafkaPort(currConf.getKafkaPort()); + newConf.setPulsarUrl(currConf.getPulsarUrl()); + newConf.setZookeeperHost(currConf.getZookeeperHost()); + newConf.setZookeeperPort(currConf.getZookeeperPort()); + newConf.setAssetDir(currConf.getAssetDir()); + newConf.setFilesDir(currConf.getFilesDir()); + + storage.createElement(newConf); + } + + @Override + public String getDescription() { + return "Migrate Consul configurations to CouchDB"; + } +} diff --git a/streampipes-service-discovery/pom.xml b/streampipes-service-discovery/pom.xml index 3f1944a47..f7bb3c11f 100644 --- a/streampipes-service-discovery/pom.xml +++ b/streampipes-service-discovery/pom.xml @@ -38,6 +38,11 @@ <artifactId>streampipes-service-discovery-api</artifactId> <version>0.93.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-service-discovery-consul</artifactId> + <version>0.93.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-storage-management</artifactId> diff --git a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java index e192ad76b..a488b4310 100644 --- a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java +++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java @@ -22,6 +22,7 @@ import org.apache.streampipes.commons.environment.Environments; import org.apache.streampipes.svcdiscovery.api.ISpKvManagement; import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; import org.apache.streampipes.svcdiscovery.api.SpConfig; +import org.apache.streampipes.svcdiscovery.consul.ConsulSpConfig; public class SpServiceDiscovery { @@ -47,8 +48,7 @@ public class SpServiceDiscovery { public static SpConfig getSpConfig(String serviceGroup, Environment environment) { - // TODO can probably be removed? - return null; + return new ConsulSpConfig(serviceGroup, environment); } } diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java index eee65bd0f..9e26868c3 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java @@ -87,4 +87,6 @@ public interface INoSqlStorage { CRUDStorage<String, SpServiceRegistrationRequest> getExtensionsServiceStorage(); CRUDStorage<String, SpServiceConfiguration> getExtensionsServiceConfigurationStorage(); + + ISpCoreConfigurationStorage getSpCoreConfigurationStorage(); } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/ISpCoreConfigurationStorage.java similarity index 67% copy from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java copy to streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/ISpCoreConfigurationStorage.java index eda79448f..829dc09ce 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetConstants.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/ISpCoreConfigurationStorage.java @@ -15,12 +15,22 @@ * limitations under the License. * */ -package org.apache.streampipes.manager.assets; -import org.apache.streampipes.config.backend.BackendConfig; +package org.apache.streampipes.storage.api; -public class AssetConstants { +import org.apache.streampipes.model.configuration.SpCoreConfiguration; - public static final String ASSET_BASE_DIR = BackendConfig.INSTANCE.getAssetDir(); +import java.util.List; +public interface ISpCoreConfigurationStorage { + + List<SpCoreConfiguration> getAll(); + + void createElement(SpCoreConfiguration element); + + SpCoreConfiguration get(); + + SpCoreConfiguration updateElement(SpCoreConfiguration element); + + void deleteElement(); } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java index 6318552f8..e3a64167b 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java @@ -44,6 +44,7 @@ import org.apache.streampipes.storage.api.IPipelineElementDescriptionStorageCach import org.apache.streampipes.storage.api.IPipelineElementTemplateStorage; import org.apache.streampipes.storage.api.IPipelineMonitoringDataStorage; import org.apache.streampipes.storage.api.IPipelineStorage; +import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage; import org.apache.streampipes.storage.api.IUserActivationTokenStorage; import org.apache.streampipes.storage.api.IUserGroupStorage; import org.apache.streampipes.storage.api.IUserStorage; @@ -53,6 +54,7 @@ import org.apache.streampipes.storage.couchdb.impl.AdapterInstanceStorageImpl; import org.apache.streampipes.storage.couchdb.impl.AssetDashboardStorageImpl; import org.apache.streampipes.storage.couchdb.impl.CategoryStorageImpl; import org.apache.streampipes.storage.couchdb.impl.ConnectionStorageImpl; +import org.apache.streampipes.storage.couchdb.impl.CoreConfigurationStorageImpl; import org.apache.streampipes.storage.couchdb.impl.DashboardStorageImpl; import org.apache.streampipes.storage.couchdb.impl.DashboardWidgetStorageImpl; import org.apache.streampipes.storage.couchdb.impl.DataExplorerDashboardStorageImpl; @@ -254,5 +256,10 @@ public enum CouchDbStorageManager implements INoSqlStorage { return new ExtensionsServiceConfigStorageImpl(); } + @Override + public ISpCoreConfigurationStorage getSpCoreConfigurationStorage() { + return new CoreConfigurationStorageImpl(); + } + } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/CoreConfigurationStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/CoreConfigurationStorageImpl.java new file mode 100644 index 000000000..7ff0a99c0 --- /dev/null +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/CoreConfigurationStorageImpl.java @@ -0,0 +1,61 @@ +/* + * 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.storage.couchdb.impl; + +import org.apache.streampipes.model.configuration.SpCoreConfiguration; +import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage; +import org.apache.streampipes.storage.couchdb.dao.AbstractDao; +import org.apache.streampipes.storage.couchdb.utils.Utils; + +import java.util.List; + +public class CoreConfigurationStorageImpl extends AbstractDao<SpCoreConfiguration> + implements ISpCoreConfigurationStorage { + + + public CoreConfigurationStorageImpl() { + super(Utils::getCouchDbGeneralConfigStorage, SpCoreConfiguration.class); + } + + @Override + public List<SpCoreConfiguration> getAll() { + return findAll(); + } + + @Override + public void createElement(SpCoreConfiguration element) { + persist(element); + } + + @Override + public SpCoreConfiguration get() { + return findWithNullIfEmpty(SpCoreConfiguration.ID); + } + + @Override + public SpCoreConfiguration updateElement(SpCoreConfiguration element) { + update(element); + return get(); + } + + @Override + public void deleteElement() { + delete(SpCoreConfiguration.ID); + } +} diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java index 30da73daf..b9b48c303 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java @@ -60,6 +60,10 @@ public class Utils { return getCouchDbGsonClient("pipelinecanvasmetadata"); } + public static CouchDbClient getCouchDbGeneralConfigStorage() { + return getCouchDbGsonClient("general-configuration"); + } + public static CouchDbClient getCouchDbCategoryClient() { return getCouchDbGsonClient("category"); } diff --git a/streampipes-user-management/src/main/java/org/apache/streampipes/user/management/jwt/JwtTokenProvider.java b/streampipes-user-management/src/main/java/org/apache/streampipes/user/management/jwt/JwtTokenProvider.java index 4abb09a3c..c4af4619e 100644 --- a/streampipes-user-management/src/main/java/org/apache/streampipes/user/management/jwt/JwtTokenProvider.java +++ b/streampipes-user-management/src/main/java/org/apache/streampipes/user/management/jwt/JwtTokenProvider.java @@ -20,13 +20,14 @@ package org.apache.streampipes.user.management.jwt; import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.environment.Environments; -import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.config.backend.model.JwtSigningMode; -import org.apache.streampipes.config.backend.model.LocalAuthConfig; import org.apache.streampipes.model.client.user.Principal; +import org.apache.streampipes.model.configuration.JwtSigningMode; +import org.apache.streampipes.model.configuration.LocalAuthConfig; +import org.apache.streampipes.model.configuration.SpCoreConfiguration; import org.apache.streampipes.security.jwt.JwtTokenGenerator; import org.apache.streampipes.security.jwt.JwtTokenUtils; import org.apache.streampipes.security.jwt.JwtTokenValidator; +import org.apache.streampipes.storage.management.StorageDispatcher; import org.apache.streampipes.user.management.model.PrincipalUserDetails; import org.apache.streampipes.user.management.util.GrantedAuthoritiesBuilder; import org.apache.streampipes.user.management.util.UserInfoUtil; @@ -51,11 +52,16 @@ public class JwtTokenProvider { public static final String CLAIM_USER = "user"; private static final Logger LOG = LoggerFactory.getLogger(JwtTokenProvider.class); - private BackendConfig config; + private SpCoreConfiguration config; private Environment env; public JwtTokenProvider() { - this.config = BackendConfig.INSTANCE; + this.config = StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getSpCoreConfigurationStorage() + .get(); + this.env = Environments.getEnvironment(); } diff --git a/streampipes-user-management/src/main/java/org/apache/streampipes/user/management/jwt/SpKeyResolver.java b/streampipes-user-management/src/main/java/org/apache/streampipes/user/management/jwt/SpKeyResolver.java index 7e0af5040..608671383 100644 --- a/streampipes-user-management/src/main/java/org/apache/streampipes/user/management/jwt/SpKeyResolver.java +++ b/streampipes-user-management/src/main/java/org/apache/streampipes/user/management/jwt/SpKeyResolver.java @@ -17,7 +17,6 @@ */ package org.apache.streampipes.user.management.jwt; -import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.model.client.user.Principal; import org.apache.streampipes.model.client.user.ServiceAccount; import org.apache.streampipes.model.client.user.UserAccount; @@ -70,7 +69,13 @@ public class SpKeyResolver implements SigningKeyResolver { } public String getPublicKeyFromConfig() { - return BackendConfig.INSTANCE.getLocalAuthConfig().getPublicKey(); + return StorageDispatcher + .INSTANCE + .getNoSqlStore() + .getSpCoreConfigurationStorage() + .get() + .getLocalAuthConfig() + .getPublicKey(); }
