add periodic fetch configuration

Issue-ID: DCAEGEN2-2007

Signed-off-by: Kai Lu <lukai@chinamobile.com>
Change-Id: Iab1c329534e37323eb7d0f1227b2d5c16e1cb50c

7 files changed, 358 insertions, 244 deletions

diff --git a/dcae-analytics/dcae-analytics-model/pom.xml b/dcae-analytics/dcae-analytics-model/pom.xml
index b16958a..337dcb4 100644
--- a/dcae-analytics/dcae-analytics-model/pom.xml
+++ b/dcae-analytics/dcae-analytics-model/pom.xml
@@ -38,9 +38,11 @@
     <name>DCAE Analytics Model</name>
     <description>Contains models (e.g. Common Event Format) which are common to DCAE Analytics</description>
 
-    <properties>
-        <main.basedir>${project.parent.basedir}</main.basedir>
-    </properties>
+	<properties>
+		<main.basedir>${project.parent.basedir}</main.basedir>
+		<sdk.version>1.1.6</sdk.version>
+		<immutable.version>2.7.1</immutable.version>
+	</properties>
 
     <dependencies>
 
@@ -75,6 +77,48 @@
             <artifactId>dcae-analytics-test</artifactId>
         </dependency>
 
-    </dependencies>
+	<dependency>
+		<groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
+		<artifactId>cbs-client</artifactId>
+		<version>${sdk.version}</version>
+	</dependency>
+	<dependency>
+		<groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
+		<artifactId>dmaap-client</artifactId>
+		<version>${sdk.version}</version>
+	</dependency>
+	<dependency>
+		<groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
+		<artifactId>common-dependency</artifactId>
+		<version>${sdk.version}</version>
+	</dependency>
+	<dependency>
+		<groupId>org.springframework</groupId>
+		<artifactId>spring-beans</artifactId>
+	</dependency>
+	<dependency>
+		<groupId>org.springframework</groupId>
+		<artifactId>spring-context</artifactId>
+	</dependency>
+	<dependency>
+		<groupId>org.springframework</groupId>
+		<artifactId>spring-webflux</artifactId>
+	</dependency>
+	<dependency>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-autoconfigure</artifactId>
+	</dependency>
+	<dependency>
+		<groupId>javax.validation</groupId>
+		<artifactId>validation-api</artifactId>
+	</dependency>
+        <dependency>
+                <groupId>org.immutables</groupId>
+                <artifactId>value</artifactId>
+                <version>${immutable.version}</version>
+                <scope>provided</scope>
+        </dependency>
+
+	</dependencies>
 
 </project>
