You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2023/08/06 16:19:40 UTC
[streampipes] 04/11: Move configs to messaging settings
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 <do...@gmail.com>
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);
}