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