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 0776c30 [STREAMPIPES-319] Use human-readable serviceId, improve
Consul connection
0776c30 is described below
commit 0776c3016a5a8af4404074538360b4cca743f96b
Author: Dominik Riemer <[email protected]>
AuthorDate: Thu Jun 24 08:03:32 2021 +0200
[STREAMPIPES-319] Use human-readable serviceId, improve Consul connection
---
.../backend/StreamPipesBackendApplication.java | 6 +-
.../init/ConnectWorkerRegistrationService.java | 25 +++--
.../container/base/StreamPipesServiceBase.java | 4 +-
.../container/init/DeclarersSingleton.java | 1 +
.../svcdiscovery/consul/AbstractConsulService.java | 10 +-
.../consul/ConsulHealthServiceManager.java | 121 +++++++++++++++++++++
.../consul/SpConsulServiceDiscovery.java | 105 ++++--------------
.../model/ConsulServiceRegistrationBody.java | 90 ---------------
.../consul/model/HealthCheckConfiguration.java | 70 ------------
.../base/StreamPipesExtensionsServiceBase.java | 8 +-
10 files changed, 179 insertions(+), 261 deletions(-)
diff --git
a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
index 99e650e..d72a4d3 100644
---
a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
+++
b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
@@ -76,13 +76,17 @@ public class StreamPipesBackendApplication extends
StreamPipesServiceBase {
try {
application.startStreamPipesService(StreamPipesBackendApplication.class,
DefaultSpServiceGroups.CORE,
- AUTO_GENERATED_SERVICE_ID,
+ application.serviceId(),
8030);
} catch (UnknownHostException e) {
LOG.error("Could not auto-resolve host address - please manually
provide the hostname using the SP_HOST environment variable");
}
}
+ private String serviceId() {
+ return DefaultSpServiceGroups.CORE + "-" + AUTO_GENERATED_SERVICE_ID;
+ }
+
@PostConstruct
public void init() {
this.executorService = Executors.newSingleThreadScheduledExecutor();
diff --git
a/streampipes-connect-container-worker/src/main/java/org/apache/streampipes/connect/container/worker/init/ConnectWorkerRegistrationService.java
b/streampipes-connect-container-worker/src/main/java/org/apache/streampipes/connect/container/worker/init/ConnectWorkerRegistrationService.java
index e23d1ed..7ecf3e5 100644
---
a/streampipes-connect-container-worker/src/main/java/org/apache/streampipes/connect/container/worker/init/ConnectWorkerRegistrationService.java
+++
b/streampipes-connect-container-worker/src/main/java/org/apache/streampipes/connect/container/worker/init/ConnectWorkerRegistrationService.java
@@ -19,23 +19,32 @@ package
org.apache.streampipes.connect.container.worker.init;
import
org.apache.streampipes.connect.container.worker.management.MasterRestClient;
import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
+import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
public class ConnectWorkerRegistrationService {
private static final Logger LOG =
LoggerFactory.getLogger(ConnectWorkerRegistrationService.class);
public void registerWorker(String serviceGroup) {
- String masterUrl = getConnectMasterUrl() + "/streampipes-backend";
boolean connected = false;
while (!connected) {
- LOG.info("Trying to connect to master: " + masterUrl);
- connected = MasterRestClient.register(masterUrl,
- new
ConnectWorkerDescriptionProvider().getContainerDescription(serviceGroup));
+ List<String> coreServices = getConnectMasterUrl();
+ if (coreServices.size() > 0) {
+ String masterUrl = getConnectMasterUrl().get(0) +
"/streampipes-backend";
+ LOG.info("Trying to connect to master: " + masterUrl);
+ connected = MasterRestClient.register(masterUrl,
+ new
ConnectWorkerDescriptionProvider().getContainerDescription(serviceGroup));
+
+ if (connected) {
+ LOG.info("Successfully connected to master: " + masterUrl + " Worker
is now running.");
+ }
+ }
if (!connected) {
LOG.info("Retrying in 5 seconds");
@@ -46,13 +55,11 @@ public class ConnectWorkerRegistrationService {
}
}
}
-
- LOG.info("Successfully connected to master: " + masterUrl + " Worker is
now running.");
}
- private String getConnectMasterUrl() {
+ private List<String> getConnectMasterUrl() {
return SpServiceDiscovery
.getServiceDiscovery()
- .getServiceEndpoints("core", false, Arrays.asList()).get(0);
+ .getServiceEndpoints(DefaultSpServiceGroups.CORE, true,
Collections.emptyList());
}
}
diff --git
a/streampipes-container-base/src/main/java/org/apache/streampipes/container/base/StreamPipesServiceBase.java
b/streampipes-container-base/src/main/java/org/apache/streampipes/container/base/StreamPipesServiceBase.java
index 6ebf29c..4969fe0 100644
---
a/streampipes-container-base/src/main/java/org/apache/streampipes/container/base/StreamPipesServiceBase.java
+++
b/streampipes-container-base/src/main/java/org/apache/streampipes/container/base/StreamPipesServiceBase.java
@@ -17,6 +17,7 @@
*/
package org.apache.streampipes.container.base;
+import org.apache.commons.lang3.RandomStringUtils;
import org.apache.streampipes.commons.networking.Networking;
import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
import
org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest;
@@ -28,13 +29,12 @@ import org.springframework.boot.SpringApplication;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.List;
-import java.util.UUID;
public abstract class StreamPipesServiceBase {
private static final Logger LOG =
LoggerFactory.getLogger(StreamPipesServiceBase.class);
- protected static final String AUTO_GENERATED_SERVICE_ID =
UUID.randomUUID().toString();
+ protected static final String AUTO_GENERATED_SERVICE_ID =
RandomStringUtils.randomAlphanumeric(6);
protected void startStreamPipesService(Class<?> serviceClass,
String serviceGroup,
diff --git
a/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java
b/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java
index 62feaec..44c61d1 100644
---
a/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java
+++
b/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java
@@ -86,6 +86,7 @@ public class DeclarersSingleton {
this.setHostName(host);
this.setPort(port);
this.addDeclarers(serviceDef.getDeclarers());
+ this.serviceId = serviceDef.getServiceId();
this.registerProtocols(serviceDef.getProtocolDefinitionFactories());
this.registerDataFormats(serviceDef.getDataFormatFactories());
this.allAdapters = serviceDef.getSpecificAdapters();
diff --git
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/AbstractConsulService.java
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/AbstractConsulService.java
index 6fb4dda..aff0f18 100644
---
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/AbstractConsulService.java
+++
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/AbstractConsulService.java
@@ -21,17 +21,13 @@ import com.orbitz.consul.Consul;
public abstract class AbstractConsulService {
- protected ConsulProvider consulProvider;
+ protected ConsulHealthServiceManager consul;
public AbstractConsulService() {
- this.consulProvider = new ConsulProvider();
+ this.consul = ConsulHealthServiceManager.INSTANCE;
}
public Consul consulInstance() {
- return this.consulProvider.consulInstance();
- }
-
- public String makeConsulEndpoint() {
- return this.consulProvider.makeConsulEndpoint();
+ return this.consul.consulInstance();
}
}
diff --git
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java
new file mode 100644
index 0000000..94573b9
--- /dev/null
+++
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java
@@ -0,0 +1,121 @@
+/*
+ * 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.svcdiscovery.consul;
+
+import com.orbitz.consul.Consul;
+import com.orbitz.consul.HealthClient;
+import com.orbitz.consul.cache.ServiceHealthCache;
+import com.orbitz.consul.model.health.ServiceHealth;
+import com.orbitz.consul.option.QueryOptions;
+import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public enum ConsulHealthServiceManager {
+
+ INSTANCE;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ConsulHealthServiceManager.class);
+
+ private final Consul client;
+ private final Map<String, ServiceHealthCache> serviceHealthCaches;
+ private final int MAX_RETRIES = 3;
+
+ ConsulHealthServiceManager() {
+ serviceHealthCaches = new HashMap<>();
+ client = new ConsulProvider().consulInstance();
+ initializeAll();
+ }
+
+ public void initializeAll() {
+ initialize(DefaultSpServiceGroups.CORE);
+ initialize(DefaultSpServiceGroups.EXT);
+ }
+
+ public void initialize(String serviceGroup) {
+ HealthClient healthClient = client.healthClient();
+ ServiceHealthCache svHealth = ServiceHealthCache.newCache(healthClient,
serviceGroup, false, 9, QueryOptions.BLANK);
+ svHealth.start();
+ try {
+ svHealth.awaitInitialized(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ serviceHealthCaches.put(serviceGroup, svHealth);
+ }
+
+ public List<String> getServiceEndpoints(String serviceGroup,
+ boolean restrictToHealthy,
+ List<String> filterByTags) {
+ List<ServiceHealth> activeServices = findService(serviceGroup, 0);
+
+ return activeServices
+ .stream()
+ .filter(service -> allFiltersSupported(service, filterByTags))
+ .filter(service -> !restrictToHealthy ||
service.getChecks().get(0).getStatus().equals("passing"))
+ .map(this::makeServiceUrl)
+ .collect(Collectors.toList());
+ }
+
+ private String makeServiceUrl(ServiceHealth service) {
+ return service.getService().getAddress() + ":" +
service.getService().getPort();
+ }
+
+ private boolean allFiltersSupported(ServiceHealth service,
+ List<String> filterByTags) {
+ return service.getService().getTags().containsAll(filterByTags);
+ }
+
+ private List<ServiceHealth> findService(String serviceGroup, int retryCount)
{
+
+ if (serviceHealthCaches.containsKey(serviceGroup)
+ && serviceHealthCaches.get(serviceGroup).getMap() != null) {
+ ServiceHealthCache cache = serviceHealthCaches.get(serviceGroup);
+ return cache
+ .getMap()
+ .values()
+ .stream()
+ .filter((value) ->
value.getService().getService().equals(serviceGroup))
+ .collect(Collectors.toList());
+ } else {
+ if (retryCount < MAX_RETRIES) {
+ try {
+ retryCount++;
+ TimeUnit.SECONDS.sleep(1);
+ return findService(serviceGroup, retryCount);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ return Collections.emptyList();
+ }
+ } else {
+ return Collections.emptyList();
+ }
+ }
+ }
+
+ public Consul consulInstance() {
+ return this.client;
+ }
+}
diff --git
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java
index 884546e..c23f760 100644
---
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java
+++
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java
@@ -19,24 +19,23 @@ package org.apache.streampipes.svcdiscovery.consul;
import com.orbitz.consul.AgentClient;
import com.orbitz.consul.Consul;
-import com.orbitz.consul.HealthClient;
+import com.orbitz.consul.model.agent.ImmutableRegCheck;
+import com.orbitz.consul.model.agent.ImmutableRegistration;
+import com.orbitz.consul.model.agent.Registration;
import com.orbitz.consul.model.health.HealthCheck;
import com.orbitz.consul.model.health.Service;
-import com.orbitz.consul.model.health.ServiceHealth;
-import org.apache.http.client.fluent.Request;
-import org.apache.http.entity.StringEntity;
-import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups;
import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags;
import
org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest;
import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag;
-import
org.apache.streampipes.svcdiscovery.consul.model.ConsulServiceRegistrationBody;
-import
org.apache.streampipes.svcdiscovery.consul.model.HealthCheckConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
@@ -46,28 +45,12 @@ public class SpConsulServiceDiscovery extends
AbstractConsulService implements I
private static final String HTTP_PROTOCOL = "http://";;
private static final String COLON = ":";
- private static final String SLASH = "/";
private static final String HEALTH_CHECK_INTERVAL = "10s";
private static final String PE_SVC_TAG = "pe";
@Override
public void registerService(SpServiceRegistrationRequest req) {
- boolean connected = false;
-
- while (!connected) {
- LOG.info("Trying to register service at Consul with svcGroup={},
svcId={} host={}, port={}. ", req.getSvcGroup(), req.getSvcId(), req.getHost(),
req.getPort());
- ConsulServiceRegistrationBody svcRegistration =
createRegistrationBody(req);
- connected = registerServiceHttpClient(svcRegistration);
-
- if (!connected) {
- LOG.info("Retrying in 1 second");
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- }
+ consulInstance().agentClient().register((createRegistrationBody(req)));
LOG.info("Successfully registered service at Consul: " + req.getSvcId());
}
@@ -91,24 +74,7 @@ public class SpConsulServiceDiscovery extends
AbstractConsulService implements I
@Override
public List<String> getServiceEndpoints(String svcGroup, boolean
restrictToHealthy, List<String> filterByTags) {
- Consul consul = consulInstance();
- HealthClient healthClient = consul.healthClient();
- List<String> endpoints = new LinkedList<>();
- List<ServiceHealth> nodes;
-
- if (!restrictToHealthy) {
- nodes = healthClient.getAllServiceInstances(svcGroup).getResponse();
- } else {
- nodes = healthClient.getHealthyServiceInstances(svcGroup).getResponse();
- }
- for (ServiceHealth node : nodes) {
- if (node.getService().getTags().containsAll(filterByTags)) {
- String endpoint = node.getService().getAddress() + ":" +
node.getService().getPort();
- LOG.info("Active " + svcGroup + " endpoint: " + endpoint);
- endpoints.add(endpoint);
- }
- }
- return endpoints;
+ return ConsulHealthServiceManager.INSTANCE.getServiceEndpoints(svcGroup,
restrictToHealthy, filterByTags);
}
@Override
@@ -143,43 +109,20 @@ public class SpConsulServiceDiscovery extends
AbstractConsulService implements I
consul.agentClient().deregister(svcId);
}
- /**
- * PUT REST call to Consul API to register new service.
- *
- * @param svcRegistration service registration object used to register
service endpoint
- * @return success or failure of service registration
- */
- private boolean registerServiceHttpClient(ConsulServiceRegistrationBody
svcRegistration) {
- try {
- String endpoint = makeConsulEndpoint();
- String body =
JacksonSerializer.getObjectMapper().writeValueAsString(svcRegistration);
-
- Request.Put(endpoint)
- .addHeader("accept", "application/json")
- .body(new StringEntity(body))
- .execute();
-
- return true;
- } catch (Exception e) {
- LOG.error("Could not register service at Consul");
- }
- return false;
- }
-
- private ConsulServiceRegistrationBody
createRegistrationBody(SpServiceRegistrationRequest req) {
- ConsulServiceRegistrationBody body = new ConsulServiceRegistrationBody();
- body.setID(req.getSvcId());
- body.setName(req.getSvcGroup());
- body.setTags(asString(req.getTags()));
- body.setAddress(HTTP_PROTOCOL + req.getHost());
- body.setPort(req.getPort());
- body.setEnableTagOverride(true);
- body.setCheck(new HealthCheckConfiguration(
- "GET",
- (HTTP_PROTOCOL + req.getHost() + COLON + req.getPort() +
req.getHealthCheckPath()),
- HEALTH_CHECK_INTERVAL,
- "60s"));
-
- return body;
+ private Registration createRegistrationBody(SpServiceRegistrationRequest
req) {
+ return ImmutableRegistration.builder()
+ .id(req.getSvcId())
+ .name(req.getSvcGroup())
+ .port(req.getPort())
+ .address(HTTP_PROTOCOL + req.getHost())
+ .check(ImmutableRegCheck.builder()
+ .http(HTTP_PROTOCOL + req.getHost() + COLON +
req.getPort() + req.getHealthCheckPath())
+ .interval(HEALTH_CHECK_INTERVAL)
+ .deregisterCriticalServiceAfter("120s")
+ .status("passing")
+ .build())
+ .tags(asString(req.getTags()))
+ .enableTagOverride(true)
+ .build();
}
}
diff --git
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/model/ConsulServiceRegistrationBody.java
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/model/ConsulServiceRegistrationBody.java
deleted file mode 100644
index 491d6da..0000000
---
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/model/ConsulServiceRegistrationBody.java
+++ /dev/null
@@ -1,90 +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.svcdiscovery.consul.model;
-
-import java.util.List;
-
-public class ConsulServiceRegistrationBody {
-
- private String ID;
- private String Name;
- private List<String> Tags;
- private String Address;
- private Integer Port;
- private Boolean EnableTagOverride;
- private HealthCheckConfiguration Check;
-
- public ConsulServiceRegistrationBody() {
- }
-
- public String getID() {
- return ID;
- }
-
- public void setID(String ID) {
- this.ID = ID;
- }
-
- public String getName() {
- return Name;
- }
-
- public void setName(String name) {
- Name = name;
- }
-
- public List<String> getTags() {
- return Tags;
- }
-
- public void setTags(List<String> tags) {
- Tags = tags;
- }
-
- public String getAddress() {
- return Address;
- }
-
- public void setAddress(String address) {
- Address = address;
- }
-
- public Integer getPort() {
- return Port;
- }
-
- public void setPort(Integer port) {
- Port = port;
- }
-
- public Boolean getEnableTagOverride() {
- return EnableTagOverride;
- }
-
- public void setEnableTagOverride(Boolean enableTagOverride) {
- EnableTagOverride = enableTagOverride;
- }
-
- public HealthCheckConfiguration getCheck() {
- return Check;
- }
-
- public void setCheck(HealthCheckConfiguration check) {
- Check = check;
- }
-}
diff --git
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/model/HealthCheckConfiguration.java
b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/model/HealthCheckConfiguration.java
deleted file mode 100644
index ed5ca91..0000000
---
a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/model/HealthCheckConfiguration.java
+++ /dev/null
@@ -1,70 +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.svcdiscovery.consul.model;
-
-public class HealthCheckConfiguration {
- private String Method;
- private String http;
- private String interval;
- private String deregister_critical_service_after;
-
- public HealthCheckConfiguration() {
- }
-
- public HealthCheckConfiguration(String method,
- String http,
- String interval,
- String deregisterAfter) {
- Method = method;
- this.http = http;
- this.interval = interval;
- this.deregister_critical_service_after = deregisterAfter;
- }
-
- public String getMethod() {
- return Method;
- }
-
- public void setMethod(String method) {
- Method = method;
- }
-
- public String getHttp() {
- return http;
- }
-
- public void setHttp(String http) {
- this.http = http;
- }
-
- public String getInterval() {
- return interval;
- }
-
- public void setInterval(String interval) {
- this.interval = interval;
- }
-
- public String getDeregister_critical_service_after() {
- return deregister_critical_service_after;
- }
-
- public void setDeregister_critical_service_after(String
deregister_critical_service_after) {
- this.deregister_critical_service_after =
deregister_critical_service_after;
- }
-}
diff --git
a/streampipes-service-extensions-base/src/main/java/org/apache/streampipes/service/extensions/base/StreamPipesExtensionsServiceBase.java
b/streampipes-service-extensions-base/src/main/java/org/apache/streampipes/service/extensions/base/StreamPipesExtensionsServiceBase.java
index c7fb289..d65d6e4 100644
---
a/streampipes-service-extensions-base/src/main/java/org/apache/streampipes/service/extensions/base/StreamPipesExtensionsServiceBase.java
+++
b/streampipes-service-extensions-base/src/main/java/org/apache/streampipes/service/extensions/base/StreamPipesExtensionsServiceBase.java
@@ -41,6 +41,8 @@ public abstract class StreamPipesExtensionsServiceBase
extends StreamPipesServic
try {
String host = getHostname();
Integer port = getPort(serviceDef.getDefaultPort());
+ String serviceId = serviceDef.getServiceGroup() + "-" +
AUTO_GENERATED_SERVICE_ID;
+ serviceDef.setServiceId(serviceId);
DeclarersSingleton.getInstance().populate(host, port, serviceDef);
startExtensionsService(this.getClass(), serviceDef);
@@ -60,7 +62,7 @@ public abstract class StreamPipesExtensionsServiceBase
extends StreamPipesServic
this.startStreamPipesService(
serviceClass,
DefaultSpServiceGroups.EXT,
- serviceDef.getServiceId(),
+ serviceId(),
serviceDef.getDefaultPort()
);
this.afterServiceRegistered(serviceDef);
@@ -69,4 +71,8 @@ public abstract class StreamPipesExtensionsServiceBase
extends StreamPipesServic
@PreDestroy
public abstract void onExit();
+ public String serviceId() {
+ return DeclarersSingleton.getInstance().getServiceId();
+ }
+
}
