ms/blueprintsprocessor/functions/message-prioritization/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 }