2 * ============LICENSE_START=======================================================
3 * dcaegen2-collectors-veshv
4 * ================================================================================
5 * Copyright (C) 2019 NOKIA
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
20 package org.onap.dcae.collectors.veshv.config.impl
22 import org.onap.dcae.collectors.veshv.config.api.ConfigurationStateListener
23 import org.onap.dcae.collectors.veshv.utils.logging.Logger
24 import org.onap.dcae.collectors.veshv.utils.logging.MappedDiagnosticContext
25 import org.onap.dcae.collectors.veshv.utils.rx.nextWithVariableInterval
26 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient
27 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests
28 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest
29 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext
30 import reactor.core.publisher.Mono
31 import reactor.retry.Retry
32 import java.time.Duration
33 import java.util.concurrent.atomic.AtomicReference
36 internal class CbsClientAdapter(private val cbsClientMono: Mono<CbsClient>,
37 private val firstRequestDelay: Duration,
38 private val configurationStateListener: ConfigurationStateListener,
39 private val mdc: MappedDiagnosticContext,
40 retrySpec: Retry<Any>) {
42 private val requestInterval = AtomicReference<Duration>(Duration.ZERO)
43 private val retry = retrySpec.doOnRetry {
44 logger.withWarn(mdc) {
45 log("Exception while creating CBS client, retrying. Reason: ${it.exception().localizedMessage}")
47 configurationStateListener.retrying()
50 fun configurationUpdates() = cbsClientMono
53 "CBS client successfully created, first request will be sent in ${firstRequestDelay.seconds} s"
57 .delayElement(firstRequestDelay)
58 .flatMapMany(::toPeriodicalConfigurations)
59 .distinctUntilChanged()
61 fun updateCbsInterval(intervalUpdate: Duration) = requestInterval.set(intervalUpdate).also {
62 logger.debug(mdc) { "CBS request interval changed to: ${intervalUpdate.seconds} s" }
65 private fun toPeriodicalConfigurations(cbsClient: CbsClient) =
66 Mono.defer { cbsClient.get(configurationRequest.withNewInvocationId()) }
67 .repeatWhen { it.nextWithVariableInterval(requestInterval::get) }
70 private val logger = Logger(CbsClientAdapter::class)
72 private val configurationRequest: CbsRequest = CbsRequests.getConfiguration(RequestDiagnosticContext.create())