Add environment configuration to kafka consumer

- HV-VES-specific environment prefix moved inside HvVes modules to
allow simpler no-prefix API for other modules
- created OptionDSL for brevity

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

6 files changed, 174 insertions, 125 deletions

diff --git a/sources/hv-collector-commandline/src/main/kotlin/org/onap/dcae/collectors/veshv/commandline/CommandLineOption.kt b/sources/hv-collector-commandline/src/main/kotlin/org/onap/dcae/collectors/veshv/commandline/CommandLineOption.kt
index d08f6c09..9d875571 100644
--- a/sources/hv-collector-commandline/src/main/kotlin/org/onap/dcae/collectors/veshv/commandline/CommandLineOption.kt
+++ b/sources/hv-collector-commandline/src/main/kotlin/org/onap/dcae/collectors/veshv/commandline/CommandLineOption.kt
@@ -23,109 +23,118 @@ import org.apache.commons.cli.Option
 
 
 enum class CommandLineOption(val option: Option, val required: Boolean = false) {
-    HEALTH_CHECK_API_PORT(
-            Option.builder("H")
-                    .longOpt("health-check-api-port")
-                    .hasArg()
-                    .desc("Health check rest api listen port")
-                    .build()
-    ),
-    CONFIGURATION_FILE(
-            Option.builder("c")
-                    .longOpt("configuration-file")
-                    .hasArg()
-                    .desc("Json file containing HV-VES configuration")
-                    .build(),
-            required = true
-    ),
-    LISTEN_PORT(
-            Option.builder("p")
-                    .longOpt("listen-port")
-                    .hasArg()
-                    .desc("Listen port")
-                    .build(),
-            required = true
-    ),
-    VES_HV_PORT(
-            Option.builder("v")
-                    .longOpt("ves-port")
-                    .hasArg()
-                    .desc("VesHvCollector port")
-                    .build(),
-            required = true
-    ),
-    VES_HV_HOST(
-            Option.builder("h")
-                    .longOpt("ves-host")
-                    .hasArg()
-                    .desc("VesHvCollector host")
-                    .build(),
-            required = true
-    ),
-    KAFKA_SERVERS(
-            Option.builder("s")
-                    .longOpt("kafka-bootstrap-servers")
-                    .hasArg()
-                    .desc("Comma-separated Kafka bootstrap servers in <host>:<port> format")
-                    .build(),
-            required = true
-    ),
-    KAFKA_TOPICS(
-            Option.builder("f")
-                    .longOpt("kafka-topics")
-                    .hasArg()
-                    .desc("Comma-separated Kafka topics")
-                    .build(),
-            required = true
-    ),
-    SSL_DISABLE(
-            Option.builder("l")
-                    .longOpt("ssl-disable")
-                    .desc("Disable SSL encryption")
-                    .build()
-    ),
-    KEY_STORE_FILE(
-            Option.builder("k")
-                    .longOpt("key-store")
-                    .hasArg()
-                    .desc("Key store in PKCS12 format")
-                    .build()
-    ),
-    KEY_STORE_PASSWORD_FILE(
-            Option.builder("kp")
-                    .longOpt("key-store-password-file")
-                    .hasArg()
-                    .desc("File with key store password")
-                    .build()
-    ),
-    TRUST_STORE_FILE(
-            Option.builder("t")
-                    .longOpt("trust-store")
-                    .hasArg()
-                    .desc("File with trusted certificate bundle in PKCS12 format")
-                    .build()
-    ),
-    TRUST_STORE_PASSWORD_FILE(
-            Option.builder("tp")
-                    .longOpt("trust-store-password-file")
-                    .hasArg()
-                    .desc("File with trust store password")
-                    .build()
-    ),
-    MAXIMUM_PAYLOAD_SIZE_BYTES(
-            Option.builder("m")
-                    .longOpt("max-payload-size")
-                    .hasArg()
-                    .desc("Maximum supported payload size in bytes")
-                    .build()
-    );
+    CONFIGURATION_FILE(required = true,
+            option = option {
+                shortOpt = "c"
+                longOpt = "configuration-file"
+                desc = "Json file containing HV-VES configuration"
+                hasArgument = true
+            }),
+    LISTEN_PORT(required = true,
+            option = option {
+                shortOpt = "p"
+                longOpt = "listen-port"
+                desc = "Listen port"
+                hasArgument = true
+            }),
+    VES_HV_PORT(required = true,
+            option = option {
+                shortOpt = "v"
+                longOpt = "ves-port"
+                desc = "VesHvCollector port"
+                hasArgument = true
+            }),
+    VES_HV_HOST(required = true,
+            option = option {
+                shortOpt = "h"
+                longOpt = "ves-host"
+                desc = "VesHvCollector host"
+                hasArgument = true
+            }),
+    KAFKA_SERVERS(required = true,
+            option = option {
+                shortOpt = "s"
+                longOpt = "kafka-bootstrap-servers"
+                desc = "Comma-separated Kafka bootstrap servers in <host>:<port> format"
+                hasArgument = true
+            }),
+    KAFKA_TOPICS(required = true,
+            option = option {
+                shortOpt = "f"
+                longOpt = "kafka-topics"
+                desc = "Comma-separated Kafka topics"
+                hasArgument = true
+            }),
+    HEALTH_CHECK_API_PORT(option {
+        shortOpt = "H"
+        longOpt = "health-check-api-port"
+        desc = "Health check rest api listen port"
+        hasArgument = true
+    }),
+    SSL_DISABLE(option {
+        shortOpt = "l"
+        longOpt = "ssl-disable"
+        desc = "Disable SSL encryption"
+    }),
+    KEY_STORE_FILE(option {
+        shortOpt = "k"
+        longOpt = "key-store"
+        desc = "Key store in PKCS12 format"
+        hasArgument = true
+    }),
+    KEY_STORE_PASSWORD_FILE(option {
+        shortOpt = "kp"
+        longOpt = "key-store-password-file"
+        desc = "File with key store password"
+        hasArgument = true
+    }),
+    TRUST_STORE_FILE(option {
+        shortOpt = "t"
+        longOpt = "trust-store"
+        desc = "File with trusted certificate bundle in PKCS12 format"
+        hasArgument = true
+    }),
+    TRUST_STORE_PASSWORD_FILE(option {
+        shortOpt = "tp"
+        longOpt = "trust-store-password-file"
+        desc = "File with trust store password"
+        hasArgument = true
+    }),
+    MAXIMUM_PAYLOAD_SIZE_BYTES(option {
+        shortOpt = "m"
+        longOpt = "max-payload-size"
+        desc = "Maximum supported payload size in bytes"
+        hasArgument = true
+    }),
+    DISABLE_PROCESSING(option {
+        shortOpt = "d"
+        longOpt = "disable-processing"
+        desc = "Message queue consumer option. Indicates whether messages should be fully processed"
+    });
 
-    fun environmentVariableName(prefix: String = DEFAULT_ENV_PREFIX): String =
+    fun environmentVariableName(prefix: String = ""): String =
             option.longOpt.toUpperCase().replace('-', '_').let { mainPart ->
-                "${prefix}_${mainPart}"
+                if (prefix.isNotBlank()) {
+                    "${prefix}_${mainPart}"
+                } else {
+                    mainPart
+                }
             }
+}
+
 
