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