diff --git a/dcae-analytics/dcae-analytics-model/src/main/java/org/onap/dcae/analytics/model/configbindingservice/ConfigBindingServiceConstants.java b/dcae-analytics/dcae-analytics-model/src/main/java/org/onap/dcae/analytics/model/configbindingservice/ConfigBindingServiceConstants.java
index 786765f..9776d70 100644
--- a/dcae-analytics/dcae-analytics-model/src/main/java/org/onap/dcae/analytics/model/configbindingservice/ConfigBindingServiceConstants.java
+++ b/dcae-analytics/dcae-analytics-model/src/main/java/org/onap/dcae/analytics/model/configbindingservice/ConfigBindingServiceConstants.java
@@ -34,6 +34,7 @@ public abstract class ConfigBindingServiceConstants {
 
     // ================== CONFIG SERVICE CONSTANTS ============================== //
     public static final String CONSUL_HOST_ENV_VARIABLE_KEY = "CONSUL_HOST";
+    public static final Integer DEFAULT_CONSUL_PORT_ENV_VARIABLE_VALUE = 8500;
     public static final String CONSUL_HOST_ENV_VARIABLE_VALUE = System.getenv(CONSUL_HOST_ENV_VARIABLE_KEY);
     public static final String CONFIG_BINDING_SERVICE_ENV_VARIABLE_KEY = "CONFIG_BINDING_SERVICE";
     public static final String CONFIG_BINDING_SERVICE_ENV_VARIABLE_VALUE =
@@ -47,12 +48,13 @@ public abstract class ConfigBindingServiceConstants {
 
     public static final String CONFIG_BINDING_SERVICE_PROPERTIES_KEY = "config-binding-service";
 
-
     public static final Set<String> SPRING_RESERVED_PROPERTIES_KEY_PREFIXES =
             Stream.of("spring", "endpoints", "server", "logging", "management").collect(Collectors.toSet());
     public static final String CONFIG_SERVICE_MESSAGE_ROUTER_VALUE = "message_router";
     // CONVERT JSON TO MAP
     public static final String KEY_SEPARATOR = ".";
+    public static final String CONFIG = "config";
+    public static final int CONFIG_SERVICE_REFRESHPERIOD = 1;
 
     // ============== CONFIG BINDING SERVICE UTILS ========================= //
     /**
diff --git a/dcae-analytics/dcae-analytics-model/src/main/java/org/onap/dcae/analytics/model/util/supplier/ConfigBindingServiceJsonSupplier.java b/dcae-analytics/dcae-analytics-model/src/main/java/org/onap/dcae/analytics/model/util/supplier/ConfigBindingServiceJsonSupplier.java
deleted file mode 100644
index b882bb0..0000000
--- a/dcae-analytics/dcae-analytics-model/src/main/java/org/onap/dcae/analytics/model/util/supplier/ConfigBindingServiceJsonSupplier.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * ================================================================================
- * Copyright (c) 2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed 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.
- * ============LICENSE_END=========================================================
- *
- */
-
-package org.onap.dcae.analytics.model.util.supplier;
-
-
-import java.util.List;
-import java.util.Optional;
-import java.util.function.Function;
-import java.util.function.Supplier;
-
-import org.onap.dcae.analytics.model.configbindingservice.ConfigBindingServiceConstants;
-import org.onap.dcae.analytics.model.configbindingservice.ConsulConfigBindingServiceQueryResponse;
-import org.onap.dcae.analytics.model.util.function.StringToURLFunction;
-import org.onap.dcae.analytics.model.util.function.URLToHttpGetFunction;
-import org.onap.dcae.analytics.model.util.json.AnalyticsModelJsonConversion;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A Function which fetches ApplicationProperties configuration from
- * Config Binding Service when deployed via docker, typically during application startup time.
- *
- * @author Rajiv Singla
- */
-public class ConfigBindingServiceJsonSupplier implements Supplier<Optional<String>> {
-
-    private static final Logger logger = LoggerFactory.getLogger(ConfigBindingServiceJsonSupplier.class);
-
-    private final Function<String, Optional<String>> fetchUrlContentFunction;
-
-    public ConfigBindingServiceJsonSupplier(final Function<String, Optional<String>> fetchUrlContentFunction) {
-        this.fetchUrlContentFunction = fetchUrlContentFunction;
-    }
-
-    public ConfigBindingServiceJsonSupplier() {
-        fetchUrlContentFunction = (String s) -> new StringToURLFunction().apply(s).flatMap(new URLToHttpGetFunction());
-    }
-
-    @Override
-    public Optional<String> get() {
-
-        logger.info("Consul Host Environment Variable: {}",
-                ConfigBindingServiceConstants.CONSUL_HOST_ENV_VARIABLE_VALUE);
-        logger.info("Config Binding Service Environment Variable: {}",
-                ConfigBindingServiceConstants.CONFIG_BINDING_SERVICE_ENV_VARIABLE_VALUE);
-        logger.info("Service Name Environment Variable: {}",
-                ConfigBindingServiceConstants.SERVICE_NAME_ENV_VARIABLE_VALUE);
-
-        if (ConfigBindingServiceConstants.CONSUL_HOST_ENV_VARIABLE_VALUE == null ||
-                ConfigBindingServiceConstants.CONFIG_BINDING_SERVICE_ENV_VARIABLE_VALUE == null ||
-                ConfigBindingServiceConstants.SERVICE_NAME_ENV_VARIABLE_VALUE == null) {
-            logger.error("Environment variables required to query Config Binding Service are not present");
-            return Optional.empty();
-        }
-
-        return Optional.of(ConfigBindingServiceConstants.CONSUL_QUERY_URL_STRING)
-                // Step 1: Query CONSUL to get the IP/PORT of CONFIG BINDING SERVICE
-                .flatMap(fetchUrlContentFunction)
-                // Step 2: Fetch the generated configurations from CONFIG BINDING SERVICE
-                .flatMap(ConfigBindingServiceJsonSupplier::parseConsulConfigBindingServiceQueryResponseJson)
-                // Step 3: create url from service address and service port
-                .flatMap(ConfigBindingServiceJsonSupplier::createConfigServiceURL)
-                // Step 4: Fetch final config binding service generated application configuration json string
-                .flatMap(fetchUrlContentFunction);
-    }
-
-    /**
-     * Creates URL using config binding service ip address and port
-     *
-     * @param consulConfigBindingServiceQueryResponse consul config binding service query response containing config
-     * binding service address and service port
-     *
-     * @return config service url to fetch service configuration
-     */
-    private static Optional<String> createConfigServiceURL(final ConsulConfigBindingServiceQueryResponse
-                                                                   consulConfigBindingServiceQueryResponse) {
-        final String configBindingServiceAddress = consulConfigBindingServiceQueryResponse.getServiceAddress();
-        final Integer configServicePort = consulConfigBindingServiceQueryResponse.getServicePort();
-
-        if (configBindingServiceAddress == null && configServicePort == null) {
-            logger.error("Config Binding Service Address & Port are not present.");
-            return Optional.empty();
-        }
-
-        return Optional.of(String.format(ConfigBindingServiceConstants.CONFIG_SERVICE_QUERY_URL_STRING,
-                configBindingServiceAddress, configServicePort,
-                ConfigBindingServiceConstants.SERVICE_NAME_ENV_VARIABLE_VALUE));
-
-    }
-
-    private static Optional<ConsulConfigBindingServiceQueryResponse>
-    parseConsulConfigBindingServiceQueryResponseJson(final String configBindingServiceQueryResponseJson) {
-        // parse json
-        final Optional<List<ConsulConfigBindingServiceQueryResponse>> configBindingServiceQueryResponseOptional =
-                AnalyticsModelJsonConversion.CONFIG_BINDING_SERVICE_LIST_JSON_FUNCTION
-                        .apply(configBindingServiceQueryResponseJson);
-
-        // check parsing is successful and at least 1 config binding query response is present
-        if (!configBindingServiceQueryResponseOptional.isPresent() ||
-                configBindingServiceQueryResponseOptional.get().isEmpty()) {
-            logger.error("No Consul config binding service information found in JSON: {} ",
-                    configBindingServiceQueryResponseJson);
-            return Optional.empty();
-        }
-
-        // return first consul query response
-        return Optional.of(configBindingServiceQueryResponseOptional.get().get(0));
-    }
-
-
-}
diff --git a/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/config/SystemConfig.java b/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/config/SystemConfig.java
new file mode 100644
index 0000000..946c2f2
--- /dev/null
+++ b/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/config/SystemConfig.java
@@ -0,0 +1,84 @@
+/*-
+ * ============LICENSE_START======================================================================
+ * Copyright (C) 2019-2020 China Mobile.  All rights reserved.
+ * ===============================================================================================
+ * Licensed 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.
+ * ============LICENSE_END========================================================================
+ */
+
+package org.onap.dcae.analytics.web.config;
+
+import java.util.Optional;
+
+import org.onap.dcae.analytics.model.configbindingservice.ConfigBindingServiceConstants;
+import org.onap.dcae.analytics.web.exception.EnvironmentLoaderException;
+
+/**
+ * 
+ * get consul config and cbs config.
+ *
+ * @author Kai Lu
+ */
+public class SystemConfig {
+
+    /**
+     * 
+     * private constructor.
+     *
+     */
+    private SystemConfig() {
+    }
+
+    /**
+     * getConsulHost.
+     *
+     * @return consulHost consulHost
+     *
+     */
+    public static String getConsulHost() throws EnvironmentLoaderException {
+        return Optional.ofNullable(ConfigBindingServiceConstants.CONSUL_HOST_ENV_VARIABLE_VALUE)
+                .orElseThrow(() -> new EnvironmentLoaderException("$CONSUL_HOST environment has not been defined"));
+    }
+
+    /**
+     * getConsultPort.
+     *
+     * @return consulPort consulPort
+     *
+     */
+    public static Integer getConsultPort() {
+        return ConfigBindingServiceConstants.DEFAULT_CONSUL_PORT_ENV_VARIABLE_VALUE;
+    }
+
+    /**
+     * getConfigBindingService.
+     *
+     * @return configBindingService configBindingService
+     *
+     */
+    public static String getConfigBindingService() throws EnvironmentLoaderException {
+        return Optional.ofNullable(ConfigBindingServiceConstants.CONFIG_BINDING_SERVICE_ENV_VARIABLE_VALUE) //
+                .orElseThrow(() -> new EnvironmentLoaderException(
+                        "$CONFIG_BINDING_SERVICE environment has not been defined"));
+    }
+
+    /**
+     * getService.
+     *
+     * @return service service
+     *
+     */
+    public static String getService() throws EnvironmentLoaderException {
+        return Optional.ofNullable(ConfigBindingServiceConstants.SERVICE_NAME_ENV_VARIABLE_VALUE).orElseThrow(
+                () -> new EnvironmentLoaderException("$HOSTNAME have not been defined as system environment"));
+    }
+
+}
diff --git a/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/exception/EnvironmentLoaderException.java b/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/exception/EnvironmentLoaderException.java
new file mode 100644
index 0000000..0a5c14d
--- /dev/null
+++ b/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/exception/EnvironmentLoaderException.java
@@ -0,0 +1,33 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019-2020 China Mobile. All rights reserved.
+ * ================================================================================
+ * Licensed 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.dcae.analytics.web.exception;
+
+/**
+ * Exception thrown when there is a problem with the Consul environment.
+ *
+ * @author Kai Lu
+ */
+public class EnvironmentLoaderException extends Exception {
+
+    private static final long serialVersionUID = 1L;
+
+    public EnvironmentLoaderException(String message) {
+        super(message);
+    }
+}
diff --git a/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/spring/ConfigBindingServiceEnvironmentPostProcessor.java b/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/spring/ConfigBindingServiceEnvironmentPostProcessor.java
index 2073127..59fca64 100644
--- a/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/spring/ConfigBindingServiceEnvironmentPostProcessor.java
+++ b/dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/spring/ConfigBindingServiceEnvironmentPostProcessor.java
@@ -19,6 +19,7 @@
 
 package org.onap.dcae.analytics.web.spring;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
@@ -29,8 +30,16 @@ import java.util.stream.Collectors;
 import org.onap.dcae.analytics.model.AnalyticsProfile;
 import org.onap.dcae.analytics.model.configbindingservice.ConfigBindingServiceConstants;
 import org.onap.dcae.analytics.model.util.function.JsonStringToMapFunction;
-import org.onap.dcae.analytics.model.util.supplier.ConfigBindingServiceJsonSupplier;
+import org.onap.dcae.analytics.web.config.SystemConfig;
 import org.onap.dcae.analytics.web.exception.AnalyticsValidationException;
+import org.onap.dcae.analytics.web.exception.EnvironmentLoaderException;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.EnvProperties;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.ImmutableEnvProperties;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.SpringApplication;
@@ -44,6 +53,13 @@ import org.springframework.core.env.StandardEnvironment;
 import org.springframework.util.ClassUtils;
 import org.springframework.web.context.support.StandardServletEnvironment;
 
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 /**
  * A custom spring framework environment post processor which can fetch and populate spring context with
  * Config Binding Service application properties.
@@ -60,63 +76,33 @@ public class ConfigBindingServiceEnvironmentPostProcessor implements Environment
 
     private static final int DEFAULT_ORDER = Ordered.HIGHEST_PRECEDENCE;
 
+    private Disposable refreshConfigTask = null;
+
+    private ConfigurableEnvironment env = null;
+
+    private Map<String, Object> filterKeyMap = null;
+
+    private String configServicePropertiesKey =
+            ConfigBindingServiceConstants.CONFIG_BINDING_SERVICE_PROPERTIES_KEY;
+
     @Override
-    public void postProcessEnvironment(final ConfigurableEnvironment environment, final SpringApplication application) {
+    public void postProcessEnvironment(final ConfigurableEnvironment environment,
+            final SpringApplication application) {
 
         final boolean isConfigServiceProfilePresent = Arrays.stream(environment.getActiveProfiles())
                 .anyMatch(p -> p.equalsIgnoreCase(AnalyticsProfile.CONFIG_BINDING_SERVICE_PROFILE_NAME));
 
         if (!isConfigServiceProfilePresent) {
-            logger.info("Config Binding Service Profile is not active. " +
-                    "Skipping Adding config binding service properties");
+            logger.info("Config Binding Service Profile is not active. "
+                    + "Skipping Adding config binding service properties");
             return;
         }
 
-        logger.info("Config Binding Service Profile is active. " +
-                "Application properties will be fetched from config binding service");
-
-        // Fetch config binding service json
-        final Optional<String> configServiceJsonOptional = new ConfigBindingServiceJsonSupplier().get();
+        logger.info("Config Binding Service Profile is active. "
+                + "Application properties will be fetched from config binding service");
 
-        if (!configServiceJsonOptional.isPresent()) {
-            final String errorMessage = "Unable to get fetch application configuration from config binding service";
-            throw new AnalyticsValidationException(errorMessage, new IllegalStateException(errorMessage));
-        }
-
-        final String configServicePropertiesKey = ConfigBindingServiceConstants.CONFIG_BINDING_SERVICE_PROPERTIES_KEY;
-
-        // convert fetch config binding service json string to Map of property key and values
-        final Map<String, Object> configPropertiesMap = configServiceJsonOptional
-                .map(new JsonStringToMapFunction(configServicePropertiesKey))
-                .orElse(Collections.emptyMap());
-
-        if (configPropertiesMap.isEmpty()) {
-
-            logger.warn("No properties found in config binding service");
-
-        } else {
-
-            // remove config service key prefix on spring reserved property key prefixes
-            final Set<String> springKeyPrefixes = ConfigBindingServiceConstants.SPRING_RESERVED_PROPERTIES_KEY_PREFIXES;
-            final Set<String> springKeys = springKeyPrefixes.stream()
-                    .map(springKeyPrefix -> configServicePropertiesKey + "." + springKeyPrefix)
-                    .collect(Collectors.toSet());
-
-            final Map<String, Object> filterKeyMap = configPropertiesMap.entrySet()
-                    .stream()
-                    .collect(Collectors.toMap(
-                            (Map.Entry<String, Object> e) ->
-                                    springKeys.stream().anyMatch(springKey -> e.getKey().startsWith(springKey)) ?
-                                            e.getKey().substring(configServicePropertiesKey.toCharArray().length + 1) :
-                                            e.getKey(),
-                            Map.Entry::getValue)
-                    );
-
-            filterKeyMap.forEach((key, value) ->
-                    logger.info("Adding property from config service in spring context: {} -> {}", key, value));
-
-            addJsonPropertySource(environment, new MapPropertySource(configServicePropertiesKey, filterKeyMap));
-        }
+        env = environment;
+        initialize();
 
     }
 
@@ -125,9 +111,8 @@ public class ConfigBindingServiceEnvironmentPostProcessor implements Environment
         return DEFAULT_ORDER;
     }
 
-
-    private void addJsonPropertySource(final ConfigurableEnvironment environment, final PropertySource<?> source) {
-        final MutablePropertySources sources = environment.getPropertySources();
+    public synchronized void addJsonPropertySource(final MutablePropertySources sources,
+            final PropertySource<?> source) {
         final String name = findPropertySource(sources);
         if (sources.contains(name)) {
             sources.addBefore(name, source);
@@ -137,12 +122,164 @@ public class ConfigBindingServiceEnvironmentPostProcessor implements Environment
     }
 
     private String findPropertySource(final MutablePropertySources sources) {
-        if (ClassUtils.isPresent(SERVLET_ENVIRONMENT_CLASS, null) &&
-                sources.contains(StandardServletEnvironment.JNDI_PROPERTY_SOURCE_NAME)) {
+        if (ClassUtils.isPresent(SERVLET_ENVIRONMENT_CLASS, null)
+                && sources.contains(StandardServletEnvironment.JNDI_PROPERTY_SOURCE_NAME)) {
             return StandardServletEnvironment.JNDI_PROPERTY_SOURCE_NAME;
 
         }
         return StandardEnvironment.SYSTEM_PROPERTIES_PROPERTY_SOURCE_NAME;
     }
 
+    /**
+     *
+     * Fetch the configuration.
+     *
+     */
+    public void initialize() {
+        stop();
+
+        refreshConfigTask = createRefreshTask() //
+                .subscribe(e -> logger.info("Refreshed configuration data"),
+                        throwable -> logger.error("Configuration refresh terminated due to exception", throwable),
+                        () -> logger.error("Configuration refresh terminated"));
+    }
+
+    /**
+     * Fetch the configuration task from CBS.
+     *
+     */
+    private Flux<String> createRefreshTask() {
+        return readEnvironmentVariables() //
+                .flatMap(this::createCbsClient) //
+                .flatMapMany(this::periodicConfigurationUpdates) //
+                .map(this::parseTcaConfig) //
+                .onErrorResume(this::onErrorResume);
+    }
+
+    /**
+     * periodicConfigurationUpdates.
+     *
+     * @param cbsClient cbsClient
+     * @return configuration refreshed
+     *
+     */
+    public Flux<JsonObject> periodicConfigurationUpdates(CbsClient cbsClient) {
+        final Duration initialDelay = Duration.ZERO;
+        final Duration refreshPeriod =
+                 Duration.ofMinutes(ConfigBindingServiceConstants.CONFIG_SERVICE_REFRESHPERIOD);
+        final CbsRequest getConfigRequest = CbsRequests.getAll(RequestDiagnosticContext.create());
+        return cbsClient.updates(getConfigRequest, initialDelay, refreshPeriod);
+    }
+
+    /**
+     *
+     * get environment variables.
+     *
+     * @return environment properties.
+     *
+     */
+    public Mono<EnvProperties> readEnvironmentVariables() {
+        logger.trace("Loading configuration from system environment variables");
+        EnvProperties envProperties;
+        try {
+            envProperties = ImmutableEnvProperties.builder() //
+                    .consulHost(SystemConfig.getConsulHost()) //
+                    .consulPort(SystemConfig.getConsultPort()) //
+                    .cbsName(SystemConfig.getConfigBindingService()) //
+                    .appName(SystemConfig.getService()) //
+                    .build();
+        } catch (EnvironmentLoaderException e) {
+            return Mono.error(e);
+        }
+        logger.trace("Evaluated environment system variables {}", envProperties);
+        return Mono.just(envProperties);
+    }
+
+    /**
+     * Stops the refreshing of the configuration.
+     *
+     */
+    public void stop() {
+        if (refreshConfigTask != null) {
+            refreshConfigTask.dispose();
+            refreshConfigTask = null;
+        }
+    }
+
+    /**
+     * periodicConfigurationUpdates.
+     *
+     * @param throwable throwable
+     * @return Mono
+     *
+     */
+    private <R> Mono<R> onErrorResume(Throwable throwable) {
+        logger.error("Could not refresh application configuration {}", throwable.toString());
+        return Mono.empty();
+    }
+
+    /**
+     * create CbsClient.
+     *
+     * @param env environment properties
+     * @return cbsclient
+     *
+     */
+    public Mono<CbsClient> createCbsClient(EnvProperties env) {
+        return CbsClientFactory.createCbsClient(env);
+    }
+
+    /**
+     * Parse configuration.
+     *
+     * @param jsonObject the TCA service's configuration
+     * @return this
+     *
+     */
+    public String parseTcaConfig(JsonObject jsonObject) {
+
+        JsonElement jsonConfig = jsonObject.get(ConfigBindingServiceConstants.CONFIG);
+
+        Optional<String> configServiceJsonOptional = Optional.of(jsonConfig.toString());
+        if (!configServiceJsonOptional.isPresent()) {
+            final String errorMessage =
+                    "Unable to get fetch application configuration from config binding service";
+            throw new AnalyticsValidationException(errorMessage,
+                    new IllegalStateException(errorMessage));
+        }
+
+        // convert fetch config binding service json string to Map of property key and
+        // values
+        Map<String, Object> configPropertiesMap = configServiceJsonOptional
+                .map(new JsonStringToMapFunction(configServicePropertiesKey)).orElse(Collections.emptyMap());
+
+        if (configPropertiesMap.isEmpty()) {
+
+            logger.warn("No properties found in config binding service");
+
+        } else {
+
+            // remove config service key prefix on spring reserved property key prefixes
+            final Set<String> springKeyPrefixes =
+                    ConfigBindingServiceConstants.SPRING_RESERVED_PROPERTIES_KEY_PREFIXES;
+            final Set<String> springKeys = springKeyPrefixes.stream()
+                    .map(springKeyPrefix -> configServicePropertiesKey + "." + springKeyPrefix)
+                    .collect(Collectors.toSet());
+
+            filterKeyMap = configPropertiesMap.entrySet().stream()
+                    .collect(Collectors.toMap((Map.Entry<String, Object> e) -> springKeys.stream()
+                            .anyMatch(springKey -> e.getKey().startsWith(springKey))
+                                    ? e.getKey().substring(configServicePropertiesKey.toCharArray().length + 1)
+                                    : e.getKey(),
+                            Map.Entry::getValue));
+
+            filterKeyMap.forEach((key, value) -> logger
+                    .info("Adding property from config service in spring context: {} -> {}", key, value));
+            MutablePropertySources sources = env.getPropertySources();
+            addJsonPropertySource(sources, new MapPropertySource(configServicePropertiesKey, filterKeyMap));
+
+        }
+        return configServiceJsonOptional.get();
+    }
+
 }
diff --git a/dcae-analytics/dcae-analytics-web/src/test/java/org/onap/dcae/analytics/web/spring/ConfigBindingServiceEnvironmentPostProcessorTest.java b/dcae-analytics/dcae-analytics-web/src/test/java/org/onap/dcae/analytics/web/spring/ConfigBindingServiceEnvironmentPostProcessorTest.java
deleted file mode 100644
index 5506342..0000000
--- a/dcae-analytics/dcae-analytics-web/src/test/java/org/onap/dcae/analytics/web/spring/ConfigBindingServiceEnvironmentPostProcessorTest.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * ================================================================================
- * Copyright (c) 2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed 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.
- * ============LICENSE_END=========================================================
- *
- */
-
-package org.onap.dcae.analytics.web.spring;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-import org.onap.dcae.analytics.model.AnalyticsProfile;
-import org.onap.dcae.analytics.web.BaseAnalyticsWebTest;
-import org.onap.dcae.analytics.web.exception.AnalyticsValidationException;
-import org.springframework.boot.SpringApplication;
-import org.springframework.core.env.ConfigurableEnvironment;
-import org.springframework.core.env.MutablePropertySources;
-
-class ConfigBindingServiceEnvironmentPostProcessorTest extends BaseAnalyticsWebTest {
-
-    @BeforeAll
-    static void beforeAll() throws Exception {
-        BaseAnalyticsWebTest.initializeConfigBindingServiceEnvironmentVariables();
-    }
-
-
-    @Test
-    void postProcessEnvironment() {
-
-        Assertions.assertThrows(AnalyticsValidationException.class, () -> {
-
-            final ConfigBindingServiceEnvironmentPostProcessor configBindingServiceEnvironmentPostProcessor =
-                    new ConfigBindingServiceEnvironmentPostProcessor();
-
-            final ConfigurableEnvironment configurableEnvironment = Mockito.mock(ConfigurableEnvironment.class);
-            final SpringApplication springApplication = Mockito.mock(SpringApplication.class);
-            final String[] activeProfiles = {AnalyticsProfile.CONFIG_BINDING_SERVICE_PROFILE_NAME};
-            Mockito.when(configurableEnvironment.getActiveProfiles()).thenReturn(activeProfiles);
-
-            configBindingServiceEnvironmentPostProcessor
-                    .postProcessEnvironment(configurableEnvironment, springApplication);
-        });
-    }
-}