Added partial config merger and updated config file format

Change-Id: I470d3a8b2d43f5b45d958320b620205457d1cf23
Issue-ID: DCAEGEN2-1370
Signed-off-by: Pawel Biniek <pawel.biniek@nokia.com>

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)
+
+}