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

   1 /*
   2  * Copyright © 2018-2019 AT&T Intellectual Property.
   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 kotlinx.coroutines.channels.Channel
  20 import org.apache.kafka.clients.consumer.ConsumerConfig
  21 import org.apache.kafka.streams.KafkaStreams
  22 import org.apache.kafka.streams.StreamsConfig
  23 import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsBasicAuthConsumerProperties
  24 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
  25 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  26 import java.util.*
  27
  28 open class KafkaStreamsBasicAuthConsumerService(private val messageConsumerProperties: KafkaStreamsBasicAuthConsumerProperties)
  29     : BlueprintMessageConsumerService {
  30
  31     val log = logger(KafkaStreamsBasicAuthConsumerService::class)
  32     lateinit var kafkaStreams: KafkaStreams
  33
  34     private fun streamsConfig(additionalConfig: Map<String, Any>? = null): Properties {
  35         val configProperties = Properties()
  36         configProperties[StreamsConfig.APPLICATION_ID_CONFIG] = messageConsumerProperties.applicationId
  37         configProperties[StreamsConfig.BOOTSTRAP_SERVERS_CONFIG] = messageConsumerProperties.bootstrapServers
  38         configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = messageConsumerProperties.autoOffsetReset
  39         configProperties[StreamsConfig.PROCESSING_GUARANTEE_CONFIG] = messageConsumerProperties.processingGuarantee
  40         // TODO("Security Implementation based on type")
  41         /** add or override already set properties */
  42         additionalConfig?.let { configProperties.putAll(it) }
  43         /** Create Kafka consumer */
  44         return configProperties
  45     }
  46
  47     override suspend fun subscribe(additionalConfig: Map<String, Any>?): Channel<String> {
  48         throw BluePrintProcessorException("not implemented")
  49     }
  50
  51     override suspend fun subscribe(topics: List<String>, additionalConfig: Map<String, Any>?): Channel<String> {
  52         throw BluePrintProcessorException("not implemented")
  53     }
  54
  55     override suspend fun consume(additionalConfig: Map<String, Any>?, consumerFunction: ConsumerFunction) {
  56         val streamsConfig = streamsConfig(additionalConfig)
  57         val kafkaStreamConsumerFunction = consumerFunction as KafkaStreamConsumerFunction
  58         val topology = kafkaStreamConsumerFunction.createTopology(messageConsumerProperties, additionalConfig)
  59         kafkaStreams = KafkaStreams(topology, streamsConfig)
  60         kafkaStreams.cleanUp()
  61         kafkaStreams.start()
  62         kafkaStreams.localThreadsMetadata().forEach { data -> log.info("Topology : $data") }
  63     }
  64
  65     override suspend fun shutDown() {
  66         if (kafkaStreams != null) {
  67             kafkaStreams.close()
  68         }
  69     }
  70 }