-    companion object {
-        private const val DEFAULT_ENV_PREFIX = "VESHV"
-    }
+private class OptionDSL {
+    lateinit var shortOpt: String
+    lateinit var longOpt: String
+    lateinit var desc: String
+    var hasArgument: Boolean = false
 }
+
+private fun option(conf: OptionDSL.() -> Unit): Option {
+    val dsl = OptionDSL().apply(conf)
+    return Option.builder(dsl.shortOpt)
+            .longOpt(dsl.longOpt)
+            .hasArg(dsl.hasArgument)
+            .desc(dsl.desc)
+            .build()
+}
\ No newline at end of file
diff --git a/sources/hv-collector-commandline/src/main/kotlin/org/onap/dcae/collectors/veshv/commandline/extensions.kt b/sources/hv-collector-commandline/src/main/kotlin/org/onap/dcae/collectors/veshv/commandline/extensions.kt
index 6d8ba3ff..20ca97ed 100644
--- a/sources/hv-collector-commandline/src/main/kotlin/org/onap/dcae/collectors/veshv/commandline/extensions.kt
+++ b/sources/hv-collector-commandline/src/main/kotlin/org/onap/dcae/collectors/veshv/commandline/extensions.kt
@@ -40,28 +40,43 @@ fun handleWrongArgumentError(programName: String, err: WrongArgumentError): Exit
     return ExitFailure(2)
 }
 
-fun CommandLine.longValue(cmdLineOpt: CommandLineOption, default: Long): Long =
-        longValue(cmdLineOpt).getOrElse { default }
+inline class EnvPrefix(val it: String)
 
-fun CommandLine.stringValue(cmdLineOpt: CommandLineOption, default: String): String =
-        optionValue(cmdLineOpt).getOrElse { default }
+private val DEFAULT_PREFIX = EnvPrefix("")
 
