diff options
Diffstat (limited to 'sources/hv-collector-configuration/src/main/kotlin/org')
7 files changed, 156 insertions, 16 deletions
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/ConfigurationModule.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/ConfigurationModule.kt index 9f8c552b..efe0aa88 100644 --- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/ConfigurationModule.kt +++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/ConfigurationModule.kt @@ -34,10 +34,14 @@ class ConfigurationModule { private val configReader = FileConfigurationReader() private val configValidator = ConfigurationValidator() + private lateinit var initialConfig: HvVesConfiguration + fun hvVesConfigurationUpdates(args: Array<String>): Flux<HvVesConfiguration> = Flux.just(cmd.parse(args)) .throwOnLeft { MissingArgumentException(it.message, it.cause) } .map { it.reader().use(configReader::loadConfig) } .map { configValidator.validate(it) } .throwOnLeft { ValidationException(it.message) } + .doOnNext { initialConfig = it } + } diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Routing.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Routing.kt index 45180a84..5ffa39df 100644 --- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Routing.kt +++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/Routing.kt @@ -23,4 +23,4 @@ import org.onap.ves.VesEventOuterClass.CommonEventHeader data class Routing(val routes: List<Route>) -data class Route(val domain: String, val targetTopic: String, val partitioning: (CommonEventHeader) -> Int = { 0 })
\ No newline at end of file +data class Route(val domain: String, val targetTopic: String, val partitioning: (CommonEventHeader) -> Int = { 0 }) diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMerger.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMerger.kt new file mode 100644 index 00000000..f044492c --- /dev/null +++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationMerger.kt @@ -0,0 +1,97 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2019 NOKIA + * ================================================================================ + * 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.dcae.collectors.veshv.config.impl + + +import arrow.core.None +import arrow.core.Option +import arrow.core.Some +import arrow.core.getOrElse +import arrow.core.toOption +import org.onap.dcae.collectors.veshv.utils.logging.LogLevel + +/** + * @author Pawel Biniek <pawel.biniek@nokia.com> + * @since March 2019 + */ +internal class ConfigurationMerger { + fun merge(base: PartialConfiguration, update: PartialConfiguration): PartialConfiguration = + PartialConfiguration( + mergeServerConfig(base.server, update.server), + mergeCbsConfig(base.cbs, update.cbs), + mergeSecurityConfig(base.security, update.security), + mergeCollectorConfig(base.collector, update.collector), + mergeLogLevel(base.logLevel, update.logLevel) + ) + + + private fun mergeServerConfig(baseOption: Option<PartialServerConfig>, + updateOption: Option<PartialServerConfig>) = + applyUpdate(baseOption, updateOption) { base, update -> + PartialServerConfig( + base.listenPort.updateToGivenOrNone(update.listenPort), + base.idleTimeoutSec.updateToGivenOrNone(update.idleTimeoutSec), + base.maxPayloadSizeBytes.updateToGivenOrNone(update.maxPayloadSizeBytes) + ) + } + + + private fun mergeCbsConfig(baseOption: Option<PartialCbsConfig>, + updateOption: Option<PartialCbsConfig>) = + applyUpdate(baseOption, updateOption) { base, update -> + PartialCbsConfig( + base.firstRequestDelaySec.updateToGivenOrNone(update.firstRequestDelaySec), + base.requestIntervalSec.updateToGivenOrNone(update.requestIntervalSec) + ) + } + + private fun mergeSecurityConfig(baseOption: Option<PartialSecurityConfig>, + updateOption: Option<PartialSecurityConfig>) = + applyUpdate(baseOption, updateOption) { base, update -> + PartialSecurityConfig( + base.keys.updateToGivenOrNone(update.keys) + ) + } + + private fun mergeCollectorConfig(baseOption: Option<PartialCollectorConfig>, + updateOption: Option<PartialCollectorConfig>) = + applyUpdate(baseOption, updateOption) { base, update -> + PartialCollectorConfig( + base.maxRequestSizeBytes.updateToGivenOrNone(update.maxRequestSizeBytes), + base.kafkaServers.updateToGivenOrNone(update.kafkaServers), + base.routing.updateToGivenOrNone(update.routing) + ) + } + + + private fun mergeLogLevel(base: Option<LogLevel>, update: Option<LogLevel>) = + base.updateToGivenOrNone(update) +} + +private fun <T> applyUpdate(base: Option<T>, update: Option<T>, overrider: (base: T, update: T) -> T) = + when { + base is Some && update is Some -> overrider(base.t, update.t).toOption() + base is Some && update is None -> base + base is None && update is Some -> update + else -> None + } + +private fun <T> Option<T>.updateToGivenOrNone(update: Option<T>) = + update.getOrElse(this::orNull).toOption() diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidator.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidator.kt index a6dc6887..c8a156c5 100644 --- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidator.kt +++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ConfigurationValidator.kt @@ -87,7 +87,7 @@ internal class ConfigurationValidator { partial.mapBinding { ServerConfiguration( it.listenPort.bind(), - Duration.ofSeconds(it.idleTimeoutSec.bind().toLong()), + it.idleTimeoutSec.bind(), it.maxPayloadSizeBytes.bind() ) } @@ -95,8 +95,8 @@ internal class ConfigurationValidator { private fun createCbsConfiguration(partial: PartialCbsConfig) = partial.mapBinding { CbsConfiguration( - Duration.ofSeconds(it.firstRequestDelaySec.bind().toLong()), - Duration.ofSeconds(it.requestIntervalSec.bind().toLong()) + it.firstRequestDelaySec.bind(), + it.requestIntervalSec.bind() ) } diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReader.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReader.kt index d4e3db85..9513107b 100644 --- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReader.kt +++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReader.kt @@ -21,10 +21,12 @@ package org.onap.dcae.collectors.veshv.config.impl import arrow.core.Option import com.google.gson.GsonBuilder +import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.DurationOfSecondsAdapter import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.OptionAdapter import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.SecurityAdapter + import java.io.Reader -import java.net.InetSocketAddress +import java.time.Duration /** * @author Pawel Biniek <pawel.biniek@nokia.com> @@ -34,6 +36,7 @@ internal class FileConfigurationReader { private val gson = GsonBuilder() .registerTypeAdapter(Option::class.java, OptionAdapter()) .registerTypeAdapter(PartialSecurityConfig::class.java, SecurityAdapter()) + .registerTypeAdapter(Duration::class.java, DurationOfSecondsAdapter()) .create() fun loadConfig(input: Reader): PartialConfiguration = diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/PartialConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/PartialConfiguration.kt index 12dac044..a27998e1 100644 --- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/PartialConfiguration.kt +++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/PartialConfiguration.kt @@ -25,6 +25,7 @@ import org.onap.dcae.collectors.veshv.config.api.model.Routing import org.onap.dcae.collectors.veshv.utils.logging.LogLevel import org.onap.dcaegen2.services.sdk.security.ssl.SecurityKeys import java.net.InetSocketAddress +import java.time.Duration /** * @author Pawel Biniek <pawel.biniek@nokia.com> @@ -34,26 +35,25 @@ internal data class PartialConfiguration( val server: Option<PartialServerConfig> = None, val cbs: Option<PartialCbsConfig> = None, val security: Option<PartialSecurityConfig> = None, -// TOD0: retrieve when ConfigurationMerger is implemented -// val collector: Option<PartialCollectorConfig> = None, + val collector: Option<PartialCollectorConfig> = None, val logLevel: Option<LogLevel> = None ) internal data class PartialServerConfig( val listenPort: Option<Int> = None, - val idleTimeoutSec: Option<Int> = None, + val idleTimeoutSec: Option<Duration> = None, val maxPayloadSizeBytes: Option<Int> = None ) internal data class PartialCbsConfig( - val firstRequestDelaySec: Option<Int> = None, - val requestIntervalSec: Option<Int> = None + val firstRequestDelaySec: Option<Duration> = None, + val requestIntervalSec: Option<Duration> = None ) internal data class PartialSecurityConfig(val keys: Option<SecurityKeys> = None) -// TOD0: retrieve when ConfigurationMerger is implemented -//internal data class PartialCollectorConfig( -// val maxRequestSizeBytes: Option<Int> = None, -// val kafkaServers: Option<List<InetSocketAddress>> = None, -// val routing: Option<Routing> = None -//) + +internal data class PartialCollectorConfig( + val maxRequestSizeBytes: Option<Int> = None, + val kafkaServers: Option<List<InetSocketAddress>> = None, + val routing: Option<Routing> = None +) diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/DurationOfSecondsAdapter.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/DurationOfSecondsAdapter.kt new file mode 100644 index 00000000..99da1102 --- /dev/null +++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/DurationOfSecondsAdapter.kt @@ -0,0 +1,36 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2019 NOKIA + * ================================================================================ + * 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.dcae.collectors.veshv.config.impl.gsonadapters + +import com.google.gson.JsonDeserializationContext +import com.google.gson.JsonDeserializer +import com.google.gson.JsonElement +import java.lang.reflect.Type +import java.time.Duration + +/** + * @author Pawel Biniek <pawel.biniek@nokia.com> + * @since March 2019 + */ +class DurationOfSecondsAdapter : JsonDeserializer<Duration> { + override fun deserialize(json: JsonElement, typeOfT: Type, context: JsonDeserializationContext) = + Duration.ofSeconds(json.asLong) + +} |