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

   1 /*
   2  *  Copyright © 2019 IBM.
   3  *  Modifications Copyright © 2018-2019 AT&T Intellectual Property.
   4  *
   5  *  Licensed under the Apache License, Version 2.0 (the "License");
   6  *  you may not use this file except in compliance with the License.
   7  *  You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  *  Unless required by applicable law or agreed to in writing, software
  12  *  distributed under the License is distributed on an "AS IS" BASIS,
  13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  *  See the License for the specific language governing permissions and
  15  *  limitations under the License.
  16  */
  17
  18 package org.onap.ccsdk.cds.blueprintsprocessor.message.service
  19
  20 import com.fasterxml.jackson.databind.JsonNode
  21 import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
  22 import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageConsumerProperties
  23 import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageProducerProperties
  24 import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsBasicAuthConsumerProperties
  25 import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties
  26 import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageLibConstants
  27 import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageProducerProperties
  28 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
  29 import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
  30 import org.springframework.stereotype.Service
  31
  32 @Service(MessageLibConstants.SERVICE_BLUEPRINT_MESSAGE_LIB_PROPERTY)
  33 open class BluePrintMessageLibPropertyService(private var bluePrintPropertiesService: BluePrintPropertiesService) {
  34
  35     fun blueprintMessageProducerService(jsonNode: JsonNode): BlueprintMessageProducerService {
  36         val messageClientProperties = messageProducerProperties(jsonNode)
  37         return blueprintMessageProducerService(messageClientProperties)
  38     }
  39
  40     fun blueprintMessageProducerService(selector: String): BlueprintMessageProducerService {
  41         val prefix = "${MessageLibConstants.PROPERTY_MESSAGE_PRODUCER_PREFIX}$selector"
  42         val messageClientProperties = messageProducerProperties(prefix)
  43         return blueprintMessageProducerService(messageClientProperties)
  44     }
  45
  46     fun messageProducerProperties(prefix: String): MessageProducerProperties {
  47         val type = bluePrintPropertiesService.propertyBeanType("$prefix.type", String::class.java)
  48         return when (type) {
  49             MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
  50                 kafkaBasicAuthMessageProducerProperties(prefix)
  51             }
  52             else -> {
  53                 throw BluePrintProcessorException("Message adaptor($type) is not supported")
  54             }
  55         }
  56     }
  57
  58     fun messageProducerProperties(jsonNode: JsonNode): MessageProducerProperties {
  59         val type = jsonNode.get("type").textValue()
  60         return when (type) {
  61             MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
  62                 JacksonUtils.readValue(jsonNode, KafkaBasicAuthMessageProducerProperties::class.java)!!
  63             }
  64             else -> {
  65                 throw BluePrintProcessorException("Message adaptor($type) is not supported")
  66             }
  67         }
  68     }
  69
  70     private fun blueprintMessageProducerService(MessageProducerProperties: MessageProducerProperties):
  71             BlueprintMessageProducerService {
  72
  73         when (MessageProducerProperties) {
  74             is KafkaBasicAuthMessageProducerProperties -> {
  75                 return KafkaBasicAuthMessageProducerService(MessageProducerProperties)
  76             }
  77             else -> {
  78                 throw BluePrintProcessorException("couldn't get Message client service for")
  79             }
  80         }
  81     }
  82
  83     private fun kafkaBasicAuthMessageProducerProperties(prefix: String): KafkaBasicAuthMessageProducerProperties {
  84         return bluePrintPropertiesService.propertyBeanType(
  85             prefix, KafkaBasicAuthMessageProducerProperties::class.java
  86         )
  87     }
  88
  89     /** Consumer Property Lib Service Implementation **/
  90
  91     /** Return Message Consumer Service for [jsonNode] definitions. */
  92     fun blueprintMessageConsumerService(jsonNode: JsonNode): BlueprintMessageConsumerService {
  93         val messageConsumerProperties = messageConsumerProperties(jsonNode)
  94         return blueprintMessageConsumerService(messageConsumerProperties)
  95     }
  96
  97     /** Return Message Consumer Service for [selector] definitions. */
  98     fun blueprintMessageConsumerService(selector: String): BlueprintMessageConsumerService {
  99         val prefix = "${MessageLibConstants.PROPERTY_MESSAGE_CONSUMER_PREFIX}$selector"
 100         val messageClientProperties = messageConsumerProperties(prefix)
 101         return blueprintMessageConsumerService(messageClientProperties)
 102     }
 103
 104     /** Return Message Consumer Properties for [prefix] definitions. */
 105     fun messageConsumerProperties(prefix: String): MessageConsumerProperties {
 106         val type = bluePrintPropertiesService.propertyBeanType("$prefix.type", String::class.java)
 107         return when (type) {
 108             MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
 109                 kafkaBasicAuthMessageConsumerProperties(prefix)
 110             }
 111             MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> {
 112                 kafkaStreamsBasicAuthMessageConsumerProperties(prefix)
 113             }
 114             else -> {
 115                 throw BluePrintProcessorException("Message adaptor($type) is not supported")
 116             }
 117         }
 118     }
 119
 120     fun messageConsumerProperties(jsonNode: JsonNode): MessageConsumerProperties {
 121         val type = jsonNode.get("type").textValue()
 122         return when (type) {
 123             MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
 124                 JacksonUtils.readValue(jsonNode, KafkaBasicAuthMessageConsumerProperties::class.java)!!
 125             }
 126             MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> {
 127                 JacksonUtils.readValue(jsonNode, KafkaStreamsBasicAuthConsumerProperties::class.java)!!
 128             }
 129             else -> {
 130                 throw BluePrintProcessorException("Message adaptor($type) is not supported")
 131             }
 132         }
 133     }
 134
 135     private fun blueprintMessageConsumerService(messageConsumerProperties: MessageConsumerProperties):
 136             BlueprintMessageConsumerService {
 137
 138         when (messageConsumerProperties) {
 139             is KafkaBasicAuthMessageConsumerProperties -> {
 140                 return KafkaBasicAuthMessageConsumerService(messageConsumerProperties)
 141             }
 142             is KafkaStreamsBasicAuthConsumerProperties -> {
 143                 return KafkaStreamsBasicAuthConsumerService(messageConsumerProperties)
 144             }
 145             else -> {
 146                 throw BluePrintProcessorException("couldn't get Message client service for")
 147             }
 148         }
 149     }
 150
 151     private fun kafkaBasicAuthMessageConsumerProperties(prefix: String): KafkaBasicAuthMessageConsumerProperties {
 152         return bluePrintPropertiesService.propertyBeanType(
 153             prefix, KafkaBasicAuthMessageConsumerProperties::class.java
 154         )
 155     }
 156
 157     private fun kafkaStreamsBasicAuthMessageConsumerProperties(prefix: String): KafkaStreamsBasicAuthConsumerProperties {
 158         return bluePrintPropertiesService.propertyBeanType(
 159             prefix, KafkaStreamsBasicAuthConsumerProperties::class.java
 160         )
 161     }
 162 }