Switch configuration provider to CBS Client - DCAE SDK

Issue-ID: DCAEGEN2-2964
Signed-off-by: Tomasz Wrobel <tomasz.wrobel@nokia.com>
Change-Id: Iaaf170d209b77b9709cc202c131f9d3bdf1033ed

11 files changed, 229 insertions, 283 deletions

diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java
index 9cfddf0..0da2b7e 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java
@@ -1,7 +1,7 @@
 /*-
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2019-2020 Nordix Foundation.
- *  Copyright (C) 2021 Nokia.
+ *  Copyright (C) 2021-2022 Nokia.
  *  Copyright (C) 2021 Samsung Electronics.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -35,7 +35,6 @@ import org.onap.dcaegen2.services.pmmapper.config.DynamicConfiguration;
 import org.onap.dcaegen2.services.pmmapper.config.EnvironmentReader;
 import org.onap.dcaegen2.services.pmmapper.config.FilesProcessingConfig;
 import org.onap.dcaegen2.services.pmmapper.datarouter.DeliveryHandler;
-import org.onap.dcaegen2.services.pmmapper.exceptions.CBSServerError;
 import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.MapperConfigException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.ProcessEventException;
@@ -53,6 +52,12 @@ import org.onap.dcaegen2.services.pmmapper.utils.IncomingEventsCache;
 import org.onap.dcaegen2.services.pmmapper.utils.MeasConverter;
 import org.onap.dcaegen2.services.pmmapper.utils.MeasSplitter;
 import org.onap.dcaegen2.services.pmmapper.utils.XMLValidator;
+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.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import org.onap.logging.ref.slf4j.ONAPLogAdapter;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
@@ -119,8 +124,8 @@ public class App {
     public App(Path templatesDirectory, Path schemasDirectory, int httpPort, int httpsPort, ConfigHandler configHandler, FilesProcessingConfig filesProcessingConfig)
         throws EnvironmentConfigException {
         try {
-            this.mapperConfig = configHandler.getMapperConfig();
-        } catch (EnvironmentConfigException | CBSServerError | MapperConfigException e) {
+            this.mapperConfig = configHandler.getInitialConfiguration();
+        } catch (MapperConfigException e) {
             logger.unwrap().error("Failed to acquire initial configuration, Application cannot start", e);
             throw new IllegalStateException("Config acquisition failed");
         }
@@ -160,7 +165,7 @@ public class App {
         this.configScheduler.schedulePeriodically(this::reconfigure, INITIAL_RECONFIGURATION_PERIOD, RECONFIGURATION_PERIOD, TimeUnit.SECONDS);
         this.healthCheckHandler = new HealthCheckHandler();
         this.deliveryHandler = new DeliveryHandler(fluxSink::next);
-        this.dynamicConfiguration = new DynamicConfiguration(Arrays.asList(mapperConfig), mapperConfig);
+        this.dynamicConfiguration = new DynamicConfiguration(Arrays.asList(mapperConfig), mapperConfig, configHandler);
         this.serverResources = Arrays.asList(healthCheckHandler, deliveryHandler, dynamicConfiguration);
         try {
             this.applicationServer = server(this.mapperConfig, this.serverResources);
@@ -219,7 +224,12 @@ public class App {
 
     public static void main(String[] args) throws EnvironmentConfigException {
         FilesProcessingConfig processingConfig = new FilesProcessingConfig(new EnvironmentReader());
-        new App(templates, schemas, HTTP_PORT, HTTPS_PORT, new ConfigHandler(), processingConfig).start();
+        final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
+        final CbsRequest request = CbsRequests.getConfiguration(diagnosticContext);
+        final CbsClientConfiguration config = CbsClientConfiguration.fromEnvironment();
+        CbsClient cbsClient = CbsClientFactory.createCbsClient(config).block();
+
+        new App(templates, schemas, HTTP_PORT, HTTPS_PORT, new ConfigHandler(cbsClient, request), processingConfig).start();
     }
 
     public static boolean filterByFileType(MeasFilterHandler filterHandler,Event event, MapperConfig config) {
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandler.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandler.java
index ae185f4..6c93d2f 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandler.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandler.java
@@ -1,6 +1,7 @@
 /*-
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2019 Nordix Foundation.
+ *  Copyright (C) 2022 Nokia.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,19 +20,19 @@
  */
 package org.onap.dcaegen2.services.pmmapper.config;
 
-import com.google.gson.Gson;
 import com.google.gson.JsonObject;
-import org.onap.dcaegen2.services.pmmapper.exceptions.CBSServerError;
 import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.MapperConfigException;
-import org.onap.dcaegen2.services.pmmapper.utils.EnvironmentConfig;
 import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
-import org.onap.dcaegen2.services.pmmapper.utils.RequestSender;
 import org.onap.dcaegen2.services.pmmapper.utils.RequiredFieldDeserializer;
 
