2 * Copyright © 2019 IBM.
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
8 * http://www.apache.org/licenses/LICENSE-2.0
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.
17 package org.onap.ccsdk.cds.blueprintsprocessor.message.service
19 import kotlinx.coroutines.channels.Channel
20 import kotlinx.coroutines.delay
21 import kotlinx.coroutines.launch
22 import kotlinx.coroutines.runBlocking
23 import org.apache.kafka.clients.CommonClientConfigs
24 import org.apache.kafka.clients.consumer.Consumer
25 import org.apache.kafka.clients.consumer.ConsumerConfig
26 import org.apache.kafka.clients.consumer.KafkaConsumer
27 import org.apache.kafka.common.serialization.StringDeserializer
28 import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageConsumerProperties
29 import org.onap.ccsdk.cds.controllerblueprints.core.logger
30 import java.time.Duration
31 import kotlin.concurrent.thread
33 class KafkaBasicAuthMessageConsumerService(
34 private val messageConsumerProperties: KafkaBasicAuthMessageConsumerProperties)
35 : BlueprintMessageConsumerService {
37 private val channel = Channel<String>()
38 private var kafkaConsumer: Consumer<String, String>? = null
39 val log = logger(KafkaBasicAuthMessageConsumerService::class)
44 fun kafkaConsumer(additionalConfig: Map<String, Any>? = null): Consumer<String, String> {
45 val configProperties = hashMapOf<String, Any>()
46 configProperties[CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG] = messageConsumerProperties.bootstrapServers
47 configProperties[ConsumerConfig.GROUP_ID_CONFIG] = messageConsumerProperties.groupId
48 configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "latest"
49 configProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
50 configProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
51 if (messageConsumerProperties.clientId != null) {
52 configProperties[ConsumerConfig.CLIENT_ID_CONFIG] = messageConsumerProperties.clientId!!
54 // TODO("Security Implementation based on type")
55 /** add or override already set properties */
56 additionalConfig?.let { configProperties.putAll(it) }
57 /** Create Kafka consumer */
58 return KafkaConsumer(configProperties)
61 override suspend fun subscribe(additionalConfig: Map<String, Any>?): Channel<String> {
62 /** get to topic names */
63 val consumerTopic = messageConsumerProperties.topic?.split(",")?.map { it.trim() }
64 check(!consumerTopic.isNullOrEmpty()) { "couldn't get topic information" }
65 return subscribe(consumerTopic, additionalConfig)
69 override suspend fun subscribe(consumerTopic: List<String>, additionalConfig: Map<String, Any>?): Channel<String> {
70 /** Create Kafka consumer */
71 kafkaConsumer = kafkaConsumer(additionalConfig)
73 checkNotNull(kafkaConsumer) {
74 "failed to create kafka consumer for " +
75 "server(${messageConsumerProperties.bootstrapServers})'s " +
76 "topics(${messageConsumerProperties.bootstrapServers})"
79 kafkaConsumer!!.subscribe(consumerTopic)
80 log.info("Successfully consumed topic($consumerTopic)")
82 thread(start = true, name = "KafkaConsumer") {
84 kafkaConsumer!!.use { kc ->
86 val consumerRecords = kc.poll(Duration.ofMillis(messageConsumerProperties.pollMillSec))
88 consumerRecords?.forEach { consumerRecord ->
89 /** execute the command block */
90 consumerRecord.value()?.let {
92 if (!channel.isClosedForSend) {
95 log.error("Channel is closed to receive message")
102 log.info("message listener shutting down.....")
108 override suspend fun shutDown() {
109 /** stop the polling loop */
111 /** Close the Channel */
113 /** TO shutdown gracefully, need to wait for the maximum poll time */
114 delay(messageConsumerProperties.pollMillSec)