Various improvements

* Kotlin upgrade
* Monad usage on APIs
* Idle timeout
* Simulator enhancements

Closes ONAP-390

Change-Id: I3c00fcfe38c722caf661ddaad428cf089eeefcaa
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
Issue-ID: DCAEGEN2-601

4 files changed, 66 insertions, 30 deletions

diff --git a/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfiguration.kt b/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfiguration.kt
index 59b91d7f..f3e97be2 100644
--- a/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfiguration.kt
+++ b/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfiguration.kt
@@ -19,16 +19,19 @@
  */
 package org.onap.dcae.collectors.veshv.main
 
-import org.apache.commons.cli.DefaultParser
 import org.apache.commons.cli.CommandLine
+import org.apache.commons.cli.DefaultParser
+import org.onap.dcae.collectors.veshv.domain.SecurityConfiguration
+import org.onap.dcae.collectors.veshv.model.ServerConfiguration
 import org.onap.dcae.collectors.veshv.utils.commandline.ArgBasedConfiguration
-import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.LISTEN_PORT
+import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.CERT_FILE
 import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.CONSUL_CONFIG_URL
+import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.DUMMY_MODE
+import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.IDLE_TIMEOUT_SEC
+import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.LISTEN_PORT
 import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.PRIVATE_KEY_FILE
-import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.CERT_FILE
 import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.TRUST_CERT_FILE