-fun CommandLine.intValue(cmdLineOpt: CommandLineOption, default: Int): Int =
-        intValue(cmdLineOpt).getOrElse { default }
+fun CommandLine.longValue(cmdLineOpt: CommandLineOption,
+                          default: Long,
+                          envPrefix: EnvPrefix = DEFAULT_PREFIX): Long =
+        longValue(cmdLineOpt, envPrefix).getOrElse { default }
 
-fun CommandLine.intValue(cmdLineOpt: CommandLineOption): Option<Int> =
-        optionValue(cmdLineOpt).map(String::toInt)
+fun CommandLine.stringValue(cmdLineOpt: CommandLineOption,
+                            default: String,
+                            envPrefix: EnvPrefix = DEFAULT_PREFIX): String =
+        optionValue(cmdLineOpt, envPrefix).getOrElse { default }
 
-fun CommandLine.longValue(cmdLineOpt: CommandLineOption): Option<Long> =
-        optionValue(cmdLineOpt).map(String::toLong)
+fun CommandLine.intValue(cmdLineOpt: CommandLineOption,
+                         default: Int,
+                         envPrefix: EnvPrefix = DEFAULT_PREFIX): Int =
+        intValue(cmdLineOpt, envPrefix).getOrElse { default }
 
-fun CommandLine.stringValue(cmdLineOpt: CommandLineOption): Option<String> =
-        optionValue(cmdLineOpt)
+fun CommandLine.intValue(cmdLineOpt: CommandLineOption,
+                         envPrefix: EnvPrefix = DEFAULT_PREFIX): Option<Int> =
+        optionValue(cmdLineOpt, envPrefix).map(String::toInt)
 
-fun CommandLine.hasOption(cmdLineOpt: CommandLineOption): Boolean =
+fun CommandLine.longValue(cmdLineOpt: CommandLineOption,
+                          envPrefix: EnvPrefix = DEFAULT_PREFIX): Option<Long> =
+        optionValue(cmdLineOpt, envPrefix).map(String::toLong)
+
+fun CommandLine.stringValue(cmdLineOpt: CommandLineOption,
+                            envPrefix: EnvPrefix = DEFAULT_PREFIX): Option<String> =
+        optionValue(cmdLineOpt, envPrefix)
+
+fun CommandLine.hasOption(cmdLineOpt: CommandLineOption,
+                          envPrefix: EnvPrefix = DEFAULT_PREFIX): Boolean =
         this.hasOption(cmdLineOpt.option.opt) ||
-                System.getenv(cmdLineOpt.environmentVariableName()) != null
+                System.getenv(cmdLineOpt.environmentVariableName(envPrefix.it)) != null
 
