ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaBasicAuthMessageProducerService.kt

   1 /*
   2  *  Copyright © 2019 IBM.
   3  *
   4  *  Licensed under the Apache License, Version 2.0 (the "License");
   5  *  you may not use this file except in compliance with the License.
   6  *  You may obtain a copy of the License at
   7  *
   8  *      http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  *  Unless required by applicable law or agreed to in writing, software
  11  *  distributed under the License is distributed on an "AS IS" BASIS,
  12  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  *  See the License for the specific language governing permissions and
  14  *  limitations under the License.
  15  */
  16
  17 package org.onap.ccsdk.cds.blueprintsprocessor.message.service
  18
  19 import org.apache.commons.lang.builder.ToStringBuilder
  20 import org.apache.kafka.clients.producer.ProducerConfig.*
  21 import org.apache.kafka.common.serialization.StringSerializer
  22 import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageProducerProperties
  23 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonType
  24 import org.slf4j.LoggerFactory
  25 import org.springframework.kafka.core.DefaultKafkaProducerFactory
  26 import org.springframework.kafka.core.KafkaTemplate
  27 import org.springframework.kafka.core.ProducerFactory
  28 import org.springframework.kafka.support.SendResult
  29 import org.springframework.util.concurrent.ListenableFutureCallback
  30
  31 class KafkaBasicAuthMessageProducerService(
  32         private val messageProducerProperties: KafkaBasicAuthMessageProducerProperties)
  33     : BlueprintMessageProducerService {
  34
  35     private val log = LoggerFactory.getLogger(KafkaBasicAuthMessageProducerService::class.java)!!
  36
  37     private var kafkaTemplate: KafkaTemplate<String, Any>? = null
  38
  39     override suspend fun sendMessageNB(message: Any): Boolean {
  40         checkNotNull(messageProducerProperties.topic) { "default topic is not configured" }
  41         return sendMessage(messageProducerProperties.topic!!, message)
  42     }
  43
  44     override suspend fun sendMessageNB(topic: String, message: Any): Boolean {
  45         val serializedMessage = when (message) {
  46             is String -> {
  47                 message
  48             }
  49             else -> {
  50                 message.asJsonType().toString()
  51             }
  52         }
  53         val future = messageTemplate().send(topic, serializedMessage)
  54
  55         future.addCallback(object : ListenableFutureCallback<SendResult<String, Any>> {
  56             override fun onSuccess(result: SendResult<String, Any>) {
  57                 log.info("message sent successfully with offset=[${result.recordMetadata.offset()}]")
  58             }
  59
  60             override fun onFailure(ex: Throwable) {
  61                 log.error("Unable to send message", ex)
  62             }
  63         })
  64         return true
  65     }
  66
  67     private fun producerFactory(additionalConfig: Map<String, Any>? = null): ProducerFactory<String, Any> {
  68         log.info("Client Properties : ${ToStringBuilder.reflectionToString(messageProducerProperties)}")
  69         val configProps = hashMapOf<String, Any>()
  70         configProps[BOOTSTRAP_SERVERS_CONFIG] = messageProducerProperties.bootstrapServers
  71         configProps[KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java
  72         configProps[VALUE_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java
  73         if (messageProducerProperties.clientId != null) {
  74             configProps[CLIENT_ID_CONFIG] = messageProducerProperties.clientId!!
  75         }
  76         // TODO("Security Implementation based on type")
  77
  78         // Add additional Properties
  79         if (additionalConfig != null) {
  80             configProps.putAll(additionalConfig)
  81         }
  82         return DefaultKafkaProducerFactory(configProps)
  83     }
  84
  85     fun messageTemplate(additionalConfig: Map<String, Any>? = null): KafkaTemplate<String, Any> {
  86         if (kafkaTemplate == null) {
  87             kafkaTemplate = KafkaTemplate(producerFactory(additionalConfig))
  88         }
  89         return kafkaTemplate!!
  90     }
  91 }
  92