2 * ============LICENSE_START=======================================================
3 * dcaegen2-collectors-veshv
4 * ================================================================================
5 * Copyright (C) 2019-2020 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 reactor.util.retry.Retry.withThrowable
33 import java.time.Duration
34 import java.util.concurrent.atomic.AtomicReference
37 internal class CbsClientAdapter(private val cbsClientMono: Mono<CbsClient>,
38 private val firstRequestDelay: Duration,
39 private val configurationStateListener: ConfigurationStateListener,
40 private val mdc: MappedDiagnosticContext,
41 retrySpec: Retry<Any>) {
43 private val requestInterval = AtomicReference<Duration>(Duration.ZERO)
44 private val retry = retrySpec.doOnRetry {
45 logger.withWarn(mdc) {
46 log("Exception while creating CBS client, retrying. Reason: ${it.exception().localizedMessage}")
48 configurationStateListener.retrying()
51 fun configurationUpdates() = cbsClientMono
54 "CBS client successfully created, first request will be sent in ${firstRequestDelay.seconds} s"
57 .retryWhen(withThrowable(retry))
58 .delayElement(firstRequestDelay)
59 .flatMapMany(::toPeriodicalConfigurations)
60 .distinctUntilChanged()
62 fun updateCbsInterval(intervalUpdate: Duration) = requestInterval.set(intervalUpdate).also {
63 logger.debug(mdc) { "CBS request interval changed to: ${intervalUpdate.seconds} s" }
66 private fun toPeriodicalConfigurations(cbsClient: CbsClient) =
67 Mono.defer { cbsClient.get(configurationRequest.withNewInvocationId()) }
68 .repeatWhen { it.nextWithVariableInterval(requestInterval::get) }
71 private val logger = Logger(CbsClientAdapter::class)
73 private val configurationRequest: CbsRequest = CbsRequests.getConfiguration(RequestDiagnosticContext.create())