This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch STREAMPIPES-319
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/STREAMPIPES-319 by this push:
new 8189b40 [STREAMPIPES-319] Refactor config definition
8189b40 is described below
commit 8189b40e05f17fa4302cb8792f06d9fcb513fe70
Author: Dominik Riemer <[email protected]>
AuthorDate: Tue Apr 20 22:23:25 2021 +0200
[STREAMPIPES-319] Refactor config definition
---
.../constants/{Envs.java => CustomEnvs.java} | 24 ++----
.../apache/streampipes/commons/constants/Envs.java | 15 ++--
streampipes-config/pom.xml | 9 +-
.../org/apache/streampipes/config/SpConfig.java | 7 +-
.../streampipes/config/consul/ConsulSpConfig.java | 95 ++++++----------------
.../streampipes/config/consul/MockSpConfig.java | 4 +-
.../model/SpServiceDefinitionBuilder.java | 16 ++--
.../apache/streampipes/rest/impl/ConsulConfig.java | 4 +-
.../svcdiscovery/api}/model/ConfigItem.java | 2 +-
.../api}/model/ConfigurationScope.java | 2 +-
.../svcdiscovery/api}/model/PeConfig.java | 2 +-
streampipes-service-discovery-consul/pom.xml | 5 --
.../svcdiscovery/consul/ConsulProvider.java | 49 +++++++++--
.../svcdiscovery/consul/SpConsulKvManagement.java | 2 +-
14 files changed, 113 insertions(+), 123 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/CustomEnvs.java
similarity index 66%
copy from
streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
copy to
streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/CustomEnvs.java
index 1e06871..b0f2e94 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/CustomEnvs.java
@@ -17,27 +17,21 @@
*/
package org.apache.streampipes.commons.constants;
-public enum Envs {
+public class CustomEnvs {
- SP_HOST("SP_HOST"),
- SP_PORT("SP_PORT");
-
- private String envVariableName;
-
- Envs(String envVariableName) {
- this.envVariableName = envVariableName;
+ public static boolean exists(String envVariable) {
+ return System.getenv().containsKey(envVariable);
}
- public boolean exists() {
- return System.getenv().containsKey(this.envVariableName);
+ public static String getEnv(String envVariable) {
+ return System.getenv(envVariable);
}
- public String getValue() {
- return System.getenv(this.envVariableName);
+ public static Integer getEnvAsInt(String envVariable) {
+ return Integer.parseInt(getEnv(envVariable));
}
- public Integer getValueAsInt() {
- return Integer.parseInt(getValue());
+ public static Boolean getEnvAsBoolean(String envVariable) {
+ return Boolean.parseBoolean(envVariable);
}
-
}
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 1e06871..dac07be 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
@@ -20,24 +20,29 @@ package org.apache.streampipes.commons.constants;
public enum Envs {
SP_HOST("SP_HOST"),
- SP_PORT("SP_PORT");
+ SP_PORT("SP_PORT"),
+ SP_CONSUL_LOCATION("CONSUL_LOCATION");
- private String envVariableName;
+ private final String envVariableName;
Envs(String envVariableName) {
this.envVariableName = envVariableName;
}
public boolean exists() {
- return System.getenv().containsKey(this.envVariableName);
+ return CustomEnvs.exists(this.envVariableName);
}
public String getValue() {
- return System.getenv(this.envVariableName);
+ return CustomEnvs.getEnv(this.envVariableName);
}
public Integer getValueAsInt() {
- return Integer.parseInt(getValue());
+ return CustomEnvs.getEnvAsInt(this.envVariableName);
+ }
+
+ public Boolean getValueAsBoolean() {
+ return CustomEnvs.getEnvAsBoolean(this.envVariableName);
}
}
diff --git a/streampipes-config/pom.xml b/streampipes-config/pom.xml
index 101103e..1abef45 100644
--- a/streampipes-config/pom.xml
+++ b/streampipes-config/pom.xml
@@ -39,6 +39,11 @@
<artifactId>streampipes-vocabulary</artifactId>
<version>0.68.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-service-discovery</artifactId>
+ <version>0.68.0-SNAPSHOT</version>
+ </dependency>
<!-- External dependencies -->
<dependency>
@@ -49,9 +54,5 @@
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</dependency>
- <dependency>
- <groupId>com.orbitz.consul</groupId>
- <artifactId>consul-client</artifactId>
- </dependency>
</dependencies>
</project>
diff --git
a/streampipes-config/src/main/java/org/apache/streampipes/config/SpConfig.java
b/streampipes-config/src/main/java/org/apache/streampipes/config/SpConfig.java
index ddf8c26..5167061 100644
---
a/streampipes-config/src/main/java/org/apache/streampipes/config/SpConfig.java
+++
b/streampipes-config/src/main/java/org/apache/streampipes/config/SpConfig.java
@@ -19,11 +19,12 @@
package org.apache.streampipes.config;
import org.apache.streampipes.config.consul.ConsulSpConfig;
-import org.apache.streampipes.config.model.ConfigItem;
-import org.apache.streampipes.config.model.ConfigurationScope;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigItem;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigurationScope;
+import org.apache.streampipes.svcdiscovery.consul.AbstractConsulService;
-public abstract class SpConfig {
+public abstract class SpConfig extends AbstractConsulService {
public SpConfig(String name) {
diff --git
a/streampipes-config/src/main/java/org/apache/streampipes/config/consul/ConsulSpConfig.java
b/streampipes-config/src/main/java/org/apache/streampipes/config/consul/ConsulSpConfig.java
index 1b2c5ba..9a3f14a 100644
---
a/streampipes-config/src/main/java/org/apache/streampipes/config/consul/ConsulSpConfig.java
+++
b/streampipes-config/src/main/java/org/apache/streampipes/config/consul/ConsulSpConfig.java
@@ -18,36 +18,31 @@
package org.apache.streampipes.config.consul;
-import com.google.gson.Gson;
+import com.fasterxml.jackson.core.JsonProcessingException;
import com.orbitz.consul.Consul;
import com.orbitz.consul.KeyValueClient;
import com.orbitz.consul.model.kv.Value;
import org.apache.streampipes.config.SpConfig;
import org.apache.streampipes.config.SpConfigChangeCallback;
-import org.apache.streampipes.config.model.ConfigItem;
-import org.apache.streampipes.config.model.ConfigurationScope;
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigItem;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigurationScope;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.net.InetSocketAddress;
-import java.net.MalformedURLException;
-import java.net.Socket;
-import java.net.URL;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
public class ConsulSpConfig extends SpConfig implements Runnable {
- private static final Logger LOG =
LoggerFactory.getLogger(ConsulSpConfig.class.getCanonicalName());
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ConsulSpConfig.class);
private static final String SLASH = "/";
- private static final String CONSUL_ENV_LOCATION = "CONSUL_LOCATION";
- private static final int CONSUL_DEFAULT_PORT = 8500;
public static final String SERVICE_ROUTE_PREFIX = "sp/v1/";
- private String serviceName;
- private KeyValueClient kvClient;
-
+ private final String serviceName;
+ private final KeyValueClient kvClient;
// TODO Implement mechanism to update the client when some configuration
parameters change in Consul
private SpConfigChangeCallback callback;
@@ -67,59 +62,6 @@ public class ConsulSpConfig extends SpConfig implements
Runnable {
new Thread(this).start();
}
- private static Consul consulInstance() {
- boolean connected = false;
- URL consulUrl = consulURL();
-
- while (!connected) {
- LOG.info("Trying to connect to Consul to register config items");
- connected = isReady(consulUrl.getHost(), consulUrl.getPort());
-
- if (!connected) {
- LOG.info("Retrying in 1 second");
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- }
- LOG.info("Successfully connected to Consul");
- return Consul.builder().withUrl(consulURL()).build();
- }
-
- private static URL consulURL() {
- Map<String, String> env = System.getenv();
- URL url = null;
-
- if (env.containsKey(CONSUL_ENV_LOCATION)) {
- try {
- url = new URL("http", env.get(CONSUL_ENV_LOCATION),
CONSUL_DEFAULT_PORT, "");
- } catch (MalformedURLException e) {
- e.printStackTrace();
- }
- } else {
- try {
- url = new URL("http", "localhost", CONSUL_DEFAULT_PORT, "");
- } catch (MalformedURLException e) {
- e.printStackTrace();
- }
- }
- return url;
- }
-
- public static boolean isReady(String host, int port) {
- try {
- InetSocketAddress sa = new InetSocketAddress(host, port);
- Socket ss = new Socket();
- ss.connect(sa, 1000);
- ss.close();
- } catch(Exception e) {
- return false;
- }
- return true;
- }
-
@Override
public void run() {
Consul consulThread = Consul.builder().build(); // connect to Consul
on localhost
@@ -239,8 +181,12 @@ public class ConsulSpConfig extends SpConfig implements
Runnable {
public <T> T getObject(String key, Class<T> clazz, T defaultValue) {
Optional<String> os = kvClient.getValueAsString(addSn(key));
if (os.isPresent()) {
- Gson gson = new Gson();
- return gson.fromJson(os.get(), clazz);
+ try {
+ return JacksonSerializer.getObjectMapper().readValue(os.get(),
clazz);
+ } catch (JsonProcessingException e) {
+ LOG.info("Could not deserialize object", e);
+ return defaultValue;
+ }
} else {
return defaultValue;
}
@@ -275,8 +221,7 @@ public class ConsulSpConfig extends SpConfig implements
Runnable {
@Override
public void setObject(String key, Object value) {
- Gson gson = new Gson();
- kvClient.putValue(addSn(key), gson.toJson(value));
+ kvClient.putValue(addSn(key), toJson(value));
}
private String addSn(String key) {
@@ -285,7 +230,7 @@ public class ConsulSpConfig extends SpConfig implements
Runnable {
private ConfigItem fromJson(String content) {
try {
- return new Gson().fromJson(content, ConfigItem.class);
+ return JacksonSerializer.getObjectMapper().readValue(content,
ConfigItem.class);
} catch (Exception e) {
// if old config is used, this is a fallback
ConfigItem configItem = new ConfigItem();
@@ -305,6 +250,12 @@ public class ConsulSpConfig extends SpConfig implements
Runnable {
}
private String toJson(Object object) {
- return new Gson().toJson(object);
+ try {
+ return
JacksonSerializer.getObjectMapper().writeValueAsString(object);
+ } catch (JsonProcessingException e) {
+ LOG.info("Could not serialize object to JSON", e);
+ return "";
+ }
}
+
}
diff --git
a/streampipes-config/src/main/java/org/apache/streampipes/config/consul/MockSpConfig.java
b/streampipes-config/src/main/java/org/apache/streampipes/config/consul/MockSpConfig.java
index 47ef43e..b7b5b6c 100644
---
a/streampipes-config/src/main/java/org/apache/streampipes/config/consul/MockSpConfig.java
+++
b/streampipes-config/src/main/java/org/apache/streampipes/config/consul/MockSpConfig.java
@@ -20,8 +20,8 @@ package org.apache.streampipes.config.consul;
import org.apache.streampipes.config.SpConfig;
import org.apache.streampipes.config.SpConfigChangeCallback;
-import org.apache.streampipes.config.model.ConfigItem;
-import org.apache.streampipes.config.model.ConfigurationScope;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigItem;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigurationScope;
public class MockSpConfig extends SpConfig {
diff --git
a/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinitionBuilder.java
b/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinitionBuilder.java
index a7fe218..7912c5e 100644
---
a/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinitionBuilder.java
+++
b/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinitionBuilder.java
@@ -17,6 +17,8 @@
*/
package org.apache.streampipes.container.model;
+import org.apache.streampipes.config.SpConfig;
+import org.apache.streampipes.config.consul.ConsulSpConfig;
import org.apache.streampipes.container.declarer.Declarer;
import org.apache.streampipes.dataformat.SpDataFormatFactory;
import org.apache.streampipes.messaging.SpProtocolDefinitionFactory;
@@ -26,6 +28,7 @@ import java.util.Arrays;
public class SpServiceDefinitionBuilder {
private SpServiceDefinition serviceDefinition;
+ private SpConfig config;
public static SpServiceDefinitionBuilder create(String serviceId,
String serviceName,
@@ -43,6 +46,7 @@ public class SpServiceDefinitionBuilder {
this.serviceDefinition.setServiceName(serviceName);
this.serviceDefinition.setServiceDescription(serviceDescription);
this.serviceDefinition.setDefaultPort(defaultPort);
+ this.config = new ConsulSpConfig(serviceId);
}
public SpServiceDefinitionBuilder withHostname(String hostname) {
@@ -50,18 +54,18 @@ public class SpServiceDefinitionBuilder {
return this;
}
- public SpServiceDefinitionBuilder addConfig(String key, String value) {
-
+ public SpServiceDefinitionBuilder addConfig(String key, String defaultValue,
String description) {
+ this.config.register(key, defaultValue, description);
return this;
}
- public SpServiceDefinitionBuilder addConfig(String key, Integer value) {
-
+ public SpServiceDefinitionBuilder addConfig(String key, Integer
defaultValue, String description) {
+ this.config.register(key, defaultValue, description);
return this;
}
- public SpServiceDefinitionBuilder addConfig(String key, Boolean value) {
-
+ public SpServiceDefinitionBuilder addConfig(String key, Boolean
defaultValue, String description) {
+ this.config.register(key, defaultValue, description);
return this;
}
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java
index 484e4f6..8d273e0 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java
@@ -23,8 +23,8 @@ import com.google.gson.JsonObject;
import org.apache.streampipes.config.backend.BackendConfig;
import org.apache.streampipes.config.backend.MessagingSettings;
import org.apache.streampipes.config.consul.ConsulSpConfig;
-import org.apache.streampipes.config.model.ConfigItem;
-import org.apache.streampipes.config.model.PeConfig;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigItem;
+import org.apache.streampipes.svcdiscovery.api.model.PeConfig;
import org.apache.streampipes.rest.shared.annotation.GsonWithIds;
import org.apache.streampipes.svcdiscovery.api.ISpKvManagement;
import org.slf4j.Logger;
diff --git
a/streampipes-config/src/main/java/org/apache/streampipes/config/model/ConfigItem.java
b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java
similarity index 97%
rename from
streampipes-config/src/main/java/org/apache/streampipes/config/model/ConfigItem.java
rename to
streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java
index 8f10cd5..edcdaf6 100644
---
a/streampipes-config/src/main/java/org/apache/streampipes/config/model/ConfigItem.java
+++
b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.config.model;
+package org.apache.streampipes.svcdiscovery.api.model;
public class ConfigItem {
diff --git
a/streampipes-config/src/main/java/org/apache/streampipes/config/model/ConfigurationScope.java
b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigurationScope.java
similarity index 94%
rename from
streampipes-config/src/main/java/org/apache/streampipes/config/model/ConfigurationScope.java
rename to
streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigurationScope.java
index 040a3f8..094cb0c 100644
---
a/streampipes-config/src/main/java/org/apache/streampipes/config/model/ConfigurationScope.java
+++
b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigurationScope.java
@@ -15,7 +15,7 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.config.model;
+package org.apache.streampipes.svcdiscovery.api.model;
public enum ConfigurationScope {
CONTAINER_STARTUP_CONFIG,
diff --git
a/streampipes-config/src/main/java/org/apache/streampipes/config/model/PeConfig.java
b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/PeConfig.java
similarity index 96%
rename from
streampipes-config/src/main/java/org/apache/streampipes/config/model/PeConfig.java
rename to
streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/PeConfig.java
index 905f169..c215a1c 100644
---
a/streampipes-config/src/main/java/org/apache/streampipes/config/model/PeConfig.java
+++
b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/PeConfig.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.config.model;
+package org.apache.streampipes.svcdiscovery.api.model;
import java.util.List;
import java.util.Map;
diff --git a/streampipes-service-discovery-consul/pom.xml
b/streampipes-service-discovery-consul/pom.xml
index 453928b..7300f0a 100644
--- a/streampipes-service-discovery-consul/pom.xml
+++ b/streampipes-service-discovery-consul/pom.xml
@@ -14,11 +14,6 @@
<dependencies>
<dependency>
<groupId>org.apache.streampipes</groupId>
- <artifactId>streampipes-config</artifactId>
- <version>0.68.0-SNAPSHOT</version>
- </dependency>
- <dependency>
- <groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-logging</artifactId>
<version>0.68.0-SNAPSHOT</version>
</dependency>
diff --git
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulProvider.java
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulProvider.java
index 6a5a704..76e5c7b 100644
---
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulProvider.java
+++
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulProvider.java
@@ -18,28 +18,51 @@
package org.apache.streampipes.svcdiscovery.consul;
import com.orbitz.consul.Consul;
+import org.apache.streampipes.commons.constants.Envs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.net.InetSocketAddress;
import java.net.MalformedURLException;
+import java.net.Socket;
import java.net.URL;
-import java.util.Map;
public class ConsulProvider {
- private static final String CONSUL_ENV_LOCATION = "CONSUL_LOCATION";
+ private static final Logger LOG =
LoggerFactory.getLogger(ConsulProvider.class);
+
private static final int CONSUL_DEFAULT_PORT = 8500;
private static final String CONSUL_URL_REGISTER_SERVICE =
"v1/agent/service/register";
public Consul consulInstance() {
+ URL consulUrl = consulURL();
+ boolean connected;
+
+ do {
+ LOG.info("Checking if consul is available...");
+ connected = checkConsulAvailable(consulUrl);
+
+ if (!connected) {
+ LOG.info("Retrying in 1 second");
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ } while(!connected);
+
+ LOG.info("Successfully connected to Consul");
return Consul.builder().withUrl(consulURL()).build();
}
private URL consulURL() {
- Map<String, String> env = System.getenv();
URL url = null;
- if (env.containsKey(CONSUL_ENV_LOCATION)) {
+ if (Envs.SP_CONSUL_LOCATION.exists()) {
try {
- url = new URL("http", env.get(CONSUL_ENV_LOCATION),
CONSUL_DEFAULT_PORT, "");
+ url = new URL("http", Envs.SP_CONSUL_LOCATION.getValue(),
CONSUL_DEFAULT_PORT, "");
} catch (MalformedURLException e) {
e.printStackTrace();
}
@@ -53,6 +76,22 @@ public class ConsulProvider {
return url;
}
+ private boolean checkConsulAvailable(URL consulUrl) {
+ try {
+ InetSocketAddress sa = new InetSocketAddress(consulUrl.getHost(),
consulUrl.getPort());
+ Socket ss = new Socket();
+ ss.connect(sa, 1000);
+ ss.close();
+
+ return true;
+ } catch (IOException e) {
+ LOG.info("Could not connect to Consul instance...");
+ return false;
+ }
+
+
+ }
+
public String makeConsulEndpoint() {
return consulURL().toString() + "/" + CONSUL_URL_REGISTER_SERVICE;
}
diff --git
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java
index cff972d..d54277c 100644
---
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java
+++
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java
@@ -22,9 +22,9 @@ import com.orbitz.consul.Consul;
import com.orbitz.consul.KeyValueClient;
import com.orbitz.consul.model.ConsulResponse;
import com.orbitz.consul.model.kv.Value;
-import org.apache.streampipes.config.model.ConfigItem;
import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.apache.streampipes.svcdiscovery.api.ISpKvManagement;
+import org.apache.streampipes.svcdiscovery.api.model.ConfigItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
