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

5 files changed, 93 insertions, 46 deletions

diff --git a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgBasedDcaeAppSimConfiguration.kt b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgBasedDcaeAppSimConfiguration.kt
index 3f539302..27edde9c 100644
--- a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgBasedDcaeAppSimConfiguration.kt
+++ b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgBasedDcaeAppSimConfiguration.kt
@@ -24,12 +24,14 @@ import org.apache.commons.cli.DefaultParser
 import org.onap.dcae.collectors.veshv.simulators.dcaeapp.config.DefaultValues.API_PORT
 import org.onap.dcae.collectors.veshv.utils.commandline.ArgBasedConfiguration
 import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption
-import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.LISTEN_PORT
 import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.KAFKA_SERVERS
 import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.KAFKA_TOPICS
+import org.onap.dcae.collectors.veshv.utils.commandline.CommandLineOption.LISTEN_PORT
 
 internal object DefaultValues {
     const val API_PORT = 8080
+    const val KAFKA_SERVERS = "kafka:9092"
+    const val KAFKA_TOPICS = "ves_hvRanMeas"
 }
 
 class ArgBasedDcaeAppSimConfiguration : ArgBasedConfiguration<DcaeAppSimConfiguration>(DefaultParser()) {
@@ -41,8 +43,8 @@ class ArgBasedDcaeAppSimConfiguration : ArgBasedConfiguration<DcaeAppSimConfigur
 
     override fun getConfiguration(cmdLine: CommandLine): DcaeAppSimConfiguration {
         val port = cmdLine.intValue(LISTEN_PORT, API_PORT)
-        val kafkaBootstrapServers = cmdLine.stringValue(KAFKA_SERVERS)
-        val kafkaTopics = cmdLine.stringValue(KAFKA_TOPICS).split(",").toSet()
+        val kafkaBootstrapServers = cmdLine.stringValue(KAFKA_SERVERS, DefaultValues.KAFKA_SERVERS)
+        val kafkaTopics = cmdLine.stringValue(KAFKA_TOPICS, DefaultValues.KAFKA_TOPICS).split(",").toSet()
         return DcaeAppSimConfiguration(
                 port,
                 kafkaBootstrapServers,
diff --git a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/kafka/KafkaSource.kt b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/kafka/KafkaSource.kt
index f7703b86..d53609ca 100644
--- a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/kafka/KafkaSource.kt
+++ b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/kafka/KafkaSource.kt
@@ -19,10 +19,10 @@
  */
 package org.onap.dcae.collectors.veshv.simulators.dcaeapp.kafka
 
+import arrow.effects.IO
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.common.serialization.ByteArrayDeserializer
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import reactor.core.publisher.Mono
 import reactor.kafka.receiver.KafkaReceiver
 import reactor.kafka.receiver.ReceiverOptions
 import java.util.*
@@ -33,10 +33,10 @@ import java.util.*
  */
 class KafkaSource(private val receiver: KafkaReceiver<ByteArray, ByteArray>) {
 
-    fun start(): Mono<Consumer> = Mono.create { sink ->
+    fun start(): IO<Consumer> = IO {
         val consumer = Consumer()
         receiver.receive().subscribe(consumer::update)
-        sink.success(consumer)
+        consumer
     }
 
     companion object {
diff --git a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/kafka/consumer.kt b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/kafka/consumer.kt
index 5f0fe7f6..66169534 100644
--- a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/kafka/consumer.kt
+++ b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/kafka/consumer.kt
@@ -19,7 +19,9 @@
  */
 package org.onap.dcae.collectors.veshv.simulators.dcaeapp.kafka
 
-import reactor.core.publisher.Mono
+import arrow.core.Option
+import arrow.effects.IO
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
 import reactor.kafka.receiver.ReceiverRecord
 
 /**
@@ -27,10 +29,11 @@ import reactor.kafka.receiver.ReceiverRecord
  * @since June 2018
  */
 
-class ConsumerState(val msgCount: Long, val lastKey: ByteArray?, val lastValue: ByteArray?)
+class ConsumerState(val msgCount: Long, val lastKey: Option<ByteArray>, val lastValue: Option<ByteArray>)
 
 interface ConsumerStateProvider {
-    fun currentState(): Mono<ConsumerState>
+    fun currentState(): ConsumerState
+    fun reset(): IO<Unit>
 }
 
 class Consumer : ConsumerStateProvider {
@@ -38,18 +41,28 @@ class Consumer : ConsumerStateProvider {
     private var lastKey: ByteArray? = null
     private var lastValue: ByteArray? = null
 
-    override fun currentState(): Mono<ConsumerState> = Mono.create { sink ->
-        val state = synchronized(this) {
-            ConsumerState(msgCount, lastKey, lastValue)
+    override fun currentState() =
+            ConsumerState(msgCount, Option.fromNullable(lastKey), Option.fromNullable(lastValue))
+
+    override fun reset() = IO {
+        synchronized(this) {
+            msgCount = 0
+            lastKey = null
+            lastValue = null
         }
-        sink.success(state)
     }
 
     fun update(record: ReceiverRecord<ByteArray, ByteArray>) {
+        logger.trace { "Updating stats for message from ${record.topic()}:${record.partition()}" }
+
         synchronized(this) {
             msgCount++
             lastKey = record.key()
             lastValue = record.value()
         }
     }
+
+    companion object {
+        private val logger = Logger(Consumer::class)
+    }
 }
diff --git a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/main.kt b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/main.kt
index c037af35..f7d44ede 100644
--- a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/main.kt
+++ b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/main.kt
@@ -19,32 +19,30 @@
  */
 package org.onap.dcae.collectors.veshv.simulators.dcaeapp
 
+import arrow.core.Failure
+import arrow.core.Success
 import org.onap.dcae.collectors.veshv.simulators.dcaeapp.config.ArgBasedDcaeAppSimConfiguration
+import org.onap.dcae.collectors.veshv.simulators.dcaeapp.config.DcaeAppSimConfiguration
 import org.onap.dcae.collectors.veshv.simulators.dcaeapp.kafka.KafkaSource
 import org.onap.dcae.collectors.veshv.simulators.dcaeapp.remote.ApiServer
-import org.onap.dcae.collectors.veshv.utils.commandline.WrongArgumentException
+import org.onap.dcae.collectors.veshv.utils.commandline.handleErrorsInMain
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
 import org.slf4j.LoggerFactory
 
 private val logger = Logger(LoggerFactory.getLogger("DCAE simulator :: main"))
 
 fun main(args: Array<String>) {
+    logger.info("Starting DCAE APP simulator")
 
-    try {
-        logger.info("Starting DCAE APP simulator")
-        val simulatorConfig = ArgBasedDcaeAppSimConfiguration().parse(args)
+    val config = ArgBasedDcaeAppSimConfiguration().parse(args)
+    when (config) {
+        is Success -> startApp(config.value).unsafeRunSync()
+        is Failure -> config.handleErrorsInMain("", logger)
+    }
+}
 
-        KafkaSource.create(simulatorConfig.kafkaBootstrapServers, simulatorConfig.kafkaTopics)
+private fun startApp(config: DcaeAppSimConfiguration) =
+        KafkaSource.create(config.kafkaBootstrapServers, config.kafkaTopics)
                 .start()
                 .map(::ApiServer)
-                .flatMap { it.start(simulatorConfig.apiPort) }
-                .block()
-    } catch (e: WrongArgumentException) {
-        e.printHelp("java org.onap.dcae.collectors.veshv.simulators.dcaeapp.MainKt")
-        System.exit(1)
-    } catch (e: Exception) {
-        logger.error(e.localizedMessage)
-        logger.debug("An error occurred when starting ves dcea app simulator", e)
-        System.exit(2)
-    }
-}
+                .flatMap { it.start(config.apiPort) }
diff --git a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/remote/ApiServer.kt b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/remote/ApiServer.kt
index fcb8e131..2fa8abec 100644
--- a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/remote/ApiServer.kt
+++ b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/remote/ApiServer.kt
@@ -19,13 +19,17 @@
  */
 package org.onap.dcae.collectors.veshv.simulators.dcaeapp.remote
 
+import arrow.core.Try
+import arrow.core.getOrElse
+import arrow.effects.IO
+import com.google.protobuf.MessageOrBuilder
 import com.google.protobuf.util.JsonFormat
 import org.onap.dcae.collectors.veshv.simulators.dcaeapp.kafka.ConsumerStateProvider
+import org.onap.ves.HVRanMeasFieldsV5.HVRanMeasFields
 import org.onap.ves.VesEventV5.VesEvent
 import ratpack.handling.Chain
 import ratpack.server.RatpackServer
 import ratpack.server.ServerConfig
-import reactor.core.publisher.Mono
 
 /**
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
@@ -34,41 +38,71 @@ import reactor.core.publisher.Mono
 class ApiServer(private val consumerState: ConsumerStateProvider) {
     private val jsonPrinter = JsonFormat.printer()
 
-    fun start(port: Int): Mono<RatpackServer> = Mono.fromCallable {
-        RatpackServer.of { server ->
+    fun start(port: Int): IO<RatpackServer> = IO {
+        RatpackServer.start { server ->
             server.serverConfig(ServerConfig.embedded().port(port))
                     .handlers(this::setupHandlers)
         }
-    }.doOnNext(RatpackServer::start)
+    }
 
     private fun setupHandlers(chain: Chain) {
         chain
                 .get("messages/count") { ctx ->
                     ctx.response.contentType(CONTENT_TEXT)
-                    consumerState.currentState()
-                            .map { it.msgCount.toString() }
-                            .subscribe(ctx.response::send)
+                    val state = consumerState.currentState()
+                    ctx.response.send(state.msgCount.toString())
                 }
 
                 .get("messages/last/key") { ctx ->
                     ctx.response.contentType(CONTENT_JSON)
-                    consumerState.currentState()
-                            .map { it.lastKey }
-                            .map { VesEvent.CommonEventHeader.parseFrom(it) }
-                            .map { jsonPrinter.print(it) }
-                            .subscribe(ctx.response::send)
+                    val state = consumerState.currentState()
+                    val resp = state.lastKey
+                            .map { Try { VesEvent.CommonEventHeader.parseFrom(it) } }
+                            .map(this::protobufToJson)
+                            .getOrElse { "null" }
+                    ctx.response.send(resp)
                 }
 
                 .get("messages/last/value") { ctx ->
                     ctx.response.contentType(CONTENT_JSON)
-                    consumerState.currentState()
-                            .map { it.lastValue }
-                            .map { VesEvent.parseFrom(it) }
-                            .map { jsonPrinter.print(it) }
-                            .subscribe(ctx.response::send)
+                    val state = consumerState.currentState()
+                    val resp = state.lastValue
+                            .map { Try { VesEvent.parseFrom(it) } }
+                            .map(this::protobufToJson)
+                            .getOrElse { "null" }
+                    ctx.response.send(resp)
+                }
+
+                .get("messages/last/hvRanMeasFields") { ctx ->
+                    ctx.response.contentType(CONTENT_JSON)
+                    val state = consumerState.currentState()
+                    val resp = state.lastValue
+                            .flatMap { Try { VesEvent.parseFrom(it) }.toOption() }
+                            .filter { it.commonEventHeader.domain == VesEvent.CommonEventHeader.Domain.HVRANMEAS }
+                            .map { Try { HVRanMeasFields.parseFrom(it.hvRanMeasFields) } }
+                            .map(this::protobufToJson)
+                            .getOrElse { "null" }
+                    ctx.response.send(resp)
+                }
+
+                .delete("messages") { ctx ->
+                    ctx.response.contentType(CONTENT_TEXT)
+                    consumerState.reset()
+                            .unsafeRunAsync {
+                                it.fold(
+                                        { ctx.response.send("NOK") },
+                                        { ctx.response.send("OK") }
+                                )
+                            }
                 }
     }
 
+    private fun protobufToJson(parseResult: Try<MessageOrBuilder>): String =
+            parseResult.fold(
+                    { ex -> "\"Failed to parse protobuf: ${ex.message}\"" },
+                    { jsonPrinter.print(it) })
+
+
     companion object {
         private const val CONTENT_TEXT = "text/plain"
         private const val CONTENT_JSON = "application/json"