-import org.onap.dcae.collectors.veshv.model.ServerConfiguration
-import org.onap.dcae.collectors.veshv.domain.SecurityConfiguration
+import java.time.Duration
 
 internal object DefaultValues {
     const val PORT = 6061
@@ -36,6 +39,7 @@ internal object DefaultValues {
     const val PRIVATE_KEY_FILE = "/etc/ves-hv/server.key"
     const val CERT_FILE = "/etc/ves-hv/server.crt"
     const val TRUST_CERT_FILE = "/etc/ves-hv/trust.crt"
+    const val IDLE_TIMEOUT_SEC = 60L
 }
 
 internal class ArgBasedServerConfiguration : ArgBasedConfiguration<ServerConfiguration>(DefaultParser()) {
@@ -44,14 +48,23 @@ internal class ArgBasedServerConfiguration : ArgBasedConfiguration<ServerConfigu
             CONSUL_CONFIG_URL,
             PRIVATE_KEY_FILE,
             CERT_FILE,
-            TRUST_CERT_FILE
+            TRUST_CERT_FILE,
+            IDLE_TIMEOUT_SEC,
+            DUMMY_MODE
     )
 
     override fun getConfiguration(cmdLine: CommandLine): ServerConfiguration {
         val port = cmdLine.intValue(LISTEN_PORT, DefaultValues.PORT)
         val configUrl = cmdLine.stringValue(CONSUL_CONFIG_URL, DefaultValues.CONFIG_URL)
+        val idleTimeoutSec = cmdLine.longValue(IDLE_TIMEOUT_SEC, DefaultValues.IDLE_TIMEOUT_SEC)
+        val dummyMode = cmdLine.hasOption(DUMMY_MODE)
         val security = createSecurityConfiguration(cmdLine)
-        return ServerConfiguration(port, configUrl, security)
+        return ServerConfiguration(
+                port = port,
+                configurationUrl = configUrl,
+                securityConfiguration = security,
+                idleTimeout = Duration.ofSeconds(idleTimeoutSec),
+                dummyMode = dummyMode)
     }
 
     private fun createSecurityConfiguration(cmdLine: CommandLine): SecurityConfiguration {
diff --git a/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/main.kt b/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/main.kt
index 1f2686ba..074a75e4 100644
--- a/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/main.kt
+++ b/hv-collector-main/src/main/kotlin/org/onap/dcae/collectors/veshv/main/main.kt
@@ -19,37 +19,49 @@
  */
 package org.onap.dcae.collectors.veshv.main
 
+import arrow.core.flatMap
 import org.onap.dcae.collectors.veshv.boundary.ConfigurationProvider
-import org.onap.dcae.collectors.veshv.utils.commandline.WrongArgumentException
-import org.onap.dcae.collectors.veshv.model.CollectorConfiguration
-import org.onap.dcae.collectors.veshv.model.ServerConfiguration
-import org.onap.dcae.collectors.veshv.model.routing
+import org.onap.dcae.collectors.veshv.boundary.Server
 import org.onap.dcae.collectors.veshv.factory.CollectorFactory
 import org.onap.dcae.collectors.veshv.factory.ServerFactory
 import org.onap.dcae.collectors.veshv.impl.adapters.AdapterFactory
+import org.onap.dcae.collectors.veshv.model.CollectorConfiguration
+import org.onap.dcae.collectors.veshv.model.ServerConfiguration
+import org.onap.dcae.collectors.veshv.model.routing
+import org.onap.dcae.collectors.veshv.utils.commandline.handleErrorsInMain
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
-import org.slf4j.LoggerFactory
-import kotlin.system.exitProcess
 
-private val logger = LoggerFactory.getLogger("main")
+private val logger = Logger("org.onap.dcae.collectors.veshv.main")
+private const val PROGRAM_NAME = "java org.onap.dcae.collectors.veshv.main.MainKt"
 
 fun main(args: Array<String>) {
-    try {
-        val serverConfiguration = ArgBasedServerConfiguration().parse(args)
-
-        val collectorProvider = CollectorFactory(
-                resolveConfigurationProvider(serverConfiguration),
-                AdapterFactory.kafkaSink(),
-                MicrometerMetrics()
-        ).createVesHvCollectorProvider()
-        ServerFactory.createNettyTcpServer(serverConfiguration, collectorProvider).start().block()
-    } catch (ex: WrongArgumentException) {
-        ex.printMessage()
-        ex.printHelp("java org.onap.dcae.collectors.veshv.main.MainKt")
-        exitProcess(1)
-    }
+    ArgBasedServerConfiguration().parse(args)
+            .toEither()
+            .map(::createServer)
+            .map(Server::start)
+            .flatMap { it.attempt().unsafeRunSync() }
+            .fold(
+                    { ex ->
+                        handleErrorsInMain(ex, PROGRAM_NAME, logger)
+                    },
+                    { handle ->
+                        logger.info("Server started. Listening on ${handle.host}:${handle.port}")
+                        handle.await().unsafeRunSync()
+                    }
+            )
 }
 
+private fun createServer(config: ServerConfiguration): Server {
+    val sink = if (config.dummyMode) AdapterFactory.loggingSink() else AdapterFactory.kafkaSink()
+    val collectorProvider = CollectorFactory(
+            resolveConfigurationProvider(config),
+            sink,
+            MicrometerMetrics()
+    ).createVesHvCollectorProvider()
+
+    return ServerFactory.createNettyTcpServer(config, collectorProvider)
+}
 
 private fun resolveConfigurationProvider(serverConfiguration: ServerConfiguration): ConfigurationProvider {
 
diff --git a/hv-collector-main/src/main/resources/logback.xml b/hv-collector-main/src/main/resources/logback.xml
index 48da3b18..5127e7ef 100644
--- a/hv-collector-main/src/main/resources/logback.xml
+++ b/hv-collector-main/src/main/resources/logback.xml
@@ -26,7 +26,10 @@
       </rollingPolicy>
     </appender>
 
-  <logger name="org.onap.dcae.collectors.veshv" level="INFO"/>
+  <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+  <logger name="org.onap.dcae.collectors.veshv.impl.wire" level="TRACE"/>
+  <logger name="org.onap.dcae.collectors.veshv.impl.adapters.kafka.KafkaSink" level="TRACE"/>
+  <logger name="org.onap.dcae.collectors.veshv.impl.adapters.LoggingSinkProvider" level="TRACE"/>
   <!--<logger name="reactor.ipc.netty" level="DEBUG"/>-->
 
   <root level="INFO">
diff --git a/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfigurationTest.kt b/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfigurationTest.kt
index 923f9d58..4c2425bc 100644
--- a/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfigurationTest.kt
+++ b/hv-collector-main/src/test/kotlin/org/onap/dcae/collectors/veshv/main/ArgBasedServerConfigurationTest.kt
@@ -19,6 +19,8 @@
  */
 package org.onap.dcae.collectors.veshv.main
 
+import arrow.core.Failure
+import arrow.core.Success
 import org.assertj.core.api.Assertions.assertThat
 import org.jetbrains.spek.api.Spek
 import org.jetbrains.spek.api.dsl.describe
@@ -44,7 +46,13 @@ object ArgBasedServerConfigurationTest : Spek({
         cut = ArgBasedServerConfiguration()
     }
 
-    fun parse(vararg cmdLine: String) = cut.parse(cmdLine)
+    fun parse(vararg cmdLine: String): ServerConfiguration {
+        val result = cut.parse(cmdLine)
+        return when (result) {
+            is Success -> result.value
+            is Failure -> throw AssertionError("Parsing result should be present")
+        }
+    }
 
     describe("parsing arguments") {
         given("all parameters are present in the long form") {