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

   1 /*
   2  * Copyright © 2018-2019 AT&T Intellectual Property.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 package org.onap.ccsdk.cds.blueprintsprocessor.message.kafka
  18
  19 /*
  20 import org.apache.kafka.streams.processor.ProcessorContext
  21 import org.apache.kafka.streams.processor.StateStore
  22 import org.apache.kafka.streams.state.StoreBuilder
  23 import org.apache.kafka.streams.state.StoreSupplier
  24 import org.onap.ccsdk.cds.blueprintsprocessor.db.BlueprintDBLibGenericService
  25 import org.onap.ccsdk.cds.blueprintsprocessor.db.primaryDBLibGenericService
  26 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  27 import org.onap.ccsdk.cds.controllerblueprints.core.service.BlueprintDependencyService
  28 import java.util.*
  29
  30
  31 class KafkaJDBCKeyStoreSupplier(private val name: String) : StoreSupplier<KafkaJDBCStore> {
  32
  33     override fun get(): KafkaJDBCStore {
  34         // Get the DBLibGenericService Instance
  35         val bluePrintDBLibGenericService = BlueprintDependencyService.primaryDBLibGenericService()
  36         return KafkaJDBCStoreImpl(name, bluePrintDBLibGenericService)
  37     }
  38
  39     override fun name(): String {
  40         return name
  41     }
  42
  43     override fun metricsScope(): String {
  44         return "jdbc-state"
  45     }
  46 }
  47
  48 class KafkaJDBCKeyStoreBuilder(private val storeSupplier: KafkaJDBCKeyStoreSupplier)
  49     : StoreBuilder<KafkaJDBCStore> {
  50
  51     private var logConfig: MutableMap<String, String> = HashMap()
  52     private var enableCaching: Boolean = false
  53     private var enableLogging = true
  54
  55     override fun logConfig(): MutableMap<String, String> {
  56         return logConfig
  57     }
  58
  59     override fun withCachingDisabled(): StoreBuilder<KafkaJDBCStore> {
  60         enableCaching = false
  61         return this
  62     }
  63
  64     override fun loggingEnabled(): Boolean {
  65         return enableLogging
  66     }
  67
  68     override fun withLoggingDisabled(): StoreBuilder<KafkaJDBCStore> {
  69         enableLogging = false
  70         return this
  71     }
  72
  73     override fun withCachingEnabled(): StoreBuilder<KafkaJDBCStore> {
  74         enableCaching = true
  75         return this
  76     }
  77
  78     override fun withLoggingEnabled(config: MutableMap<String, String>?): StoreBuilder<KafkaJDBCStore> {
  79         enableLogging = true
  80         return this
  81     }
  82
  83     override fun name(): String {
  84         return "KafkaJDBCKeyStoreBuilder"
  85     }
  86
  87     override fun build(): KafkaJDBCStore {
  88         return storeSupplier.get()
  89     }
  90 }
  91
  92 interface KafkaJDBCStore : StateStore {
  93
  94     suspend fun query(sql: String, params: Map<String, Any>): List<Map<String, Any>>
  95
  96     suspend fun update(sql: String, params: Map<String, Any>): Int
  97 }
  98
  99
 100 class KafkaJDBCStoreImpl(private val name: String,
 101                          private val bluePrintDBLibGenericService: BlueprintDBLibGenericService)
 102     : KafkaJDBCStore {
 103
 104     private val log = logger(KafkaJDBCStoreImpl::class)
 105
 106     override fun isOpen(): Boolean {
 107         log.info("isOpen...")
 108         return true
 109     }
 110
 111     override fun init(context: ProcessorContext, root: StateStore) {
 112         log.info("init...")
 113     }
 114
 115     override fun flush() {
 116         log.info("flush...")
 117     }
 118
 119     override fun close() {
 120         log.info("Close...")
 121     }
 122
 123     override fun name(): String {
 124         return name
 125     }
 126
 127     override fun persistent(): Boolean {
 128         return true
 129     }
 130
 131     override suspend fun query(sql: String, params: Map<String, Any>): List<Map<String, Any>> {
 132         log.info("Query : $sql")
 133         log.info("Params : $params")
 134         return bluePrintDBLibGenericService.query(sql, params)
 135     }
 136
 137     override suspend fun update(sql: String, params: Map<String, Any>): Int {
 138         log.info("Query : $sql")
 139         log.info("Params : $params")
 140         return bluePrintDBLibGenericService.update(sql, params)
 141     }
 142 }
 143 */