ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/SampleMessagePrioritizationService.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.service
  18
  19 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationService
  20 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationStateService
  21 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
  22 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
  23 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.ids
  24 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.AbstractKafkaMessagePrioritizationService
  25 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.nats.AbstractNatsMessagePrioritizationService
  26 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.orderByHighestPriority
  27 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  28
  29 /** Sample Prioritization Service, Define spring service injector to register in application*/
  30 open class SampleMessagePrioritizationService(private val messagePrioritizationStateService: MessagePrioritizationStateService) :
  31     AbstractMessagePrioritizationService(messagePrioritizationStateService) {
  32
  33     /** Child overriding this implementation , if necessary */
  34     override suspend fun handleAggregation(messages: List<MessagePrioritization>) {
  35         val sampleMessagePrioritizationHandler = SampleMessagePrioritizationHandler(
  36             this, messagePrioritizationStateService
  37         )
  38         sampleMessagePrioritizationHandler.handleAggregation(messages)
  39     }
  40
  41     /** If consumer wants specific correlation with respect to group and types, then populate the specific types,
  42      * otherwise correlation happens with group and correlationId */
  43     override fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? {
  44         val sampleMessagePrioritizationHandler = SampleMessagePrioritizationHandler(
  45             this, messagePrioritizationStateService
  46         )
  47         return sampleMessagePrioritizationHandler.getGroupCorrelationTypes(messagePrioritization)
  48     }
  49 }
  50
  51 open class SampleKafkaMessagePrioritizationService(private val messagePrioritizationStateService: MessagePrioritizationStateService) :
  52     AbstractKafkaMessagePrioritizationService(messagePrioritizationStateService) {
  53
  54     /** Child overriding this implementation , if necessary */
  55     override suspend fun handleAggregation(messages: List<MessagePrioritization>) {
  56         val sampleMessagePrioritizationHandler = SampleMessagePrioritizationHandler(
  57             this, messagePrioritizationStateService
  58         )
  59         sampleMessagePrioritizationHandler.handleAggregation(messages)
  60     }
  61
  62     /** If consumer wants specific correlation with respect to group and types, then populate the specific types,
  63      * otherwise correlation happens with group and correlationId */
  64     override fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? {
  65         val sampleMessagePrioritizationHandler = SampleMessagePrioritizationHandler(
  66             this, messagePrioritizationStateService
  67         )
  68         return sampleMessagePrioritizationHandler.getGroupCorrelationTypes(messagePrioritization)
  69     }
  70 }
  71
  72 open class SampleNatsMessagePrioritizationService(private val messagePrioritizationStateService: MessagePrioritizationStateService) :
  73     AbstractNatsMessagePrioritizationService(messagePrioritizationStateService) {
  74
  75     /** Child overriding this implementation , if necessary */
  76     override suspend fun handleAggregation(messages: List<MessagePrioritization>) {
  77         val sampleMessagePrioritizationHandler = SampleMessagePrioritizationHandler(
  78             this, messagePrioritizationStateService
  79         )
  80         sampleMessagePrioritizationHandler.handleAggregation(messages)
  81     }
  82
  83     /** If consumer wants specific correlation with respect to group and types, then populate the specific types,
  84      * otherwise correlation happens with group and correlationId */
  85     override fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? {
  86         val sampleMessagePrioritizationHandler = SampleMessagePrioritizationHandler(
  87             this, messagePrioritizationStateService
  88         )
  89         return sampleMessagePrioritizationHandler.getGroupCorrelationTypes(messagePrioritization)
  90     }
  91 }
  92
  93 class SampleMessagePrioritizationHandler(
  94     private val messagePrioritizationService: MessagePrioritizationService,
  95     private val messagePrioritizationStateService: MessagePrioritizationStateService
  96 ) {
  97
  98     private val log = logger(SampleMessagePrioritizationHandler::class)
  99
 100     suspend fun handleAggregation(messages: List<MessagePrioritization>) {
 101         log.info("messages(${messages.ids()}) aggregated")
 102         /** Sequence based on Priority and Updated Date */
 103         val sequencedMessage = messages.orderByHighestPriority()
 104         /** Update all messages to aggregated state */
 105         messagePrioritizationStateService.setMessagesState(
 106             sequencedMessage.ids(),
 107             MessageState.AGGREGATED.name
 108         )
 109         messagePrioritizationService.output(sequencedMessage)
 110     }
 111
 112     fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? {
 113         return when (messagePrioritization.group) {
 114             /** Dummy Implementation, This can also be read from file and stored as cached map **/
 115             "group-typed" -> arrayListOf("type-0", "type-1", "type-2")
 116             "pass-typed" -> arrayListOf(messagePrioritization.type)
 117             else -> null
 118         }
 119     }
 120 }