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 e14fadccac06d6e5eef6c7da9087a25a087805d1 Author: Dominik Riemer <[email protected]> AuthorDate: Fri Aug 4 23:48:11 2023 +0200 Move configs to messaging settings --- .../apache/streampipes/commons/constants/Envs.java | 1 + .../commons/environment/DefaultEnvironment.java | 5 + .../commons/environment/Environment.java | 2 + .../streampipes/config/backend/BackendConfig.java | 1 + .../config/backend/DefaultMessagingSettings.java | 68 ------------ .../connect/management/util/GroundingUtils.java | 25 ++--- .../export/utils/EventGroundingProcessor.java | 28 ++--- .../configuration/DefaultMessagingSettings.java | 76 ++++++++++++++ .../model/configuration/MessagingSettings.java | 115 +++++++++++++++++++++ .../model/configuration/SpCoreConfiguration.java | 105 ------------------- .../manager/matching/ProtocolSelector.java | 33 +++--- .../migrations/v093/ConsulConfigMigration.java | 25 ++--- 12 files changed, 257 insertions(+), 227 deletions(-) diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java index 23f16489f..ab1a87890 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java @@ -34,6 +34,7 @@ public enum Envs { DefaultEnvValues.LOCALHOST), SP_CONSUL_PORT("SP_CONSUL_PORT", DefaultEnvValues.CONSUL_PORT_DEFAULT), SP_KAFKA_RETENTION_MS("SP_KAFKA_RETENTION_MS", DefaultEnvValues.SP_KAFKA_RETENTION_MS_DEFAULT), + SP_PRIORITIZED_PROTOCOL("SP_PRIORITIZED_PROTOCOL", "kafka"), SP_JWT_SECRET("JWT_SECRET"), SP_JWT_SIGNING_MODE("SP_JWT_SIGNING_MODE"), SP_JWT_PRIVATE_KEY_LOC("SP_JWT_PRIVATE_KEY_LOC"), diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java index 4f6d71cb1..9b5e695e3 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java @@ -149,6 +149,11 @@ public class DefaultEnvironment implements Environment { return new StringEnvironmentVariable(Envs.SP_KAFKA_RETENTION_MS); } + @Override + public StringEnvironmentVariable getPrioritizedProtocol() { + return new StringEnvironmentVariable(Envs.SP_PRIORITIZED_PROTOCOL); + } + @Override public BooleanEnvironmentVariable getSetupInstallPipelineElements() { return new BooleanEnvironmentVariable(Envs.SP_SETUP_INSTALL_PIPELINE_ELEMENTS); diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java index 7217d261a..8abc73154 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java @@ -89,6 +89,8 @@ public interface Environment { // Messaging StringEnvironmentVariable getKafkaRetentionTimeMs(); + StringEnvironmentVariable getPrioritizedProtocol(); + // Setup BooleanEnvironmentVariable getSetupInstallPipelineElements(); 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 b3abc1f00..bb4b46874 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,6 +22,7 @@ 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.model.configuration.DefaultMessagingSettings; import org.apache.streampipes.model.configuration.EmailConfig; import org.apache.streampipes.model.configuration.GeneralConfig; import org.apache.streampipes.model.configuration.LocalAuthConfig; 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 deleted file mode 100644 index 8e2180382..000000000 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/DefaultMessagingSettings.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -package org.apache.streampipes.config.backend; - -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; - -public class DefaultMessagingSettings { - - public static MessagingSettings make() { - List<SpProtocol> protocolList; - if (System.getenv(BackendConfigKeys.PRIORITIZED_PROTOCOL) != null) { - switch (System.getenv(BackendConfigKeys.PRIORITIZED_PROTOCOL).toLowerCase()) { - case "mqtt": - protocolList = - Arrays.asList(SpProtocol.MQTT, SpProtocol.KAFKA, SpProtocol.JMS, SpProtocol.NATS, SpProtocol.PULSAR); - break; - case "kafka": - protocolList = - Arrays.asList(SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.NATS, SpProtocol.PULSAR); - break; - case "jms": - protocolList = - Arrays.asList(SpProtocol.JMS, SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.NATS, SpProtocol.PULSAR); - break; - case "nats": - protocolList = - Arrays.asList(SpProtocol.NATS, SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.PULSAR); - break; - case "pulsar": - protocolList = - Arrays.asList(SpProtocol.PULSAR, SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.NATS); - break; - default: - protocolList = - Arrays.asList(SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.NATS, SpProtocol.PULSAR); - } - } else { - protocolList = - Arrays.asList(SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.NATS, SpProtocol.PULSAR); - } - - return new MessagingSettings( - 1638400, 5000012, 20, 2, - Arrays.asList(SpDataFormat.JSON, SpDataFormat.CBOR, SpDataFormat.FST, SpDataFormat.SMILE), - protocolList); - } -} 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 8f51d12de..74c771358 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 @@ -38,44 +38,45 @@ public class GroundingUtils { public static EventGrounding createEventGrounding() { EventGrounding eventGrounding = new EventGrounding(); - var cfg = Utils + var messagingSettings = Utils .getCoreConfigStorage() - .get(); + .get() + .getMessagingSettings(); String topic = TOPIC_PREFIX + UUID.randomUUID().toString(); TopicDefinition topicDefinition = new SimpleTopicDefinition(topic); SpProtocol prioritizedProtocol = - cfg.getMessagingSettings().getPrioritizedProtocols().get(0); + messagingSettings.getPrioritizedProtocols().get(0); if (isPrioritized(prioritizedProtocol, JmsTransportProtocol.class)) { eventGrounding.setTransportProtocol( makeJmsTransportProtocol( - cfg.getJmsHost(), - cfg.getJmsPort(), + messagingSettings.getJmsHost(), + messagingSettings.getJmsPort(), topicDefinition)); } else if (isPrioritized(prioritizedProtocol, KafkaTransportProtocol.class)) { eventGrounding.setTransportProtocol( makeKafkaTransportProtocol( - cfg.getKafkaHost(), - cfg.getKafkaPort(), + messagingSettings.getKafkaHost(), + messagingSettings.getKafkaPort(), topicDefinition)); } else if (isPrioritized(prioritizedProtocol, MqttTransportProtocol.class)) { eventGrounding.setTransportProtocol( makeMqttTransportProtocol( - cfg.getMqttHost(), - cfg.getMqttPort(), + messagingSettings.getMqttHost(), + messagingSettings.getMqttPort(), topicDefinition)); } else if (isPrioritized(prioritizedProtocol, NatsTransportProtocol.class)) { eventGrounding.setTransportProtocol( makeNatsTransportProtocol( - cfg.getNatsHost(), - cfg.getNatsPort(), + messagingSettings.getNatsHost(), + messagingSettings.getNatsPort(), topicDefinition)); } else if (isPrioritized(prioritizedProtocol, PulsarTransportProtocol.class)) { eventGrounding.setTransportProtocol( makePulsarTransportProtocol( - cfg.getPulsarUrl(), + messagingSettings.getPulsarUrl(), topicDefinition ) ); 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 72bd19e23..83d508560 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,7 +19,7 @@ package org.apache.streampipes.export.utils; import org.apache.streampipes.commons.exceptions.SpRuntimeException; -import org.apache.streampipes.model.configuration.SpCoreConfiguration; +import org.apache.streampipes.model.configuration.MessagingSettings; import org.apache.streampipes.model.configuration.SpProtocol; import org.apache.streampipes.model.grounding.KafkaTransportProtocol; import org.apache.streampipes.model.grounding.MqttTransportProtocol; @@ -30,18 +30,18 @@ import org.apache.streampipes.storage.management.StorageDispatcher; public class EventGroundingProcessor { - SpCoreConfiguration cfg; - SpProtocol configuredProtocol; + private final MessagingSettings messagingSettings; + private final SpProtocol configuredProtocol; public EventGroundingProcessor() { - this.cfg = StorageDispatcher + this.messagingSettings = StorageDispatcher .INSTANCE .getNoSqlStore() .getSpCoreConfigurationStorage() - .get(); + .get() + .getMessagingSettings(); - this.configuredProtocol = cfg - .getMessagingSettings() + this.configuredProtocol = messagingSettings .getPrioritizedProtocols().get(0); } @@ -77,28 +77,28 @@ public class EventGroundingProcessor { private KafkaTransportProtocol makeKafkaProtocol() { var protocol = new KafkaTransportProtocol(); - protocol.setBrokerHostname(cfg.getKafkaHost()); - protocol.setKafkaPort(cfg.getKafkaPort()); + protocol.setBrokerHostname(messagingSettings.getKafkaHost()); + protocol.setKafkaPort(messagingSettings.getKafkaPort()); return protocol; } private MqttTransportProtocol makeMqttProtocol() { var protocol = new MqttTransportProtocol(); - protocol.setBrokerHostname(cfg.getMqttHost()); - protocol.setPort(cfg.getMqttPort()); + protocol.setBrokerHostname(messagingSettings.getMqttHost()); + protocol.setPort(messagingSettings.getMqttPort()); return protocol; } private NatsTransportProtocol makeNatsProtocol() { var protocol = new NatsTransportProtocol(); - protocol.setBrokerHostname(cfg.getNatsHost()); - protocol.setPort(cfg.getNatsPort()); + protocol.setBrokerHostname(messagingSettings.getNatsHost()); + protocol.setPort(messagingSettings.getNatsPort()); return protocol; } private PulsarTransportProtocol makePulsarProtocol() { var protocol = new PulsarTransportProtocol(); - protocol.setBrokerHostname(cfg.getNatsHost()); + protocol.setBrokerHostname(messagingSettings.getNatsHost()); return protocol; } diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/DefaultMessagingSettings.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/DefaultMessagingSettings.java new file mode 100644 index 000000000..079f20e25 --- /dev/null +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/DefaultMessagingSettings.java @@ -0,0 +1,76 @@ +/* + * 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 org.apache.streampipes.commons.environment.Environments; + +import java.util.Arrays; +import java.util.List; + +public class DefaultMessagingSettings { + + public static MessagingSettings make() { + List<SpProtocol> protocolList; + var env = Environments.getEnvironment(); + if (env.getPrioritizedProtocol().exists()) { + protocolList = switch (env.getPrioritizedProtocol().getValueOrDefault().toLowerCase()) { + case "mqtt" -> + Arrays.asList(SpProtocol.MQTT, SpProtocol.KAFKA, SpProtocol.JMS, SpProtocol.NATS, SpProtocol.PULSAR); + case "kafka" -> + Arrays.asList(SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.NATS, SpProtocol.PULSAR); + case "jms" -> + Arrays.asList(SpProtocol.JMS, SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.NATS, SpProtocol.PULSAR); + case "nats" -> + Arrays.asList(SpProtocol.NATS, SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.PULSAR); + case "pulsar" -> + Arrays.asList(SpProtocol.PULSAR, SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.NATS); + default -> Arrays.asList(SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.NATS, SpProtocol.PULSAR); + }; + } else { + protocolList = + Arrays.asList(SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS, SpProtocol.NATS, SpProtocol.PULSAR); + } + + var defaultSettings = new MessagingSettings( + 1638400, + 5000012, + 20, + 2, + Arrays.asList(SpDataFormat.JSON, SpDataFormat.CBOR, SpDataFormat.FST, SpDataFormat.SMILE), + protocolList); + + defaultSettings.setJmsHost("activemq"); + defaultSettings.setJmsPort(61616); + + defaultSettings.setMqttHost("mosquitto"); + defaultSettings.setMqttPort(1883); + + defaultSettings.setNatsHost("nats"); + defaultSettings.setNatsPort(4222); + + defaultSettings.setKafkaHost("kafka"); + defaultSettings.setKafkaPort(9092); + + defaultSettings.setPulsarUrl("pulsar://localhost:6650"); + + defaultSettings.setZookeeperHost("zookeeper"); + defaultSettings.setZookeeperPort(2181); + return defaultSettings; + } +} diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java index 6796db780..b5c00c2aa 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java @@ -17,6 +17,7 @@ */ package org.apache.streampipes.model.configuration; +import java.util.ArrayList; import java.util.List; public class MessagingSettings { @@ -29,7 +30,24 @@ public class MessagingSettings { private List<SpDataFormat> prioritizedFormats; private List<SpProtocol> prioritizedProtocols; + 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 List<String> supportedProtocols; public MessagingSettings(Integer batchSize, Integer messageMaxBytes, @@ -43,6 +61,7 @@ public class MessagingSettings { this.acks = acks; this.prioritizedFormats = prioritizedFormats; this.prioritizedProtocols = prioritizedProtocols; + this.supportedProtocols = new ArrayList<>(); } public MessagingSettings() { @@ -96,4 +115,100 @@ public class MessagingSettings { public void setPrioritizedProtocols(List<SpProtocol> prioritizedProtocols) { this.prioritizedProtocols = prioritizedProtocols; } + + 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 List<String> getSupportedProtocols() { + return supportedProtocols; + } + + public void setSupportedProtocols(List<String> supportedProtocols) { + this.supportedProtocols = supportedProtocols; + } } 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 index cf192d567..32ea21d4e 100644 --- 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 @@ -32,23 +32,6 @@ public class SpCoreConfiguration { 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; @@ -105,94 +88,6 @@ public class SpCoreConfiguration { 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; } 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 ea6e0b649..e1eb685f6 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 @@ -22,7 +22,7 @@ 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.configuration.SpCoreConfiguration; +import org.apache.streampipes.model.configuration.MessagingSettings; import org.apache.streampipes.model.configuration.SpProtocol; import org.apache.streampipes.model.grounding.JmsTransportProtocol; import org.apache.streampipes.model.grounding.KafkaTransportProtocol; @@ -41,19 +41,20 @@ public class ProtocolSelector extends GroundingSelector { private final String outputTopic; private final List<SpProtocol> prioritizedProtocols; - private final SpCoreConfiguration cfg; + private final MessagingSettings messagingSettings; public ProtocolSelector(NamedStreamPipesEntity source, Set<InvocableStreamPipesEntity> targets) { super(source, targets); this.outputTopic = TopicGenerator.generateRandomTopic(); - this.cfg = StorageDispatcher + this.messagingSettings = StorageDispatcher .INSTANCE .getNoSqlStore() .getSpCoreConfigurationStorage() - .get(); + .get() + .getMessagingSettings(); this.prioritizedProtocols = - cfg.getMessagingSettings().getPrioritizedProtocols(); + messagingSettings.getPrioritizedProtocols(); } public TransportProtocol getPreferredProtocol() { @@ -77,7 +78,7 @@ public class ProtocolSelector extends GroundingSelector { return natsTopic(); } else if (prioritizedProtocol.getProtocolClass().equals(PulsarTransportProtocol.class.getCanonicalName()) && supportsProtocol(PulsarTransportProtocol.class)) { - return new PulsarTransportProtocol(cfg.getPulsarUrl(), + return new PulsarTransportProtocol(messagingSettings.getPulsarUrl(), new SimpleTopicDefinition(outputTopic)); } } @@ -87,35 +88,35 @@ public class ProtocolSelector extends GroundingSelector { private TransportProtocol mqttTopic() { return new MqttTransportProtocol( - cfg.getMqttHost(), - cfg.getMqttPort(), + messagingSettings.getMqttHost(), + messagingSettings.getMqttPort(), outputTopic ); } private TransportProtocol jmsTopic() { return new JmsTransportProtocol( - cfg.getJmsHost(), - cfg.getJmsPort(), + messagingSettings.getJmsHost(), + messagingSettings.getJmsPort(), outputTopic ); } private TransportProtocol natsTopic() { return new NatsTransportProtocol( - cfg.getNatsHost(), - cfg.getNatsPort(), + messagingSettings.getNatsHost(), + messagingSettings.getNatsPort(), outputTopic ); } private TransportProtocol kafkaTopic() { return new KafkaTransportProtocol( - cfg.getKafkaHost(), - cfg.getKafkaPort(), + messagingSettings.getKafkaHost(), + messagingSettings.getKafkaPort(), outputTopic, - cfg.getZookeeperHost(), - cfg.getZookeeperPort() + messagingSettings.getZookeeperHost(), + messagingSettings.getZookeeperPort() ); } 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 index eebc1c356..899c9a5ca 100644 --- 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 @@ -42,25 +42,26 @@ public class ConsulConfigMigration implements Migration { var currConf = BackendConfig.INSTANCE; var newConf = new SpCoreConfiguration(); - newConf.setMessagingSettings(currConf.getMessagingSettings()); + var messagingSettings = 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()); + messagingSettings.setJmsHost(currConf.getJmsHost()); + messagingSettings.setJmsPort(currConf.getJmsPort()); + messagingSettings.setMqttHost(currConf.getMqttHost()); + messagingSettings.setMqttPort(currConf.getMqttPort()); + messagingSettings.setNatsHost(currConf.getNatsHost()); + messagingSettings.setNatsPort(currConf.getNatsPort()); + messagingSettings.setKafkaHost(currConf.getKafkaHost()); + messagingSettings.setKafkaPort(currConf.getKafkaPort()); + messagingSettings.setPulsarUrl(currConf.getPulsarUrl()); + messagingSettings.setZookeeperHost(currConf.getZookeeperHost()); + messagingSettings.setZookeeperPort(currConf.getZookeeperPort()); newConf.setAssetDir(currConf.getAssetDir()); newConf.setFilesDir(currConf.getFilesDir()); + newConf.setMessagingSettings(messagingSettings); storage.createElement(newConf); }
