ms/blueprintsprocessor/functions/message-prioritization/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/nats/AbstractNatsMessagePrioritizationService.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 org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationStateService
  20 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
  21 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
  22 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.ids
  23 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.AbstractMessagePrioritizationService
  24 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessageProcessorUtils
  25 import org.onap.ccsdk.cds.blueprintsprocessor.nats.service.BluePrintNatsService
  26 import org.onap.ccsdk.cds.blueprintsprocessor.nats.utils.NatsClusterUtils
  27 import org.onap.ccsdk.cds.controllerblueprints.core.asByteArray
  28 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonType
  29 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  30
  31 abstract class AbstractNatsMessagePrioritizationService(
  32     private val messagePrioritizationStateService: MessagePrioritizationStateService
  33 ) : AbstractMessagePrioritizationService(messagePrioritizationStateService) {
  34
  35     private val log = logger(AbstractNatsMessagePrioritizationService::class)
  36
  37     lateinit var bluePrintNatsService: BluePrintNatsService
  38
  39     override suspend fun output(messages: List<MessagePrioritization>) {
  40         log.info("$$$$$ received in output processor id(${messages.ids()})")
  41         checkNotNull(prioritizationConfiguration.natsConfiguration) { "failed to initialize NATS configuration" }
  42         check(::bluePrintNatsService.isInitialized) { "failed to initialize NATS services" }
  43
  44         val outputSubject = prioritizationConfiguration.natsConfiguration!!.outputSubject
  45         messages.forEach { message ->
  46             val updatedMessage =
  47                 messagePrioritizationStateService.updateMessageState(message.id, MessageState.COMPLETED.name)
  48
  49             /** send to the output subject */
  50             bluePrintNatsService.publish(
  51                 NatsClusterUtils.currentApplicationSubject(outputSubject),
  52                 updatedMessage.asJsonType().asByteArray()
  53             )
  54         }
  55     }
  56
  57     override suspend fun updateExpiredMessages() {
  58         checkNotNull(prioritizationConfiguration.natsConfiguration) { "failed to initialize NATS configuration" }
  59         check(::bluePrintNatsService.isInitialized) { "failed to initialize NATS services" }
  60
  61         val expiryConfiguration = prioritizationConfiguration.expiryConfiguration
  62         val outputSubject = prioritizationConfiguration.natsConfiguration!!.expiredSubject
  63         val clusterLock = MessageProcessorUtils.prioritizationExpiryLock()
  64         try {
  65             val fetchMessages = messagePrioritizationStateService
  66                 .getExpiryEligibleMessages(expiryConfiguration.maxPollRecord)
  67             val expiredIds = fetchMessages?.ids()
  68             if (!expiredIds.isNullOrEmpty()) {
  69                 messagePrioritizationStateService.updateMessagesState(expiredIds, MessageState.EXPIRED.name)
  70                 fetchMessages.forEach { expiredMessage ->
  71                     expiredMessage.state = MessageState.EXPIRED.name
  72                     /** send to the output subject */
  73                     bluePrintNatsService.publish(
  74                         NatsClusterUtils.currentApplicationSubject(outputSubject),
  75                         expiredMessage.asJsonType().asByteArray()
  76                     )
  77                 }
  78             }
  79         } catch (e: Exception) {
  80             log.error("failed in updating expired messages", e)
  81         } finally {
  82             MessageProcessorUtils.prioritizationUnLock(clusterLock)
  83         }
  84     }
  85 }