Revert "Remove CBS and Consul dependencies from CBSClient lib"kohn

This reverts commit 592f371519d0647b7ae9a0a2fc601c757133cf0d.

This commit belongs to London/master branch only

Change-Id: Ie92be8fa8eda72462d7411646bbef6b52760d29a
Signed-off-by: Vijay Venkatesh Kumar <vv770d@att.com>
Issue-ID: DCAEGEN2-3236
Signed-off-by: Vijay Venkatesh Kumar <vv770d@att.com>

18 files changed, 1265 insertions, 25 deletions

diff --git a/Changelog.md b/Changelog.md
index 69247c0c..31994d17 100644
--- a/Changelog.md
+++ b/Changelog.md
@@ -4,10 +4,6 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
-## [1.9.0] - 2022/09/07
-### Added
-    - [DCAEGEN2-3098] (https://jira.onap.org/browse/DCAEGEN2-3098) - Remove Consul and CBS property dependencies from CBS-client SDK
-
 ## [1.8.10] - 2022/07/29
 ### Added
     - [DCAEGEN2-3220] (https://jira.onap.org/browse/DCAEGEN2-3220) - Fix SDK Vulnerability. Top up Spring-Boot version to 2.7.2
diff --git a/pom.xml b/pom.xml
index 6771ea9c..68b37200 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2,7 +2,6 @@
 <!--
 ============LICENSE_START=======================================================
 Copyright (c) 2022 Nokia. All rights reserved.
-Copyright (C) 2022 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
@@ -96,13 +95,13 @@ language governing permissions and limitations under the License.
         <sonar.coverage.jacoco.xmlReportPaths>
             ${project.reporting.outputDirectory}/jacoco-ut/jacoco.xml
         </sonar.coverage.jacoco.xmlReportPaths>
-        <revision>1.9.0-SNAPSHOT</revision>
+        <revision>1.8.10-SNAPSHOT</revision>
     </properties>
 
     <modules>
         <module>rest-services</module>
         <module>services</module>
-        <module>security</module> 
+        <module>security</module>
         <module>standardization</module>
         <module>services/common</module>
     </modules>
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactory.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactory.java
index f0293564..e43cf312 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactory.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactory.java
@@ -4,7 +4,6 @@
  * ================================================================================
  * Copyright (C) 2019-2021 Nokia. All rights reserved.
  * Copyright (C) 2021 Wipro Limited.
- * Copyright (C) 2022 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.
@@ -25,6 +24,8 @@ import org.jetbrains.annotations.NotNull;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClientFactory;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.CbsClientConfigMap;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.CbsClientRest;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.CbsLookup;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
 import org.onap.dcaegen2.services.sdk.security.ssl.TrustStoreKeys;
 import org.slf4j.Logger;
@@ -74,6 +75,12 @@ public class CbsClientFactory {
         CbsClientConfiguration configuration) {
             CbsClientConfigMap cbsClientConfigMap = new CbsClientConfigMap(configuration.configMapFilePath(),
                     configuration.policySyncFilePath(), configuration.appName());
-        return Mono.just(cbsClientConfigMap);
+        return cbsClientConfigMap.verifyConfigMapFile() ? Mono.just(cbsClientConfigMap) :
+                getConfigFromCBS(httpClient, configuration);
+    }
+
+    private static Mono<CbsClient> getConfigFromCBS(RxHttpClient httpClient, CbsClientConfiguration configuration) {
+        return new CbsLookup().lookup(configuration)
+                .map(addr ->new CbsClientRest(httpClient, configuration.appName(), addr, configuration.protocol()));
     }
 }
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientRest.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientRest.java
new file mode 100644
index 00000000..3eda6d8a
--- /dev/null
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientRest.java
@@ -0,0 +1,88 @@
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2019-2021 Nokia. 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.dcaegen2.services.sdk.rest.services.cbs.client.impl;
+
+import com.google.gson.JsonObject;
+import org.jetbrains.annotations.NotNull;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+
+import java.net.InetSocketAddress;
+import java.net.MalformedURLException;
+import java.net.URL;
+
+public class CbsClientRest implements CbsClient {
+    private static final Logger LOGGER = LoggerFactory.getLogger(CbsClientRest.class);
+    private final RxHttpClient httpClient;
+    private final String serviceName;
+    private final InetSocketAddress cbsAddress;
+    private final String protocol;
+
+    public CbsClientRest(RxHttpClient httpClient, String serviceName, InetSocketAddress cbsAddress, String protocol) {
+        this.httpClient = httpClient;
+        this.serviceName = serviceName;
+        this.cbsAddress = cbsAddress;
+        this.protocol = protocol;
+    }
+
+    @Override
+    public @NotNull Mono<JsonObject> get(CbsRequest request) {
+        return Mono.fromCallable(() -> constructUrl(request).toString())
+                .doOnNext(this::logRequestUrl)
+                .map(url -> ImmutableHttpRequest.builder()
+                        .method(HttpMethod.GET)
+                        .url(url)
+                        .diagnosticContext(request.diagnosticContext())
+                        .build())
+                .flatMap(httpClient::call)
+                .doOnNext(HttpResponse::throwIfUnsuccessful)
+                .map(resp -> resp.bodyAsJson(JsonObject.class))
+                .map(CbsClientEnvironmentParsing::processEnvironmentVariables)
+                .doOnNext(this::logCbsResponse);
+    }
+
+    private void logRequestUrl(String url) {
+        LOGGER.debug("Calling {} for configuration", url);
+    }
+
+    private void logCbsResponse(JsonObject json) {
+        LOGGER.info("Got successful response from Config Binding Service");
+        LOGGER.debug("CBS response: {}", json);
+    }
+
+    private URL constructUrl(CbsRequest request) {
+        try {
+            return new URL(
+                    this.protocol,
+                    cbsAddress.getHostString(),
+                    cbsAddress.getPort(),
+                    request.requestPath().getForService(serviceName));
+        } catch (MalformedURLException e) {
+            throw new IllegalArgumentException("Invalid CBS URL", e);
+        }
+    }
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookup.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookup.java
new file mode 100644
index 00000000..c07ed8e4
--- /dev/null
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookup.java
@@ -0,0 +1,53 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. 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.dcaegen2.services.sdk.rest.services.cbs.client.impl;
+
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+
+import java.net.InetSocketAddress;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since February 2019
+ */
+public class CbsLookup {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(CbsLookup.class);
+
+    public Mono<InetSocketAddress> lookup(CbsClientConfiguration configuration) {
+        return Mono.just(createCbsAddress(configuration))
+                .doOnNext(this::logCbsServiceAddress);
+    }
+
+    private InetSocketAddress createCbsAddress(CbsClientConfiguration configuration) {
+        return InetSocketAddress.createUnresolved(
+                configuration.hostname(),
+                configuration.port());
+    }
+
+    private void logCbsServiceAddress(InetSocketAddress address) {
+        LOGGER.info("Config Binding Service address: {}", address);
+    }
+
+}
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/model/CbsClientConfiguration.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/model/CbsClientConfiguration.java
index fa265d85..6eec6674 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/model/CbsClientConfiguration.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/model/CbsClientConfiguration.java
@@ -4,7 +4,6 @@
  * =========================================================
  * Copyright (C) 2019-2021 Nokia. All rights reserved.
  * Copyright (C) 2021 Wipro Limited.
- * Copyright (C) 2022 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.
@@ -51,7 +50,7 @@ public interface CbsClientConfiguration {
 
     String TRUST_JKS = "trust.jks";
     String TRUST_PASS = "trust.pass";
-
+    Integer PORT_FOR_CBS_OVER_TLS = 10443;
 
     /**
      * Name of environment variable containing path to the cacert.pem file.
@@ -59,6 +58,16 @@ public interface CbsClientConfiguration {
     String DCAE_CA_CERT_PATH = "DCAE_CA_CERTPATH";
 
     /**
+     * Name of environment variable containing Config Binding Service network hostname.
+     */
+    String ENV_CBS_HOSTNAME = "CONFIG_BINDING_SERVICE";
+
+    /**
+     * Name of environment variable containing Config Binding Service network port.
+     */
+    String ENV_CBS_PORT = "CONFIG_BINDING_SERVICE_SERVICE_PORT";
+
+    /**
      * Name of environment variable containing current application name.
      */
     String ENV_APP_NAME = "HOSTNAME";
@@ -73,16 +82,61 @@ public interface CbsClientConfiguration {
      */
     String ENV_CBS_CLIENT_POLICY_PATH = "CBS_CLIENT_POLICY_PATH";
 
+    /**
+     * Name of environment variable containing Consul host name.
+     *
+     * @deprecated CBS lookup in Consul service should not be needed,
+     * instead {@link #ENV_CBS_HOSTNAME} should be used directly.
+     */
+    @Deprecated
+    String ENV_CONSUL_HOST = "CONSUL_HOST";
+
+    /**
+     * Name of environment variable containing Config Binding Service <em>service name</em> as registered in Consul
+     * services API.
+     *
+     * @deprecated CBS lookup in Consul service should not be needed,
+     * instead {@link #ENV_CBS_HOSTNAME} should be used directly.
+     */
+    @Deprecated
+    String ENV_CBS_NAME = "CONFIG_BINDING_SERVICE";
+
+    @Value.Parameter
+    @Nullable
+    String hostname();
+
+    @Value.Parameter
+    @Nullable
+    Integer port();
 
     @Value.Parameter
     String appName();
 
+    @Value.Parameter
+    @Nullable
+    String protocol();
+
     @Value.Default
     default @Nullable TrustStoreKeys trustStoreKeys() {
         return null;
     }
 
     @Value.Default
+    @Deprecated
+    default String consulHost() {
+        return "consul-server";
+    }
+    @Value.Default
+    @Deprecated
+    default Integer consulPort() {
+        return 8500;
+    }
+    @Value.Default
+    @Deprecated
+    default String cbsName() {
+        return "config-binding-service";
+    }
+    @Value.Default
     default String configMapFilePath() {
         return "/app-config/application_config.yaml";
     }
@@ -91,7 +145,6 @@ public interface CbsClientConfiguration {
         return "/etc/policies/policies.json";
     }
 
-
     /**
      * Creates CbsClientConfiguration from system environment variables.
      *
@@ -102,6 +155,7 @@ public interface CbsClientConfiguration {
         String pathToCaCert = System.getenv(DCAE_CA_CERT_PATH);
 
         ImmutableCbsClientConfiguration.Builder configBuilder = ImmutableCbsClientConfiguration.builder()
+                .hostname(getEnv(ENV_CBS_HOSTNAME))
                 .appName(getEnv(ENV_APP_NAME));
 
         Optional.ofNullable(System.getenv(ENV_CBS_CLIENT_CONFIG_PATH))
@@ -109,7 +163,41 @@ public interface CbsClientConfiguration {
 
         Optional.ofNullable(System.getenv(ENV_CBS_CLIENT_POLICY_PATH))
             .ifPresent(configBuilder::policySyncFilePath);
-        return configBuilder.build();
+
+        return Optional.ofNullable(pathToCaCert).filter(certPath -> !"".equals(certPath))
+                .map(certPath -> createSslHttpConfig(configBuilder, certPath))
+                .orElseGet(() -> createPlainHttpConfig(configBuilder));
+    }
+
+    static CbsClientConfiguration createPlainHttpConfig(ImmutableCbsClientConfiguration.Builder configBuilder) {
+        LOGGER.info("CBS client will use plain http protocol.");
+        return configBuilder
+                .protocol("http")
+                .port(Integer.valueOf(getEnv(ENV_CBS_PORT)))
+                .build();
+    }
+
+    static CbsClientConfiguration createSslHttpConfig(ImmutableCbsClientConfiguration.Builder configBuilder,
+                                                      String pathToCaCert) {
+        LOGGER.info("CBS client will use http over TLS.");
+        return configBuilder
+                .trustStoreKeys(crateSecurityKeysFromEnvironment(createPathToJksFile(pathToCaCert)))
+                .port(PORT_FOR_CBS_OVER_TLS)
+                .protocol("https")
+                .build();
+    }
+
+    static TrustStoreKeys crateSecurityKeysFromEnvironment(String pathToCerts) {
+        LOGGER.info("Path to cert files: {}", pathToCerts + "/");
+        validateIfFilesExist(pathToCerts);
+        return ImmutableTrustStoreKeys.builder()
+                .trustStore(SecurityKeysStore.fromPath(Paths.get(pathToCerts + "/" + TRUST_JKS)))
+                .trustStorePassword(Passwords.fromPath(Paths.get(pathToCerts + "/" + TRUST_PASS)))
+                .build();
+    }
+
+    static String createPathToJksFile(String pathToCaCertPemFile) {
+        return pathToCaCertPemFile.substring(0, pathToCaCertPemFile.lastIndexOf("/"));
     }
 
     static String getEnv(String envName) {
@@ -124,5 +212,12 @@ public interface CbsClientConfiguration {
         }
     }
 
+    static void validateIfFilesExist(String pathToFile) {
+        boolean areFilesExist = Files.exists(Paths.get(pathToFile + "/" + TRUST_JKS)) &&
+                Files.exists(Paths.get(pathToFile + "/" + TRUST_PASS));
 
+        if (!areFilesExist) {
+            throw new CbsClientConfigurationException("Required files do not exist in " + pathToFile + " directory.");
+        }
+    }
 }
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationClient.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationClient.java
new file mode 100644
index 00000000..e1dd7aff
--- /dev/null
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationClient.java
@@ -0,0 +1,114 @@
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2018 NOKIA 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.dcaegen2.services.sdk.rest.services.cbs.client.providers;
+
+import com.google.gson.JsonObject;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.ImmutableCbsClientConfiguration;
+import reactor.core.publisher.Mono;
+
+/**
+ * Complete CloudConfiguration HTTPClient API.
+ *
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 11/16/18
+ * @version 1.0.0
+ * @since 1.0.0
+ */
+public final class CloudConfigurationClient implements CloudConfigurationProvider {
+
+    private final CloudConfigurationProvider cloudConfigurationProvider;
+
+    /**
+     * Default constructor for CloudConfigurationClient, set CloudConfigurationProvider cloudConfigurationProvider
+     * property by calling: {@link ReactiveCloudConfigurationProvider}.
+     * Calls other constructor in this class {@link #CloudConfigurationClient(CloudConfigurationProvider)}.
+     */
+    public CloudConfigurationClient() {
+        this(new ReactiveCloudConfigurationProvider());
+    }
+
+    /**
+     * Constructor for CloudConfigurationClient, set loudConfigurationProvider cloudConfigurationProvider property
+     * by passing them in constructor {@link org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.CloudConfigurationProvider}
+     * implementation client.
+     *
+     * @param cloudConfigurationProvider - client provider for calling ConfigBindingService
+     */
+    public CloudConfigurationClient(
+        CloudConfigurationProvider cloudConfigurationProvider) {
+        this.cloudConfigurationProvider = cloudConfigurationProvider;
+    }
+
+    /**
+     * Documentation in {@link org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.CloudConfigurationProvider}
+     *
+     * @param consulHost - Hostname/IPAddress of consul Database
+     * @param consulPort - Port number of consul Database
+     * @param cbsName - ConfigBindingService url
+     * @param appName - ApplicationName for each config will be returned
+     */
+    @Override
+    public Mono<JsonObject> callForServiceConfigurationReactive(String consulHost, int consulPort, String cbsName,
+        String appName) {
+        return cloudConfigurationProvider.callForServiceConfigurationReactive(
+            ImmutableCbsClientConfiguration.builder().consulHost(consulHost)
+                .consulPort(consulPort).cbsName(cbsName)
+                .appName(appName).build());
+    }
+
+    /**
+     * Documentation in {@link org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.CloudConfigurationProvider}.
+     *
+     * @param cbsClientConfiguration - Object holds consulPort, consulURL, configBindingSeriveName, applicationName which have
+     * been defined in dcaegen2 cloud environment.
+     */
+    @Override
+    public Mono<JsonObject> callForServiceConfigurationReactive(CbsClientConfiguration cbsClientConfiguration) {
+        return cloudConfigurationProvider.callForServiceConfigurationReactive(cbsClientConfiguration);
+    }
+
+    /**
+     * Documentation in {@link org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.CloudConfigurationProvider}.
+     *
+     * @param consulHost - Hostname/IPAddress of consul Database
+     * @param consulPort - Port number of consul Database
+     * @param cbsName - ConfigBindingService url
+     * @param appName - ApplicationName for each config will be returned
+     */
+    @Override
+    public JsonObject callForServiceConfiguration(String consulHost, int consulPort, String cbsName, String appName) {
+        return cloudConfigurationProvider.callForServiceConfigurationReactive(
+                ImmutableCbsClientConfiguration.builder().consulHost(consulHost)
+                .consulPort(consulPort).cbsName(cbsName)
+                .appName(appName).build()).block();
+    }
+
+    /**
+     * Documentation in {@link org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.CloudConfigurationProvider}.
+     *
+     * @param cbsClientConfiguration - Object holds consulPort, consulURL, configBindingSeriveName, applicationName which have
+     */
+    @Override
+    public JsonObject callForServiceConfiguration(CbsClientConfiguration cbsClientConfiguration) {
+        return cloudConfigurationProvider.callForServiceConfigurationReactive(cbsClientConfiguration).block();
+    }
+}
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationProvider.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationProvider.java
index 37e59ca5..c0dbd5f9 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationProvider.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationProvider.java
@@ -1,9 +1,8 @@
 /*
-* ============LICENSE_START=======================================================
+ * ============LICENSE_START=======================================================
  * DCAEGEN2-SERVICES-SDK
  * ================================================================================
  * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
- * Copyright (C) 2022 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.
@@ -65,6 +64,27 @@ public interface CloudConfigurationProvider {
         String appName);
 
 
+    /*callForServiceConfiguration*/
 
+    /**
+     * Getting configuration for appName from ConfigBindingService.
+     *
+     * @param consulHost - Hostname/IPAddress of consul Database
+     * @param consulPort - Port number of consul Database
+     * @param cbsName - ConfigBindingService url
+     * @param appName - ApplicationName for each config will be returned
+     * @return configuration for specified application in dcaegen2 cloud infrastructure.
+     */
+    JsonObject callForServiceConfiguration(String consulHost, int consulPort, String cbsName, String appName);
+
+    /*callForServiceConfiguration*/
+
+    /**
+     * Getting configuration for appName from ConfigBindingService.
+     *
+     * @param cbsClientConfiguration - Object holds consulPort, consulURL, configBindingSeriveName, applicationName which have
+     * @return configuration for specified application in dcaegen2 cloud infrastructure.
+     */
+    JsonObject callForServiceConfiguration(CbsClientConfiguration cbsClientConfiguration);
 
 }
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProvider.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProvider.java
new file mode 100644
index 00000000..a50edd38
--- /dev/null
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProvider.java
@@ -0,0 +1,139 @@
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2018-2019 NOKIA 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.dcaegen2.services.sdk.rest.services.cbs.client.providers;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClientFactory;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.uri.URI;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 11/15/18
+ */
+public final class ReactiveCloudConfigurationProvider implements CloudConfigurationProvider {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ReactiveCloudConfigurationProvider.class);
+    private static final String EXCEPTION_MESSAGE = "Unsupported method call: ";
+
+    private final RxHttpClient rxHttpClient;
+
+    public ReactiveCloudConfigurationProvider() {
+        this(RxHttpClientFactory.create());
+    }
+
+    ReactiveCloudConfigurationProvider(RxHttpClient rxHttpClient) {
+        this.rxHttpClient = rxHttpClient;
+    }
+
+    @Override
+    public Mono<JsonObject> callForServiceConfigurationReactive(CbsClientConfiguration configuration) {
+        return callConsulForConfigBindingServiceEndpoint(configuration)
+            .flatMap(this::callConfigBindingServiceForConfiguration);
+    }
+
+    @Override
+    public Mono<JsonObject> callForServiceConfigurationReactive(String consulHost, int consulPort, String cbsName,
+        String appName) {
+        throw new UnsupportedOperationException(EXCEPTION_MESSAGE + this);
+    }
+
+    @Override
+    public JsonObject callForServiceConfiguration(String consulHost, int consulPort, String cbsName, String appName) {
+        throw new UnsupportedOperationException(EXCEPTION_MESSAGE + this);
+    }
+
+    @Override
+    public JsonObject callForServiceConfiguration(CbsClientConfiguration configuration) {
+        throw new UnsupportedOperationException(EXCEPTION_MESSAGE + this);
+    }
+
+    private Mono<String> callConsulForConfigBindingServiceEndpoint(CbsClientConfiguration configuration) {
+        LOGGER.info("Retrieving Config Binding Service endpoint from Consul");
+
+        HttpRequest httpRequest = ImmutableHttpRequest.builder()
+                .url(getConsulUrl(configuration)).method(HttpMethod.GET).build();
+
+        return rxHttpClient.call(httpRequest)
+                .map(resp -> resp.bodyAsJson(JsonArray.class))
+                .flatMap(jsonArray ->
+                        this.createConfigBindingServiceUrl(
+                                jsonArray,
+                                configuration.appName())
+                );
+    }
+
+    private String getConsulUrl(CbsClientConfiguration configuration) {
+        return getUri(configuration.consulHost(), configuration.consulPort(), "/v1/catalog/service",
+            configuration.cbsName());
+    }
+
+    private Mono<JsonObject> callConfigBindingServiceForConfiguration(String configBindingServiceUri) {
+        LOGGER.info("Retrieving configuration");
+        HttpRequest httpRequest = ImmutableHttpRequest.builder()
+                .url(configBindingServiceUri).method(HttpMethod.GET).build();
+
+        return rxHttpClient.call(httpRequest)
+                .map(httpResponse -> httpResponse.bodyAsJson(JsonObject.class));
+    }
+
+
+    private Mono<String> createConfigBindingServiceUrl(JsonArray jsonArray, String appName) {
+        return getConfigBindingObject(jsonArray)
+            .flatMap(jsonObject -> buildConfigBindingServiceUrl(jsonObject, appName));
+    }
+
+    private Mono<String> buildConfigBindingServiceUrl(JsonObject jsonObject, String appName) {
+        return Mono.just(getUri(jsonObject.get("ServiceAddress").getAsString(),
+            jsonObject.get("ServicePort").getAsInt(), "/service_component", appName));
+    }
+
+    private Mono<JsonObject> getConfigBindingObject(JsonArray jsonArray) {
+        try {
+            if (jsonArray.size() > 0) {
+                return Mono.just(jsonArray.get(0).getAsJsonObject());
+            } else {
+                throw new IllegalStateException("JSON Array was empty");
+            }
+        } catch (IllegalStateException e) {
+            LOGGER.warn("Failed to retrieve JSON Object from array", e);
+            return Mono.error(e);
+        }
+    }
+
+    private String getUri(String host, Integer port, String... paths) {
+        return new URI.URIBuilder()
+            .scheme("http")
+            .host(host)
+            .port(port)
+            .path(String.join("/", paths))
+            .build().toString();
+    }
+
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientConfigurationTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientConfigurationTest.java
index 7a683dac..2cc221ef 100644
--- a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientConfigurationTest.java
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientConfigurationTest.java
@@ -3,7 +3,6 @@
  * DCAEGEN2-SERVICES-SDK
  * =========================================================
  * Copyright (C) 2019-2021 Nokia. All rights reserved.
- * Copyright (C) 2022 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.
@@ -42,7 +41,11 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
  */
 class CbsClientConfigurationTest {
 
-    private static final String ENV_APPNAME = "HOSTNAME";
+    private static final String ENV_DCAE_CA_CERTPATH = "DCAE_CA_CERTPATH";
+    private static final String ENV_CONFIG_BINDING_SERVICE = "CONFIG_BINDING_SERVICE";
+    private static final String ENV_CONFIG_BINDING_SERVICE_SERVICE_PORT = "CONFIG_BINDING_SERVICE_SERVICE_PORT";
+    private static final String ENV_HOSTNAME = "HOSTNAME";
+    private static final String ENV_CONSUL_HOST = "CONSUL_HOST";
     private static final String ENV_CBS_CLIENT_CONFIG_PATH = "CBS_CLIENT_CONFIG_PATH";
     private static final String ENV_CBS_CLIENT_POLICY_PATH = "CBS_CLIENT_POLICY_PATH";
 
@@ -51,7 +54,38 @@ class CbsClientConfigurationTest {
 
     @BeforeEach
     void setUp(){
-        envs.clear(ENV_APPNAME, ENV_CBS_CLIENT_CONFIG_PATH, ENV_CBS_CLIENT_POLICY_PATH);
+        envs.clear(ENV_DCAE_CA_CERTPATH, ENV_CONFIG_BINDING_SERVICE, ENV_CONFIG_BINDING_SERVICE_SERVICE_PORT,
+            ENV_HOSTNAME, ENV_CONSUL_HOST, ENV_CBS_CLIENT_CONFIG_PATH, ENV_CBS_CLIENT_POLICY_PATH);
+    }
+
+    @Test
+    void fromEnvironment_shouldReturnConfigurationForConnectionWithoutTls_when_DCAE_CA_CERTPATH_isEmpty() {
+        // given
+        createBasicValidEnvsConfiguration();
+        envs.set(ENV_DCAE_CA_CERTPATH, "");
+
+        // when
+        CbsClientConfiguration configuration = CbsClientConfiguration.fromEnvironment();
+
+        // then
+        assertThat(configuration.trustStoreKeys()).isEqualTo(null);
+        assertThat(configuration.protocol()).isEqualTo("http");
+    }
+
+    @Test
+    void fromEnvironment_shouldReturnConfigurationForConnectionOverTls_when_DCAE_CA_CERTPATH_isSet() throws URISyntaxException {
+        // given
+        envs.set(ENV_DCAE_CA_CERTPATH, preparePathToCertFile());
+        envs.set(ENV_CONFIG_BINDING_SERVICE, "config-binding-service");
+        envs.set(ENV_HOSTNAME, "dcae-prh");
+        envs.set(ENV_CONSUL_HOST, "consul-server.onap");
+
+        // when
+        CbsClientConfiguration configuration = CbsClientConfiguration.fromEnvironment();
+
+        // then
+        assertThat(configuration.trustStoreKeys()).isNotNull();
+        assertThat(configuration.protocol()).isEqualTo("https");
     }
 
     @Test
@@ -96,6 +130,22 @@ class CbsClientConfigurationTest {
         assertThat(configuration.policySyncFilePath()).isEqualTo("/etc/policies/policies.json");
     }
 
+    @Test
+    void fromEnvironment_shouldReturn_CbsClientConfigurationException_When_DCAE_CA_CERTPATH_is_Null() {
+        // given
+        envs.set(ENV_DCAE_CA_CERTPATH, null);
+        envs.set(ENV_CONFIG_BINDING_SERVICE_SERVICE_PORT, "9090");
+        envs.set(ENV_CONFIG_BINDING_SERVICE, "config-binding-service");
+        envs.set(ENV_HOSTNAME, "dcae-prh");
+        envs.set(ENV_CONSUL_HOST, "consul-server.onap");
+
+        // when
+        CbsClientConfiguration configuration = CbsClientConfiguration.fromEnvironment();
+
+        // then
+        assertThat(configuration.trustStoreKeys()).isNull();
+        assertThat(configuration.protocol()).isEqualTo("http");
+    }
 
     @Test
     void fromEnvironment_shouldReturn_CbsClientConfigurationException_WhenAllEnvVariablesAreMissing() {
@@ -103,7 +153,57 @@ class CbsClientConfigurationTest {
                 .isThrownBy(CbsClientConfiguration::fromEnvironment);
     }
 
+    @Test
+    void fromEnvironment_shouldReturn_CbsClientConfigurationException_When_DCAE_CA_CERTPATH_isWrong() {
+        // given
+        envs.set(ENV_DCAE_CA_CERTPATH, "/home/cacert.pem");
+        envs.set(ENV_HOSTNAME, "dcae-prh");
+        envs.set(ENV_CONFIG_BINDING_SERVICE, "config-binding-service");
+        envs.set(ENV_CONSUL_HOST, "consul-server.onap");
+
+        // then
+        assertThatExceptionOfType(CbsClientConfigurationException.class)
+                .isThrownBy(CbsClientConfiguration::fromEnvironment)
+                .withMessageContaining("Required files do not exist in /home directory");
+    }
+
+    @Test
+    void fromEnvironment_shouldReturn_CbsClientConfigurationException_When_HOSTNAME_isMissing() throws URISyntaxException {
+        // given
+        envs.set(ENV_HOSTNAME, "");
+        envs.set(ENV_DCAE_CA_CERTPATH, preparePathToCertFile());
+        envs.set(ENV_CONFIG_BINDING_SERVICE, "config-binding-service");
+        envs.set(ENV_CONSUL_HOST, "consul-server.onap");
+
+        // then
+        assertThatExceptionOfType(CbsClientConfigurationException.class)
+                .isThrownBy(CbsClientConfiguration::fromEnvironment)
+                .withMessageContaining("Cannot read HOSTNAME from environment.");
+    }
+
+    @Test
+    void fromEnvironment_shouldReturn_CbsClientConfigurationException_When_CONFIG_BINDING_SERVICE_SERVICE_PORT_isEmpty() {
+        // given
+        envs.set(ENV_CONFIG_BINDING_SERVICE_SERVICE_PORT, "");
+        envs.set(ENV_DCAE_CA_CERTPATH, "");
+        envs.set(ENV_HOSTNAME, "dcae-prh");
+        envs.set(ENV_CONFIG_BINDING_SERVICE, "config-binding-service");
+        envs.set(ENV_CONSUL_HOST, "consul-server.onap");
+
+        // then
+        assertThatExceptionOfType(CbsClientConfigurationException.class)
+                .isThrownBy(CbsClientConfiguration::fromEnvironment)
+                .withMessageContaining("Cannot read CONFIG_BINDING_SERVICE_SERVICE_PORT from environment.");
+    }
+
     private void createBasicValidEnvsConfiguration() {
-        envs.set(ENV_APPNAME, "dcae-prh");
+        envs.set(ENV_CONFIG_BINDING_SERVICE, "config-binding-service");
+        envs.set(ENV_CONFIG_BINDING_SERVICE_SERVICE_PORT, "10000");
+        envs.set(ENV_HOSTNAME, "dcae-prh");
+        envs.set(ENV_CONSUL_HOST, "consul-server.onap");
+    }
+
+    private String preparePathToCertFile() throws URISyntaxException {
+        return Paths.get(Passwords.class.getResource("/test-certs/cacert.pem").toURI()) + "";
     }
-}
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactoryTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactoryTest.java
index bc4e675e..43577f4a 100644
--- a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactoryTest.java
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactoryTest.java
@@ -3,7 +3,6 @@
  * DCAEGEN2-SERVICES-SDK
  * =========================================================
  * Copyright (C) 2020 Nokia. All rights reserved.
- * Copyright (C) 2022 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.
@@ -40,7 +39,15 @@ class CbsClientFactoryTest {
     void shouldAllowMultipleSubscriptions() throws URISyntaxException {
         //given
         ImmutableCbsClientConfiguration sampleConfiguration = ImmutableCbsClientConfiguration.builder()
+            .protocol("https")
             .appName("dcae-component")
+            .trustStoreKeys(ImmutableTrustStoreKeys.builder()
+                .trustStore(SecurityKeysStore.fromPath(
+                    Paths.get(CbsClientFactoryTest.class.getResource("/test-certs/trust.jks").toURI())))
+                .trustStorePassword(Passwords.fromResource("/test-certs/trust.pass"))
+                .build())
+            .hostname("config-binding-service")
+            .port(10443)
             .build();
 
         //when
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsRequestsTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsRequestsTest.java
new file mode 100644
index 00000000..d2229a52
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsRequestsTest.java
@@ -0,0 +1,73 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. 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.dcaegen2.services.sdk.rest.services.cbs.client.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since March 2019
+ */
+class CbsRequestsTest {
+
+    private final RequestDiagnosticContext diagCtx = RequestDiagnosticContext.create();
+    private final String serviceName = "srv-name";
+
+    @Test
+    void getConfiguration() {
+        // given
+        final CbsRequest cut = CbsRequests.getConfiguration(diagCtx);
+
+        // when
+        final String result = cut.requestPath().getForService(serviceName);
+
+        // then
+        assertThat(result).isEqualTo("/service_component/srv-name");
+    }
+
+    @Test
+    void getByKey() {
+        // given
+        final CbsRequest cut = CbsRequests.getByKey(diagCtx, "configKey");
+
+        // when
+        final String result = cut.requestPath().getForService(serviceName);
+
+        // then
+        assertThat(result).isEqualTo("/configKey/srv-name");
+    }
+
+    @Test
+    void getAll() {
+        // given
+        final CbsRequest cut = CbsRequests.getAll(diagCtx);
+
+        // when
+        final String result = cut.requestPath().getForService(serviceName);
+
+        // then
+        assertThat(result).isEqualTo("/service_component_all/srv-name");
+    }
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java
index 57bf9b3d..db881a2e 100644
--- a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java
@@ -3,7 +3,6 @@
  * DCAEGEN2-SERVICES-SDK
  * =========================================================
  * Copyright (C) 2019-2021 Nokia. All rights reserved.
- * Copyright (C) 2022 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.
@@ -96,6 +95,64 @@ class CbsClientImplIT {
         server.close();
     }
 
+    @Test
+    void testCbsClientWithSingleCall() {
+        // given
+        envs.set("AAF_USER", "admin");
+        envs.set("AAF_PASSWORD", "admin_secret");
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleConfigurationCbsSource);
+        final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
+
+        // when
+        final Mono<JsonObject> result = sut.flatMap(cbsClient -> cbsClient.get(request));
+
+        // then
+        StepVerifier.create(result.map(this::sampleConfigValue))
+                .expectNext(EXPECTED_CONFIG_VALUE_FROM_CBS)
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
+
+    @Test
+    void testCbsClientWithPeriodicCall() {
+        // given
+        envs.set("AAF_USER", "admin");
+        envs.set("AAF_PASSWORD", "admin_secret");
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleConfigurationCbsSource);
+        final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
+
+        // when
+        final Flux<JsonObject> result = sut
+                .flatMapMany(cbsClient -> cbsClient.get(request, Duration.ZERO, Duration.ofMillis(10)));
+
+        // then
+        final int itemsToTake = 5;
+        StepVerifier.create(result.take(itemsToTake).map(this::sampleConfigValue))
+                .expectNextSequence(Stream.of(EXPECTED_CONFIG_VALUE_FROM_CBS).cycle(itemsToTake))
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
+
+    @Test
+    void testCbsClientWithUpdatesCall() {
+        // given
+        envs.set("AAF_USER", "admin");
+        envs.set("AAF_PASSWORD", "admin_secret");
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleConfigurationCbsSource);
+        final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
+        final Duration period = Duration.ofMillis(10);
+
+        // when
+        final Flux<JsonObject> result = sut
+                .flatMapMany(cbsClient -> cbsClient.updates(request, Duration.ZERO, period));
+
+        // then
+        final Duration timeToCollectItemsFor = period.multipliedBy(50);
+        StepVerifier.create(result.take(timeToCollectItemsFor).map(this::sampleConfigValue))
+                .expectNext(EXPECTED_CONFIG_VALUE_FROM_CBS)
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
 
     @Test
     void testCbsClientWithConfigRetrievedFromFileMissingEnv() {
@@ -134,11 +191,159 @@ class CbsClientImplIT {
                 .verify(Duration.ofSeconds(5));
     }
 
+    @Test
+    void testCbsClientWithStreamsParsing() {
+        // given
+        envs.set("AAF_USER", "admin");
+        envs.set("AAF_PASSWORD", "admin_secret");
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleConfigurationCbsSource);
+        final StreamFromGsonParser<KafkaSink> kafkaSinkParser = StreamFromGsonParsers.kafkaSinkParser();
+        final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
+
+        // when
+        final Mono<KafkaSink> result = sut.flatMap(cbsClient -> cbsClient.get(request))
+                .map(json ->
+                        DataStreams.namedSinks(json).map(kafkaSinkParser::unsafeParse).head()
+                );
+
+        // then
+        StepVerifier.create(result)
+                .consumeNextWith(kafkaSink -> {
+                    assertThat(kafkaSink.name()).isEqualTo("perf3gpp");
+                    assertThat(kafkaSink.bootstrapServers()).isEqualTo("dmaap-mr-kafka:6060");
+                    assertThat(kafkaSink.topicName()).isEqualTo("HVVES_PERF3GPP");
+                })
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
+
+    @Test
+    void testCbsClientWithStreamsParsingUsingSwitch() {
+        // given
+        envs.set("AAF_USER", "admin");
+        envs.set("AAF_PASSWORD", "admin_secret");
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleConfigurationCbsSource);
+        final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
+        // TODO: Use these parsers below
+        final StreamFromGsonParser<KafkaSink> kafkaSinkParser = StreamFromGsonParsers.kafkaSinkParser();
+        final StreamFromGsonParser<MessageRouterSink> mrSinkParser = StreamFromGsonParsers.messageRouterSinkParser();
+
+        // when
+        final Mono<Void> result = sut.flatMap(cbsClient -> cbsClient.get(request))
+                .map(json -> {
+                    final Stream<RawDataStream<JsonObject>> sinks = DataStreams.namedSinks(json);
+
+                    final Stream<KafkaSink> allKafkaSinks = sinks.filter(streamOfType(KAFKA))
+                            .map(kafkaSinkParser::unsafeParse);
+                    final Stream<MessageRouterSink> allMrSinks = sinks.filter(streamOfType(MESSAGE_ROUTER))
+                            .map(mrSinkParser::unsafeParse);
+
+                    assertThat(allKafkaSinks.size())
+                            .describedAs("Number of kafka sinks")
+                            .isEqualTo(2);
+                    assertThat(allMrSinks.size())
+                            .describedAs("Number of DMAAP-MR sinks")
+                            .isEqualTo(1);
+
+                    return true;
+                })
+                .then();
+
+        // then
+        StepVerifier.create(result)
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
+
+    @Test
+    void testCbsClientWithStreamsParsingWhenUsingInvalidParser() {
+        // given
+        envs.set("AAF_USER", "admin");
+        envs.set("AAF_PASSWORD", "admin_secret");
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleConfigurationCbsSource);
+        final StreamFromGsonParser<KafkaSource> kafkaSourceParser = StreamFromGsonParsers.kafkaSourceParser();
+        final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
+
+        // when
+        final Mono<KafkaSource> result = sut.flatMap(cbsClient -> cbsClient.get(request))
+                .map(json ->
+                        DataStreams.namedSources(json).map(kafkaSourceParser::unsafeParse).head()
+                );
+
+        // then
+        StepVerifier.create(result)
+                .expectErrorSatisfies(ex -> {
+                    assertThat(ex).isInstanceOf(StreamParsingException.class);
+                    assertThat(ex).hasMessageContaining("Invalid stream type");
+                    assertThat(ex).hasMessageContaining(MESSAGE_ROUTER.toString());
+                    assertThat(ex).hasMessageContaining(KAFKA.toString());
+                })
+                .verify(Duration.ofSeconds(5));
+    }
+
+    @Test
+    void testCbsClientWithSingleAllRequest() {
+        // given
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleConfigurationCbsSource);
+        final CbsRequest request = CbsRequests.getAll(RequestDiagnosticContext.create());
+
+        // when
+        final Mono<JsonObject> result = sut.flatMap(cbsClient -> cbsClient.get(request));
+
+        // then
+        StepVerifier.create(result)
+                .assertNext(json -> {
+                    assertThat(json.get("config")).isNotNull();
+                    assertThat(json.get("policies")).isNotNull();
+                    assertThat(json.get("sampleKey")).isNotNull();
+                })
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
+
+
+    @Test
+    void testCbsClientWithSingleKeyRequest() {
+        // given
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleConfigurationCbsSource);
+        final CbsRequest request = CbsRequests.getByKey(RequestDiagnosticContext.create(), "sampleKey");
+
+        // when
+        final Mono<JsonObject> result = sut.flatMap(cbsClient -> cbsClient.get(request));
+
+        // then
+        StepVerifier.create(result)
+                .assertNext(json -> {
+                    assertThat(json.get("key")).isNotNull();
+                    assertThat(json.get("key").getAsString()).isEqualTo("value");
+                })
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
+
+    @Test
+    void testCbsClientWhenTheConfigurationWasNotFound() {
+        // given
+        final CbsClientConfiguration unknownAppEnv = ImmutableCbsClientConfiguration.copyOf(sampleConfigurationCbsSource).withAppName("unknown_app");
+        final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(unknownAppEnv);
+        final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create());
+
+        // when
+        final Mono<JsonObject> result = sut.flatMap(cbsClient -> cbsClient.get(request));
+
+        // then
+        StepVerifier.create(result)
+                .expectError(HttpException.class)
+                .verify(Duration.ofSeconds(5));
+    }
 
     @NotNull
     private static ImmutableCbsClientConfiguration.Builder getConfigBuilder() {
         return ImmutableCbsClientConfiguration.builder()
-                .appName("dcae-component");
+                .protocol("http")
+                .appName("dcae-component")
+                .hostname(server.host())
+                .port(server.port());
     }
 
     private String sampleConfigValue(JsonObject obj) {
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientRestTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientRestTest.java
new file mode 100644
index 00000000..6368fbac
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientRestTest.java
@@ -0,0 +1,77 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019-2021 Nokia. 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.dcaegen2.services.sdk.rest.services.cbs.client.impl;
+
+import com.google.gson.JsonObject;
+import io.vavr.collection.HashMultimap;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.ImmutableHttpResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
+import reactor.core.publisher.Mono;
+import java.net.InetSocketAddress;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since February 2019
+ */
+class CbsClientRestTest {
+    private final RxHttpClient httpClient = mock(RxHttpClient.class);
+
+    @Test
+    void shouldFetchUsingProperUrl() {
+        // given
+        InetSocketAddress cbsAddress = InetSocketAddress.createUnresolved("cbshost", 6969);
+        String serviceName = "dcaegen2-ves-collector";
+        final CbsClient cut = new CbsClientRest(httpClient, serviceName, cbsAddress, "http");
+        final HttpResponse httpResponse = ImmutableHttpResponse.builder()
+                .url("http://xxx")
+                .statusCode(200)
+                .rawBody("{}".getBytes())
+                .headers(HashMultimap.withSeq().empty())
+                .build();
+        given(httpClient.call(any(HttpRequest.class))).willReturn(Mono.just(httpResponse));
+        RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
+
+        // when
+        final JsonObject result = cut.get(CbsRequests.getConfiguration(diagnosticContext)).block();
+
+        // then
+        final String expectedUrl = "http://cbshost:6969/service_component/dcaegen2-ves-collector";
+        verify(httpClient).call(ImmutableHttpRequest.builder()
+                .method(HttpMethod.GET)
+                .url(expectedUrl)
+                .diagnosticContext(diagnosticContext)
+                .build());
+        assertThat(result.toString()).isEqualTo(httpResponse.bodyAsString());
+    }
+}
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookupTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookupTest.java
new file mode 100644
index 00000000..70f31c8b
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookupTest.java
@@ -0,0 +1,54 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. 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.dcaegen2.services.sdk.rest.services.cbs.client.impl;
+
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.ImmutableCbsClientConfiguration;
+
+import java.net.InetSocketAddress;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since February 2019
+ */
+class CbsLookupTest {
+
+    private static final String cbsHostname = "cbs-service";
+    private static final int cbsPort = 10000;
+    private final CbsClientConfiguration configuration = ImmutableCbsClientConfiguration.builder()
+            .hostname(cbsHostname)
+            .port(cbsPort)
+            .appName("whatever").build();
+    private final CbsLookup cut = new CbsLookup();
+
+    @Test
+    void lookupShouldReturnValidSocketAddressFromEnvironment() {
+        // when
+        final InetSocketAddress result = cut.lookup(configuration).block();
+
+        // then
+        assertThat(result.getHostString()).isEqualTo(cbsHostname);
+        assertThat(result.getPort()).isEqualTo(cbsPort);
+    }
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationClientTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationClientTest.java
new file mode 100644
index 00000000..389e6e8a
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/CloudConfigurationClientTest.java
@@ -0,0 +1,78 @@
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2020 Nokia 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.dcaegen2.services.sdk.rest.services.cbs.client.providers;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+class CloudConfigurationClientTest {
+    private static final String CONFIGURATION_MOCK = "{\"test\":1}";
+    private static final JsonObject CONFIGURATION_JSON_MOCK = new Gson()
+        .fromJson(CONFIGURATION_MOCK, JsonObject.class);
+
+    private final CloudConfigurationProvider provider = mock(CloudConfigurationProvider.class);
+    private final CbsClientConfiguration configuration = mock(CbsClientConfiguration.class);
+
+    private CloudConfigurationClient client;
+
+    @BeforeEach
+    void setUp() {
+        client = new CloudConfigurationClient(provider);
+        when(provider.callForServiceConfigurationReactive(any(CbsClientConfiguration.class)))
+            .thenReturn(Mono.just(CONFIGURATION_JSON_MOCK));
+    }
+
+    @Test
+    void callForServiceConfigurationReactiveWithManyParamsShouldReturnConfigurationObjectMono() {
+        StepVerifier.create(client.callForServiceConfigurationReactive("hostName", 4444, "cbsName1", "appName1"))
+            .expectSubscription()
+            .expectNext(CONFIGURATION_JSON_MOCK).verifyComplete();
+    }
+
+    @Test
+    void callForServiceConfigurationReactiveWithOneParamShouldReturnConfigurationObjectMono() {
+        StepVerifier.create(client.callForServiceConfigurationReactive(configuration))
+            .expectSubscription()
+            .expectNext(CONFIGURATION_JSON_MOCK).verifyComplete();
+    }
+
+    @Test
+    void callForServiceConfigurationWithManyParamsShouldReturnConfigurationObject() {
+        JsonObject json = client.callForServiceConfiguration("hostName", 4444, "cbsName1", "appName1");
+        assertEquals(CONFIGURATION_JSON_MOCK, json);
+    }
+
+    @Test
+    void callForServiceConfigurationWithOneParamShouldReturnConfigurationObject() {
+        JsonObject json = client.callForServiceConfiguration(configuration);
+        assertEquals(CONFIGURATION_JSON_MOCK, json);
+    }
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProviderTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProviderTest.java
new file mode 100644
index 00000000..de0870d0
--- /dev/null
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProviderTest.java
@@ -0,0 +1,135 @@
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2018-2019 NOKIA 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.dcaegen2.services.sdk.rest.services.cbs.client.providers;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import java.io.IOException;
+import java.util.List;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClient;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.GsonUtils;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.ImmutableCbsClientConfiguration;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+/**
+ * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 11/15/18
+ */
+class ReactiveCloudConfigurationProviderTest {
+
+    private static final Gson gson = new Gson();
+    private static final String CONFIGURATION_MOCK = "{\"test\":1}";
+    private static final JsonObject CONFIGURATION_JSON_MOCK = gson
+            .fromJson(CONFIGURATION_MOCK, JsonObject.class);
+
+    private final RxHttpClient httpClient = mock(RxHttpClient.class);
+    private final JsonArray configBindingService = GsonUtils.readObjectArrayFromResource("/sample_config_binding_service.json");
+
+    private CbsClientConfiguration cbsClientConfiguration = ImmutableCbsClientConfiguration.builder()
+            .appName("dcae-prh")
+            .cbsName("config-binding-service")
+            .consulHost("consul")
+            .consulPort(8500)
+            .build();
+
+    private HttpResponse response;
+    private ReactiveCloudConfigurationProvider provider;
+
+    ReactiveCloudConfigurationProviderTest() throws IOException {
+    }
+
+
+    @BeforeEach
+    void setUp() {
+        response = mock(HttpResponse.class);
+        provider = new ReactiveCloudConfigurationProvider(httpClient);
+    }
+
+    @Test
+    void shouldReturnPrhConfiguration(){
+        //when
+        when(httpClient.call(any(HttpRequest.class))).thenReturn(Mono.just(response));
+        when(response.bodyAsJson(JsonArray.class)).thenReturn(configBindingService);
+        when(response.bodyAsJson(JsonObject.class)).thenReturn(CONFIGURATION_JSON_MOCK);
+
+
+        //then
+        StepVerifier.create(provider.callForServiceConfigurationReactive(cbsClientConfiguration))
+                .expectSubscription()
+                .expectNext(CONFIGURATION_JSON_MOCK).verifyComplete();
+    }
+
+    @Test
+    void shouldRequestCorrectUrl(){
+        // given
+        String consulRequestUrl = "http://consul:8500/v1/catalog/service/config-binding-service";
+        String configRequestUrl = "http://config-binding-service:10000/service_component/dcae-prh";
+
+        //when
+        when(httpClient.call(any(HttpRequest.class))).thenReturn(Mono.just(response));
+        when(response.bodyAsJson(JsonArray.class)).thenReturn(configBindingService);
+        when(response.bodyAsJson(JsonObject.class)).thenReturn(CONFIGURATION_JSON_MOCK);
+
+
+        //then
+        StepVerifier.create(provider.callForServiceConfigurationReactive(cbsClientConfiguration))
+                .expectSubscription()
+                .expectNext(CONFIGURATION_JSON_MOCK).verifyComplete();
+
+
+        ArgumentCaptor<HttpRequest> httpReq = ArgumentCaptor
+                .forClass(HttpRequest.class);
+        verify(httpClient, times(2)).call(httpReq.capture());
+
+        List<HttpRequest> allRequests = httpReq.getAllValues();
+        assertThat(allRequests.get(0).url()).isEqualTo(consulRequestUrl);
+        assertThat(allRequests.get(1).url()).isEqualTo(configRequestUrl);
+    }
+
+    @Test
+    void shouldReturnMonoErrorWhenConsuleDoesntHaveConfigBindingServiceEntry() {
+        // given
+        JsonArray emptyArray = gson.fromJson("[]", JsonArray.class);
+
+        //when
+        when(httpClient.call(any(HttpRequest.class))).thenReturn(Mono.just(response));
+        when(response.bodyAsJson(JsonArray.class)).thenReturn(emptyArray);
+
+
+        //then
+        StepVerifier.create(provider.callForServiceConfigurationReactive(cbsClientConfiguration))
+                .expectSubscription()
+                .expectError(IllegalStateException.class).verify();
+    }
+}
\ No newline at end of file
diff --git a/version.properties b/version.properties
index a1653f6b..51dd6a7c 100644
--- a/version.properties
+++ b/version.properties
@@ -1,6 +1,6 @@
 major=1
-minor=9
-patch=0
+minor=8
+patch=10
 base_version=${major}.${minor}.${patch}
 release_version=${base_version}
 snapshot_version=${base_version}-SNAPSHOT