1 files changed, 9 insertions, 45 deletions
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/CbsConfigurationProvider.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/CbsConfigurationProvider.kt
index 4982c732..6efa38e6 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/CbsConfigurationProvider.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/CbsConfigurationProvider.kt
@@ -22,56 +22,31 @@ package org.onap.dcae.collectors.veshv.config.impl
 import arrow.core.toOption
 import com.google.gson.JsonObject
 import org.onap.dcae.collectors.veshv.config.api.ConfigurationStateListener
-import org.onap.dcae.collectors.veshv.config.api.model.CbsConfiguration
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
 import org.onap.dcae.collectors.veshv.utils.logging.MappedDiagnosticContext
 import org.onap.dcae.collectors.veshv.utils.logging.onErrorLog
 import org.onap.dcae.collectors.veshv.utils.reader
 import org.onap.dcaegen2.services.sdk.model.streams.StreamType.KAFKA
 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
-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.cbs.client.api.streams.DataStreams
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParser
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamPredicates.streamOfType
-import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext
 import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
-import reactor.retry.Jitter
 import reactor.retry.Retry
 
 /**
  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
  * @since May 2018
  */
-internal class CbsConfigurationProvider(private val cbsClientMono: Mono<CbsClient>,
-                                        private val cbsConfiguration: CbsConfiguration,
+internal class CbsConfigurationProvider(private val cbsClientAdapter: CbsClientAdapter,
                                         private val configParser: JsonConfigurationParser,
-                                        private val streamParser: StreamFromGsonParser<KafkaSink>,
                                         private val configurationStateListener: ConfigurationStateListener,
                                         private val mdc: MappedDiagnosticContext,
-                                        retrySpec: Retry<Any>
-
+                                        retrySpec: Retry<Any>,
+                                        private val streamParser: StreamFromGsonParser<KafkaSink> =
+                                                StreamFromGsonParsers.kafkaSinkParser()
 ) {
-    constructor(cbsClientMono: Mono<CbsClient>,
-                cbsConfig: CbsConfiguration,
-                configParser: JsonConfigurationParser,
-                configurationStateListener: ConfigurationStateListener,
-                mdc: MappedDiagnosticContext) :
-            this(
-                    cbsClientMono,
-                    cbsConfig,
-                    configParser,
-                    StreamFromGsonParsers.kafkaSinkParser(),
-                    configurationStateListener,
-                    mdc,
-                    Retry.any<Any>()
-                            .retryMax(MAX_RETRIES)
-                            .fixedBackoff(cbsConfig.requestInterval)
-                            .jitter(Jitter.random())
-            )
-
     private val retry = retrySpec.doOnRetry {
         logger.withWarn(mdc) {
             log("Exception from configuration provider client, retrying subscription", it.exception())
@@ -80,22 +55,12 @@ internal class CbsConfigurationProvider(private val cbsClientMono: Mono<CbsClien
     }
 
     operator fun invoke(): Flux<PartialConfiguration> =
-            cbsClientMono
-                    .doOnNext { logger.info(mdc) { "CBS client successfully created" } }
-                    .onErrorLog(logger, mdc) { "Failed to retrieve CBS client" }
+            cbsClientAdapter.configurationUpdates(mdc)
+                    .doOnNext { logger.info(mdc) { "Received new configuration:\n$it" } }
+                    .map(::parseConfiguration)
+                    .doOnNext { logger.info(mdc) { "Successfully parsed configuration json to:\n$it" } }
+                    .onErrorLog(logger, mdc) { "Error while creating configuration" }
                     .retryWhen(retry)
-                    .doFinally { logger.trace(mdc) { "CBS client subscription finished" } }
-                    .flatMapMany(::handleUpdates)
-
-    private fun handleUpdates(cbsClient: CbsClient) = cbsClient
-            .updates(CbsRequests.getConfiguration(RequestDiagnosticContext.create()),
-                    cbsConfiguration.firstRequestDelay,
-                    cbsConfiguration.requestInterval)
-            .doOnNext { logger.info(mdc) { "Received new configuration:\n$it" } }
-            .map(::parseConfiguration)
-            .doOnNext { logger.info(mdc) { "Successfully parsed configuration json to:\n$it" } }
-            .onErrorLog(logger, mdc) { "Error while creating configuration" }
-            .retryWhen(retry)
 
     private fun parseConfiguration(json: JsonObject) =
             configParser
@@ -110,7 +75,6 @@ internal class CbsConfigurationProvider(private val cbsClientMono: Mono<CbsClien
                     .toList()
 
     companion object {
-        private const val MAX_RETRIES = 5L
         private val logger = Logger(CbsConfigurationProvider::class)
     }
 }