Use AAF credentials from stream definition

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

2 files changed, 34 insertions, 30 deletions

diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/kafka.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSenderOptionsFactory.kt
index b16ad109..1c4acf64 100644
--- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/kafka.kt
+++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSenderOptionsFactory.kt
@@ -17,36 +17,32 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.collectors.veshv.impl
+package org.onap.dcae.collectors.veshv.impl.adapters.kafka
 
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.common.config.SaslConfigs
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.security.plain.internals.PlainSaslServer
+import org.jetbrains.annotations.Nullable
 import org.onap.dcae.collectors.veshv.domain.VesMessage
-import org.onap.dcae.collectors.veshv.impl.adapters.kafka.ProtobufSerializer
-import org.onap.dcae.collectors.veshv.impl.adapters.kafka.VesMessageSerializer
-import org.onap.dcaegen2.services.sdk.model.streams.SinkStream
+import org.onap.dcae.collectors.veshv.utils.applyIf
+import org.onap.dcaegen2.services.sdk.model.streams.AafCredentials
 import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
 import org.onap.ves.VesEventOuterClass.CommonEventHeader
-import reactor.kafka.sender.KafkaSender
 import reactor.kafka.sender.SenderOptions
 
+internal object KafkaSenderOptionsFactory {
 
-private const val MAXIMUM_REQUEST_SIZE_MULTIPLIER = 1.2f
-private const val BUFFER_MEMORY_MULTIPLIER = 32
-private const val MINIMUM_BUFFER_MEMORY = 32 * 1024 * 1024
+    private const val MAXIMUM_REQUEST_SIZE_MULTIPLIER = 1.2f
+    private const val BUFFER_MEMORY_MULTIPLIER = 32
+    private const val MINIMUM_BUFFER_MEMORY = 32 * 1024 * 1024
 
-private const val LOGIN_MODULE_CLASS = "org.apache.kafka.common.security.plain.PlainLoginModule"
-private const val USERNAME = "admin"
-private const val PASSWORD = "admin_secret"
-private const val JAAS_CONFIG = "$LOGIN_MODULE_CLASS required username=$USERNAME password=$PASSWORD;"
-private val SASL_PLAINTEXT = (SecurityProtocol.SASL_PLAINTEXT as Enum<SecurityProtocol>).name
+    private const val LOGIN_MODULE_CLASS = "org.apache.kafka.common.security.plain.PlainLoginModule"
+    private val SASL_PLAINTEXT = (SecurityProtocol.SASL_PLAINTEXT as Enum<SecurityProtocol>).name
 
-internal fun createKafkaSender(sinkStream: SinkStream) =
-        (sinkStream as KafkaSink).let { kafkaSink ->
-            KafkaSender.create(SenderOptions.create<CommonEventHeader, VesMessage>()
+    fun createSenderOptions(kafkaSink: KafkaSink): SenderOptions<CommonEventHeader, VesMessage> =
+            SenderOptions.create<CommonEventHeader, VesMessage>()
                     .producerProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaSink.bootstrapServers())
                     .producerProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize(kafkaSink))
                     .producerProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory(kafkaSink))
@@ -55,15 +51,22 @@ internal fun createKafkaSender(sinkStream: SinkStream) =
                     .producerProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1)
                     .producerProperty(ProducerConfig.RETRIES_CONFIG, 1)
                     .producerProperty(ProducerConfig.ACKS_CONFIG, "1")
-                    .producerProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SASL_PLAINTEXT)
-                    .producerProperty(SaslConfigs.SASL_MECHANISM, PlainSaslServer.PLAIN_MECHANISM)
-                    .producerProperty(SaslConfigs.SASL_JAAS_CONFIG, JAAS_CONFIG)
                     .stopOnError(false)
-            )
-        }
+                    .applyIf(kafkaSink.aafCredentials() != null) {
+                        producerProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SASL_PLAINTEXT)
+                                .producerProperty(SaslConfigs.SASL_MECHANISM, PlainSaslServer.PLAIN_MECHANISM)
+                                .producerProperty(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig(kafkaSink.aafCredentials()!!))
+                    }
 
-private fun maxRequestSize(kafkaSink: KafkaSink) =
-        (MAXIMUM_REQUEST_SIZE_MULTIPLIER * kafkaSink.maxPayloadSizeBytes()).toInt()
+    private fun jaasConfig(aafCredentials: AafCredentials) =
+            """$LOGIN_MODULE_CLASS required username="${aafCredentials.username().jaasEscape()}" password="${aafCredentials.password().jaasEscape()}";"""
 
-private fun bufferMemory(kafkaSink: KafkaSink) =
-        Integer.max(MINIMUM_BUFFER_MEMORY, BUFFER_MEMORY_MULTIPLIER * kafkaSink.maxPayloadSizeBytes())
+    private fun String?.jaasEscape() = this?.replace("\"", "\\\"")
+
+    private fun maxRequestSize(kafkaSink: KafkaSink) =
+            (MAXIMUM_REQUEST_SIZE_MULTIPLIER * kafkaSink.maxPayloadSizeBytes()).toInt()
+
+    private fun bufferMemory(kafkaSink: KafkaSink) =
+            Integer.max(MINIMUM_BUFFER_MEMORY, BUFFER_MEMORY_MULTIPLIER * kafkaSink.maxPayloadSizeBytes())
+
+}
diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt
index 2973fa8d..58363a26 100644
--- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt
+++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt
@@ -21,11 +21,11 @@ package org.onap.dcae.collectors.veshv.impl.adapters.kafka
 
 import org.onap.dcae.collectors.veshv.boundary.SinkFactory
 import org.onap.dcae.collectors.veshv.domain.VesMessage
-import org.onap.dcae.collectors.veshv.impl.createKafkaSender
 import org.onap.dcae.collectors.veshv.domain.logging.ClientContext
 import org.onap.dcae.collectors.veshv.domain.logging.ServiceContext
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
 import org.onap.dcaegen2.services.sdk.model.streams.SinkStream
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
 import org.onap.ves.VesEventOuterClass.CommonEventHeader
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
@@ -39,13 +39,11 @@ import java.util.Collections.synchronizedMap
  */
 internal class KafkaSinkFactory : SinkFactory {
     private val messageSinks = synchronizedMap(
-            mutableMapOf<SinkStream, KafkaSender<CommonEventHeader, VesMessage>>()
+            mutableMapOf<KafkaSink, KafkaSender<CommonEventHeader, VesMessage>>()
     )
 
     override fun invoke(stream: SinkStream, ctx: ClientContext) = lazy {
-        messageSinks.computeIfAbsent(stream, ::createKafkaSender).let {
-            KafkaPublisher(it, ctx)
-        }
+        KafkaPublisher(messageSinks.computeIfAbsent(stream as KafkaSink, this::createKafkaSender), ctx)
     }
 
     override fun close(): Mono<Void> =
@@ -57,6 +55,9 @@ internal class KafkaSinkFactory : SinkFactory {
                         logger.info(ServiceContext::mdc) { "Message sinks flushed and closed" }
                     }
 
+    private fun createKafkaSender(stream: KafkaSink) =
+            KafkaSender.create(KafkaSenderOptionsFactory.createSenderOptions(stream))
+
     companion object {
         private val logger = Logger(KafkaSinkFactory::class)
     }