Extract HV-VES configuration module

To avoid cyclic dependency between "modules trio"
configuration-core-ssl some classes from core.model were also extracted.

Change-Id: Ie11029ae3500964f67f4d72279ddd68cdb2a1f0c
Issue-ID: DCAEGEN2-1332
Signed-off-by: Filip Krzywka <filip.krzywka@nokia.com>

14 files changed, 724 insertions, 0 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
new file mode 100644
index 00000000..874cc5b0
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/ConfigurationModule.kt
@@ -0,0 +1,31 @@
+/*
+ * ============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.api
+
+import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
+import org.onap.dcae.collectors.veshv.config.impl.ArgVesHvConfiguration
+import reactor.core.publisher.Flux
+
+class ConfigurationModule {
+    fun createConfigurationFromCommandLine(args: Array<String>) =
+            ArgVesHvConfiguration().parse(args)
+
+    fun hvVesConfigurationUpdates(): Flux<ServerConfiguration> = Flux.never<ServerConfiguration>()
+}
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/CollectorConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/CollectorConfiguration.kt
new file mode 100644
index 00000000..7999451e
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/CollectorConfiguration.kt
@@ -0,0 +1,26 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 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.api.model
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+data class CollectorConfiguration(val routing: Routing)
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ConfigurationProviderParams.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ConfigurationProviderParams.kt
new file mode 100644
index 00000000..847279d9
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ConfigurationProviderParams.kt
@@ -0,0 +1,29 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018-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.api.model
+
+import java.time.Duration
+
+/**
+ * @author Jakub Dudycz <jakub.dudycz@nokia.com>
+ * @since July 2018
+ */
+data class ConfigurationProviderParams(val firstRequestDelay: Duration,
+                                       val requestInterval: Duration)
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/KafkaConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/KafkaConfiguration.kt
new file mode 100644
index 00000000..f9dff203
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/KafkaConfiguration.kt
@@ -0,0 +1,26 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 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.api.model
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since December 2018
+ */
+data class KafkaConfiguration(val bootstrapServers: String, val maximalRequestSizeBytes: Int)
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ServerConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ServerConfiguration.kt
new file mode 100644
index 00000000..8fbc649b
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/ServerConfiguration.kt
@@ -0,0 +1,42 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 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.api.model
+
+import org.onap.dcae.collectors.veshv.ssl.boundary.SecurityConfiguration
+import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
+import java.net.InetSocketAddress
+import java.time.Duration
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since May 2018
+ */
+data class ServerConfiguration(
+        val serverListenAddress: InetSocketAddress,
+        val kafkaConfiguration: KafkaConfiguration,
+        val configurationProviderParams: ConfigurationProviderParams,
+        val securityConfiguration: SecurityConfiguration,
+        val idleTimeout: Duration,
+        val healthCheckApiListenAddress: InetSocketAddress,
+        val maximumPayloadSizeBytes: Int,
+        val logLevel: LogLevel,
+        val dummyMode: Boolean = false
+)
+
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
new file mode 100644
index 00000000..847f35ad
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/api/model/routing.kt
@@ -0,0 +1,85 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018 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.api.model
+
+import arrow.core.Option
+import org.onap.dcae.collectors.veshv.domain.RoutedMessage
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.ves.VesEventOuterClass.CommonEventHeader
+
+data class Routing(val routes: List<Route>) {
+
+    fun routeFor(commonHeader: CommonEventHeader): Option<Route> =
+            Option.fromNullable(routes.find { it.applies(commonHeader) })
+}
+
+data class Route(val domain: String, val targetTopic: String, val partitioning: (CommonEventHeader) -> Int = {0}) {
+
+    fun applies(commonHeader: CommonEventHeader) = commonHeader.domain == domain
+
+    operator fun invoke(message: VesMessage): RoutedMessage =
+            RoutedMessage(targetTopic, partitioning(message.header), message)
+}
+
+
+/*
+Configuration DSL
+ */
+
+fun routing(init: RoutingBuilder.() -> Unit): RoutingBuilder {
+    val conf = RoutingBuilder()
+    conf.init()
+    return conf
+}
+
+class RoutingBuilder {
+    private val routes: MutableList<RouteBuilder> = mutableListOf()
+
+    fun defineRoute(init: RouteBuilder.() -> Unit): RouteBuilder {
+        val rule = RouteBuilder()
+        rule.init()
+        routes.add(rule)
+        return rule
+    }
+
+    fun build() = Routing(routes.map { it.build() }.toList())
+}
+
+class RouteBuilder {
+
+    private lateinit var domain: String
+    private lateinit var targetTopic: String
+    private lateinit var partitioning: (CommonEventHeader) -> Int
+
+    fun fromDomain(domain: String) : RouteBuilder = apply {
+        this.domain = domain
+    }
+
+    fun toTopic(targetTopic: String) : RouteBuilder = apply {
+        this.targetTopic = targetTopic
+    }
+
+    fun withFixedPartitioning(num: Int = 0) : RouteBuilder = apply {
+        partitioning = { num }
+    }
+
+    fun build() = Route(domain, targetTopic, partitioning)
+
+}
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ArgVesHvConfiguration.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ArgVesHvConfiguration.kt
new file mode 100644
index 00000000..08346d30
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/ArgVesHvConfiguration.kt
@@ -0,0 +1,151 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2018-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.Option
+import arrow.core.fix
+import arrow.core.getOrElse
+import arrow.instances.option.monad.monad
+import arrow.typeclasses.binding
+import org.apache.commons.cli.CommandLine
+import org.apache.commons.cli.DefaultParser
+import org.onap.dcae.collectors.veshv.commandline.ArgBasedConfiguration
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.CONFIGURATION_REQUEST_INTERVAL
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.CONFIGURATION_FIRST_REQUEST_DELAY
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.DUMMY_MODE
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.HEALTH_CHECK_API_PORT
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.IDLE_TIMEOUT_SEC
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.KAFKA_SERVERS
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.KEY_STORE_FILE
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.KEY_STORE_PASSWORD
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.LISTEN_PORT
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.LOG_LEVEL
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.MAXIMUM_PAYLOAD_SIZE_BYTES
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.SSL_DISABLE
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.TRUST_STORE_FILE
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.TRUST_STORE_PASSWORD
+import org.onap.dcae.collectors.veshv.commandline.hasOption
+import org.onap.dcae.collectors.veshv.commandline.intValue
+import org.onap.dcae.collectors.veshv.commandline.longValue
+import org.onap.dcae.collectors.veshv.commandline.stringValue
+import org.onap.dcae.collectors.veshv.config.api.model.ConfigurationProviderParams
+import org.onap.dcae.collectors.veshv.config.api.model.KafkaConfiguration
+import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
+import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
+import org.onap.dcae.collectors.veshv.ssl.boundary.createSecurityConfiguration
+import org.onap.dcae.collectors.veshv.utils.arrow.doOnFailure
+import org.onap.dcae.collectors.veshv.utils.logging.LogLevel
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import java.net.InetSocketAddress
+import java.time.Duration
+
+
+internal class ArgVesHvConfiguration : ArgBasedConfiguration<ServerConfiguration>(DefaultParser()) {
+    override val cmdLineOptionsList = listOf(
+            KAFKA_SERVERS,
+            HEALTH_CHECK_API_PORT,
+            LISTEN_PORT,
+            CONFIGURATION_FIRST_REQUEST_DELAY,
+            CONFIGURATION_REQUEST_INTERVAL,
+            SSL_DISABLE,
+            KEY_STORE_FILE,
+            KEY_STORE_PASSWORD,
+            TRUST_STORE_FILE,
+            TRUST_STORE_PASSWORD,
+            IDLE_TIMEOUT_SEC,
+            MAXIMUM_PAYLOAD_SIZE_BYTES,
+            DUMMY_MODE,
+            LOG_LEVEL
+    )
+
+    override fun getConfiguration(cmdLine: CommandLine): Option<ServerConfiguration> =
+            Option.monad().binding {
+                val healthCheckApiPort = cmdLine.intValue(
+                        HEALTH_CHECK_API_PORT,
+                        DefaultValues.HEALTH_CHECK_API_PORT
+                )
+                val kafkaServers = cmdLine.stringValue(KAFKA_SERVERS).bind()
+                val listenPort = cmdLine.intValue(LISTEN_PORT).bind()
+                val idleTimeoutSec = cmdLine.longValue(IDLE_TIMEOUT_SEC, DefaultValues.IDLE_TIMEOUT_SEC)
+                val maxPayloadSizeBytes = cmdLine.intValue(
+                        MAXIMUM_PAYLOAD_SIZE_BYTES,
+                        DefaultValues.MAX_PAYLOAD_SIZE_BYTES
+                )
+                val dummyMode = cmdLine.hasOption(DUMMY_MODE)
+                val security = createSecurityConfiguration(cmdLine)
+                        .doOnFailure { ex ->
+                            logger.withError { log("Could not read security keys", ex) }
+                        }
+                        .toOption()
+                        .bind()
+                val logLevel = cmdLine.stringValue(LOG_LEVEL, DefaultValues.LOG_LEVEL)
+                val configurationProviderParams = createConfigurationProviderParams(cmdLine).bind()
+                ServerConfiguration(
+                        serverListenAddress = InetSocketAddress(listenPort),
+                        kafkaConfiguration = KafkaConfiguration(kafkaServers, maxPayloadSizeBytes),
+                        healthCheckApiListenAddress = InetSocketAddress(healthCheckApiPort),
+                        configurationProviderParams = configurationProviderParams,
+                        securityConfiguration = security,
+                        idleTimeout = Duration.ofSeconds(idleTimeoutSec),
+                        maximumPayloadSizeBytes = maxPayloadSizeBytes,
+                        dummyMode = dummyMode,
+                        logLevel = determineLogLevel(logLevel)
+                )
+            }.fix()
+
+    private fun createConfigurationProviderParams(cmdLine: CommandLine): Option<ConfigurationProviderParams> =
+            Option.monad().binding {
+                val firstRequestDelay = cmdLine.longValue(
+                        CONFIGURATION_FIRST_REQUEST_DELAY,
+                        DefaultValues.CONFIGURATION_FIRST_REQUEST_DELAY
+                )
+                val requestInterval = cmdLine.longValue(
+                        CONFIGURATION_REQUEST_INTERVAL,
+                        DefaultValues.CONFIGURATION_REQUEST_INTERVAL
+                )
+                ConfigurationProviderParams(
+                        Duration.ofSeconds(firstRequestDelay),
+                        Duration.ofSeconds(requestInterval)
+                )
+            }.fix()
+
+    private fun determineLogLevel(logLevel: String) = LogLevel.optionFromString(logLevel)
+            .getOrElse {
+                logger.warn {
+                    "Failed to parse $logLevel as $LOG_LEVEL command line. " +
+                            "Using default log level (${DefaultValues.LOG_LEVEL})"
+                }
+                LogLevel.valueOf(DefaultValues.LOG_LEVEL)
+            }
+
+
+    internal object DefaultValues {
+        const val HEALTH_CHECK_API_PORT = 6060
+        const val CONFIGURATION_FIRST_REQUEST_DELAY = 10L
+        const val CONFIGURATION_REQUEST_INTERVAL = 5L
+        const val IDLE_TIMEOUT_SEC = 60L
+        const val MAX_PAYLOAD_SIZE_BYTES = WireFrameMessage.DEFAULT_MAX_PAYLOAD_SIZE_BYTES
+        val LOG_LEVEL = LogLevel.INFO.name
+    }
+
+    companion object {
+        private val logger = Logger(ArgVesHvConfiguration::class)
+    }
+}
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
new file mode 100644
index 00000000..452b9215
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/FileConfigurationReader.kt
@@ -0,0 +1,50 @@
+/*
+ * ============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.Option
+import com.google.gson.GsonBuilder
+import org.onap.dcae.collectors.veshv.config.api.model.Route
+import org.onap.dcae.collectors.veshv.config.api.model.Routing
+import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.AddressAdapter
+import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.OptionAdapter
+import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.RouteAdapter
+import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.RoutingAdapter
+import org.onap.dcae.collectors.veshv.config.impl.gsonadapters.SecurityKeysAdapter
+import org.onap.dcaegen2.services.sdk.security.ssl.SecurityKeys
+import java.io.Reader
+import java.net.InetSocketAddress
+
+/**
+ * @author Pawel Biniek <pawel.biniek@nokia.com>
+ * @since February 2019
+ */
+internal class FileConfigurationReader {
+    private val gson = GsonBuilder()
+            .registerTypeAdapter(InetSocketAddress::class.java, AddressAdapter())
+            .registerTypeAdapter(Route::class.java, RouteAdapter())
+            .registerTypeAdapter(Routing::class.java, RoutingAdapter())
+            .registerTypeAdapter(Option::class.java, OptionAdapter())
+            .registerTypeAdapter(SecurityKeys::class.java, SecurityKeysAdapter())
+            .create()
+
+    fun loadConfig(input: Reader): PartialConfiguration =
+            gson.fromJson(input, PartialConfiguration::class.java)
+}
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
new file mode 100644
index 00000000..8d9cca73
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/PartialConfiguration.kt
@@ -0,0 +1,60 @@
+/*
+ * ============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.Option
+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
+
+/**
+ * @author Pawel Biniek <pawel.biniek@nokia.com>
+ * @since February 2019
+ */
+internal data class PartialConfiguration(
+        val server : Option<PartialServerConfig>,
+        val cbs : Option<PartialCbsConfig>,
+        val security : Option<PartialSecurityConfig>,
+        val kafka : Option<PartialKafkaConfig>,
+        val logLevel : Option<LogLevel>
+)
+
+internal data class PartialServerConfig(
+        val healthCheckApiPort : Option<Int>,
+        val listenPort : Option<Int>,
+        val idleTimeoutSec : Option<Int>,
+        val maximumPayloadSizeBytes : Option<Int>,
+        val dummyMode : Option<Boolean>
+)
+
+internal data class PartialCbsConfig(
+        val firstRequestDelaySec : Option<Int>,
+        val requestIntervalSec : Option<Int>
+)
+
+internal data class PartialSecurityConfig(
+        val sslDisable : Option<Boolean>,
+        val keys : Option<SecurityKeys>)
+
+internal data class PartialKafkaConfig(
+    val kafkaServers : Option<Array<InetSocketAddress>>,
+    val routing : Option<Routing>
+)
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/AddressAdapter.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/AddressAdapter.kt
new file mode 100644
index 00000000..6d20da4a
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/AddressAdapter.kt
@@ -0,0 +1,50 @@
+/*
+ * ============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.net.InetSocketAddress
+
+
+/**
+ * @author Pawel Biniek <pawel.biniek@nokia.com>
+ * @since February 2019
+ */
+internal class AddressAdapter : JsonDeserializer<InetSocketAddress> {
+    override fun deserialize(
+            json: JsonElement,
+            typeOfT: Type,
+            context: JsonDeserializationContext?): InetSocketAddress
+        {
+            val portStart = json.asString.lastIndexOf(":")
+            if (portStart > 0) {
+                val address = json.asString.substring(0, portStart)
+                val port = json.asString.substring(portStart + 1)
+                return InetSocketAddress(address, port.toInt())
+            } else throw InvalidAddressException("Cannot parse '" + json.asString + "' to address")
+        }
+
+    class InvalidAddressException(reason:String) : RuntimeException(reason)
+}
+
+
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/OptionAdapter.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/OptionAdapter.kt
new file mode 100644
index 00000000..f7fdc3c3
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/OptionAdapter.kt
@@ -0,0 +1,40 @@
+/*
+ * ============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 arrow.core.Option
+import com.google.gson.JsonDeserializationContext
+import com.google.gson.JsonDeserializer
+import com.google.gson.JsonElement
+import java.lang.reflect.ParameterizedType
+import java.lang.reflect.Type
+
+/**
+ * @author Pawel Biniek <pawel.biniek@nokia.com>
+ * @since March 2019
+ */
+internal class OptionAdapter : JsonDeserializer<Option<Any>> {
+    override fun deserialize(json: JsonElement, typeOfT: Type, context: JsonDeserializationContext): Option<Any> {
+        val parametrizedType = typeOfT as ParameterizedType
+        val typeParameter = parametrizedType.actualTypeArguments.first()
+        return Option.fromNullable(context.deserialize<Any>(json, typeParameter))
+    }
+
+}
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/RouteAdapter.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/RouteAdapter.kt
new file mode 100644
index 00000000..25cb8861
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/RouteAdapter.kt
@@ -0,0 +1,43 @@
+/*
+ * ============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 org.onap.dcae.collectors.veshv.config.api.model.Route
+import org.onap.dcae.collectors.veshv.config.api.model.RouteBuilder
+import java.lang.reflect.Type
+
+/**
+ * @author Pawel Biniek <pawel.biniek@nokia.com>
+ * @since March 2019
+ */
+internal class RouteAdapter : JsonDeserializer<Route> {
+    override fun deserialize(json: JsonElement, typeOfT: Type, context: JsonDeserializationContext?): Route {
+        val jobj = json.asJsonObject
+        return RouteBuilder()
+                .fromDomain(jobj["fromDomain"].asString)
+                .toTopic(jobj["toTopic"].asString)
+                .withFixedPartitioning()
+                .build()
+    }
+
+}
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/RoutingAdapter.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/RoutingAdapter.kt
new file mode 100644
index 00000000..4b299098
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/RoutingAdapter.kt
@@ -0,0 +1,40 @@
+/*
+ * ============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 com.google.gson.reflect.TypeToken
+import org.onap.dcae.collectors.veshv.config.api.model.Route
+import org.onap.dcae.collectors.veshv.config.api.model.Routing
+import java.lang.reflect.Type
+
+/**
+ * @author Pawel Biniek <pawel.biniek@nokia.com>
+ * @since March 2019
+ */
+internal class RoutingAdapter : JsonDeserializer<Routing> {
+    override fun deserialize(json: JsonElement, typeOfT: Type, context: JsonDeserializationContext): Routing {
+        val parametrizedType = TypeToken.getParameterized(List::class.java, Route::class.java).type
+        return Routing(context.deserialize<List<Route>>(json, parametrizedType))
+    }
+
+}
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/SecurityKeysAdapter.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/SecurityKeysAdapter.kt
new file mode 100644
index 00000000..f9c0b4db
--- /dev/null
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/gsonadapters/SecurityKeysAdapter.kt
@@ -0,0 +1,51 @@
+/*
+ * ============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 org.onap.dcaegen2.services.sdk.security.ssl.ImmutableSecurityKeys
+import org.onap.dcaegen2.services.sdk.security.ssl.ImmutableSecurityKeysStore
+import org.onap.dcaegen2.services.sdk.security.ssl.Passwords
+import org.onap.dcaegen2.services.sdk.security.ssl.SecurityKeys
+import java.io.File
+import java.lang.reflect.Type
+
+/**
+ * @author Pawel Biniek <pawel.biniek@nokia.com>
+ * @since March 2019
+ */
+internal class SecurityKeysAdapter : JsonDeserializer<SecurityKeys> {
+    override fun deserialize(json: JsonElement, typeOfT: Type, context: JsonDeserializationContext?): SecurityKeys {
+        val obj = json.asJsonObject
+        return ImmutableSecurityKeys.builder()
+                .keyStore(ImmutableSecurityKeysStore.of(
+                        File(obj["keyStoreFile"].asString).toPath()))
+                .keyStorePassword(
+                        Passwords.fromString(obj["keyStorePassword"].asString))
+                .trustStore(ImmutableSecurityKeysStore.of(
+                        File(obj["trustStoreFile"].asString).toPath()))
+                .trustStorePassword(
+                        Passwords.fromString(obj["trustStorePassword"].asString))
+                .build()
+    }
+
+}