diff options
Diffstat (limited to 'ms/blueprintsprocessor/modules/commons/message-lib')
3 files changed, 209 insertions, 0 deletions
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 new file mode 100644 index 000000000..4c6c0acdd --- /dev/null +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/AbstractKafkaTopologyComponents.kt @@ -0,0 +1,65 @@ +/* + * 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.kafka + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.runBlocking +import org.apache.kafka.streams.processor.Processor +import org.apache.kafka.streams.processor.ProcessorContext +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<K, V> : Processor<K, V> { + + private val log = logger(AbstractBluePrintMessageProcessor::class) + + lateinit var processorContext: ProcessorContext + + + override fun process(key: K, value: V) = runBlocking(Dispatchers.IO) { + try { + processNB(key, value) + } catch (e: Exception) { + log.error("failed in processor(${this.javaClass.simpleName}) message(${this.javaClass.simpleName} :", e) + } + } + + override fun init(context: ProcessorContext) { + log.info("initializing processor (${this.javaClass.simpleName})") + this.processorContext = context + + } + + override fun close() { + log.info("closing processor (${this.javaClass.simpleName})") + } + + abstract suspend fun processNB(key: K, value: V) +} + +/** CDS Kafka Stream Punctuator abstract class to implement */ +abstract class AbstractBluePrintMessagePunctuator : Punctuator { + lateinit var processorContext: ProcessorContext + + + override fun punctuate(timestamp: Long) = runBlocking(Dispatchers.IO) { + punctuateNB(timestamp) + } + + abstract suspend fun punctuateNB(timestamp: Long) +}
\ No newline at end of file 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 new file mode 100644 index 000000000..86ccd74a2 --- /dev/null +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/kafka/KafkaJDBCStores.kt @@ -0,0 +1,143 @@ +/* + * 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.kafka + +/* +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.primaryDBLibGenericService +import org.onap.ccsdk.cds.controllerblueprints.core.logger +import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService +import java.util.* + + +class KafkaJDBCKeyStoreSupplier(private val name: String) : StoreSupplier<KafkaJDBCStore> { + + override fun get(): KafkaJDBCStore { + // Get the DBLibGenericService Instance + val bluePrintDBLibGenericService = BluePrintDependencyService.primaryDBLibGenericService() + return KafkaJDBCStoreImpl(name, bluePrintDBLibGenericService) + } + + override fun name(): String { + return name + } + + override fun metricsScope(): String { + return "jdbc-state" + } +} + +class KafkaJDBCKeyStoreBuilder(private val storeSupplier: KafkaJDBCKeyStoreSupplier) + : StoreBuilder<KafkaJDBCStore> { + + private var logConfig: MutableMap<String, String> = HashMap() + private var enableCaching: Boolean = false + private var enableLogging = true + + override fun logConfig(): MutableMap<String, String> { + return logConfig + } + + override fun withCachingDisabled(): StoreBuilder<KafkaJDBCStore> { + enableCaching = false + return this + } + + override fun loggingEnabled(): Boolean { + return enableLogging + } + + override fun withLoggingDisabled(): StoreBuilder<KafkaJDBCStore> { + enableLogging = false + return this + } + + override fun withCachingEnabled(): StoreBuilder<KafkaJDBCStore> { + enableCaching = true + return this + } + + override fun withLoggingEnabled(config: MutableMap<String, String>?): StoreBuilder<KafkaJDBCStore> { + enableLogging = true + return this + } + + override fun name(): String { + return "KafkaJDBCKeyStoreBuilder" + } + + override fun build(): KafkaJDBCStore { + return storeSupplier.get() + } +} + +interface KafkaJDBCStore : StateStore { + + suspend fun query(sql: String, params: Map<String, Any>): List<Map<String, Any>> + + suspend fun update(sql: String, params: Map<String, Any>): Int +} + + +class KafkaJDBCStoreImpl(private val name: String, + private val bluePrintDBLibGenericService: BluePrintDBLibGenericService) + : KafkaJDBCStore { + + private val log = logger(KafkaJDBCStoreImpl::class) + + override fun isOpen(): Boolean { + log.info("isOpen...") + return true + } + + override fun init(context: ProcessorContext, root: StateStore) { + log.info("init...") + } + + override fun flush() { + log.info("flush...") + } + + override fun close() { + log.info("Close...") + } + + override fun name(): String { + return name + } + + override fun persistent(): Boolean { + return true + } + + override suspend fun query(sql: String, params: Map<String, Any>): List<Map<String, Any>> { + log.info("Query : $sql") + log.info("Params : $params") + return bluePrintDBLibGenericService.query(sql, params) + } + + override suspend fun update(sql: String, params: Map<String, Any>): Int { + log.info("Query : $sql") + log.info("Params : $params") + return bluePrintDBLibGenericService.update(sql, params) + } +} +*/ diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsBasicAuthConsumerService.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsBasicAuthConsumerService.kt index 229e462da..d0297df4c 100644 --- a/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsBasicAuthConsumerService.kt +++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsBasicAuthConsumerService.kt @@ -56,6 +56,7 @@ open class KafkaStreamsBasicAuthConsumerService(private val messageConsumerPrope val streamsConfig = streamsConfig(additionalConfig) val kafkaStreamConsumerFunction = consumerFunction as KafkaStreamConsumerFunction val topology = kafkaStreamConsumerFunction.createTopology(messageConsumerProperties, additionalConfig) + log.info("Kafka streams topology : ${topology.describe()}") kafkaStreams = KafkaStreams(topology, streamsConfig) kafkaStreams.cleanUp() kafkaStreams.start() |