-private fun CommandLine.optionValue(cmdLineOpt: CommandLineOption) = Option.fromNullablesChain(
+private fun CommandLine.optionValue(cmdLineOpt: CommandLineOption, envPrefix: EnvPrefix) = Option.fromNullablesChain(
         getOptionValue(cmdLineOpt.option.opt),
-        { System.getenv(cmdLineOpt.environmentVariableName()) })
+        { System.getenv(cmdLineOpt.environmentVariableName(envPrefix.it)) })
+
diff --git a/sources/hv-collector-commandline/src/test/kotlin/org/onap/dcae/collectors/veshv/commandline/CommandLineOptionTest.kt b/sources/hv-collector-commandline/src/test/kotlin/org/onap/dcae/collectors/veshv/commandline/CommandLineOptionTest.kt
index 6614e77f..e6776974 100644
--- a/sources/hv-collector-commandline/src/test/kotlin/org/onap/dcae/collectors/veshv/commandline/CommandLineOptionTest.kt
+++ b/sources/hv-collector-commandline/src/test/kotlin/org/onap/dcae/collectors/veshv/commandline/CommandLineOptionTest.kt
@@ -53,8 +53,8 @@ class CommandLineOptionTest : Spek({
                 on("calling environmentVariableName") {
                     val result = opt.environmentVariableName()
 
-                    it("should return prefixed upper snake cased long option name") {
-                        assertThat(result).isEqualTo("VESHV_SSL_DISABLE")
+                    it("should return upper snake cased long option name without prefix") {
+                        assertThat(result).isEqualTo("SSL_DISABLE")
                     }
                 }
             }
diff --git a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/HvVesCommandLineParser.kt b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/HvVesCommandLineParser.kt
index c6730a4c..27560642 100644
--- a/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/HvVesCommandLineParser.kt
+++ b/sources/hv-collector-configuration/src/main/kotlin/org/onap/dcae/collectors/veshv/config/impl/HvVesCommandLineParser.kt
@@ -29,6 +29,7 @@ import org.apache.commons.cli.DefaultParser
 import org.apache.commons.cli.Options
 import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.CONFIGURATION_FILE
 import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.HEALTH_CHECK_API_PORT
+import org.onap.dcae.collectors.veshv.commandline.EnvPrefix
 import org.onap.dcae.collectors.veshv.commandline.WrongArgumentError
 import org.onap.dcae.collectors.veshv.commandline.intValue
 import org.onap.dcae.collectors.veshv.commandline.stringValue
@@ -42,7 +43,7 @@ internal class HvVesCommandLineParser(private val parser: CommandLineParser = De
 
     fun getConfigurationFile(args: Array<out String>): Either<WrongArgumentError, File> =
             parse(args) {
-                it.stringValue(CONFIGURATION_FILE).map(::File)
+                it.stringValue(CONFIGURATION_FILE, HV_VES_ENV_PREFIX).map(::File)
             }.toEither {
                 WrongArgumentError(
                         message = "Base configuration filepath missing on command line",
@@ -51,7 +52,7 @@ internal class HvVesCommandLineParser(private val parser: CommandLineParser = De
 
     fun getHealthcheckPort(args: Array<out String>): Int =
             parse(args) {
-                it.intValue(HEALTH_CHECK_API_PORT)
+                it.intValue(HEALTH_CHECK_API_PORT, HV_VES_ENV_PREFIX)
             }.getOrElse {
                 logger.info { "Healthcheck port missing on command line, using default: $DEFAULT_HEALTHCHECK_PORT" }
                 DEFAULT_HEALTHCHECK_PORT
@@ -76,6 +77,7 @@ internal class HvVesCommandLineParser(private val parser: CommandLineParser = De
                     .let { parser.parse(it, args) }
 
     companion object {
+        private val HV_VES_ENV_PREFIX = EnvPrefix("VESHV")
         private const val DEFAULT_HEALTHCHECK_PORT: Int = 6060
         private val logger = Logger(HvVesCommandLineParser::class)
     }
diff --git a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/config/ArgKafkaConsumerConfiguration.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/config/ArgKafkaConsumerConfiguration.kt
index 4d65e916..be7b5cca 100644
--- a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/config/ArgKafkaConsumerConfiguration.kt
+++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/config/ArgKafkaConsumerConfiguration.kt
@@ -24,17 +24,35 @@ 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
-import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.LISTEN_PORT
+import org.onap.dcae.collectors.veshv.commandline.CommandLineOption.*
+import org.onap.dcae.collectors.veshv.commandline.hasOption
 import org.onap.dcae.collectors.veshv.commandline.intValue
+import org.onap.dcae.collectors.veshv.commandline.stringValue
 import org.onap.dcae.collectors.veshv.utils.arrow.OptionUtils.binding
 import java.net.InetSocketAddress
 
 internal class ArgKafkaConsumerConfiguration : ArgBasedConfiguration<KafkaConsumerConfiguration>(DefaultParser()) {
-    override val cmdLineOptionsList: List<CommandLineOption> = listOf(LISTEN_PORT)
+    override val cmdLineOptionsList: List<CommandLineOption> = listOf(
+            LISTEN_PORT,
+            KAFKA_TOPICS,
+            KAFKA_SERVERS,
+            DISABLE_PROCESSING
+    )
 
     override fun getConfiguration(cmdLine: CommandLine): Option<KafkaConsumerConfiguration> =
             binding {
                 val listenPort = cmdLine.intValue(LISTEN_PORT).bind()
-                KafkaConsumerConfiguration(InetSocketAddress(listenPort))
+                val kafkaTopics = cmdLine.stringValue(KAFKA_TOPICS)
+                        .map { it.split(',').toSet() }
+                        .bind()
+                val kafkaBootstrapServers = cmdLine.stringValue(KAFKA_SERVERS).bind()
+                val disableProcessing = cmdLine.hasOption(DISABLE_PROCESSING)
+
+                KafkaConsumerConfiguration(
+                        InetSocketAddress(listenPort),
+                        kafkaTopics,
+                        kafkaBootstrapServers,
+                        disableProcessing
+                )
             }
 }
diff --git a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/config/KafkaConsumerConfiguration.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/config/KafkaConsumerConfiguration.kt
index ef06a0d1..cdd4c30a 100644
--- a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/config/KafkaConsumerConfiguration.kt
+++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/config/KafkaConsumerConfiguration.kt
@@ -21,4 +21,9 @@ package org.onap.dcae.collectors.veshv.kafkaconsumer.config
 
 import java.net.InetSocketAddress
 
-internal data class KafkaConsumerConfiguration(val apiAddress: InetSocketAddress)
+internal data class KafkaConsumerConfiguration(
+        val apiAddress: InetSocketAddress,
+        val kafkaTopics: Set<String>,
+        val kafkaBootstrapServers: String,
+        val disableProcessing: Boolean
+)