ms/blueprintsprocessor/functions/message-prioritization/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.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.apache.kafka.streams.processor.ProcessorSupplier
  20 import org.onap.ccsdk.cds.blueprintsprocessor.core.cluster.optionalClusterService
  21 import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterLock
  22 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
  23 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.AbstractMessagePrioritizeProcessor
  24 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.toFormatedCorrelation
  25 import org.onap.ccsdk.cds.controllerblueprints.core.BlueprintProcessorException
  26 import org.onap.ccsdk.cds.controllerblueprints.core.service.BlueprintDependencyService
  27
  28 object MessageProcessorUtils {
  29
  30     /** Utility to create the cluster lock for message [messagePrioritization] prioritization procssing.*/
  31     suspend fun prioritizationGrouplock(messagePrioritization: MessagePrioritization): ClusterLock? {
  32         val clusterService = BlueprintDependencyService.optionalClusterService()
  33
  34         return if (clusterService != null && clusterService.clusterJoined() &&
  35             !messagePrioritization.correlationId.isNullOrBlank()
  36         ) {
  37             // Get the correlation key in ascending order, even it it is misplaced
  38             val correlationId = messagePrioritization.toFormatedCorrelation()
  39             val lockName = "prioritize::${messagePrioritization.group}::$correlationId"
  40             val clusterLock = clusterService.clusterLock(lockName)
  41             clusterLock.lock()
  42             if (!clusterLock.isLocked()) throw BlueprintProcessorException("failed to lock($lockName)")
  43             clusterLock
  44         } else null
  45     }
  46
  47     /** Utility to create the cluster lock for expiry scheduler*/
  48     suspend fun prioritizationExpiryLock(): ClusterLock? {
  49         val clusterService = BlueprintDependencyService.optionalClusterService()
  50         return if (clusterService != null && clusterService.clusterJoined()) {
  51             val lockName = "prioritize-expiry"
  52             val clusterLock = clusterService.clusterLock(lockName)
  53             clusterLock.lock()
  54             if (!clusterLock.isLocked()) throw BlueprintProcessorException("failed to lock($lockName)")
  55             clusterLock
  56         } else null
  57     }
  58
  59     /** Utility to create the cluster lock for expiry scheduler*/
  60     suspend fun prioritizationCleanLock(): ClusterLock? {
  61         val clusterService = BlueprintDependencyService.optionalClusterService()
  62         return if (clusterService != null && clusterService.clusterJoined()) {
  63             val lockName = "prioritize-clean"
  64             val clusterLock = clusterService.clusterLock(lockName)
  65             clusterLock.lock()
  66             if (!clusterLock.isLocked()) throw BlueprintProcessorException("failed to lock($lockName)")
  67             clusterLock
  68         } else null
  69     }
  70
  71     /** Utility used to cluster unlock for message [clusterLock] */
  72     suspend fun prioritizationUnLock(clusterLock: ClusterLock?) {
  73         if (clusterLock != null) {
  74             clusterLock.unLock()
  75             clusterLock.close()
  76         }
  77     }
  78
  79     /** Get the Kafka Supplier for processor lookup [name] **/
  80     fun <K, V> bluePrintProcessorSupplier(name: String): ProcessorSupplier<K, V> {
  81         return ProcessorSupplier<K, V> {
  82             // Dynamically resolve the Prioritization Processor
  83             BlueprintDependencyService.instance<AbstractMessagePrioritizeProcessor<K, V>>(name)
  84         }
  85     }
  86 }