ms/blueprintsprocessor/functions/message-prioritization/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/MessagePrioritizationStateServiceImpl.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.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.db.PrioritizationMessageRepository
  23 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.toFormatedCorrelation
  24 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
  25 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  26 import org.onap.ccsdk.cds.controllerblueprints.core.utils.addDate
  27 import org.onap.ccsdk.cds.controllerblueprints.core.utils.controllerDate
  28 import org.springframework.data.domain.PageRequest
  29 import org.springframework.stereotype.Service
  30 import org.springframework.transaction.annotation.Transactional
  31 import java.util.Date
  32
  33 @Service
  34 open class MessagePrioritizationStateServiceImpl(
  35     private val prioritizationMessageRepository: PrioritizationMessageRepository
  36 ) : MessagePrioritizationStateService {
  37
  38     private val log = logger(MessagePrioritizationStateServiceImpl::class)
  39
  40     @Transactional
  41     override suspend fun saveMessage(message: MessagePrioritization): MessagePrioritization {
  42         if (!message.correlationId.isNullOrBlank()) {
  43             message.correlationId = message.toFormatedCorrelation()
  44         }
  45         message.updatedDate = Date()
  46         return prioritizationMessageRepository.save(message)
  47     }
  48
  49     override suspend fun getMessage(id: String): MessagePrioritization {
  50         return prioritizationMessageRepository.findById(id).orElseGet(null)
  51             ?: throw BluePrintProcessorException("couldn't find message for id($id)")
  52     }
  53
  54     override suspend fun getMessages(ids: List<String>): List<MessagePrioritization>? {
  55         return prioritizationMessageRepository.findAllById(ids)
  56     }
  57
  58     override suspend fun getExpiryEligibleMessages(count: Int): List<MessagePrioritization>? {
  59         return prioritizationMessageRepository
  60             .findByStateInAndExpiredDate(
  61                 arrayListOf(MessageState.NEW.name, MessageState.WAIT.name),
  62                 Date(), PageRequest.of(0, count)
  63             )
  64     }
  65
  66     override suspend fun getMessageForStatesNotExpiredIn(group: String, states: List<String>, count: Int):
  67         List<MessagePrioritization>? {
  68             return prioritizationMessageRepository.findByGroupAndStateInAndNotExpiredDate(
  69                 group,
  70                 states, Date(), PageRequest.of(0, count)
  71             )
  72         }
  73
  74     override suspend fun getMessageForStatesExpired(group: String, states: List<String>, count: Int):
  75         List<MessagePrioritization>? {
  76             return prioritizationMessageRepository.findByGroupAndStateInAndExpiredDate(
  77                 group,
  78                 states, Date(), PageRequest.of(0, count)
  79             )
  80         }
  81
  82     override suspend fun getExpiredMessages(expiryDate: Date, count: Int): List<MessagePrioritization>? {
  83         return prioritizationMessageRepository.findByExpiredDate(
  84             expiryDate, PageRequest.of(0, count)
  85         )
  86     }
  87
  88     override suspend fun getExpiredMessages(group: String, expiryDate: Date, count: Int):
  89         List<MessagePrioritization>? {
  90             return prioritizationMessageRepository.findByGroupAndExpiredDate(
  91                 group,
  92                 expiryDate, PageRequest.of(0, count)
  93             )
  94         }
  95
  96     override suspend fun getCorrelatedMessages(
  97         group: String,
  98         states: List<String>,
  99         types: List<String>?,
 100         correlationIds: String
 101     ): List<MessagePrioritization>? {
 102         return if (!types.isNullOrEmpty()) {
 103             prioritizationMessageRepository.findByGroupAndTypesAndCorrelationId(group, states, types, correlationIds)
 104         } else {
 105             prioritizationMessageRepository.findByGroupAndCorrelationId(group, states, correlationIds)
 106         }
 107     }
 108
 109     @Transactional
 110     override suspend fun updateMessagesState(ids: List<String>, state: String) {
 111         ids.forEach {
 112             val updated = updateMessageState(it, state)
 113             log.info("message($it) update to state(${updated.state})")
 114         }
 115     }
 116
 117     @Transactional
 118     override suspend fun setMessageState(id: String, state: String) {
 119         prioritizationMessageRepository.setStateForMessageId(id, state, Date())
 120     }
 121
 122     @Transactional
 123     override suspend fun setMessagesPriority(ids: List<String>, priority: String) {
 124         prioritizationMessageRepository.setPriorityForMessageIds(ids, priority, Date())
 125     }
 126
 127     @Transactional
 128     override suspend fun setMessagesState(ids: List<String>, state: String) {
 129         prioritizationMessageRepository.setStateForMessageIds(ids, state, Date())
 130     }
 131
 132     @Transactional
 133     override suspend fun setMessageStateANdError(id: String, state: String, error: String) {
 134         prioritizationMessageRepository.setStateAndErrorForMessageId(id, state, error, Date())
 135     }
 136
 137     @Transactional
 138     override suspend fun updateMessageState(id: String, state: String): MessagePrioritization {
 139         val updateMessage = getMessage(id).apply {
 140             this.updatedDate = Date()
 141             this.state = state
 142         }
 143         return saveMessage(updateMessage)
 144     }
 145
 146     @Transactional
 147     override suspend fun setMessageStateAndAggregatedIds(id: String, state: String, aggregatedIds: List<String>) {
 148         val groupedIds = aggregatedIds.joinToString(",")
 149         prioritizationMessageRepository.setStateAndAggregatedMessageIds(id, state, groupedIds, Date())
 150     }
 151
 152     override suspend fun deleteMessage(id: String) {
 153         prioritizationMessageRepository.deleteById(id)
 154         log.info("Prioritization Messages $id deleted successfully.")
 155     }
 156
 157     override suspend fun deleteMessages(ids: List<String>) {
 158         prioritizationMessageRepository.deleteByIds(ids)
 159         log.info("Prioritization Messages $ids deleted successfully.")
 160     }
 161
 162     override suspend fun deleteExpiredMessage(retentionDays: Int) {
 163         val expiryCheckDate = controllerDate().addDate(retentionDays)
 164         prioritizationMessageRepository.deleteByExpiryDate(expiryCheckDate)
 165     }
 166
 167     override suspend fun deleteMessageByGroup(group: String) {
 168         prioritizationMessageRepository.deleteGroup(group)
 169         log.info("Prioritization Messages group($group) deleted successfully.")
 170     }
 171
 172     override suspend fun deleteMessageStates(group: String, states: List<String>) {
 173         prioritizationMessageRepository.deleteGroupAndStateIn(group, states)
 174         log.info("Prioritization Messages group($group) with states($states) deleted successfully.")
 175     }
 176 }