From adcd4f2bc695840e9ecbc05003bc52c675f22fec Mon Sep 17 00:00:00 2001 From: KAPIL SINGAL Date: Fri, 22 Jan 2021 11:49:51 -0500 Subject: Renaming Files having BluePrint to have Blueprint Replacing BluePrint with Blueprint throughout Issue-ID: CCSDK-3098 Signed-off-by: KAPIL SINGAL Change-Id: Ibee8bad07ae7d9287073db2d4f2f2cd730fa8b96 --- .../message/BluePrintMessageExtensions.kt | 31 --- .../message/BluePrintMessageLibConfiguration.kt | 71 ------ .../message/BluePrintMessageLibData.kt | 264 --------------------- .../message/BlueprintMessageExtensions.kt | 31 +++ .../message/BlueprintMessageLibConfiguration.kt | 71 ++++++ .../message/BlueprintMessageLibData.kt | 264 +++++++++++++++++++++ .../message/MessagePropertiesDSL.kt | 94 ++++---- .../kafka/AbstractKafkaTopologyComponents.kt | 6 +- .../message/kafka/KafkaJDBCStores.kt | 8 +- .../service/BluePrintMessageLibPropertyService.kt | 221 ----------------- .../service/BlueprintMessageConsumerService.kt | 6 +- .../service/BlueprintMessageLibPropertyService.kt | 221 +++++++++++++++++ .../message/service/KafkaStreamsConsumerService.kt | 6 +- .../message/service/MessageLoggerService.kt | 14 +- .../message/MessagePropertiesDSLTest.kt | 18 +- .../service/BlueprintMessageConsumerServiceTest.kt | 12 +- .../service/BlueprintMessageProducerServiceTest.kt | 18 +- .../service/KafkaStreamsConsumerServiceTest.kt | 12 +- .../message/service/MessageLoggerServiceTest.kt | 4 +- .../message/service/MockKafkaTopologyComponents.kt | 4 +- 20 files changed, 688 insertions(+), 688 deletions(-) delete mode 100644 ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageExtensions.kt delete mode 100644 ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibConfiguration.kt delete mode 100644 ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt create mode 100644 ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageExtensions.kt create mode 100644 ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageLibConfiguration.kt create mode 100644 ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageLibData.kt delete mode 100644 ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BluePrintMessageLibPropertyService.kt create mode 100644 ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageLibPropertyService.kt (limited to 'ms/blueprintsprocessor/modules/commons/message-lib') diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageExtensions.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageExtensions.kt deleted file mode 100644 index 509689eca..000000000 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageExtensions.kt +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Copyright © 2018-2019 AT&T Intellectual Property. - * - * 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 - -import org.apache.kafka.common.header.Headers -import org.apache.kafka.common.header.internals.RecordHeader -import java.nio.charset.Charset - -fun T?.toMap(): MutableMap { - val map: MutableMap = hashMapOf() - this?.forEach { map[it.key()] = String(it.value(), Charset.defaultCharset()) } - return map -} - -fun Headers.addHeader(key: String, value: String) { - this.add(RecordHeader(key, value.toByteArray())) -} diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibConfiguration.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibConfiguration.kt deleted file mode 100644 index 659295a6b..000000000 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibConfiguration.kt +++ /dev/null @@ -1,71 +0,0 @@ -/* - * Copyright © 2019 IBM. - * Modifications Copyright © 2018-2019 AT&T Intellectual Property. - * - * 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 - -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 - -@Configuration -@ComponentScan -@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_CONSUMER_PREFIX = "blueprintsprocessor.messageconsumer." - const val PROPERTY_MESSAGE_PRODUCER_PREFIX = "blueprintsprocessor.messageproducer." - const val TYPE_KAFKA_BASIC_AUTH = "kafka-basic-auth" - const val TYPE_KAFKA_SCRAM_SSL_AUTH = "kafka-scram-ssl-auth" - const val TYPE_KAFKA_SSL_AUTH = "kafka-ssl-auth" - const val TYPE_KAFKA_STREAMS_BASIC_AUTH = "kafka-streams-basic-auth" - const val TYPE_KAFKA_STREAMS_SSL_AUTH = "kafka-streams-ssl-auth" - const val TYPE_KAFKA_STREAMS_SCRAM_SSL_AUTH = "kafka-streams-scram-ssl-auth" - } -} diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt deleted file mode 100644 index 67dba1f19..000000000 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BluePrintMessageLibData.kt +++ /dev/null @@ -1,264 +0,0 @@ -/* - * Copyright © 2019 IBM. - * Modifications Copyright © 2018-2019 AT&T Intellectual Property. - * - * 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 - -import org.apache.kafka.clients.CommonClientConfigs -import org.apache.kafka.clients.consumer.ConsumerConfig -import org.apache.kafka.clients.producer.ProducerConfig -import org.apache.kafka.common.config.SaslConfigs -import org.apache.kafka.common.config.SslConfigs -import org.apache.kafka.common.security.auth.SecurityProtocol -import org.apache.kafka.common.security.scram.ScramLoginModule -import org.apache.kafka.common.serialization.ByteArrayDeserializer -import org.apache.kafka.common.serialization.ByteArraySerializer -import org.apache.kafka.common.serialization.StringDeserializer -import org.apache.kafka.common.serialization.StringSerializer -import org.apache.kafka.streams.StreamsConfig - -/** Common Properties **/ -abstract class CommonProperties { - - lateinit var type: String - lateinit var topic: String - lateinit var bootstrapServers: String - - open fun getConfig(): HashMap { - val configProps = hashMapOf() - configProps[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = bootstrapServers - return configProps - } -} - -/** Message Producer */ -/** Message Producer Properties **/ -abstract class MessageProducerProperties : CommonProperties() - -/** Basic Auth */ -open class KafkaBasicAuthMessageProducerProperties : MessageProducerProperties() { - - var clientId: String? = null - var acks: String = "all" // strongest producing guarantee - var maxBlockMs: Int = 250 // max blocking time in ms to send a message - var reconnectBackOffMs: Int = 60 * 60 * 1000 // time in ms before retrying connection (1 hour) - var enableIdempotence: Boolean = true // ensure we don't push duplicates - - override fun getConfig(): HashMap { - val configProps = super.getConfig() - configProps[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java - configProps[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = ByteArraySerializer::class.java - configProps[ProducerConfig.ACKS_CONFIG] = acks - configProps[ProducerConfig.MAX_BLOCK_MS_CONFIG] = maxBlockMs - configProps[ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG] = reconnectBackOffMs - configProps[ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG] = enableIdempotence - if (clientId != null) { - configProps[ProducerConfig.CLIENT_ID_CONFIG] = clientId!! - } - return configProps - } -} - -/** SSL Auth */ -open class KafkaSslAuthMessageProducerProperties : KafkaBasicAuthMessageProducerProperties() { - - lateinit var truststore: String - lateinit var truststorePassword: String - var truststoreType: String = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE - var keystore: String? = null - var keystorePassword: String? = null - var keystoreType: String = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE - var sslEndpointIdentificationAlgorithm: String = SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM - - override fun getConfig(): HashMap { - val configProps = super.getConfig() - configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SSL.toString() - configProps[SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG] = truststoreType - configProps[SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG] = truststore!! - configProps[SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG] = truststorePassword!! - if (keystore != null) { - configProps[SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG] = keystore!! - configProps[SslConfigs.SSL_KEYSTORE_TYPE_CONFIG] = keystoreType - configProps[SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG] = keystorePassword!! - } - configProps[SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG] = sslEndpointIdentificationAlgorithm - - return configProps - } -} - -/** (SASL) SCRAM SSL Auth */ -class KafkaScramSslAuthMessageProducerProperties : KafkaSslAuthMessageProducerProperties() { - - var saslMechanism: String = "SCRAM-SHA-512" - lateinit var scramUsername: String - lateinit var scramPassword: String - - override fun getConfig(): HashMap { - val configProps = super.getConfig() - configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SASL_SSL.toString() - configProps[SaslConfigs.SASL_MECHANISM] = saslMechanism - configProps[SaslConfigs.SASL_JAAS_CONFIG] = "${ScramLoginModule::class.java.canonicalName} required " + - "username=\"${scramUsername}\" " + - "password=\"${scramPassword}\";" - return configProps - } -} - -/** Consumer */ -abstract class MessageConsumerProperties : CommonProperties() -/** Kafka Streams */ -/** Streams properties */ - -/** Basic Auth */ -open class KafkaStreamsBasicAuthConsumerProperties : MessageConsumerProperties() { - - lateinit var applicationId: String - var autoOffsetReset: String = "latest" - var processingGuarantee: String = StreamsConfig.EXACTLY_ONCE - - override fun getConfig(): HashMap { - val configProperties = super.getConfig() - configProperties[StreamsConfig.APPLICATION_ID_CONFIG] = applicationId - configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = autoOffsetReset - configProperties[StreamsConfig.PROCESSING_GUARANTEE_CONFIG] = processingGuarantee - return configProperties - } -} - -/** SSL Auth */ -open class KafkaStreamsSslAuthConsumerProperties : KafkaStreamsBasicAuthConsumerProperties() { - - lateinit var truststore: String - lateinit var truststorePassword: String - var truststoreType: String = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE - var keystore: String? = null - var keystorePassword: String? = null - var keystoreType: String = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE - var sslEndpointIdentificationAlgorithm: String = SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM - - override fun getConfig(): HashMap { - val configProps = super.getConfig() - configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SSL.toString() - configProps[SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG] = truststoreType - configProps[SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG] = truststore!! - configProps[SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG] = truststorePassword!! - if (keystore != null) { - configProps[SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG] = keystore!! - configProps[SslConfigs.SSL_KEYSTORE_TYPE_CONFIG] = keystoreType - configProps[SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG] = keystorePassword!! - } - configProps[SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG] = sslEndpointIdentificationAlgorithm - return configProps - } -} - -/** (SASL) SCRAM SSL Auth */ -class KafkaStreamsScramSslAuthConsumerProperties : KafkaStreamsSslAuthConsumerProperties() { - - var saslMechanism: String = "SCRAM-SHA-512" - lateinit var scramUsername: String - lateinit var scramPassword: String - - override fun getConfig(): HashMap { - val configProps = super.getConfig() - configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SASL_SSL.toString() - configProps[SaslConfigs.SASL_MECHANISM] = saslMechanism - configProps[SaslConfigs.SASL_JAAS_CONFIG] = "${ScramLoginModule::class.java.canonicalName} required " + - "username=\"${scramUsername}\" " + - "password=\"${scramPassword}\";" - return configProps - } -} - -/** Message Consumer */ -/** Message Consumer Properties **/ -/** Basic Auth */ -open class KafkaBasicAuthMessageConsumerProperties : MessageConsumerProperties() { - - lateinit var groupId: String - lateinit var clientId: String - var autoCommit: Boolean = true - var autoOffsetReset: String = "latest" - var pollMillSec: Long = 1000 - var pollRecords: Int = -1 - - override fun getConfig(): HashMap { - val configProperties = super.getConfig() - configProperties[ConsumerConfig.GROUP_ID_CONFIG] = groupId - configProperties[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = autoCommit - /** - * earliest: automatically reset the offset to the earliest offset - * latest: automatically reset the offset to the latest offset - */ - configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = autoOffsetReset - configProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java - configProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = ByteArrayDeserializer::class.java - configProperties[ConsumerConfig.CLIENT_ID_CONFIG] = clientId - - /** To handle Back pressure, Get only configured record for processing */ - if (pollRecords > 0) { - configProperties[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = pollRecords - } - - return configProperties - } -} - -/** SSL Auth */ -open class KafkaSslAuthMessageConsumerProperties : KafkaBasicAuthMessageConsumerProperties() { - - lateinit var truststore: String - lateinit var truststorePassword: String - var truststoreType: String = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE - var keystore: String? = null - var keystorePassword: String? = null - var keystoreType: String = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE - var sslEndpointIdentificationAlgorithm: String = SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM - - override fun getConfig(): HashMap { - val configProps = super.getConfig() - configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SSL.toString() - configProps[SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG] = truststoreType - configProps[SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG] = truststore!! - configProps[SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG] = truststorePassword!! - if (keystore != null) { - configProps[SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG] = keystore!! - configProps[SslConfigs.SSL_KEYSTORE_TYPE_CONFIG] = keystoreType - configProps[SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG] = keystorePassword!! - } - configProps[SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG] = sslEndpointIdentificationAlgorithm - return configProps - } -} - -/** (SASL) SCRAM SSL Auth */ -class KafkaScramSslAuthMessageConsumerProperties : KafkaSslAuthMessageConsumerProperties() { - - var saslMechanism: String = "SCRAM-SHA-512" - lateinit var scramUsername: String - lateinit var scramPassword: String - - override fun getConfig(): HashMap { - val configProps = super.getConfig() - configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SASL_SSL.toString() - configProps[SaslConfigs.SASL_MECHANISM] = saslMechanism - configProps[SaslConfigs.SASL_JAAS_CONFIG] = "${ScramLoginModule::class.java.canonicalName} required " + - "username=\"${scramUsername}\" " + - "password=\"${scramPassword}\";" - return configProps - } -} diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageExtensions.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageExtensions.kt new file mode 100644 index 000000000..509689eca --- /dev/null +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageExtensions.kt @@ -0,0 +1,31 @@ +/* + * Copyright © 2018-2019 AT&T Intellectual Property. + * + * 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 + +import org.apache.kafka.common.header.Headers +import org.apache.kafka.common.header.internals.RecordHeader +import java.nio.charset.Charset + +fun T?.toMap(): MutableMap { + val map: MutableMap = hashMapOf() + this?.forEach { map[it.key()] = String(it.value(), Charset.defaultCharset()) } + return map +} + +fun Headers.addHeader(key: String, value: String) { + this.add(RecordHeader(key, value.toByteArray())) +} diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageLibConfiguration.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageLibConfiguration.kt new file mode 100644 index 000000000..88931d13f --- /dev/null +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageLibConfiguration.kt @@ -0,0 +1,71 @@ +/* + * Copyright © 2019 IBM. + * Modifications Copyright © 2018-2019 AT&T Intellectual Property. + * + * 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 + +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 + +@Configuration +@ComponentScan +@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_CONSUMER_PREFIX = "blueprintsprocessor.messageconsumer." + const val PROPERTY_MESSAGE_PRODUCER_PREFIX = "blueprintsprocessor.messageproducer." + const val TYPE_KAFKA_BASIC_AUTH = "kafka-basic-auth" + const val TYPE_KAFKA_SCRAM_SSL_AUTH = "kafka-scram-ssl-auth" + const val TYPE_KAFKA_SSL_AUTH = "kafka-ssl-auth" + const val TYPE_KAFKA_STREAMS_BASIC_AUTH = "kafka-streams-basic-auth" + const val TYPE_KAFKA_STREAMS_SSL_AUTH = "kafka-streams-ssl-auth" + const val TYPE_KAFKA_STREAMS_SCRAM_SSL_AUTH = "kafka-streams-scram-ssl-auth" + } +} diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageLibData.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageLibData.kt new file mode 100644 index 000000000..67dba1f19 --- /dev/null +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/BlueprintMessageLibData.kt @@ -0,0 +1,264 @@ +/* + * Copyright © 2019 IBM. + * Modifications Copyright © 2018-2019 AT&T Intellectual Property. + * + * 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 + +import org.apache.kafka.clients.CommonClientConfigs +import org.apache.kafka.clients.consumer.ConsumerConfig +import org.apache.kafka.clients.producer.ProducerConfig +import org.apache.kafka.common.config.SaslConfigs +import org.apache.kafka.common.config.SslConfigs +import org.apache.kafka.common.security.auth.SecurityProtocol +import org.apache.kafka.common.security.scram.ScramLoginModule +import org.apache.kafka.common.serialization.ByteArrayDeserializer +import org.apache.kafka.common.serialization.ByteArraySerializer +import org.apache.kafka.common.serialization.StringDeserializer +import org.apache.kafka.common.serialization.StringSerializer +import org.apache.kafka.streams.StreamsConfig + +/** Common Properties **/ +abstract class CommonProperties { + + lateinit var type: String + lateinit var topic: String + lateinit var bootstrapServers: String + + open fun getConfig(): HashMap { + val configProps = hashMapOf() + configProps[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = bootstrapServers + return configProps + } +} + +/** Message Producer */ +/** Message Producer Properties **/ +abstract class MessageProducerProperties : CommonProperties() + +/** Basic Auth */ +open class KafkaBasicAuthMessageProducerProperties : MessageProducerProperties() { + + var clientId: String? = null + var acks: String = "all" // strongest producing guarantee + var maxBlockMs: Int = 250 // max blocking time in ms to send a message + var reconnectBackOffMs: Int = 60 * 60 * 1000 // time in ms before retrying connection (1 hour) + var enableIdempotence: Boolean = true // ensure we don't push duplicates + + override fun getConfig(): HashMap { + val configProps = super.getConfig() + configProps[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java + configProps[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = ByteArraySerializer::class.java + configProps[ProducerConfig.ACKS_CONFIG] = acks + configProps[ProducerConfig.MAX_BLOCK_MS_CONFIG] = maxBlockMs + configProps[ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG] = reconnectBackOffMs + configProps[ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG] = enableIdempotence + if (clientId != null) { + configProps[ProducerConfig.CLIENT_ID_CONFIG] = clientId!! + } + return configProps + } +} + +/** SSL Auth */ +open class KafkaSslAuthMessageProducerProperties : KafkaBasicAuthMessageProducerProperties() { + + lateinit var truststore: String + lateinit var truststorePassword: String + var truststoreType: String = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE + var keystore: String? = null + var keystorePassword: String? = null + var keystoreType: String = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE + var sslEndpointIdentificationAlgorithm: String = SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM + + override fun getConfig(): HashMap { + val configProps = super.getConfig() + configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SSL.toString() + configProps[SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG] = truststoreType + configProps[SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG] = truststore!! + configProps[SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG] = truststorePassword!! + if (keystore != null) { + configProps[SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG] = keystore!! + configProps[SslConfigs.SSL_KEYSTORE_TYPE_CONFIG] = keystoreType + configProps[SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG] = keystorePassword!! + } + configProps[SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG] = sslEndpointIdentificationAlgorithm + + return configProps + } +} + +/** (SASL) SCRAM SSL Auth */ +class KafkaScramSslAuthMessageProducerProperties : KafkaSslAuthMessageProducerProperties() { + + var saslMechanism: String = "SCRAM-SHA-512" + lateinit var scramUsername: String + lateinit var scramPassword: String + + override fun getConfig(): HashMap { + val configProps = super.getConfig() + configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SASL_SSL.toString() + configProps[SaslConfigs.SASL_MECHANISM] = saslMechanism + configProps[SaslConfigs.SASL_JAAS_CONFIG] = "${ScramLoginModule::class.java.canonicalName} required " + + "username=\"${scramUsername}\" " + + "password=\"${scramPassword}\";" + return configProps + } +} + +/** Consumer */ +abstract class MessageConsumerProperties : CommonProperties() +/** Kafka Streams */ +/** Streams properties */ + +/** Basic Auth */ +open class KafkaStreamsBasicAuthConsumerProperties : MessageConsumerProperties() { + + lateinit var applicationId: String + var autoOffsetReset: String = "latest" + var processingGuarantee: String = StreamsConfig.EXACTLY_ONCE + + override fun getConfig(): HashMap { + val configProperties = super.getConfig() + configProperties[StreamsConfig.APPLICATION_ID_CONFIG] = applicationId + configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = autoOffsetReset + configProperties[StreamsConfig.PROCESSING_GUARANTEE_CONFIG] = processingGuarantee + return configProperties + } +} + +/** SSL Auth */ +open class KafkaStreamsSslAuthConsumerProperties : KafkaStreamsBasicAuthConsumerProperties() { + + lateinit var truststore: String + lateinit var truststorePassword: String + var truststoreType: String = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE + var keystore: String? = null + var keystorePassword: String? = null + var keystoreType: String = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE + var sslEndpointIdentificationAlgorithm: String = SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM + + override fun getConfig(): HashMap { + val configProps = super.getConfig() + configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SSL.toString() + configProps[SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG] = truststoreType + configProps[SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG] = truststore!! + configProps[SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG] = truststorePassword!! + if (keystore != null) { + configProps[SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG] = keystore!! + configProps[SslConfigs.SSL_KEYSTORE_TYPE_CONFIG] = keystoreType + configProps[SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG] = keystorePassword!! + } + configProps[SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG] = sslEndpointIdentificationAlgorithm + return configProps + } +} + +/** (SASL) SCRAM SSL Auth */ +class KafkaStreamsScramSslAuthConsumerProperties : KafkaStreamsSslAuthConsumerProperties() { + + var saslMechanism: String = "SCRAM-SHA-512" + lateinit var scramUsername: String + lateinit var scramPassword: String + + override fun getConfig(): HashMap { + val configProps = super.getConfig() + configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SASL_SSL.toString() + configProps[SaslConfigs.SASL_MECHANISM] = saslMechanism + configProps[SaslConfigs.SASL_JAAS_CONFIG] = "${ScramLoginModule::class.java.canonicalName} required " + + "username=\"${scramUsername}\" " + + "password=\"${scramPassword}\";" + return configProps + } +} + +/** Message Consumer */ +/** Message Consumer Properties **/ +/** Basic Auth */ +open class KafkaBasicAuthMessageConsumerProperties : MessageConsumerProperties() { + + lateinit var groupId: String + lateinit var clientId: String + var autoCommit: Boolean = true + var autoOffsetReset: String = "latest" + var pollMillSec: Long = 1000 + var pollRecords: Int = -1 + + override fun getConfig(): HashMap { + val configProperties = super.getConfig() + configProperties[ConsumerConfig.GROUP_ID_CONFIG] = groupId + configProperties[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = autoCommit + /** + * earliest: automatically reset the offset to the earliest offset + * latest: automatically reset the offset to the latest offset + */ + configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = autoOffsetReset + configProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java + configProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = ByteArrayDeserializer::class.java + configProperties[ConsumerConfig.CLIENT_ID_CONFIG] = clientId + + /** To handle Back pressure, Get only configured record for processing */ + if (pollRecords > 0) { + configProperties[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = pollRecords + } + + return configProperties + } +} + +/** SSL Auth */ +open class KafkaSslAuthMessageConsumerProperties : KafkaBasicAuthMessageConsumerProperties() { + + lateinit var truststore: String + lateinit var truststorePassword: String + var truststoreType: String = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE + var keystore: String? = null + var keystorePassword: String? = null + var keystoreType: String = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE + var sslEndpointIdentificationAlgorithm: String = SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM + + override fun getConfig(): HashMap { + val configProps = super.getConfig() + configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SSL.toString() + configProps[SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG] = truststoreType + configProps[SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG] = truststore!! + configProps[SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG] = truststorePassword!! + if (keystore != null) { + configProps[SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG] = keystore!! + configProps[SslConfigs.SSL_KEYSTORE_TYPE_CONFIG] = keystoreType + configProps[SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG] = keystorePassword!! + } + configProps[SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG] = sslEndpointIdentificationAlgorithm + return configProps + } +} + +/** (SASL) SCRAM SSL Auth */ +class KafkaScramSslAuthMessageConsumerProperties : KafkaSslAuthMessageConsumerProperties() { + + var saslMechanism: String = "SCRAM-SHA-512" + lateinit var scramUsername: String + lateinit var scramPassword: String + + override fun getConfig(): HashMap { + val configProps = super.getConfig() + configProps[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = SecurityProtocol.SASL_SSL.toString() + configProps[SaslConfigs.SASL_MECHANISM] = saslMechanism + configProps[SaslConfigs.SASL_JAAS_CONFIG] = "${ScramLoginModule::class.java.canonicalName} required " + + "username=\"${scramUsername}\" " + + "password=\"${scramPassword}\";" + return configProps + } +} diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/MessagePropertiesDSL.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/MessagePropertiesDSL.kt index 8b31de9b9..527374f02 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/MessagePropertiesDSL.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/MessagePropertiesDSL.kt @@ -17,8 +17,8 @@ package org.onap.ccsdk.cds.blueprintsprocessor.message import com.fasterxml.jackson.databind.JsonNode -import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants -import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintTypes +import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintConstants +import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintTypes import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive import org.onap.ccsdk.cds.controllerblueprints.core.asJsonType import org.onap.ccsdk.cds.controllerblueprints.core.data.RelationshipType @@ -30,48 +30,48 @@ import org.onap.ccsdk.cds.controllerblueprints.core.dsl.relationshipType /** Relationships Types DSL for Message Producer */ fun ServiceTemplateBuilder.relationshipTypeConnectsToMessageProducer() { - val relationshipType = BluePrintTypes.relationshipTypeConnectsToMessageProducer() + val relationshipType = BlueprintTypes.relationshipTypeConnectsToMessageProducer() if (this.relationshipTypes == null) this.relationshipTypes = hashMapOf() this.relationshipTypes!![relationshipType.id!!] = relationshipType } -fun BluePrintTypes.relationshipTypeConnectsToMessageProducer(): RelationshipType { +fun BlueprintTypes.relationshipTypeConnectsToMessageProducer(): RelationshipType { return relationshipType( - id = BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_PRODUCER, - version = BluePrintConstants.DEFAULT_VERSION_NUMBER, - derivedFrom = BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO, + id = BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_PRODUCER, + version = BlueprintConstants.DEFAULT_VERSION_NUMBER, + derivedFrom = BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO, description = "Relationship connects to through message producer." ) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintConstants.DATA_TYPE_MAP, + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintConstants.DATA_TYPE_MAP, true, "Connection Config details." ) - validTargetTypes(arrayListOf(BluePrintConstants.MODEL_TYPE_CAPABILITY_TYPE_ENDPOINT)) + validTargetTypes(arrayListOf(BlueprintConstants.MODEL_TYPE_CAPABILITY_TYPE_ENDPOINT)) } } fun ServiceTemplateBuilder.relationshipTypeConnectsToMessageConsumer() { - val relationshipType = BluePrintTypes.relationshipTypeConnectsToMessageConsumer() + val relationshipType = BlueprintTypes.relationshipTypeConnectsToMessageConsumer() if (this.relationshipTypes == null) this.relationshipTypes = hashMapOf() this.relationshipTypes!![relationshipType.id!!] = relationshipType } -fun BluePrintTypes.relationshipTypeConnectsToMessageConsumer(): RelationshipType { +fun BlueprintTypes.relationshipTypeConnectsToMessageConsumer(): RelationshipType { return relationshipType( - id = BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_CONSUMER, - version = BluePrintConstants.DEFAULT_VERSION_NUMBER, - derivedFrom = BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO, + id = BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_CONSUMER, + version = BlueprintConstants.DEFAULT_VERSION_NUMBER, + derivedFrom = BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO, description = "Relationship type connects to message consumer." ) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintConstants.DATA_TYPE_MAP, + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintConstants.DATA_TYPE_MAP, true, "Connection Config details." ) - validTargetTypes(arrayListOf(BluePrintConstants.MODEL_TYPE_CAPABILITY_TYPE_ENDPOINT)) + validTargetTypes(arrayListOf(BlueprintConstants.MODEL_TYPE_CAPABILITY_TYPE_ENDPOINT)) } } @@ -90,46 +90,46 @@ fun TopologyTemplateBuilder.relationshipTemplateMessageProducer( class MessageProducerRelationshipTemplateBuilder(name: String, description: String) : RelationshipTemplateBuilder( name, - BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_PRODUCER, description + BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_PRODUCER, description ) { fun kafkaBasicAuth(block: KafkaBasicAuthMessageProducerPropertiesAssignmentBuilder.() -> Unit) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintTypes.kafkaBasicAuthMessageProducerProperties(block) + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintTypes.kafkaBasicAuthMessageProducerProperties(block) ) } fun kafkaSslAuth(block: KafkaSslAuthMessageProducerPropertiesAssignmentBuilder.() -> Unit) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintTypes.kafkaSslAuthMessageProducerProperties(block) + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintTypes.kafkaSslAuthMessageProducerProperties(block) ) } fun kafkaScramSslAuth(block: KafkaScramSslAuthMessageProducerPropertiesAssignmentBuilder.() -> Unit) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintTypes.kafkaScramSslAuthMessageProducerProperties(block) + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintTypes.kafkaScramSslAuthMessageProducerProperties(block) ) } } -fun BluePrintTypes.kafkaBasicAuthMessageProducerProperties(block: KafkaBasicAuthMessageProducerPropertiesAssignmentBuilder.() -> Unit): JsonNode { +fun BlueprintTypes.kafkaBasicAuthMessageProducerProperties(block: KafkaBasicAuthMessageProducerPropertiesAssignmentBuilder.() -> Unit): JsonNode { val assignments = KafkaBasicAuthMessageProducerPropertiesAssignmentBuilder().apply(block).build() assignments[KafkaBasicAuthMessageProducerProperties::type.name] = MessageLibConstants.TYPE_KAFKA_BASIC_AUTH.asJsonPrimitive() return assignments.asJsonType() } -fun BluePrintTypes.kafkaSslAuthMessageProducerProperties(block: KafkaSslAuthMessageProducerPropertiesAssignmentBuilder.() -> Unit): JsonNode { +fun BlueprintTypes.kafkaSslAuthMessageProducerProperties(block: KafkaSslAuthMessageProducerPropertiesAssignmentBuilder.() -> Unit): JsonNode { val assignments = KafkaSslAuthMessageProducerPropertiesAssignmentBuilder().apply(block).build() assignments[KafkaSslAuthMessageProducerProperties::type.name] = MessageLibConstants.TYPE_KAFKA_SSL_AUTH.asJsonPrimitive() return assignments.asJsonType() } -fun BluePrintTypes.kafkaScramSslAuthMessageProducerProperties(block: KafkaScramSslAuthMessageProducerPropertiesAssignmentBuilder.() -> Unit): JsonNode { +fun BlueprintTypes.kafkaScramSslAuthMessageProducerProperties(block: KafkaScramSslAuthMessageProducerPropertiesAssignmentBuilder.() -> Unit): JsonNode { val assignments = KafkaScramSslAuthMessageProducerPropertiesAssignmentBuilder().apply(block).build() assignments[KafkaScramSslAuthMessageProducerProperties::type.name] = MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH.asJsonPrimitive() @@ -245,88 +245,88 @@ fun TopologyTemplateBuilder.relationshipTemplateMessageConsumer( class MessageConsumerRelationshipTemplateBuilder(name: String, description: String) : RelationshipTemplateBuilder( name, - BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_CONSUMER, description + BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_CONSUMER, description ) { fun kafkaBasicAuth(block: KafkaBasicAuthMessageConsumerPropertiesAssignmentBuilder.() -> Unit) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintTypes.kafkaBasicAuthMessageConsumerProperties(block) + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintTypes.kafkaBasicAuthMessageConsumerProperties(block) ) } fun kafkaSslAuth(block: KafkaSslAuthMessageConsumerPropertiesAssignmentBuilder.() -> Unit) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintTypes.kafkaSslAuthMessageConsumerProperties(block) + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintTypes.kafkaSslAuthMessageConsumerProperties(block) ) } fun kafkaScramSslAuth(block: KafkaScramSslAuthMessageConsumerPropertiesAssignmentBuilder.() -> Unit) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintTypes.kafkaScramSslAuthMessageConsumerProperties(block) + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintTypes.kafkaScramSslAuthMessageConsumerProperties(block) ) } fun kafkaStreamsBasicAuth(block: KafkaStreamsBasicAuthConsumerPropertiesAssignmentBuilder.() -> Unit) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintTypes.kafkaStreamsBasicAuthConsumerProperties(block) + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintTypes.kafkaStreamsBasicAuthConsumerProperties(block) ) } fun kafkaStreamsSslAuth(block: KafkaStreamsSslAuthConsumerPropertiesAssignmentBuilder.() -> Unit) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintTypes.kafkaStreamsSslAuthConsumerProperties(block) + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintTypes.kafkaStreamsSslAuthConsumerProperties(block) ) } fun kafkaStreamsScramSslAuth(block: KafkaStreamsScramSslAuthConsumerPropertiesAssignmentBuilder.() -> Unit) { property( - BluePrintConstants.PROPERTY_CONNECTION_CONFIG, - BluePrintTypes.kafkaStreamsScramSslAuthConsumerProperties(block) + BlueprintConstants.PROPERTY_CONNECTION_CONFIG, + BlueprintTypes.kafkaStreamsScramSslAuthConsumerProperties(block) ) } } -fun BluePrintTypes.kafkaBasicAuthMessageConsumerProperties(block: KafkaBasicAuthMessageConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { +fun BlueprintTypes.kafkaBasicAuthMessageConsumerProperties(block: KafkaBasicAuthMessageConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { val assignments = KafkaBasicAuthMessageConsumerPropertiesAssignmentBuilder().apply(block).build() assignments[KafkaBasicAuthMessageConsumerProperties::type.name] = MessageLibConstants.TYPE_KAFKA_BASIC_AUTH.asJsonPrimitive() return assignments.asJsonType() } -fun BluePrintTypes.kafkaSslAuthMessageConsumerProperties(block: KafkaSslAuthMessageConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { +fun BlueprintTypes.kafkaSslAuthMessageConsumerProperties(block: KafkaSslAuthMessageConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { val assignments = KafkaSslAuthMessageConsumerPropertiesAssignmentBuilder().apply(block).build() assignments[KafkaSslAuthMessageConsumerProperties::type.name] = MessageLibConstants.TYPE_KAFKA_SSL_AUTH.asJsonPrimitive() return assignments.asJsonType() } -fun BluePrintTypes.kafkaScramSslAuthMessageConsumerProperties(block: KafkaScramSslAuthMessageConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { +fun BlueprintTypes.kafkaScramSslAuthMessageConsumerProperties(block: KafkaScramSslAuthMessageConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { val assignments = KafkaScramSslAuthMessageConsumerPropertiesAssignmentBuilder().apply(block).build() assignments[KafkaScramSslAuthMessageConsumerProperties::type.name] = MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH.asJsonPrimitive() return assignments.asJsonType() } -fun BluePrintTypes.kafkaStreamsBasicAuthConsumerProperties(block: KafkaStreamsBasicAuthConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { +fun BlueprintTypes.kafkaStreamsBasicAuthConsumerProperties(block: KafkaStreamsBasicAuthConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { val assignments = KafkaStreamsBasicAuthConsumerPropertiesAssignmentBuilder().apply(block).build() assignments[KafkaStreamsBasicAuthConsumerProperties::type.name] = MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH.asJsonPrimitive() return assignments.asJsonType() } -fun BluePrintTypes.kafkaStreamsSslAuthConsumerProperties(block: KafkaStreamsSslAuthConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { +fun BlueprintTypes.kafkaStreamsSslAuthConsumerProperties(block: KafkaStreamsSslAuthConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { val assignments = KafkaStreamsSslAuthConsumerPropertiesAssignmentBuilder().apply(block).build() assignments[KafkaStreamsSslAuthConsumerProperties::type.name] = MessageLibConstants.TYPE_KAFKA_STREAMS_SSL_AUTH.asJsonPrimitive() return assignments.asJsonType() } -fun BluePrintTypes.kafkaStreamsScramSslAuthConsumerProperties(block: KafkaStreamsScramSslAuthConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { +fun BlueprintTypes.kafkaStreamsScramSslAuthConsumerProperties(block: KafkaStreamsScramSslAuthConsumerPropertiesAssignmentBuilder.() -> Unit): JsonNode { val assignments = KafkaStreamsScramSslAuthConsumerPropertiesAssignmentBuilder().apply(block).build() assignments[KafkaStreamsScramSslAuthConsumerProperties::type.name] = MessageLibConstants.TYPE_KAFKA_STREAMS_SCRAM_SSL_AUTH.asJsonPrimitive() diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/AbstractKafkaTopologyComponents.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/AbstractKafkaTopologyComponents.kt index 72a70893a..23eca18a3 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/AbstractKafkaTopologyComponents.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/AbstractKafkaTopologyComponents.kt @@ -24,9 +24,9 @@ import org.apache.kafka.streams.processor.Punctuator import org.onap.ccsdk.cds.controllerblueprints.core.logger /** CDS Kafka Stream Processor abstract class to implement */ -abstract class AbstractBluePrintMessageProcessor : Processor { +abstract class AbstractBlueprintMessageProcessor : Processor { - private val log = logger(AbstractBluePrintMessageProcessor::class) + private val log = logger(AbstractBlueprintMessageProcessor::class) lateinit var processorContext: ProcessorContext @@ -51,7 +51,7 @@ abstract class AbstractBluePrintMessageProcessor : Processor { } /** CDS Kafka Stream Punctuator abstract class to implement */ -abstract class AbstractBluePrintMessagePunctuator : Punctuator { +abstract class AbstractBlueprintMessagePunctuator : Punctuator { lateinit var processorContext: ProcessorContext diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/KafkaJDBCStores.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/KafkaJDBCStores.kt index 86ccd74a2..ecf5870ab 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/KafkaJDBCStores.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/KafkaJDBCStores.kt @@ -21,10 +21,10 @@ import org.apache.kafka.streams.processor.ProcessorContext import org.apache.kafka.streams.processor.StateStore import org.apache.kafka.streams.state.StoreBuilder import org.apache.kafka.streams.state.StoreSupplier -import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibGenericService +import org.onap.ccsdk.cds.blueprintsprocessor.db.BlueprintDBLibGenericService import org.onap.ccsdk.cds.blueprintsprocessor.db.primaryDBLibGenericService import org.onap.ccsdk.cds.controllerblueprints.core.logger -import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService +import org.onap.ccsdk.cds.controllerblueprints.core.service.BlueprintDependencyService import java.util.* @@ -32,7 +32,7 @@ class KafkaJDBCKeyStoreSupplier(private val name: String) : StoreSupplier { - bluePrintPropertiesService.propertyBeanType( - prefix, KafkaBasicAuthMessageProducerProperties::class.java - ) - } - MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { - bluePrintPropertiesService.propertyBeanType( - prefix, KafkaSslAuthMessageProducerProperties::class.java - ) - } - MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { - bluePrintPropertiesService.propertyBeanType( - prefix, KafkaScramSslAuthMessageProducerProperties::class.java - ) - } - else -> { - throw BluePrintProcessorException("Message adaptor($type) is not supported") - } - } - } - - fun messageProducerProperties(jsonNode: JsonNode): MessageProducerProperties { - val type = jsonNode.get("type").textValue() - return when (type) { - MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> { - JacksonUtils.readValue(jsonNode, KafkaBasicAuthMessageProducerProperties::class.java)!! - } - MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { - JacksonUtils.readValue(jsonNode, KafkaSslAuthMessageProducerProperties::class.java)!! - } - MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { - JacksonUtils.readValue(jsonNode, KafkaScramSslAuthMessageProducerProperties::class.java)!! - } - else -> { - throw BluePrintProcessorException("Message adaptor($type) is not supported") - } - } - } - - /** 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 = bluePrintPropertiesService.propertyBeanType("$prefix.type", String::class.java) - return when (type) { - /** Message Consumer */ - MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> { - bluePrintPropertiesService.propertyBeanType( - prefix, KafkaBasicAuthMessageConsumerProperties::class.java - ) - } - MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { - bluePrintPropertiesService.propertyBeanType( - prefix, KafkaSslAuthMessageConsumerProperties::class.java - ) - } - MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { - bluePrintPropertiesService.propertyBeanType( - prefix, KafkaScramSslAuthMessageConsumerProperties::class.java - ) - } - /** Stream Consumer */ - MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> { - bluePrintPropertiesService.propertyBeanType( - prefix, KafkaStreamsBasicAuthConsumerProperties::class.java - ) - } - MessageLibConstants.TYPE_KAFKA_STREAMS_SSL_AUTH -> { - bluePrintPropertiesService.propertyBeanType( - prefix, KafkaStreamsSslAuthConsumerProperties::class.java - ) - } - MessageLibConstants.TYPE_KAFKA_STREAMS_SCRAM_SSL_AUTH -> { - bluePrintPropertiesService.propertyBeanType( - prefix, KafkaStreamsScramSslAuthConsumerProperties::class.java - ) - } - else -> { - throw BluePrintProcessorException("Message adaptor($type) is not supported") - } - } - } - - fun messageConsumerProperties(jsonNode: JsonNode): MessageConsumerProperties { - val type = jsonNode.get("type").textValue() - return when (type) { - /** Message Consumer */ - MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> { - JacksonUtils.readValue(jsonNode, KafkaBasicAuthMessageConsumerProperties::class.java)!! - } - MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { - JacksonUtils.readValue(jsonNode, KafkaSslAuthMessageConsumerProperties::class.java)!! - } - MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { - JacksonUtils.readValue(jsonNode, KafkaScramSslAuthMessageConsumerProperties::class.java)!! - } - /** Stream Consumer */ - MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> { - JacksonUtils.readValue(jsonNode, KafkaStreamsBasicAuthConsumerProperties::class.java)!! - } - MessageLibConstants.TYPE_KAFKA_STREAMS_SSL_AUTH -> { - JacksonUtils.readValue(jsonNode, KafkaStreamsSslAuthConsumerProperties::class.java)!! - } - MessageLibConstants.TYPE_KAFKA_STREAMS_SCRAM_SSL_AUTH -> { - JacksonUtils.readValue(jsonNode, KafkaStreamsScramSslAuthConsumerProperties::class.java)!! - } - else -> { - throw BluePrintProcessorException("Message adaptor($type) is not supported") - } - } - } - - private fun blueprintMessageConsumerService(messageConsumerProperties: MessageConsumerProperties): - BlueprintMessageConsumerService { - - when (messageConsumerProperties.type) { - /** Message Consumer */ - MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> { - return KafkaMessageConsumerService( - messageConsumerProperties as KafkaBasicAuthMessageConsumerProperties - ) - } - MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { - return KafkaMessageConsumerService( - messageConsumerProperties as KafkaSslAuthMessageConsumerProperties - ) - } - MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { - return KafkaMessageConsumerService( - messageConsumerProperties as KafkaScramSslAuthMessageConsumerProperties - ) - } - /** Stream Consumer */ - MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> { - return KafkaStreamsConsumerService( - messageConsumerProperties as KafkaStreamsBasicAuthConsumerProperties - ) - } - MessageLibConstants.TYPE_KAFKA_STREAMS_SSL_AUTH -> { - return KafkaStreamsConsumerService( - messageConsumerProperties as KafkaStreamsSslAuthConsumerProperties - ) - } - MessageLibConstants.TYPE_KAFKA_STREAMS_SCRAM_SSL_AUTH -> { - return KafkaStreamsConsumerService( - messageConsumerProperties as KafkaStreamsScramSslAuthConsumerProperties - ) - } - else -> { - throw BluePrintProcessorException("couldn't get message client service for ${messageConsumerProperties.type}") - } - } - } -} diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerService.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerService.kt index b39d89bfd..887a7a7b8 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerService.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerService.kt @@ -23,7 +23,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord import org.apache.kafka.clients.consumer.ConsumerRecords import org.apache.kafka.streams.Topology import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties -import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException +import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintProcessorException /** Consumer Function Interfaces */ interface ConsumerFunction @@ -47,7 +47,7 @@ interface BlueprintMessageConsumerService { /** Consume with [additionalConfig], so that we can execute dynamic function [consumerFunction] */ suspend fun consume(additionalConfig: Map?, consumerFunction: ConsumerFunction) { - throw BluePrintProcessorException("Not Implemented") + throw BlueprintProcessorException("Not Implemented") } /** Consume the [topics] with [additionalConfig], so that we can execute dynamic function [consumerFunction] */ @@ -56,7 +56,7 @@ interface BlueprintMessageConsumerService { additionalConfig: Map?, consumerFunction: ConsumerFunction ) { - throw BluePrintProcessorException("Not Implemented") + throw BlueprintProcessorException("Not Implemented") } /** close the channel, consumer and other resources */ diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageLibPropertyService.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageLibPropertyService.kt new file mode 100644 index 000000000..9e0c537cf --- /dev/null +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageLibPropertyService.kt @@ -0,0 +1,221 @@ +/* + * Copyright © 2019 IBM. + * Modifications Copyright © 2018-2019 AT&T Intellectual Property. + * + * 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 com.fasterxml.jackson.databind.JsonNode +import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertiesService +import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageConsumerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageProducerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaScramSslAuthMessageConsumerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaScramSslAuthMessageProducerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaSslAuthMessageConsumerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaSslAuthMessageProducerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsBasicAuthConsumerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsScramSslAuthConsumerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsSslAuthConsumerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties +import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageLibConstants +import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageProducerProperties +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 bluePrintPropertiesService: BlueprintPropertiesService) { + + fun blueprintMessageProducerService(jsonNode: JsonNode): BlueprintMessageProducerService { + val messageClientProperties = messageProducerProperties(jsonNode) + return KafkaMessageProducerService(messageClientProperties) + } + + fun blueprintMessageProducerService(selector: String): BlueprintMessageProducerService { + val prefix = "${MessageLibConstants.PROPERTY_MESSAGE_PRODUCER_PREFIX}$selector" + val messageClientProperties = messageProducerProperties(prefix) + return KafkaMessageProducerService(messageClientProperties) + } + + fun messageProducerProperties(prefix: String): MessageProducerProperties { + val type = bluePrintPropertiesService.propertyBeanType("$prefix.type", String::class.java) + return when (type) { + MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> { + bluePrintPropertiesService.propertyBeanType( + prefix, KafkaBasicAuthMessageProducerProperties::class.java + ) + } + MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { + bluePrintPropertiesService.propertyBeanType( + prefix, KafkaSslAuthMessageProducerProperties::class.java + ) + } + MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { + bluePrintPropertiesService.propertyBeanType( + prefix, KafkaScramSslAuthMessageProducerProperties::class.java + ) + } + else -> { + throw BlueprintProcessorException("Message adaptor($type) is not supported") + } + } + } + + fun messageProducerProperties(jsonNode: JsonNode): MessageProducerProperties { + val type = jsonNode.get("type").textValue() + return when (type) { + MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> { + JacksonUtils.readValue(jsonNode, KafkaBasicAuthMessageProducerProperties::class.java)!! + } + MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { + JacksonUtils.readValue(jsonNode, KafkaSslAuthMessageProducerProperties::class.java)!! + } + MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { + JacksonUtils.readValue(jsonNode, KafkaScramSslAuthMessageProducerProperties::class.java)!! + } + else -> { + throw BlueprintProcessorException("Message adaptor($type) is not supported") + } + } + } + + /** 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 = bluePrintPropertiesService.propertyBeanType("$prefix.type", String::class.java) + return when (type) { + /** Message Consumer */ + MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> { + bluePrintPropertiesService.propertyBeanType( + prefix, KafkaBasicAuthMessageConsumerProperties::class.java + ) + } + MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { + bluePrintPropertiesService.propertyBeanType( + prefix, KafkaSslAuthMessageConsumerProperties::class.java + ) + } + MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { + bluePrintPropertiesService.propertyBeanType( + prefix, KafkaScramSslAuthMessageConsumerProperties::class.java + ) + } + /** Stream Consumer */ + MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> { + bluePrintPropertiesService.propertyBeanType( + prefix, KafkaStreamsBasicAuthConsumerProperties::class.java + ) + } + MessageLibConstants.TYPE_KAFKA_STREAMS_SSL_AUTH -> { + bluePrintPropertiesService.propertyBeanType( + prefix, KafkaStreamsSslAuthConsumerProperties::class.java + ) + } + MessageLibConstants.TYPE_KAFKA_STREAMS_SCRAM_SSL_AUTH -> { + bluePrintPropertiesService.propertyBeanType( + prefix, KafkaStreamsScramSslAuthConsumerProperties::class.java + ) + } + else -> { + throw BlueprintProcessorException("Message adaptor($type) is not supported") + } + } + } + + fun messageConsumerProperties(jsonNode: JsonNode): MessageConsumerProperties { + val type = jsonNode.get("type").textValue() + return when (type) { + /** Message Consumer */ + MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> { + JacksonUtils.readValue(jsonNode, KafkaBasicAuthMessageConsumerProperties::class.java)!! + } + MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { + JacksonUtils.readValue(jsonNode, KafkaSslAuthMessageConsumerProperties::class.java)!! + } + MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { + JacksonUtils.readValue(jsonNode, KafkaScramSslAuthMessageConsumerProperties::class.java)!! + } + /** Stream Consumer */ + MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> { + JacksonUtils.readValue(jsonNode, KafkaStreamsBasicAuthConsumerProperties::class.java)!! + } + MessageLibConstants.TYPE_KAFKA_STREAMS_SSL_AUTH -> { + JacksonUtils.readValue(jsonNode, KafkaStreamsSslAuthConsumerProperties::class.java)!! + } + MessageLibConstants.TYPE_KAFKA_STREAMS_SCRAM_SSL_AUTH -> { + JacksonUtils.readValue(jsonNode, KafkaStreamsScramSslAuthConsumerProperties::class.java)!! + } + else -> { + throw BlueprintProcessorException("Message adaptor($type) is not supported") + } + } + } + + private fun blueprintMessageConsumerService(messageConsumerProperties: MessageConsumerProperties): + BlueprintMessageConsumerService { + + when (messageConsumerProperties.type) { + /** Message Consumer */ + MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> { + return KafkaMessageConsumerService( + messageConsumerProperties as KafkaBasicAuthMessageConsumerProperties + ) + } + MessageLibConstants.TYPE_KAFKA_SSL_AUTH -> { + return KafkaMessageConsumerService( + messageConsumerProperties as KafkaSslAuthMessageConsumerProperties + ) + } + MessageLibConstants.TYPE_KAFKA_SCRAM_SSL_AUTH -> { + return KafkaMessageConsumerService( + messageConsumerProperties as KafkaScramSslAuthMessageConsumerProperties + ) + } + /** Stream Consumer */ + MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> { + return KafkaStreamsConsumerService( + messageConsumerProperties as KafkaStreamsBasicAuthConsumerProperties + ) + } + MessageLibConstants.TYPE_KAFKA_STREAMS_SSL_AUTH -> { + return KafkaStreamsConsumerService( + messageConsumerProperties as KafkaStreamsSslAuthConsumerProperties + ) + } + MessageLibConstants.TYPE_KAFKA_STREAMS_SCRAM_SSL_AUTH -> { + return KafkaStreamsConsumerService( + messageConsumerProperties as KafkaStreamsScramSslAuthConsumerProperties + ) + } + else -> { + throw BlueprintProcessorException("couldn't get message client service for ${messageConsumerProperties.type}") + } + } + } +} diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerService.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerService.kt index 4340e4815..beaa995df 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerService.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerService.kt @@ -20,7 +20,7 @@ import kotlinx.coroutines.channels.Channel import org.apache.kafka.clients.consumer.ConsumerRecord import org.apache.kafka.streams.KafkaStreams import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties -import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException +import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintProcessorException import org.onap.ccsdk.cds.controllerblueprints.core.logger import java.util.Properties @@ -41,11 +41,11 @@ open class KafkaStreamsConsumerService(private val messageConsumerProperties: Me } override suspend fun subscribe(additionalConfig: Map?): Channel> { - throw BluePrintProcessorException("not implemented") + throw BlueprintProcessorException("not implemented") } override suspend fun subscribe(topics: List, additionalConfig: Map?): Channel> { - throw BluePrintProcessorException("not implemented") + throw BlueprintProcessorException("not implemented") } override suspend fun consume(additionalConfig: Map?, consumerFunction: ConsumerFunction) { diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerService.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerService.kt index 90b850017..7e5f35d15 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerService.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerService.kt @@ -21,7 +21,7 @@ import org.apache.kafka.common.header.Headers import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.CommonHeader import org.onap.ccsdk.cds.blueprintsprocessor.message.addHeader import org.onap.ccsdk.cds.blueprintsprocessor.message.toMap -import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants +import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintConstants import org.onap.ccsdk.cds.controllerblueprints.core.defaultToEmpty import org.onap.ccsdk.cds.controllerblueprints.core.defaultToUUID import org.onap.ccsdk.cds.controllerblueprints.core.logger @@ -46,9 +46,9 @@ class MessageLoggerService { fun messageConsuming(consumerRecord: ConsumerRecord<*, *>) { val headers = consumerRecord.headers().toMap() - val requestID = headers[BluePrintConstants.ONAP_REQUEST_ID].defaultToUUID() - val invocationID = headers[BluePrintConstants.ONAP_INVOCATION_ID].defaultToUUID() - val partnerName = headers[BluePrintConstants.ONAP_PARTNER_NAME] ?: "UNKNOWN" + val requestID = headers[BlueprintConstants.ONAP_REQUEST_ID].defaultToUUID() + val invocationID = headers[BlueprintConstants.ONAP_INVOCATION_ID].defaultToUUID() + val partnerName = headers[BlueprintConstants.ONAP_PARTNER_NAME] ?: "UNKNOWN" messageConsuming(requestID, invocationID, partnerName, consumerRecord) } @@ -83,9 +83,9 @@ class MessageLoggerService { */ fun messageProducing(requestHeader: Headers) { val localhost = InetAddress.getLocalHost() - requestHeader.addHeader(BluePrintConstants.ONAP_REQUEST_ID, MDC.get("InvocationID").defaultToUUID()) - requestHeader.addHeader(BluePrintConstants.ONAP_INVOCATION_ID, UUID.randomUUID().toString()) - requestHeader.addHeader(BluePrintConstants.ONAP_PARTNER_NAME, BluePrintConstants.APP_NAME) + requestHeader.addHeader(BlueprintConstants.ONAP_REQUEST_ID, MDC.get("InvocationID").defaultToUUID()) + requestHeader.addHeader(BlueprintConstants.ONAP_INVOCATION_ID, UUID.randomUUID().toString()) + requestHeader.addHeader(BlueprintConstants.ONAP_PARTNER_NAME, BlueprintConstants.APP_NAME) requestHeader.addHeader("ClientIPAddress", localhost.hostAddress) } diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/MessagePropertiesDSLTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/MessagePropertiesDSLTest.kt index b68678baf..8130016a8 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/MessagePropertiesDSLTest.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/MessagePropertiesDSLTest.kt @@ -18,7 +18,7 @@ package org.onap.ccsdk.cds.blueprintsprocessor.message import org.apache.kafka.streams.StreamsConfig import org.junit.Test -import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants +import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintConstants import org.onap.ccsdk.cds.controllerblueprints.core.dsl.relationshipTypeConnectsTo import org.onap.ccsdk.cds.controllerblueprints.core.dsl.serviceTemplate import kotlin.test.assertEquals @@ -67,12 +67,12 @@ class MessagePropertiesDSLTest { assertNotNull(relationshipTypes, "failed to get relationship types") assertEquals(2, relationshipTypes.size, "relationshipTypes doesn't match") assertNotNull( - relationshipTypes[BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO], - "failed to get ${BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO}" + relationshipTypes[BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO], + "failed to get ${BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO}" ) assertNotNull( - relationshipTypes[BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_PRODUCER], - "failed to get ${BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_PRODUCER}" + relationshipTypes[BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_PRODUCER], + "failed to get ${BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_PRODUCER}" ) } @@ -138,12 +138,12 @@ class MessagePropertiesDSLTest { assertNotNull(relationshipTypes, "failed to get relationship types") assertEquals(2, relationshipTypes.size, "relationshipTypes doesn't match") assertNotNull( - relationshipTypes[BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO], - "failed to get ${BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO}" + relationshipTypes[BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO], + "failed to get ${BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO}" ) assertNotNull( - relationshipTypes[BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_CONSUMER], - "failed to get ${BluePrintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_CONSUMER}" + relationshipTypes[BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_CONSUMER], + "failed to get ${BlueprintConstants.MODEL_TYPE_RELATIONSHIPS_CONNECTS_TO_MESSAGE_CONSUMER}" ) } } diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt index a69f9f51a..f240f76c0 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt @@ -40,9 +40,9 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer import org.apache.kafka.common.serialization.StringDeserializer import org.junit.Test import org.junit.runner.RunWith -import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService -import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration -import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration +import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertiesService +import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration +import org.onap.ccsdk.cds.blueprintsprocessor.message.BlueprintMessageLibConfiguration import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageLibConstants import org.onap.ccsdk.cds.controllerblueprints.core.logger @@ -60,8 +60,8 @@ import kotlin.test.assertTrue @DirtiesContext @ContextConfiguration( classes = [ - BluePrintMessageLibConfiguration::class, - BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class + BlueprintMessageLibConfiguration::class, + BlueprintPropertyConfiguration::class, BlueprintPropertiesService::class ] ) @TestPropertySource( @@ -98,7 +98,7 @@ open class BlueprintMessageConsumerServiceTest { val log = logger(BlueprintMessageConsumerServiceTest::class) @Autowired - lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService + lateinit var bluePrintMessageLibPropertyService: BlueprintMessageLibPropertyService @Test fun testKafkaBasicAuthConsumerService() { diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt index f88caa173..2293ceec3 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt @@ -32,9 +32,9 @@ import org.apache.kafka.common.security.scram.ScramLoginModule import org.apache.kafka.common.serialization.ByteArraySerializer import org.apache.kafka.common.serialization.StringSerializer import org.junit.runner.RunWith -import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService -import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration -import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration +import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertiesService +import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration +import org.onap.ccsdk.cds.blueprintsprocessor.message.BlueprintMessageLibConfiguration import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageLibConstants import org.springframework.beans.factory.annotation.Autowired import org.springframework.test.annotation.DirtiesContext @@ -50,8 +50,8 @@ import kotlin.test.assertTrue @DirtiesContext @ContextConfiguration( classes = [ - BluePrintMessageLibConfiguration::class, - BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class + BlueprintMessageLibConfiguration::class, + BlueprintPropertyConfiguration::class, BlueprintPropertiesService::class ] ) @TestPropertySource( @@ -72,7 +72,7 @@ import kotlin.test.assertTrue open class BlueprintMessageProducerServiceTest { @Autowired - lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService + lateinit var bluePrintMessageLibPropertyService: BlueprintMessageLibPropertyService @Test fun testKafkaScramSslAuthProducerService() { @@ -87,11 +87,11 @@ open class BlueprintMessageProducerServiceTest { every { mockKafkaTemplate.send(any(), any()) } returns responseMock - val spyBluePrintMessageProducerService = spyk(blueprintMessageProducerService, recordPrivateCalls = true) + val spyBlueprintMessageProducerService = spyk(blueprintMessageProducerService, recordPrivateCalls = true) - every { spyBluePrintMessageProducerService.messageTemplate(any()) } returns mockKafkaTemplate + every { spyBlueprintMessageProducerService.messageTemplate(any()) } returns mockKafkaTemplate - val response = spyBluePrintMessageProducerService.sendMessage("mykey", "Testing message") + val response = spyBlueprintMessageProducerService.sendMessage("mykey", "Testing message") assertTrue(response, "failed to get command response") } } diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerServiceTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerServiceTest.kt index f488a4c74..70968ef0c 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerServiceTest.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerServiceTest.kt @@ -26,9 +26,9 @@ import org.apache.kafka.streams.processor.ProcessorSupplier import org.apache.kafka.streams.state.Stores import org.junit.Test import org.junit.runner.RunWith -import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService -import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration -import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration +import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertiesService +import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration +import org.onap.ccsdk.cds.blueprintsprocessor.message.BlueprintMessageLibConfiguration import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsBasicAuthConsumerProperties import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties import org.springframework.beans.factory.annotation.Autowired @@ -42,8 +42,8 @@ import kotlin.test.assertNotNull @DirtiesContext @ContextConfiguration( classes = [ - BluePrintMessageLibConfiguration::class, - BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class + BlueprintMessageLibConfiguration::class, + BlueprintPropertyConfiguration::class, BlueprintPropertiesService::class ] ) @TestPropertySource( @@ -72,7 +72,7 @@ import kotlin.test.assertNotNull class KafkaStreamsConsumerServiceTest { @Autowired - lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService + lateinit var bluePrintMessageLibPropertyService: BlueprintMessageLibPropertyService @Test fun testProperties() { diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerServiceTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerServiceTest.kt index 3dce3344f..dc36ac675 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerServiceTest.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MessageLoggerServiceTest.kt @@ -23,7 +23,7 @@ import org.apache.kafka.common.header.internals.RecordHeaders import org.junit.Test import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.CommonHeader import org.onap.ccsdk.cds.blueprintsprocessor.message.toMap -import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants +import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintConstants import org.slf4j.MDC import kotlin.test.assertEquals @@ -51,7 +51,7 @@ class MessageLoggerServiceTest { val mockHeaders = RecordHeaders() messageLoggerService.messageProducing(mockHeaders) val map = mockHeaders.toMap() - assertEquals("1234-12", map[BluePrintConstants.ONAP_REQUEST_ID]) + assertEquals("1234-12", map[BlueprintConstants.ONAP_REQUEST_ID]) messageLoggerService.messageConsumingExisting() } diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MockKafkaTopologyComponents.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MockKafkaTopologyComponents.kt index aa38b6e5c..97a81f1cd 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MockKafkaTopologyComponents.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/MockKafkaTopologyComponents.kt @@ -22,7 +22,7 @@ import org.apache.kafka.common.serialization.Serializer import org.apache.kafka.streams.processor.Processor import org.apache.kafka.streams.processor.ProcessorContext import org.apache.kafka.streams.state.KeyValueStore -import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException +import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintProcessorException import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString import org.onap.ccsdk.cds.controllerblueprints.core.logger import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils @@ -48,7 +48,7 @@ open class PriorityMessageSerde : Serde { return object : Deserializer { override fun deserialize(topic: String, data: ByteArray): PriorityMessage { return JacksonUtils.readValue(String(data), PriorityMessage::class.java) - ?: throw BluePrintProcessorException("failed to convert") + ?: throw BlueprintProcessorException("failed to convert") } override fun configure(configs: MutableMap?, isKey: Boolean) { -- cgit 1.2.3-korg