ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtils.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.utils
  18
  19 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.CorrelationCheckResponse
  20 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
  21 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.toFormatedCorrelation
  22 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.toTypeNCorrelation
  23 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
  24
  25 object MessageCorrelationUtils {
  26
  27     /** Assumption is message is of same group **/
  28     fun correlatedMessages(collectedMessages: List<MessagePrioritization>): CorrelationCheckResponse {
  29         val correlationCheckResponse = CorrelationCheckResponse(message = "not correlated")
  30         if (collectedMessages.size > 1) {
  31             val filteredMessage = collectedMessages.filter { !it.correlationId.isNullOrBlank() }
  32             if (filteredMessage.isNotEmpty()) {
  33                 val groupedMessage = filteredMessage.groupBy { it.toFormatedCorrelation() }
  34                 if (groupedMessage.size == 1) {
  35                     correlationCheckResponse.correlated = true
  36                     correlationCheckResponse.message = null
  37                 }
  38             }
  39         } else {
  40             correlationCheckResponse.message = "received only one message for that group"
  41         }
  42         return correlationCheckResponse
  43     }
  44
  45     /** Assumption is message is of same group and checking for required types **/
  46     fun correlatedMessagesWithTypes(collectedMessages: List<MessagePrioritization>, types: List<String>?)
  47             : CorrelationCheckResponse {
  48
  49         return if (!types.isNullOrEmpty() && collectedMessages.size > 1) {
  50
  51             val unknownMessageTypes = collectedMessages.filter { !types.contains(it.type) }.map { it.id }
  52             if (!unknownMessageTypes.isNullOrEmpty()) {
  53                 throw BluePrintProcessorException("Messages($unknownMessageTypes) is not in type of($types)")
  54             }
  55
  56             val copyTypes = types.toTypedArray().copyOf().toMutableList()
  57
  58             val filteredMessage = collectedMessages.filter {
  59                 !it.correlationId.isNullOrBlank()
  60                         && types.contains(it.type)
  61             }
  62             var correlatedKeys: MutableSet<String> = mutableSetOf()
  63             if (filteredMessage.isNotEmpty()) {
  64                 val correlatedMap = filteredMessage.groupBy { it.toTypeNCorrelation() }
  65                 val foundType = correlatedMap.keys.map { it.type }
  66                 copyTypes.removeAll(foundType)
  67                 correlatedKeys = correlatedMap.keys.map {
  68                     it.correlationId
  69                 }.toMutableSet()
  70             }
  71             /** Check if any Types missing and same correlation id for all types */
  72             return if (copyTypes.isEmpty()) {
  73                 if (correlatedKeys.size == 1) CorrelationCheckResponse(correlated = true)
  74                 else CorrelationCheckResponse(message = "not matching correlation keys($correlatedKeys)")
  75             } else {
  76                 CorrelationCheckResponse(message = "couldn't find types($copyTypes)")
  77             }
  78         } else {
  79             return correlatedMessages(collectedMessages)
  80         }
  81     }
  82 }