2 * Copyright © 2019 IBM.
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
8 * http://www.apache.org/licenses/LICENSE-2.0
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.
17 package org.onap.ccsdk.cds.blueprintsprocessor.selfservice.api
19 import kotlinx.coroutines.channels.consumeEach
20 import kotlinx.coroutines.launch
21 import kotlinx.coroutines.runBlocking
22 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
23 import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService
24 import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BlueprintMessageConsumerService
25 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
26 import org.onap.ccsdk.cds.controllerblueprints.core.jsonAsType
27 import org.onap.ccsdk.cds.controllerblueprints.core.logger
28 import org.onap.ccsdk.cds.controllerblueprints.core.updateErrorMessage
29 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
30 import org.springframework.boot.context.event.ApplicationReadyEvent
31 import org.springframework.context.event.EventListener
32 import org.springframework.stereotype.Service
33 import java.util.concurrent.Phaser
34 import javax.annotation.PreDestroy
36 @ConditionalOnProperty(
37 name = ["blueprintsprocessor.messageconsumer.self-service-api.kafkaEnable"],
41 open class BluePrintProcessingKafkaConsumer(
42 private val bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService,
43 private val executionServiceHandler: ExecutionServiceHandler
46 val log = logger(BluePrintProcessingKafkaConsumer::class)
48 private val ph = Phaser(1)
50 private lateinit var blueprintMessageConsumerService: BlueprintMessageConsumerService
53 const val CONSUMER_SELECTOR = "self-service-api"
54 const val PRODUCER_SELECTOR = "self-service-api"
57 @EventListener(ApplicationReadyEvent::class)
58 fun setupMessageListener() = runBlocking {
61 "Setting up message consumer($CONSUMER_SELECTOR)" +
62 "message producer($PRODUCER_SELECTOR)..."
65 /** Get the Message Consumer Service **/
66 blueprintMessageConsumerService = try {
67 bluePrintMessageLibPropertyService
68 .blueprintMessageConsumerService(CONSUMER_SELECTOR)
69 } catch (e: BluePrintProcessorException) {
70 val errorMsg = "Failed creating Kafka consumer message service."
71 throw e.updateErrorMessage(SelfServiceApiDomains.SELF_SERVICE_API, errorMsg,
72 "Wrong Kafka selector provided or internal error in Kafka service.")
73 } catch (e: Exception) {
74 throw BluePrintProcessorException("failed to create consumer service ${e.message}")
77 /** Get the Message Producer Service **/
78 val blueprintMessageProducerService = try {
79 bluePrintMessageLibPropertyService
80 .blueprintMessageProducerService(PRODUCER_SELECTOR)
81 } catch (e: BluePrintProcessorException) {
82 val errorMsg = "Failed creating Kafka producer message service."
83 throw e.updateErrorMessage(SelfServiceApiDomains.SELF_SERVICE_API, errorMsg,
84 "Wrong Kafka selector provided or internal error in Kafka service.")
85 } catch (e: Exception) {
86 throw BluePrintProcessorException("failed to create producer service ${e.message}")
90 /** Subscribe to the consumer topics */
91 val additionalConfig: MutableMap<String, Any> = hashMapOf()
92 val channel = blueprintMessageConsumerService.subscribe(additionalConfig)
93 channel.consumeEach { message ->
97 log.trace("Consumed Message : $message")
98 val executionServiceInput = message.jsonAsType<ExecutionServiceInput>()
99 val executionServiceOutput = executionServiceHandler.doProcess(executionServiceInput)
100 blueprintMessageProducerService.sendMessage(executionServiceOutput)
101 } catch (e: Exception) {
102 log.error("failed in processing the consumed message : $message", e)
104 ph.arriveAndDeregister()
109 } catch (e: Exception) {
111 "failed to start message consumer($CONSUMER_SELECTOR) " +
112 "message producer($PRODUCER_SELECTOR) ", e
118 fun shutdownMessageListener() = runBlocking {
121 "Shutting down message consumer($CONSUMER_SELECTOR)" +
122 "message producer($PRODUCER_SELECTOR)..."
124 blueprintMessageConsumerService.shutDown()
125 ph.arriveAndAwaitAdvance()
126 } catch (e: Exception) {
127 log.error("failed to shutdown message listener($CONSUMER_SELECTOR)", e)