+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.onap.logging.ref.slf4j.ONAPLogAdapter;
 import org.slf4j.LoggerFactory;
 import com.google.gson.GsonBuilder;
+import reactor.core.publisher.Mono;
 
 /**
  * Handles the retrieval of the component spec-based PM-Mapper Configuration
@@ -39,24 +40,22 @@ import com.google.gson.GsonBuilder;
  */
 
 public class ConfigHandler {
+
     private static final ONAPLogAdapter logger = new ONAPLogAdapter(LoggerFactory.getLogger(ConfigHandler.class));
-    private RequestSender sender;
-    private EnvironmentConfig environmentConfig;
 
-    /**
-     * Creates a ConfigHandler.
-     */
-    public ConfigHandler() {
-        this(new RequestSender(), new EnvironmentConfig());
-    }
+    private final CbsClient cbsClient;
+    private final CbsRequest cbsRequest;
+
+    private MapperConfig mapperConfig;
 
     /**
-     * @see ConfigHandler#ConfigHandler()
-     * @param sender A RequestSender
+     * Creates a ConfigHandler based on Cbs Client and Cbs Request provided by DCAE SDK
+     * @param CbsClient A Cbs Client
+     * @param CbsRequest A Cbs Request
      */
-    public ConfigHandler(RequestSender sender, EnvironmentConfig environmentConfig) {
-        this.sender = sender;
-        this.environmentConfig = environmentConfig;
+    public ConfigHandler(CbsClient cbsClient, CbsRequest cbsRequest){
+        this.cbsClient = cbsClient;
+        this.cbsRequest = cbsRequest;
     }
 
     /**
@@ -65,32 +64,53 @@ public class ConfigHandler {
      * @throws EnvironmentConfigException
      */
     public MapperConfig getMapperConfig() throws EnvironmentConfigException {
-        String mapperConfigJson = "";
-        String cbsSocketAddress = this.environmentConfig.getCBSHostName() + ":" + this.environmentConfig.getCBSPort();
-        String requestURL = "http://" + cbsSocketAddress + "/service_component/" + this.environmentConfig.getServiceName();
-        try {
-            logger.unwrap().info("Fetching pm-mapper configuration from Configbinding Service");
-            mapperConfigJson = sender.send(requestURL);
-        } catch (Exception exception) {
-            throw new CBSServerError("Error connecting to Configbinding Service: ", exception);
+
+        Mono.just(cbsClient)
+            .flatMap(client -> client.get(cbsRequest))
+            .subscribe(
+                this::handleConfigurationFromConsul,
+                this::handleError
+            );
+
+        if (mapperConfig == null) {
+            logger.unwrap().error("Mapper configuration is not initialized");
+            throw new EnvironmentConfigException("Mapper configuration is not initialized");
         }
-        return convertMapperConfigToObject(mapperConfigJson);
+        return mapperConfig;
     }
 
-    private MapperConfig convertMapperConfigToObject(String mapperConfigJson) {
-        MapperConfig mapperConfig;
+    /**
+     * Retrieves Initial PM-Mapper Configuration from DCAE's ConfigBinding Service.
+     *
+     * @throws MapperConfigException
+     */
+    public MapperConfig getInitialConfiguration() {
+        logger.unwrap().info("Attempt to get initial configuration");
+            JsonObject jsonObject = Mono.just(cbsClient)
+                .flatMap(client -> client.get(cbsRequest))
+                .block();
+            handleConfigurationFromConsul(jsonObject);
+        return mapperConfig;
+    }
+
+    void handleConfigurationFromConsul(JsonObject jsonObject) {
+        logger.unwrap().info("Attempt to process configuration object");
+
         try {
-            JsonObject config = new Gson().fromJson(mapperConfigJson, JsonObject.class);
             mapperConfig = new GsonBuilder()
-                    .registerTypeAdapter(MapperConfig.class, new RequiredFieldDeserializer<MapperConfig>())
-                    .create()
-                    .fromJson(config, MapperConfig.class);
+                .registerTypeAdapter(MapperConfig.class, new RequiredFieldDeserializer<MapperConfig>())
+                .create()
+                .fromJson(jsonObject, MapperConfig.class);
         } catch (Exception exception) {
-            String exceptionMessage = "Error parsing configuration, mapper config:\n" + mapperConfigJson;
+            String exceptionMessage = "Error parsing configuration, mapper config:\n" + mapperConfig;
+            logger.unwrap().error(exceptionMessage);
             throw new MapperConfigException(exceptionMessage, exception);
         }
-        logger.unwrap().info("Received pm-mapper configuration from ConfigBinding Service");
-        logger.unwrap().debug("Mapper configuration:\n{}", mapperConfig);
-        return mapperConfig;
+        logger.unwrap().info("PM-mapper configuration processed successful");
+        logger.unwrap().info("Mapper configuration:\n{}", mapperConfig);
+    }
+
+    private void handleError(Throwable throwable) {
+        logger.unwrap().error("Unexpected error occurred during fetching configuration", throwable);
     }
 }
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java
index eee7d27..2cff1e3 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java
@@ -1,6 +1,7 @@
 /*-
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2019 Nordix Foundation.
+ *  Copyright (C) 2022 Nokia.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -44,11 +45,11 @@ public class DynamicConfiguration extends ServerResource {
      * @param configurables list of objects to reconfigure
      * @param originalConfig original config to compare against.
      */
-    public DynamicConfiguration(List<Configurable> configurables, MapperConfig originalConfig) {
+    public DynamicConfiguration(List<Configurable> configurables, MapperConfig originalConfig, ConfigHandler configHandler) {
         super(RECONFIGURE_ENDPOINT);
         this.configurables = configurables;
         this.originalConfig = originalConfig;
-        this.configHandler = new ConfigHandler();
+        this.configHandler = configHandler;
     }
 
     private void applyConfiguration(MapperConfig updatedConfig) throws ReconfigurationException {
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/exceptions/CBSServerError.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/exceptions/CBSServerError.java
deleted file mode 100644
index 58262ba..0000000
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/exceptions/CBSServerError.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*-

- * ============LICENSE_START=======================================================

- *  Copyright (C) 2019 Nordix Foundation.

- * ================================================================================

- * 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.

- *

- * SPDX-License-Identifier: Apache-2.0

- * ============LICENSE_END=========================================================

- */

-package org.onap.dcaegen2.services.pmmapper.exceptions;

-

-public class CBSServerError extends RuntimeException {

-    public CBSServerError(String message, Throwable cause) {

-        super(message, cause);

-    }

-}

diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/EnvironmentConfig.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/EnvironmentConfig.java
deleted file mode 100644
index 8457989..0000000
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/utils/EnvironmentConfig.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*-

- * ============LICENSE_START=======================================================

- *  Copyright (C) 2019 Nordix Foundation.

- * ================================================================================

- * 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.

- *

- * SPDX-License-Identifier: Apache-2.0

- * ============LICENSE_END=========================================================

- */

-package org.onap.dcaegen2.services.pmmapper.utils;

-

-import java.util.Optional;

-

-import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;

-

-public class EnvironmentConfig {

-    public static final int DEFAULT_CBS_PORT = 10000;

-    public static final String ENV_CBS_HOST_KEY = "CONFIG_BINDING_SERVICE_SERVICE_HOST";

-    public static final String ENV_CBS_PORT_KEY = "CONFIG_BINDING_SERVICE_SERVICE_PORT";

-    public static final String ENV_SERVICE_NAME_KEY = "HOSTNAME";

-

-    public String getServiceName() throws EnvironmentConfigException {

-        return Optional.ofNullable(System.getenv(ENV_SERVICE_NAME_KEY))

-                .orElseThrow(() -> new EnvironmentConfigException(

-                        ENV_SERVICE_NAME_KEY+ " environment variable must be defined prior to pm-mapper initialization."));

-    }

-

-    public String getCBSHostName() throws EnvironmentConfigException {

-        return Optional.ofNullable(System.getenv(ENV_CBS_HOST_KEY))

-                .orElseThrow(() -> new EnvironmentConfigException(

-                        ENV_CBS_HOST_KEY+ " environment variable must be defined prior to pm-mapper initialization."));

-    }

-

-    public Integer getCBSPort() {

-         return Optional.ofNullable(System.getenv(ENV_CBS_PORT_KEY))

-                    .map(Integer::valueOf).orElse(DEFAULT_CBS_PORT);

-    }

-}

diff --git a/src/test/java/org/onap/dcaegen2/pmmapper/messagerouter/VESPublisherTest.java b/src/test/java/org/onap/dcaegen2/pmmapper/messagerouter/VESPublisherTest.java
index 47e09e9..0641c55 100644
--- a/src/test/java/org/onap/dcaegen2/pmmapper/messagerouter/VESPublisherTest.java
+++ b/src/test/java/org/onap/dcaegen2/pmmapper/messagerouter/VESPublisherTest.java
@@ -1,7 +1,7 @@
 /*-
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2019 Nordix Foundation.
- *  Copyright (C) 2021 Nokia.
+ *  Copyright (C) 2021-2022 Nokia.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,12 +27,10 @@ import org.onap.dcaegen2.services.pmmapper.messagerouter.VESPublisher;
 import org.onap.dcaegen2.services.pmmapper.model.Event;
 import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
 import org.onap.dcaegen2.services.pmmapper.utils.DmaapRequestSender;
-import org.onap.dcaegen2.services.pmmapper.utils.EnvironmentConfig;
 import org.onap.dcaegen2.services.sdk.model.streams.ImmutableAafCredentials;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
 import org.powermock.core.classloader.annotations.PowerMockIgnore;
-import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
@@ -49,7 +47,6 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 @RunWith(PowerMockRunner.class)
-@PrepareForTest(EnvironmentConfig.class)
 @PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
 public class VESPublisherTest {
 
diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
index 617cbd1..32f7eb0 100644
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
@@ -1,7 +1,7 @@
 /*-
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2019-2020 Nordix Foundation.
- *  Copyright (C) 2021 Nokia.
+ *  Copyright (C) 2021-2022 Nokia.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -47,8 +47,8 @@ import com.google.gson.Gson;
 import io.undertow.server.HttpServerExchange;
 import io.undertow.util.StatusCodes;
 import org.junit.jupiter.api.BeforeEach;
+import org.mockito.MockSettings;
 import org.onap.dcaegen2.services.pmmapper.config.ConfigHandler;
-import org.onap.dcaegen2.services.pmmapper.exceptions.CBSServerError;
 import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.MapperConfigException;
 
@@ -120,7 +120,7 @@ class AppTest {
 
         MapperConfig mockConfig = Mockito.spy(mapperConfig);
         when(mockConfig.getEnableHttp()).thenReturn(false);
-        when(configHandler.getMapperConfig()).thenReturn(mockConfig);
+        when(configHandler.getInitialConfiguration()).thenReturn(mockConfig);
         objUnderTest = new App(template, schema, 0, 0, configHandler, processingConfig);
         objUnderTest.start();
         assertEquals(1, objUnderTest.getApplicationServer().getListenerInfo().size());
@@ -132,7 +132,7 @@ class AppTest {
     void testEnabledHTTPServer() throws Exception {
         MapperConfig mockConfig = Mockito.spy(mapperConfig);
         when(mockConfig.getEnableHttp()).thenReturn(true);
-        when(configHandler.getMapperConfig()).thenReturn(mockConfig);
+        when(configHandler.getInitialConfiguration()).thenReturn(mockConfig);
         objUnderTest = new App(template, schema, 0, 0, configHandler, processingConfig);
         objUnderTest.start();
         assertEquals(2, objUnderTest.getApplicationServer().getListenerInfo().size());
@@ -141,17 +141,17 @@ class AppTest {
     }
 
     @Test
-    void testConfigFailure() throws EnvironmentConfigException, CBSServerError, MapperConfigException {
-        when(configHandler.getMapperConfig()).thenThrow(MapperConfigException.class);
+    void testConfigFailure() throws MapperConfigException {
+        when(configHandler.getInitialConfiguration()).thenThrow(MapperConfigException.class);
         assertThrows(IllegalStateException.class, () -> new App(template, schema, 0, 0, configHandler, processingConfig));
 
     }
 
     @Test
-    void testServerCreationFailure() throws EnvironmentConfigException, CBSServerError, MapperConfigException {
+    void testServerCreationFailure() throws MapperConfigException {
         MapperConfig mockConfig = Mockito.spy(mapperConfig);
         when(mockConfig.getKeyStorePath()).thenReturn("not_a_file");
-        when(configHandler.getMapperConfig()).thenReturn(mockConfig);
+        when(configHandler.getInitialConfiguration()).thenReturn(mockConfig);
         assertThrows(IllegalStateException.class, () -> new App(template, schema, 0, 0, configHandler, processingConfig));
 
     }
@@ -330,7 +330,7 @@ class AppTest {
     void filesProcessingConfiguration_IsReadInMainApp() throws Exception {
         MapperConfig mockConfig = Mockito.spy(mapperConfig);
         when(mockConfig.getEnableHttp()).thenReturn(true);
-        when(configHandler.getMapperConfig()).thenReturn(mockConfig);
+        when(configHandler.getInitialConfiguration()).thenReturn(mockConfig);
         objUnderTest = new App(template, schema, 0, 0, configHandler, processingConfig);
         objUnderTest.start();
 
diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandlerTests.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandlerTests.java
index fd144cc..a1538ee 100644
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandlerTests.java
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/config/ConfigHandlerTests.java
@@ -1,6 +1,7 @@
 /*-
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2019 Nordix Foundation.
+ *  Copyright (C) 2022 Nokia.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,12 +21,11 @@
 
 package org.onap.dcaegen2.services.pmmapper.config;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.Mockito.anyString;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
-import java.net.UnknownHostException;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -39,105 +39,134 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
-import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
-import org.onap.dcaegen2.services.pmmapper.exceptions.CBSConfigException;
-import org.onap.dcaegen2.services.pmmapper.exceptions.CBSServerError;
 import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;
 import org.onap.dcaegen2.services.pmmapper.exceptions.MapperConfigException;
-import org.onap.dcaegen2.services.pmmapper.exceptions.RequestFailure;
-import org.onap.dcaegen2.services.pmmapper.utils.EnvironmentConfig;
 import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
-import org.onap.dcaegen2.services.pmmapper.utils.RequestSender;
-
 
 import com.google.gson.Gson;
 import com.google.gson.JsonObject;
 
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
+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 reactor.core.publisher.Mono;
 import utils.FileUtils;
 import utils.LoggingUtils;
 
+
 @ExtendWith(MockitoExtension.class)
 class ConfigHandlerTests {
     private static String validMapperConfig;
-    private static String HOSTNAME = "pm-mapper-service-name";
-    private static String CBS_HOST = "cbs_host";
-    private static int CBS_PORT = 10000;
-    private static Path invalidConfigsDirectory = Paths.get("src/test/resources/invalid_configs/");
+    private static String validMapperConfigChanged;
 
-    private Gson gson = new Gson();
+    private static final Path INVALID_CONFIGS_DIRECTORY = Paths.get("src/test/resources/invalid_configs/");
+    private static final String EXPECTED_ERROR_MESSAGE_IN_LOG = "Error parsing configuration";
+    private static final String EXPECTED_CHANGED_VALUE = "https://dmaap-dr-node:8443/delete_changed";
 
-    @Mock
-    private RequestSender sender;
+    private final Gson gson = new Gson();
+    private final CbsClient cbsClient = mock(CbsClient.class);
+    private final CbsRequest cbsRequest = mock(CbsRequest.class);
 
-    @Mock
-    private static EnvironmentConfig config;
 
     @BeforeAll
-    static void beforeAll() throws Exception {
+    static void beforeAll() {
         validMapperConfig = FileUtils.getFileContents("valid_mapper_config.json");
-        config = mock(EnvironmentConfig.class);
-        when(config.getServiceName()).thenReturn(HOSTNAME);
-        when(config.getCBSPort()).thenReturn(CBS_PORT);
+        validMapperConfigChanged = FileUtils.getFileContents("valid_mapper_config_after_change.json");
     }
 
     @BeforeEach
-    void setup() throws Exception {
-        when(config.getCBSHostName()).thenReturn(CBS_HOST);
+    void setup() {
+        Mono<JsonObject> just = createMonoJsonObject(validMapperConfig);
+        when(cbsClient.get(any())).thenReturn(just);
     }
 
     @Test
     void getMapperConfig_success() throws Exception {
-        when(config.getCBSHostName()).thenReturn(CBS_HOST);
-        when(config.getServiceName()).thenReturn(HOSTNAME);
-        when(config.getCBSPort()).thenReturn(CBS_PORT);
+        MapperConfig expectedConfig = gson.fromJson(validMapperConfig, MapperConfig.class);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(ConfigHandler.class);
-        String validCbsUrlMapperConfig = "http://" + CBS_HOST + ":" + CBS_PORT + "/service_component/" + HOSTNAME;
-        when(sender.send(validCbsUrlMapperConfig)).thenReturn(validMapperConfig);
-        MapperConfig actualConfig = getMapperConfig();
-        JsonObject expectedConfigJson = gson.fromJson(validMapperConfig, JsonObject.class);
-        MapperConfig expectedConfig = gson.fromJson(expectedConfigJson, MapperConfig.class);
+        MapperConfig actualConfig = new ConfigHandler(cbsClient, cbsRequest).getMapperConfig();
+
         assertEquals(expectedConfig.getPublisherTopicUrl(), actualConfig.getPublisherTopicUrl());
         assertEquals(expectedConfig.getPublisherUserName(), actualConfig.getPublisherUserName());
         assertEquals(expectedConfig.getPublisherPassword(), actualConfig.getPublisherPassword());
         assertEquals(expectedConfig, actualConfig);
         logAppender.stop();
+
     }
 
     @Test
-    void configbinding_server_error() throws Exception {
-        when(sender.send(anyString())).thenThrow(RequestFailure.class);
-        assertThrows(CBSServerError.class, this::getMapperConfig);
+    void configuration_should_can_be_changed() throws EnvironmentConfigException {
+        MapperConfig expectedConfig = gson.fromJson(validMapperConfig, MapperConfig.class);
+        Mono<JsonObject> just = createMonoJsonObject(validMapperConfig);
+
+        when(cbsClient.get(any())).thenReturn(just);
+        ConfigHandler configHandler = new ConfigHandler(cbsClient, cbsRequest);
+
+        MapperConfig actualConfig = configHandler.getInitialConfiguration();
+        assertEquals(expectedConfig.getDmaapDRDeleteEndpoint(), actualConfig.getDmaapDRDeleteEndpoint());
+
+        Mono<JsonObject> justChanged = createMonoJsonObject(validMapperConfigChanged);
+        when(cbsClient.get(any())).thenReturn(justChanged);
+        MapperConfig changedConfig = configHandler.getMapperConfig();
+
+        System.out.println(changedConfig.getDmaapDRDeleteEndpoint());
+        assertEquals(EXPECTED_CHANGED_VALUE, changedConfig.getDmaapDRDeleteEndpoint());
     }
 
     @Test
-    void configbinding_server_host_missing() throws Exception {
-        when(config.getCBSHostName()).thenThrow(EnvironmentConfigException.class);
-        assertThrows(EnvironmentConfigException.class, this::getMapperConfig);
+    void should_throw_exception_when_configuration_is_not_initialized() {
+        String wrongConfigJson = "{\"test\": \"test\"}";
+        Mono<JsonObject> just = createMonoJsonObject(wrongConfigJson);
+
+        when(cbsClient.get(any())).thenReturn(just);
+
+        ConfigHandler configHandler = new ConfigHandler(cbsClient, cbsRequest);
+        assertThrows(EnvironmentConfigException.class, configHandler::getMapperConfig);
     }
 
     @Test
-    void mapper_parse_invalid_json_mapper_config() throws Exception {
-        when(sender.send(anyString())).thenReturn("mapper config with incorrect format");
-        assertThrows(MapperConfigException.class, this::getMapperConfig);
+    void mapper_parse_invalid_json_mapper_config() {
+        String wrongConfigJson = "{\"test\": \"test\"}";
+        Mono<JsonObject> just = createMonoJsonObject(wrongConfigJson);
+
+        when(cbsClient.get(any())).thenReturn(just);
+        ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(ConfigHandler.class);
+        ConfigHandler configHandler = new ConfigHandler(cbsClient, cbsRequest);
+
+        assertThrows(MapperConfigException.class, configHandler::getInitialConfiguration);
+        assertConfigurationErrorIsLogged(logAppender);
+        logAppender.stop();
     }
 
     @ParameterizedTest
     @MethodSource("getInvalidConfigs")
     void parse_valid_json_bad_values_mapper_config(String mapperConfig) throws Exception {
-        when(sender.send(anyString())).thenReturn(mapperConfig);
-        assertThrows(MapperConfigException.class, this::getMapperConfig);
+        Mono<JsonObject> just = createMonoJsonObject(mapperConfig);
+
+        when(cbsClient.get(any())).thenReturn(just);
+        ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(ConfigHandler.class);
+        ConfigHandler configHandler = new ConfigHandler(cbsClient, cbsRequest);
+
+        assertThrows(MapperConfigException.class, configHandler::getInitialConfiguration);
+        assertConfigurationErrorIsLogged(logAppender);
+        logAppender.stop();
+    }
+
+    private void assertConfigurationErrorIsLogged(ListAppender<ILoggingEvent> logAppender) {
+        boolean containMessage = logAppender.list.stream()
+            .anyMatch(iLoggingEvent -> iLoggingEvent.getFormattedMessage().contains(EXPECTED_ERROR_MESSAGE_IN_LOG));
+        assertTrue(containMessage);
     }
 
-    private MapperConfig getMapperConfig()
-            throws UnknownHostException, EnvironmentConfigException, CBSConfigException, Exception {
-        return new ConfigHandler(sender, config).getMapperConfig();
+    private Mono<JsonObject> createMonoJsonObject(String stringJson) {
+        JsonObject configJson = new Gson().fromJson(stringJson, JsonObject.class);
+        return Mono.just(configJson);
     }
 
     private static List<String> getInvalidConfigs() throws IOException {
-        return FileUtils.getFilesFromDirectory(invalidConfigsDirectory);
+        return FileUtils.getFilesFromDirectory(INVALID_CONFIGS_DIRECTORY);
     }
 }
diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfigurationTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfigurationTest.java
index 905d18a..f8ad406 100644
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfigurationTest.java
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfigurationTest.java
@@ -1,6 +1,7 @@
 /*-
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2019 Nordix Foundation.
+ *  Copyright (C) 2022 Nokia.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,6 +21,7 @@
 
 package org.onap.dcaegen2.services.pmmapper.config;
 
+import com.google.gson.Gson;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
 import io.undertow.server.HttpServerExchange;
@@ -28,16 +30,16 @@ import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.stubbing.Answer;
 
 import org.onap.dcaegen2.services.pmmapper.exceptions.ReconfigurationException;
-import org.onap.dcaegen2.services.pmmapper.utils.EnvironmentConfig;
 import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
-import org.onap.dcaegen2.services.pmmapper.utils.RequestSender;
 
+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 reactor.core.publisher.Mono;
 import utils.ConfigUtils;
 import java.util.ArrayList;
 import utils.FileUtils;
@@ -61,24 +63,23 @@ class DynamicConfigurationTest {
     private MapperConfig originalMapperConfig;
     private static ConfigHandler configHandler;
 
-    @Mock
-    private static RequestSender sender;
-
-    @Mock
-    private static EnvironmentConfig config;
+    private final CbsClient cbsClient = mock(CbsClient.class);
+    private final CbsRequest cbsRequest = mock(CbsRequest.class);
 
     @BeforeAll
-     static void setupBeforeAll() throws Exception {
+     static void setupBeforeAll() {
         mapperConfig = FileUtils.getFileContents(VALID_MAPPER_CONFIG_FILE);
     }
 
     @BeforeEach
-    void setup() throws Exception {
-        configHandler = new ConfigHandler(sender, config);
-        when(sender.send(any())).thenReturn(mapperConfig);
+    void setup() {
+        Mono<JsonObject> just = createMonoJsonObject(mapperConfig);
+        when(cbsClient.get(any())).thenReturn(just);
+
+        configHandler = new ConfigHandler(cbsClient, cbsRequest);
         originalMapperConfig = ConfigUtils.getMapperConfigFromFile(VALID_MAPPER_CONFIG_FILE);
         configurables = new ArrayList<>();
-        objUnderTest = new DynamicConfiguration(configurables, originalMapperConfig);
+        objUnderTest = new DynamicConfiguration(configurables, originalMapperConfig, configHandler);
     }
 
     @Test
@@ -97,7 +98,9 @@ class DynamicConfigurationTest {
         configurables.add(configurable);
         JsonObject modifiedConfig = new JsonParser().parse(mapperConfig).getAsJsonObject();
         modifiedConfig.addProperty("dmaap_dr_delete_endpoint","http://modified-delete-endpoint/1");
-        when(sender.send(any())).thenReturn(modifiedConfig.toString());
+
+        Mono<JsonObject> just = Mono.just(modifiedConfig);
+        when(cbsClient.get(any())).thenReturn(just);
         MapperConfig modifiedMapperConfig = configHandler.getMapperConfig();
         objUnderTest.setConfigHandler(configHandler);
         doAnswer(new Answer() {
@@ -127,8 +130,9 @@ class DynamicConfigurationTest {
         JsonObject modifiedConfig = new JsonParser().parse(mapperConfig).getAsJsonObject();
         modifiedConfig.addProperty("dmaap_dr_delete_endpoint","http://modified-delete-endpoint/1");
 
-        when(sender.send(any()))
-                .thenReturn(modifiedConfig.toString());
+        Mono<JsonObject> just = Mono.just(modifiedConfig);
+        when(cbsClient.get(any())).thenReturn(just);
+
         MapperConfig modifiedMapperConfig = configHandler.getMapperConfig();
         objUnderTest.setConfigHandler(configHandler);
 
@@ -143,9 +147,11 @@ class DynamicConfigurationTest {
     @Test
     void testMapperConfigReconfiguration() throws Exception {
         JsonObject modifiedConfigJson = new JsonParser().parse(mapperConfig).getAsJsonObject();
-        modifiedConfigJson.addProperty("dmaap_dr_delete_endpoint","http://modified-delete-endpoint/1");
-        when(sender.send(any()))
-                .thenReturn(modifiedConfigJson.toString());
+        modifiedConfigJson.addProperty("dmaap_dr_delete_endpoint", "http://modified-delete-endpoint/1");
+
+        Mono<JsonObject> just = Mono.just(modifiedConfigJson);
+        when(cbsClient.get(any())).thenReturn(just);
+
         MapperConfig modifiedConfig = configHandler.getMapperConfig();
         originalMapperConfig.reconfigure(modifiedConfig);
         assertEquals(originalMapperConfig, modifiedConfig);
@@ -153,9 +159,16 @@ class DynamicConfigurationTest {
 
     @Test
     void testMapperConfigReconfigurationNoChange() throws Exception {
-        when(sender.send(any())).thenReturn(mapperConfig);
         MapperConfig inboundConfig = configHandler.getMapperConfig();
+
         originalMapperConfig.reconfigure(inboundConfig);
+
         assertEquals(originalMapperConfig, inboundConfig);
     }
+
+    private Mono<JsonObject> createMonoJsonObject(String stringJson) {
+        JsonObject configJson = new Gson().fromJson(stringJson, JsonObject.class);
+        return Mono.just(configJson);
+    }
+
 }
diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/utils/EnvironmentConfigTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/utils/EnvironmentConfigTest.java
deleted file mode 100644
index e3c32c8..0000000
--- a/src/test/java/org/onap/dcaegen2/services/pmmapper/utils/EnvironmentConfigTest.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*-

- * ============LICENSE_START=======================================================

- *  Copyright (C) 2019 Nordix Foundation.

- * ================================================================================

- * 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.

- *

- * SPDX-License-Identifier: Apache-2.0

- * ============LICENSE_END=========================================================

- */

-

-package org.onap.dcaegen2.services.pmmapper.utils;

-import static org.junit.jupiter.api.Assertions.assertThrows;

-import static org.junit.Assert.assertEquals;

-import org.junit.Before;

-import org.junit.Test;

-import org.junit.runner.RunWith;

-import org.onap.dcaegen2.services.pmmapper.exceptions.EnvironmentConfigException;

-import org.powermock.api.mockito.PowerMockito;

-import org.powermock.core.classloader.annotations.PrepareForTest;

-import org.powermock.modules.junit4.PowerMockRunner;

-

-@RunWith(PowerMockRunner.class)

-@PrepareForTest(EnvironmentConfig.class)

-public class EnvironmentConfigTest {

-    private EnvironmentConfig objUnderTest;

-

-    @Before

-    public void before() throws Exception {

-        PowerMockito.mockStatic(System.class);

-        objUnderTest = new EnvironmentConfig();

-    }

-

-    @Test

-    public void environmentConfig_is_present_success() throws EnvironmentConfigException {

-        String CBS_HOST = "cbs_host";

-        PowerMockito.when(System.getenv(EnvironmentConfig.ENV_CBS_HOST_KEY)).thenReturn(CBS_HOST);

-        assertEquals(CBS_HOST, objUnderTest.getCBSHostName());

-    }

-

-    @Test

-    public void environmentConfig_host_not_present() throws EnvironmentConfigException {

-        PowerMockito.when(System.getenv(EnvironmentConfig.ENV_CBS_HOST_KEY)).thenReturn(null);

-        assertThrows(EnvironmentConfigException.class, objUnderTest::getCBSHostName);

-    }

-

-    @Test

-    public void environmentConfig_hostname_present() throws EnvironmentConfigException {

-        PowerMockito.when(System.getenv(EnvironmentConfig.ENV_SERVICE_NAME_KEY)).thenCallRealMethod();

-        assertThrows(EnvironmentConfigException.class, objUnderTest::getCBSHostName);

-    }

-

-    @Test

-    public void environmentConfig_default_port_is_used() throws EnvironmentConfigException {

-        PowerMockito.when(System.getenv(EnvironmentConfig.ENV_CBS_PORT_KEY)).thenReturn(null);

-        assertEquals(Integer.valueOf(EnvironmentConfig.DEFAULT_CBS_PORT), objUnderTest.getCBSPort());

-    }

-

-    @Test

-    public void environmentConfig_port_invalid() throws EnvironmentConfigException {

-        PowerMockito.when(System.getenv(EnvironmentConfig.ENV_CBS_PORT_KEY)).thenReturn("Invalid_port number");

-        assertThrows(EnvironmentConfigException.class, objUnderTest::getCBSHostName);

-    }

-

-    @Test

-    public void environmentConfig_service_name_missing() {

-        PowerMockito.when(System.getenv(EnvironmentConfig.ENV_SERVICE_NAME_KEY)).thenReturn(null);

-        assertThrows(EnvironmentConfigException.class, objUnderTest::getServiceName);

-    }

-    @Test

-    public void environmentConfig_service_name_success() throws EnvironmentConfigException {

-        String serviceName = "we the best service";

-        PowerMockito.when(System.getenv(EnvironmentConfig.ENV_SERVICE_NAME_KEY)).thenReturn(serviceName);

-        assertEquals(serviceName, objUnderTest.getServiceName());

-    }

-}

diff --git a/src/test/resources/valid_mapper_config_after_change.json b/src/test/resources/valid_mapper_config_after_change.json
new file mode 100644
index 0000000..c33a6bd
--- /dev/null
+++ b/src/test/resources/valid_mapper_config_after_change.json
@@ -0,0 +1,35 @@
+{

+  "pm-mapper-filter": "{\"filters\": [{\"pmDefVsn\": \"V9\", \"nfType\": \"NrRadio\", \"vendor\": \"Nokia\", \"measTypes\": [\"A\", \"B\"]}]}",

+  "key_store_path": "src/test/resources/testkeystore.jks",

+  "key_store_pass_path": "src/test/resources/password",

+  "trust_store_path": "src/test/resources/testkeystore.jks",

+  "trust_store_pass_path": "src/test/resources/password",

+  "dmaap_dr_delete_endpoint": "https://dmaap-dr-node:8443/delete_changed",

+  "dmaap_dr_feed_name": "bulk_pm_feed",

+  "aaf_identity": "dcae@dcae.onap.org",

+  "aaf_password": "iheartrainbows44",

+  "enable_http": false,

+  "streams_publishes": {

+    "dmaap_publisher": {

+      "type": "message_router",

+      "dmaap_info": {

+        "topic_url": "https://message-router:3905/events/org.onap.dmaap.mr.VES_PM",

+        "client_role": "org.onap.dcae.pmPublisher",

+        "location": "san-francisco",

+        "client_id": "1562763644939"

+      }

+    }

+  },

+  "streams_subscribes": {

+    "dmaap_subscriber": {

+      "type": "data_router",

+      "dmaap_info": {

+        "username": "username",

+        "password": "password",

+        "location": "san-francisco",

+        "delivery_url": "https://dcae-pm-mapper:8443/delivery",

+        "subscriber_id": 1

+      }

+    }

+  }

+}