ms/blueprintsprocessor/functions/message-prioritization/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/nats/NatsMessagePrioritizationConsumer.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.functions.message.prioritization.nats
  18
  19 import io.nats.streaming.MessageHandler
  20 import io.nats.streaming.Subscription
  21 import kotlinx.coroutines.runBlocking
  22 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationService
  23 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
  24 import org.onap.ccsdk.cds.blueprintsprocessor.nats.asJsonType
  25 import org.onap.ccsdk.cds.blueprintsprocessor.nats.service.BlueprintNatsLibPropertyService
  26 import org.onap.ccsdk.cds.blueprintsprocessor.nats.service.BlueprintNatsService
  27 import org.onap.ccsdk.cds.blueprintsprocessor.nats.utils.NatsClusterUtils
  28 import org.onap.ccsdk.cds.blueprintsprocessor.nats.utils.SubscriptionOptionsUtils
  29 import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintProcessorException
  30 import org.onap.ccsdk.cds.controllerblueprints.core.asType
  31 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  32 import org.onap.ccsdk.cds.controllerblueprints.core.utils.ClusterUtils
  33
  34 open class NatsMessagePrioritizationConsumer(
  35     private val bluePrintNatsLibPropertyService: BlueprintNatsLibPropertyService,
  36     private val natsMessagePrioritizationService: MessagePrioritizationService
  37 ) {
  38
  39     private val log = logger(NatsMessagePrioritizationConsumer::class)
  40
  41     lateinit var bluePrintNatsService: BlueprintNatsService
  42     private lateinit var subscription: Subscription
  43
  44     suspend fun startConsuming() {
  45         val prioritizationConfiguration = natsMessagePrioritizationService.getConfiguration()
  46         val natsConfiguration = prioritizationConfiguration.natsConfiguration
  47             ?: throw BlueprintProcessorException("couldn't get NATS consumer configuration")
  48
  49         check((natsMessagePrioritizationService is AbstractNatsMessagePrioritizationService)) {
  50             "messagePrioritizationService is not of type AbstractNatsMessagePrioritizationService."
  51         }
  52         bluePrintNatsService = consumerService(natsConfiguration.connectionSelector)
  53         natsMessagePrioritizationService.bluePrintNatsService = bluePrintNatsService
  54         val inputSubject = NatsClusterUtils.currentApplicationSubject(natsConfiguration.inputSubject)
  55         val loadBalanceGroup = ClusterUtils.applicationName()
  56         val messageHandler = createMessageHandler()
  57         val subscriptionOptions = SubscriptionOptionsUtils.durable(NatsClusterUtils.currentNodeDurable(inputSubject))
  58         subscription = bluePrintNatsService.loadBalanceSubscribe(
  59             inputSubject,
  60             loadBalanceGroup,
  61             messageHandler,
  62             subscriptionOptions
  63         )
  64         log.info(
  65             "Nats prioritization consumer listening on subject($inputSubject) on loadBalance group($loadBalanceGroup)."
  66         )
  67     }
  68
  69     suspend fun shutDown() {
  70         if (::subscription.isInitialized) {
  71             subscription.unsubscribe()
  72         }
  73         log.info("Nats prioritization consumer listener shutdown complete")
  74     }
  75
  76     private fun consumerService(selector: String): BlueprintNatsService {
  77         return bluePrintNatsLibPropertyService.bluePrintNatsService(selector)
  78     }
  79
  80     private fun createMessageHandler(): MessageHandler {
  81         return MessageHandler { message ->
  82             try {
  83                 val messagePrioritization = message.asJsonType().asType(MessagePrioritization::class.java)
  84                 runBlocking {
  85                     natsMessagePrioritizationService.prioritize(messagePrioritization)
  86                 }
  87             } catch (e: Exception) {
  88                 log.error("failed to process prioritize message", e)
  89             }
  90         }
  91     }
  92 }