DCAE APP simulator rework

- Extract message parameters parsing logic to standalone class in utils
- Make DCAE APP simulator store whole received messages history
- Add validation endpoint
- Add new messege type: FIXED_PAYLOAD

Closes ONAP-686

Change-Id: I865804716ad5e46a7503a8eee70cfe9ac75a6c1e
Signed-off-by: Jakub Dudycz <jakub.dudycz@nokia.com>
Issue-ID: DCAEGEN2-601

3 files changed, 111 insertions, 79 deletions

diff --git a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgDcaeAppSimConfiguration.kt b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgDcaeAppSimConfiguration.kt
index 264033e3..065cdf92 100644
--- a/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgDcaeAppSimConfiguration.kt
+++ b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgDcaeAppSimConfiguration.kt
@@ -28,7 +28,9 @@ import org.apache.commons.cli.CommandLine
 import org.apache.commons.cli.DefaultParser
 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.*
+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
 
 class ArgDcaeAppSimConfiguration : ArgBasedConfiguration<DcaeAppSimConfiguration>(DefaultParser()) {
     override val cmdLineOptionsList: List<CommandLineOption> = listOf(
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 869c5ab6..08bb149f 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,17 +19,24 @@
  */
 package org.onap.dcae.collectors.veshv.simulators.dcaeapp.kafka
 
-import arrow.core.Option
 import arrow.effects.IO
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
 import reactor.kafka.receiver.ReceiverRecord
+import java.util.concurrent.ConcurrentLinkedQueue
 
 /**
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  * @since June 2018
  */
+class ConsumerState(private val messages: ConcurrentLinkedQueue<ByteArray>){
+    val messagesCount: Int by lazy {
+        messages.size
+    }
 
-class ConsumerState(val msgCount: Long, val lastKey: Option<ByteArray>, val lastValue: Option<ByteArray>)
+    val consumedMessages: List<ByteArray> by lazy {
+        messages.toList()
+    }
+}
 
 interface ConsumerStateProvider {
     fun currentState(): ConsumerState
@@ -37,31 +44,21 @@ interface ConsumerStateProvider {
 }
 
 class Consumer : ConsumerStateProvider {
-    private var msgCount = 0L
-    private var lastKey: ByteArray? = null
-    private var lastValue: ByteArray? = null
 
-    override fun currentState() =
-            ConsumerState(msgCount, Option.fromNullable(lastKey), Option.fromNullable(lastValue))
+    private var consumedMessages: ConcurrentLinkedQueue<ByteArray> = ConcurrentLinkedQueue()
 
-    override fun reset() = IO {
-        synchronized(this) {
-            msgCount = 0
-            lastKey = null
-            lastValue = null
-        }
+    override fun currentState(): ConsumerState = ConsumerState(consumedMessages)
+
+    override fun reset(): IO<Unit> = IO {
+        consumedMessages.clear()
     }
 
     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()
-        }
+        consumedMessages.add(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/remote/ApiServer.kt b/hv-collector-dcae-app-simulator/src/main/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/remote/ApiServer.kt
index d1d90b00..cb1484b7 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,28 +19,31 @@
  */
 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.domain.PayloadWireFrameMessage
 import org.onap.dcae.collectors.veshv.simulators.dcaeapp.kafka.ConsumerFactory
 import org.onap.dcae.collectors.veshv.simulators.dcaeapp.kafka.ConsumerStateProvider
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import org.onap.ves.HVRanMeasFieldsV5.HVRanMeasFields
+import org.onap.dcae.collectors.veshv.utils.messages.MessageParametersParser
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageGenerator
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageParameters
+import org.onap.dcae.collectors.veshv.ves.message.generator.api.MessageType.FIXED_PAYLOAD
 import org.onap.ves.VesEventV5.VesEvent
 import ratpack.handling.Chain
+import ratpack.handling.Context
 import ratpack.server.RatpackServer
 import ratpack.server.ServerConfig
+import reactor.core.publisher.Mono
+import javax.json.Json
 
 /**
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  * @since May 2018
  */
-class ApiServer(private val consumerFactory: ConsumerFactory) {
+class ApiServer(private val consumerFactory: ConsumerFactory,
+                private val messageParametersParser: MessageParametersParser = MessageParametersParser()) {
 
     private lateinit var consumerState: ConsumerStateProvider
-    private val jsonPrinter = JsonFormat.printer()
 
     fun start(port: Int, kafkaTopics: Set<String>): IO<RatpackServer> = IO {
         consumerState = consumerFactory.createConsumerForTopics(kafkaTopics)
@@ -57,60 +60,92 @@ class ApiServer(private val consumerFactory: ConsumerFactory) {
                         val topics = extractTopics(it.text)
                         logger.info("Received new configuration. Creating consumer for topics: $topics")
                         consumerState = consumerFactory.createConsumerForTopics(topics)
-                        ctx.response.contentType(CONTENT_TEXT)
-                        ctx.response.send("OK")
+                        ctx.response
+                                .status(STATUS_OK)
+                                .send()
                     }
 
                 }
-
-                .get("messages/count") { ctx ->
-                    ctx.response.contentType(CONTENT_TEXT)
-                    val state = consumerState.currentState()
-                    ctx.response.send(state.msgCount.toString())
-                }
-
-                .get("messages/last/key") { ctx ->
-                    ctx.response.contentType(CONTENT_JSON)
-                    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)
-                    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") }
-                                )
+                                        { ctx.response.status(STATUS_INTERNAL_SERVER_ERROR) },
+                                        { ctx.response.status(STATUS_OK) }
+                                ).send()
                             }
                 }
+                .get("messages/all/count") { ctx ->
+                    val state = consumerState.currentState()
+                    ctx.response
+                            .contentType(CONTENT_TEXT)
+                            .send(state.messagesCount.toString())
+                }
+                .post("messages/all/validate") { ctx ->
+                    ctx.request.body
+                            .map { Json.createReader(it.inputStream).readArray() }
+                            .map { messageParametersParser.parse(it) }
+                            .map { generateEvents(ctx, it) }
+                            .then { (generatedEvents, shouldValidatePayloads) ->
+                                generatedEvents
+                                        .doOnSuccess { sendResponse(ctx, it, shouldValidatePayloads) }
+                                        .block()
+                            }
+                }
+    }
+
+    private fun generateEvents(ctx: Context, parameters: List<MessageParameters>):
+            Pair<Mono<List<VesEvent>>, Boolean> = Pair(
+
+            doGenerateEvents(parameters).doOnError {
+                logger.error("Error occurred when generating messages: $it")
+                ctx.response
+                        .status(STATUS_INTERNAL_SERVER_ERROR)
+                        .send()
+            },
+            parameters.all { it.messageType == FIXED_PAYLOAD }
+    )
+
+    private fun doGenerateEvents(parameters: List<MessageParameters>): Mono<List<VesEvent>> = MessageGenerator.INSTANCE
+            .createMessageFlux(parameters)
+            .map(PayloadWireFrameMessage::payload)
+            .map { decode(it.unsafeAsArray()) }
+            .collectList()
+
+
+    private fun decode(bytes: ByteArray): VesEvent = VesEvent.parseFrom(bytes)
+
+
+    private fun sendResponse(ctx: Context,
+                             generatedEvents: List<VesEvent>,
+                             shouldValidatePayloads: Boolean) =
+            resolveResponseStatusCode(
+                    generated = generatedEvents,
+                    consumed = decodeConsumedEvents(),
+                    validatePayloads = shouldValidatePayloads
+            ).let { ctx.response.status(it).send() }
+
+
+    private fun decodeConsumedEvents(): List<VesEvent> = consumerState
+            .currentState()
+            .consumedMessages
+            .map(::decode)
+
+
+    private fun resolveResponseStatusCode(generated: List<VesEvent>,
+                                          consumed: List<VesEvent>,
+                                          validatePayloads: Boolean): Int =
+            if (validatePayloads) {
+                if (generated == consumed) STATUS_OK else STATUS_BAD_REQUEST
+            } else {
+                validateHeaders(consumed, generated)
+            }
+
+    private fun validateHeaders(consumed: List<VesEvent>, generated: List<VesEvent>): Int {
+        val consumedHeaders = consumed.map { it.commonEventHeader }
+        val generatedHeaders = generated.map { it.commonEventHeader }
+        return if (generatedHeaders == consumedHeaders) STATUS_OK else STATUS_BAD_REQUEST
     }
 
     private fun extractTopics(it: String): Set<String> =
@@ -118,16 +153,14 @@ class ApiServer(private val consumerFactory: ConsumerFactory) {
                     .split(",")
                     .toSet()
 
-    private fun protobufToJson(parseResult: Try<MessageOrBuilder>): String =
-            parseResult.fold(
-                    { ex -> "\"Failed to parse protobuf: ${ex.message}\"" },
-                    { jsonPrinter.print(it) })
-
-
     companion object {
         private val logger = Logger(ApiServer::class)
-
         private const val CONTENT_TEXT = "text/plain"
-        private const val CONTENT_JSON = "application/json"
+
+        private const val STATUS_OK = 200
+        private const val STATUS_BAD_REQUEST = 400
+        private const val STATUS_INTERNAL_SERVER_ERROR = 500
     }
 }
+
+