ms/blueprintsprocessor/modules/commons/message-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerService.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 kotlinx.coroutines.channels.Channel
  21 import org.apache.kafka.clients.consumer.Consumer
  22 import org.apache.kafka.clients.consumer.ConsumerRecords
  23 import org.apache.kafka.streams.Topology
  24 import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties
  25 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
  26
  27 /** Consumer Function Interfaces */
  28 interface ConsumerFunction
  29
  30 interface BlueprintMessageConsumerService {
  31
  32     suspend fun subscribe(): Channel<String> {
  33         return subscribe(null)
  34     }
  35
  36     /** Subscribe to the Kafka channel with [additionalConfig] */
  37     suspend fun subscribe(additionalConfig: Map<String, Any>?): Channel<String>
  38
  39     /** Subscribe to the Kafka channel with [additionalConfig] for dynamic [topics]*/
  40     suspend fun subscribe(topics: List<String>, additionalConfig: Map<String, Any>? = null): Channel<String>
  41
  42     /** Consume and execute dynamic function [consumerFunction] */
  43     suspend fun consume(consumerFunction: ConsumerFunction) {
  44         consume(null, consumerFunction)
  45     }
  46
  47     /** Consume with [additionalConfig], so that we can execute dynamic function [consumerFunction] */
  48     suspend fun consume(additionalConfig: Map<String, Any>?, consumerFunction: ConsumerFunction) {
  49         throw BluePrintProcessorException("Not Implemented")
  50     }
  51
  52     /** Consume the [topics] with [additionalConfig], so that we can execute dynamic function [consumerFunction] */
  53     suspend fun consume(
  54         topics: List<String>,
  55         additionalConfig: Map<String, Any>?,
  56         consumerFunction: ConsumerFunction
  57     ) {
  58         throw BluePrintProcessorException("Not Implemented")
  59     }
  60
  61     /** close the channel, consumer and other resources */
  62     suspend fun shutDown()
  63 }
  64
  65 /** Consumer dynamic implementation interface */
  66 interface KafkaConsumerRecordsFunction : ConsumerFunction {
  67
  68     suspend fun invoke(
  69         messageConsumerProperties: MessageConsumerProperties,
  70         consumer: Consumer<*, *>,
  71         consumerRecords: ConsumerRecords<*, *>
  72     )
  73 }
  74
  75 interface KafkaStreamConsumerFunction : ConsumerFunction {
  76     suspend fun createTopology(
  77         messageConsumerProperties: MessageConsumerProperties,
  78         additionalConfig: Map<String, Any>?
  79     ): Topology
  80 }