package org.onap.ccsdk.cds.blueprintsprocessor.message
+import com.fasterxml.jackson.databind.JsonNode
+import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService
+import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BlueprintMessageConsumerService
+import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BlueprintMessageProducerService
+import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.context.annotation.ComponentScan
import org.springframework.context.annotation.Configuration
@EnableConfigurationProperties
open class BluePrintMessageLibConfiguration
+/**
+ * Exposed Dependency Service by this Message Lib Module
+ */
+fun BluePrintDependencyService.messageLibPropertyService(): BluePrintMessageLibPropertyService =
+ instance(MessageLibConstants.SERVICE_BLUEPRINT_MESSAGE_LIB_PROPERTY)
+
+/** Extension functions for message producer service **/
+fun BluePrintDependencyService.messageProducerService(selector: String): BlueprintMessageProducerService {
+ return messageLibPropertyService().blueprintMessageProducerService(selector)
+}
+
+
+fun BluePrintDependencyService.messageProducerService(jsonNode: JsonNode): BlueprintMessageProducerService {
+ return messageLibPropertyService().blueprintMessageProducerService(jsonNode)
+}
+
+/** Extension functions for message consumer service **/
+fun BluePrintDependencyService.messageConsumerService(selector: String): BlueprintMessageConsumerService {
+ return messageLibPropertyService().blueprintMessageConsumerService(selector)
+}
+
+fun BluePrintDependencyService.messageConsumerService(jsonNode: JsonNode): BlueprintMessageConsumerService {
+ return messageLibPropertyService().blueprintMessageConsumerService(jsonNode)
+}
+
class MessageLibConstants {
companion object {
const val SERVICE_BLUEPRINT_MESSAGE_LIB_PROPERTY = "blueprint-message-lib-property-service"
- const val PROPERTY_MESSAGE_CLIENT_PREFIX = "blueprintsprocessor.messageclient."
+ //TODO("Change to .messageconsumer in application.properties")
+ const val PROPERTY_MESSAGE_CONSUMER_PREFIX = "blueprintsprocessor.messageclient."
+ const val PROPERTY_MESSAGE_PRODUCER_PREFIX = "blueprintsprocessor.messageproducer."
const val TYPE_KAFKA_BASIC_AUTH = "kafka-basic-auth"
}
}
\ No newline at end of file
package org.onap.ccsdk.cds.blueprintsprocessor.message
-
+/** Producer Properties **/
open class MessageProducerProperties
lateinit var bootstrapServers: String
var topic: String? = null
var clientId: String? = null
-}
\ No newline at end of file
+}
+
+/** Consumer Properties **/
+
+open class MessageConsumerProperties
+
+open class KafkaMessageConsumerProperties : MessageConsumerProperties() {
+ lateinit var bootstrapServers: String
+ lateinit var groupId: String
+ var consumerTopic: String? = null
+ var pollMillSec: Long = 100
+}
+
+open class KafkaBasicAuthMessageConsumerProperties : KafkaMessageConsumerProperties()
import com.fasterxml.jackson.databind.JsonNode
import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageProducerProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageLibConstants
-import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageProducerProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.message.*
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
import org.springframework.stereotype.Service
@Service(MessageLibConstants.SERVICE_BLUEPRINT_MESSAGE_LIB_PROPERTY)
open class BluePrintMessageLibPropertyService(private var bluePrintProperties: BluePrintProperties) {
- fun blueprintMessageClientService(jsonNode: JsonNode): BlueprintMessageProducerService {
- val messageClientProperties = messageClientProperties(jsonNode)
- return blueprintMessageClientService(messageClientProperties)
+ fun blueprintMessageProducerService(jsonNode: JsonNode): BlueprintMessageProducerService {
+ val messageClientProperties = messageProducerProperties(jsonNode)
+ return blueprintMessageProducerService(messageClientProperties)
}
- fun blueprintMessageClientService(selector: String): BlueprintMessageProducerService {
- val prefix = "${MessageLibConstants.PROPERTY_MESSAGE_CLIENT_PREFIX}$selector"
- val messageClientProperties = messageClientProperties(prefix)
- return blueprintMessageClientService(messageClientProperties)
+ fun blueprintMessageProducerService(selector: String): BlueprintMessageProducerService {
+ val prefix = "${MessageLibConstants.PROPERTY_MESSAGE_PRODUCER_PREFIX}$selector"
+ val messageClientProperties = messageProducerProperties(prefix)
+ return blueprintMessageProducerService(messageClientProperties)
}
- fun messageClientProperties(prefix: String): MessageProducerProperties {
+ fun messageProducerProperties(prefix: String): MessageProducerProperties {
val type = bluePrintProperties.propertyBeanType("$prefix.type", String::class.java)
return when (type) {
MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
- kafkaBasicAuthMessageClientProperties(prefix)
+ kafkaBasicAuthMessageProducerProperties(prefix)
}
else -> {
throw BluePrintProcessorException("Message adaptor($type) is not supported")
}
}
- fun messageClientProperties(jsonNode: JsonNode): MessageProducerProperties {
+ fun messageProducerProperties(jsonNode: JsonNode): MessageProducerProperties {
val type = jsonNode.get("type").textValue()
return when (type) {
MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
}
}
- private fun blueprintMessageClientService(MessageProducerProperties: MessageProducerProperties)
+ private fun blueprintMessageProducerService(MessageProducerProperties: MessageProducerProperties)
: BlueprintMessageProducerService {
when (MessageProducerProperties) {
}
}
- private fun kafkaBasicAuthMessageClientProperties(prefix: String): KafkaBasicAuthMessageProducerProperties {
+ private fun kafkaBasicAuthMessageProducerProperties(prefix: String): KafkaBasicAuthMessageProducerProperties {
return bluePrintProperties.propertyBeanType(
prefix, KafkaBasicAuthMessageProducerProperties::class.java)
}
+ /** Consumer Property Lib Service Implementation **/
+
+ /** Return Message Consumer Service for [jsonNode] definitions. */
+ fun blueprintMessageConsumerService(jsonNode: JsonNode): BlueprintMessageConsumerService {
+ val messageConsumerProperties = messageConsumerProperties(jsonNode)
+ return blueprintMessageConsumerService(messageConsumerProperties)
+ }
+
+ /** Return Message Consumer Service for [selector] definitions. */
+ fun blueprintMessageConsumerService(selector: String): BlueprintMessageConsumerService {
+ val prefix = "${MessageLibConstants.PROPERTY_MESSAGE_CONSUMER_PREFIX}$selector"
+ val messageClientProperties = messageConsumerProperties(prefix)
+ return blueprintMessageConsumerService(messageClientProperties)
+ }
+
+ /** Return Message Consumer Properties for [prefix] definitions. */
+ fun messageConsumerProperties(prefix: String): MessageConsumerProperties {
+ val type = bluePrintProperties.propertyBeanType("$prefix.type", String::class.java)
+ return when (type) {
+ MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
+ kafkaBasicAuthMessageConsumerProperties(prefix)
+ }
+ else -> {
+ throw BluePrintProcessorException("Message adaptor($type) is not supported")
+ }
+ }
+ }
+
+ fun messageConsumerProperties(jsonNode: JsonNode): MessageConsumerProperties {
+ val type = jsonNode.get("type").textValue()
+ return when (type) {
+ MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
+ JacksonUtils.readValue(jsonNode, KafkaBasicAuthMessageConsumerProperties::class.java)!!
+ }
+ else -> {
+ throw BluePrintProcessorException("Message adaptor($type) is not supported")
+ }
+ }
+ }
+
+ private fun blueprintMessageConsumerService(messageConsumerProperties: MessageConsumerProperties)
+ : BlueprintMessageConsumerService {
+
+ when (messageConsumerProperties) {
+ is KafkaBasicAuthMessageConsumerProperties -> {
+ return KafkaBasicAuthMessageConsumerService(messageConsumerProperties)
+ }
+ else -> {
+ throw BluePrintProcessorException("couldn't get Message client service for")
+ }
+ }
+ }
+
+ private fun kafkaBasicAuthMessageConsumerProperties(prefix: String): KafkaBasicAuthMessageConsumerProperties {
+ return bluePrintProperties.propertyBeanType(
+ prefix, KafkaBasicAuthMessageConsumerProperties::class.java)
+ }
+
}
--- /dev/null
+/*
+ * Copyright © 2019 IBM.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.message.service
+
+import kotlinx.coroutines.channels.Channel
+
+interface BlueprintMessageConsumerService {
+
+ /** Subscribe to the Kafka channel with [additionalConfig] */
+ suspend fun subscribe(additionalConfig: Map<String, Any>?): Channel<String>
+
+ /** Subscribe to the Kafka channel with [additionalConfig] for dynamic [topics]*/
+ suspend fun subscribe(topics: List<String>, additionalConfig: Map<String, Any>? = null): Channel<String>
+
+ /** close the channel, consumer and other resources */
+ suspend fun shutDown()
+
+}
\ No newline at end of file
--- /dev/null
+/*
+ * Copyright © 2019 IBM.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.message.service
+
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import org.apache.kafka.clients.CommonClientConfigs
+import org.apache.kafka.clients.consumer.Consumer
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.common.serialization.StringDeserializer
+import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageConsumerProperties
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import java.time.Duration
+import kotlin.concurrent.thread
+
+class KafkaBasicAuthMessageConsumerService(
+ private val messageConsumerProperties: KafkaBasicAuthMessageConsumerProperties)
+ : BlueprintMessageConsumerService {
+
+ private val channel = Channel<String>()
+ private var kafkaConsumer: Consumer<String, String>? = null
+ val log = logger(KafkaBasicAuthMessageConsumerService::class)
+
+ @Volatile
+ var keepGoing = true
+
+ fun kafkaConsumer(additionalConfig: Map<String, Any>? = null): Consumer<String, String> {
+ val configProperties = hashMapOf<String, Any>()
+ configProperties[CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG] = messageConsumerProperties.bootstrapServers
+ configProperties[ConsumerConfig.GROUP_ID_CONFIG] = messageConsumerProperties.groupId
+ configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "latest"
+ configProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
+ configProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
+ /** add or override already set properties */
+ additionalConfig?.let { configProperties.putAll(it) }
+ /** Create Kafka consumer */
+ return KafkaConsumer(configProperties)
+ }
+
+ override suspend fun subscribe(additionalConfig: Map<String, Any>?): Channel<String> {
+ /** get to topic names */
+ val consumerTopic = messageConsumerProperties.consumerTopic?.split(",")?.map { it.trim() }
+ check(!consumerTopic.isNullOrEmpty()) { "couldn't get topic information" }
+ return subscribe(consumerTopic, additionalConfig)
+ }
+
+
+ override suspend fun subscribe(consumerTopic: List<String>, additionalConfig: Map<String, Any>?): Channel<String> {
+ /** Create Kafka consumer */
+ kafkaConsumer = kafkaConsumer(additionalConfig)
+ checkNotNull(kafkaConsumer) {
+ "failed to create kafka consumer for " +
+ "server(${messageConsumerProperties.bootstrapServers})'s " +
+ "topics(${messageConsumerProperties.bootstrapServers})"
+ }
+
+ kafkaConsumer!!.subscribe(consumerTopic)
+ log.info("Successfully consumed topic($consumerTopic)")
+
+ val listenerThread = thread(start = true, name = "KafkaConsumer") {
+ keepGoing = true
+ kafkaConsumer!!.use { kc ->
+ while (keepGoing) {
+ val consumerRecords = kc.poll(Duration.ofMillis(messageConsumerProperties.pollMillSec))
+ runBlocking {
+ consumerRecords?.forEach { consumerRecord ->
+ /** execute the command block */
+ consumerRecord.value()?.let {
+ launch {
+ if (!channel.isClosedForSend) {
+ channel.send(it)
+ } else {
+ log.error("Channel is closed to receive message")
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ }
+ log.info("Successfully consumed in thread(${listenerThread})")
+ return channel
+ }
+
+ override suspend fun shutDown() {
+ /** Close the Channel */
+ channel.close()
+ /** stop the polling loop */
+ keepGoing = false
+ if (kafkaConsumer != null) {
+ /** sunsubscribe the consumer */
+ kafkaConsumer!!.unsubscribe()
+ }
+ }
+}
--- /dev/null
+/*
+ * Copyright © 2019 IBM.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.message.service
+
+import io.mockk.every
+import io.mockk.spyk
+import kotlinx.coroutines.channels.consumeEach
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.apache.kafka.clients.consumer.MockConsumer
+import org.apache.kafka.clients.consumer.OffsetResetStrategy
+import org.apache.kafka.common.TopicPartition
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.test.annotation.DirtiesContext
+import org.springframework.test.context.ContextConfiguration
+import org.springframework.test.context.TestPropertySource
+import org.springframework.test.context.junit4.SpringRunner
+import kotlin.test.assertNotNull
+
+
+@RunWith(SpringRunner::class)
+@DirtiesContext
+@ContextConfiguration(classes = [BluePrintMessageLibConfiguration::class,
+ BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+@TestPropertySource(properties =
+["blueprintsprocessor.messageclient.sample.type=kafka-basic-auth",
+ "blueprintsprocessor.messageclient.sample.bootstrapServers=127:0.0.1:9092",
+ "blueprintsprocessor.messageclient.sample.groupId=sample-group",
+ "blueprintsprocessor.messageclient.sample.consumerTopic=default-topic"
+])
+open class BlueprintMessageConsumerServiceTest {
+
+ @Autowired
+ lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
+
+ @Test
+ fun testKafkaBasicAuthConsumerService() {
+ runBlocking {
+ val blueprintMessageConsumerService = bluePrintMessageLibPropertyService
+ .blueprintMessageConsumerService("sample") as KafkaBasicAuthMessageConsumerService
+ assertNotNull(blueprintMessageConsumerService, "failed to get blueprintMessageConsumerService")
+
+ val spyBlueprintMessageConsumerService = spyk(blueprintMessageConsumerService, recordPrivateCalls = true)
+
+ val topic = "default-topic"
+ val partitions: MutableList<TopicPartition> = arrayListOf()
+ val topicsCollection: MutableList<String> = arrayListOf()
+ partitions.add(TopicPartition(topic, 1))
+ val partitionsBeginningMap: MutableMap<TopicPartition, Long> = mutableMapOf()
+ val partitionsEndMap: MutableMap<TopicPartition, Long> = mutableMapOf()
+
+ val records: Long = 10
+ partitions.forEach { partition ->
+ partitionsBeginningMap[partition] = 0L
+ partitionsEndMap[partition] = records
+ topicsCollection.add(partition.topic())
+ }
+ val mockKafkaConsumer = MockConsumer<String, String>(OffsetResetStrategy.EARLIEST)
+ mockKafkaConsumer.subscribe(topicsCollection)
+ mockKafkaConsumer.rebalance(partitions)
+ mockKafkaConsumer.updateBeginningOffsets(partitionsBeginningMap)
+ mockKafkaConsumer.updateEndOffsets(partitionsEndMap)
+ for (i in 1..10) {
+ val record = ConsumerRecord<String, String>(topic, 1, i.toLong(), "key_$i",
+ "I am message $i")
+ mockKafkaConsumer.addRecord(record)
+ }
+
+ every { spyBlueprintMessageConsumerService.kafkaConsumer(any()) } returns mockKafkaConsumer
+ val channel = spyBlueprintMessageConsumerService.subscribe(null)
+ launch {
+ channel.consumeEach {
+ println("Received message : $it")
+ }
+ }
+ //delay(100)
+ spyBlueprintMessageConsumerService.shutDown()
+ }
+ }
+}
\ No newline at end of file
@ContextConfiguration(classes = [BluePrintMessageLibConfiguration::class,
BlueprintPropertyConfiguration::class, BluePrintProperties::class])
@TestPropertySource(properties =
-["blueprintsprocessor.messageclient.sample.type=kafka-basic-auth",
- "blueprintsprocessor.messageclient.sample.bootstrapServers=127:0.0.1:9092",
- "blueprintsprocessor.messageclient.sample.topic=default-topic",
- "blueprintsprocessor.messageclient.sample.clientId=default-client-id"
+["blueprintsprocessor.messageproducer.sample.type=kafka-basic-auth",
+ "blueprintsprocessor.messageproducer.sample.bootstrapServers=127:0.0.1:9092",
+ "blueprintsprocessor.messageproducer.sample.topic=default-topic",
+ "blueprintsprocessor.messageproducer.sample.clientId=default-client-id"
])
open class BlueprintMessageProducerServiceTest {
lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
@Test
- fun testKafkaBasicAuthClientService() {
+ fun testKafkaBasicAuthProducertService() {
runBlocking {
- val bluePrintMessageClientService = bluePrintMessageLibPropertyService
- .blueprintMessageClientService("sample") as KafkaBasicAuthMessageProducerService
+ val blueprintMessageProducerService = bluePrintMessageLibPropertyService
+ .blueprintMessageProducerService("sample") as KafkaBasicAuthMessageProducerService
val mockKafkaTemplate = mockk<KafkaTemplate<String, Any>>()
every { mockKafkaTemplate.send(any(), any()) } returns future
- val spyBluePrintMessageClientService = spyk(bluePrintMessageClientService, recordPrivateCalls = true)
+ val spyBluePrintMessageProducerService = spyk(blueprintMessageProducerService, recordPrivateCalls = true)
- every { spyBluePrintMessageClientService.messageTemplate(any()) } returns mockKafkaTemplate
+ every { spyBluePrintMessageProducerService.messageTemplate(any()) } returns mockKafkaTemplate
- val response = spyBluePrintMessageClientService.sendMessage("Testing message")
+ val response = spyBluePrintMessageProducerService.sendMessage("Testing message")
assertTrue(response, "failed to get command response")
}
}
<!-- encoders are assigned the type
ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
<encoder>
- <pattern>%d{HH:mm:ss.SSS} %-5level %logger{100} - %msg%n</pattern>
+ <pattern>%d{HH:mm:ss.SSS} %-5level [%thread] %logger{50} - %msg%n</pattern>
</encoder>
</appender>
import org.springframework.kafka.annotation.KafkaListener
import org.springframework.stereotype.Service
+//TODO("Implement with property service and remove spring bindings")
@ConditionalOnProperty(name = ["blueprintsprocessor.messageclient.self-service-api.kafkaEnable"], havingValue = "true")
@Service
open class MessagingController(private val propertyService: BluePrintMessageLibPropertyService,
val executionServiceOutput = executionServiceHandler.doProcess(executionServiceInput)
- if (executionServiceOutput.status.code == EXECUTION_STATUS) {
- val bluePrintMessageClientService = propertyService
- .blueprintMessageClientService(PREFIX)
+ if (executionServiceOutput.status.code == EXECUTION_STATUS) {
+ val blueprintMessageProducerService = propertyService.blueprintMessageProducerService(PREFIX)
- val payload = executionServiceOutput.payload
+ val payload = executionServiceOutput.payload
- log.info("The payload to publish is {}", payload)
+ log.info("The payload to publish is {}", payload)
- bluePrintMessageClientService.sendMessage(payload)
- }
- else {
- log.error("Fail to process the given event due to {}", executionServiceOutput.status.errorMessage)
- }
+ blueprintMessageProducerService.sendMessage(payload)
+ } else {
+ log.error("Fail to process the given event due to {}", executionServiceOutput.status.errorMessage)
+ }
}
}
Code Review / ccsdk / cds.git / commitdiff