From 67689405071acdad2b26d5112b3662605e474ce9 Mon Sep 17 00:00:00 2001 From: Piotr Jaszczyk Date: Thu, 14 Jun 2018 09:48:46 +0200 Subject: Various improvements * Kotlin upgrade * Monad usage on APIs * Idle timeout * Simulator enhancements Closes ONAP-390 Change-Id: I3c00fcfe38c722caf661ddaad428cf089eeefcaa Signed-off-by: Piotr Jaszczyk Issue-ID: DCAEGEN2-601 --- .../config/ArgBasedDcaeAppSimConfiguration.kt | 8 ++- .../veshv/simulators/dcaeapp/kafka/KafkaSource.kt | 6 +- .../veshv/simulators/dcaeapp/kafka/consumer.kt | 27 ++++++--- .../collectors/veshv/simulators/dcaeapp/main.kt | 30 +++++----- .../veshv/simulators/dcaeapp/remote/ApiServer.kt | 68 ++++++++++++++++------ .../config/ArgBasedDcaeAppSimConfigurationTest.kt | 36 ++++++++---- 6 files changed, 119 insertions(+), 56 deletions(-) (limited to 'hv-collector-dcae-app-simulator/src') 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(DefaultParser()) { @@ -41,8 +43,8 @@ class ArgBasedDcaeAppSimConfiguration : ArgBasedConfiguration) { - fun start(): Mono = Mono.create { sink -> + fun start(): IO = 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, val lastValue: Option) interface ConsumerStateProvider { - fun currentState(): Mono + fun currentState(): ConsumerState + fun reset(): IO } class Consumer : ConsumerStateProvider { @@ -38,18 +41,28 @@ class Consumer : ConsumerStateProvider { private var lastKey: ByteArray? = null private var lastValue: ByteArray? = null - override fun currentState(): Mono = 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) { + 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) { + 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 @@ -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 = Mono.fromCallable { - RatpackServer.of { server -> + fun start(port: Int): IO = 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): 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" diff --git a/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgBasedDcaeAppSimConfigurationTest.kt b/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgBasedDcaeAppSimConfigurationTest.kt index 817df8e0..d99de17b 100644 --- a/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgBasedDcaeAppSimConfigurationTest.kt +++ b/hv-collector-dcae-app-simulator/src/test/kotlin/org/onap/dcae/collectors/veshv/simulators/dcaeapp/config/ArgBasedDcaeAppSimConfigurationTest.kt @@ -19,13 +19,14 @@ */ package org.onap.dcae.collectors.veshv.simulators.dcaeapp.config +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 import org.jetbrains.spek.api.dsl.given import org.jetbrains.spek.api.dsl.it import org.onap.dcae.collectors.veshv.utils.commandline.WrongArgumentException -import kotlin.test.assertFailsWith internal class ArgBasedDcaeAppSimConfigurationTest : Spek({ @@ -38,7 +39,21 @@ internal class ArgBasedDcaeAppSimConfigurationTest : Spek({ cut = ArgBasedDcaeAppSimConfiguration() } - fun parse(vararg cmdLine: String) = cut.parse(cmdLine) + fun parseExpectingSuccess(vararg cmdLine: String): DcaeAppSimConfiguration { + val result = cut.parse(cmdLine) + return when (result) { + is Success -> result.value + is Failure -> throw AssertionError("Parsing result should be present") + } + } + + fun parseExpectingFailure(vararg cmdLine: String): Throwable { + val result = cut.parse(cmdLine) + return when (result) { + is Success -> throw AssertionError("parsing should have failed") + is Failure -> result.exception + } + } describe("parsing arguments") { lateinit var result: DcaeAppSimConfiguration @@ -46,7 +61,7 @@ internal class ArgBasedDcaeAppSimConfigurationTest : Spek({ given("all parameters are present in the long form") { beforeEachTest { - result = parse("--listen-port", "6969", + result = parseExpectingSuccess("--listen-port", "6969", "--kafka-bootstrap-servers", kafkaBootstrapServers, "--kafka-topics", kafkaTopics ) @@ -71,7 +86,9 @@ internal class ArgBasedDcaeAppSimConfigurationTest : Spek({ given("some parameters are present in the short form") { beforeEachTest { - result = parse("-p", "666", "--kafka-bootstrap-servers", kafkaBootstrapServers, "-f", kafkaTopics) + result = parseExpectingSuccess("-p", "666", + "--kafka-bootstrap-servers", kafkaBootstrapServers, + "-f", kafkaTopics) } it("should set proper port") { @@ -92,7 +109,7 @@ internal class ArgBasedDcaeAppSimConfigurationTest : Spek({ given("all optional parameters are absent") { beforeEachTest { - result = parse("-s", kafkaBootstrapServers, "-f", kafkaTopics) + result = parseExpectingSuccess("-s", kafkaBootstrapServers, "-f", kafkaTopics) } it("should set default port") { @@ -100,21 +117,20 @@ internal class ArgBasedDcaeAppSimConfigurationTest : Spek({ } } - describe("required parameter is absent") { given("kafka topics are missing") { it("should throw exception") { - assertFailsWith { parse("-s", kafkaBootstrapServers) } + assertThat(parseExpectingFailure("-s", kafkaBootstrapServers)) + .isInstanceOf(WrongArgumentException::class.java) } } given("kafka bootstrap servers are missing") { it("should throw exception") { - assertFailsWith { parse("-f", kafkaTopics) } + assertThat(parseExpectingFailure("-f", kafkaTopics)) + .isInstanceOf(WrongArgumentException::class.java) } } } } - - }) \ No newline at end of file -- cgit 1.2.3-korg