diff options
Diffstat (limited to 'ms/blueprintsprocessor/functions/message-prioritizaion/src')
11 files changed, 206 insertions, 50 deletions
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractTopologyComponents.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractTopologyComponents.kt index d89f71364..d114da521 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractTopologyComponents.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractTopologyComponents.kt @@ -34,7 +34,7 @@ abstract class AbstractMessagePrioritizeProcessor<K, V> : AbstractBluePrintMessa this.processorContext = context /** Get the State service to update in store */ this.messagePrioritizationStateService = BluePrintDependencyService - .instance(MessagePrioritizationStateService::class) + .messagePrioritizationStateService() } diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumer.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumer.kt index ef9d5a058..967cc190e 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumer.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumer.kt @@ -19,6 +19,7 @@ package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.Topology import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessagePrioritizationSerde +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessageProcessorUtils.bluePrintProcessorSupplier import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsBasicAuthConsumerProperties import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService @@ -74,9 +75,12 @@ open class MessagePrioritizationConsumer( Serdes.String().serializer(), MessagePrioritizationSerde().serializer(), MessagePrioritizationConstants.PROCESSOR_PRIORITIZE) + /** To receive completed and error messages */ topology.addSink(MessagePrioritizationConstants.SINK_OUTPUT, prioritizationConfiguration.outputTopic, Serdes.String().serializer(), MessagePrioritizationSerde().serializer(), + MessagePrioritizationConstants.PROCESSOR_PRIORITIZE, + MessagePrioritizationConstants.PROCESSOR_AGGREGATE, MessagePrioritizationConstants.PROCESSOR_OUTPUT) // Output will be sent to the group-output topic from Processor API diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationData.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationData.kt index d874cef92..3358a5643 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationData.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationData.kt @@ -28,8 +28,8 @@ enum class MessageState(val id: String) { EXPIRED("expired"), PRIORITIZED("prioritized"), AGGREGATED("aggregated"), - IGNORED("ignored"), COMPLETED("completed"), + ERROR("error") } open class PrioritizationConfiguration : Serializable { @@ -59,7 +59,6 @@ open class UpdateStateRequest : Serializable { lateinit var id: String var group: String? = null var state: String? = null - var notifyMessage: String? = null } data class CorrelationCheckResponse(var message: String? = null, diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizeExtensions.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizeExtensions.kt index 94fedf4df..ec061ad47 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizeExtensions.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizeExtensions.kt @@ -16,21 +16,27 @@ package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization -import org.apache.kafka.streams.processor.ProcessorSupplier import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService +import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.AbstractComponentFunction import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService -fun <K, V> bluePrintProcessorSupplier(name: String, prioritizationConfiguration: PrioritizationConfiguration) - : ProcessorSupplier<K, V> { - return ProcessorSupplier<K, V> { - // Dynamically resolve the Prioritization Processor - val processorInstance = BluePrintDependencyService.instance<AbstractMessagePrioritizeProcessor<K, V>>(name) - processorInstance.prioritizationConfiguration = prioritizationConfiguration - processorInstance - } -} +/** + * Register the MessagePrioritizationStateService and exposed dependency + */ +fun BluePrintDependencyService.messagePrioritizationStateService(): MessagePrioritizationStateService = + instance(MessagePrioritizationStateService::class) +/** + * Expose messagePrioritizationStateService to AbstractComponentFunction + */ +fun AbstractComponentFunction.messagePrioritizationStateService() = + BluePrintDependencyService.messagePrioritizationStateService() + +/** + * MessagePrioritization correlation extensions + */ fun MessagePrioritization.toFormatedCorrelation(): String { val ascendingKey = this.correlationId!!.split(",") .map { it.trim() }.sorted().joinToString(",") diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/api/MessagePrioritizationApi.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/api/MessagePrioritizationApi.kt new file mode 100644 index 000000000..382cb9c8a --- /dev/null +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/api/MessagePrioritizationApi.kt @@ -0,0 +1,55 @@ +/* + * Copyright © 2018-2019 AT&T Intellectual Property. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.api + +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.UpdateStateRequest +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService +import org.onap.ccsdk.cds.blueprintsprocessor.rest.service.monoMdc +import org.springframework.http.MediaType +import org.springframework.web.bind.annotation.* + +@RestController +@RequestMapping(value = ["/api/v1/message-prioritization"]) +open class MessagePrioritizationApi(private val messagePrioritizationStateService: MessagePrioritizationStateService) { + + @GetMapping(path = ["/ping"], produces = [MediaType.APPLICATION_JSON_VALUE]) + @ResponseBody + fun ping(): String = "Success" + + + @GetMapping(path = ["/{id}"], produces = [MediaType.APPLICATION_JSON_VALUE]) + @ResponseBody + fun messagePrioritization(@PathVariable(value = "id") id: String) = monoMdc { + messagePrioritizationStateService.getMessage(id) + } + + @PostMapping(path = ["/"], produces = [MediaType.APPLICATION_JSON_VALUE], + consumes = [MediaType.APPLICATION_JSON_VALUE]) + @ResponseBody + fun saveMessagePrioritization(@RequestBody messagePrioritization: MessagePrioritization) = monoMdc { + messagePrioritizationStateService.saveMessage(messagePrioritization) + } + + @PostMapping(path = ["/update-state"], produces = [MediaType.APPLICATION_JSON_VALUE], + consumes = [MediaType.APPLICATION_JSON_VALUE]) + fun updateMessagePrioritizationState(@RequestBody updateMessageState: UpdateStateRequest) = + monoMdc { + messagePrioritizationStateService.setMessageState(updateMessageState.id, + updateMessageState.state!!) + } +}
\ No newline at end of file diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/MessagePrioritizationRepositories.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/MessagePrioritizationRepositories.kt index 307d932a9..69c81079d 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/MessagePrioritizationRepositories.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/MessagePrioritizationRepositories.kt @@ -71,19 +71,27 @@ interface PrioritizationMessageRepository : JpaRepository<MessagePrioritization, @Modifying @Transactional - @Query("UPDATE MessagePrioritization pm SET pm.state = :state WHERE pm.id = :id") - fun setStatusForMessageId(id: String, state: String): Int + @Query("UPDATE MessagePrioritization SET state = :state, updatedDate = :currentDate " + + "WHERE id = :id") + fun setStateForMessageId(id: String, state: String, currentDate: Date): Int @Modifying @Transactional - @Query("UPDATE MessagePrioritization pm SET pm.state = :state WHERE pm.id IN :ids") - fun setStatusForMessageIds(ids: List<String>, state: String): Int + @Query("UPDATE MessagePrioritization SET state = :state, updatedDate = :currentDate " + + "WHERE id IN :ids") + fun setStateForMessageIds(ids: List<String>, state: String, currentDate: Date): Int @Modifying @Transactional - @Query("UPDATE MessagePrioritization pm SET pm.aggregatedMessageIds = :aggregatedMessageIds " + - "WHERE pm.id = :id") - fun setAggregatedMessageIds(id: String, aggregatedMessageIds: String): Int + @Query("UPDATE MessagePrioritization SET state = :state, error = :error, updatedDate = :currentDate " + + "WHERE id = :id") + fun setStateAndErrorForMessageId(id: String, state: String, error: String, currentDate: Date): Int + + @Modifying + @Transactional + @Query("UPDATE MessagePrioritization SET state = :state, " + + "aggregatedMessageIds = :aggregatedMessageIds, updatedDate = :currentDate WHERE id = :id") + fun setStateAndAggregatedMessageIds(id: String, state: String, aggregatedMessageIds: String, currentDate: Date): Int @Modifying @Transactional diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/PrioritizationMessageEntity.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/PrioritizationMessageEntity.kt index 4973cdf6e..1825f91c2 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/PrioritizationMessageEntity.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/PrioritizationMessageEntity.kt @@ -49,6 +49,10 @@ open class MessagePrioritization { var message: String? = null @Lob + @Column(name = "error", nullable = true) + var error: String? = null + + @Lob @Column(name = "aggregated_message_ids", nullable = true) var aggregatedMessageIds: String? = null diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/MessagePrioritizationStateService.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/MessagePrioritizationStateService.kt index e4369fc20..8424226c2 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/MessagePrioritizationStateService.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/MessagePrioritizationStateService.kt @@ -33,15 +33,20 @@ interface MessagePrioritizationStateService { suspend fun getMessage(id: String): MessagePrioritization + suspend fun getMessages(ids: List<String>): List<MessagePrioritization>? + suspend fun getExpiryEligibleMessages(count: Int): List<MessagePrioritization>? - suspend fun getMessageForStatesNotExpiredIn(group: String, states: List<String>, count: Int): List<MessagePrioritization>? + suspend fun getMessageForStatesNotExpiredIn(group: String, states: List<String>, count: Int) + : List<MessagePrioritization>? - suspend fun getMessageForStatesExpired(group: String, states: List<String>, count: Int): List<MessagePrioritization>? + suspend fun getMessageForStatesExpired(group: String, states: List<String>, count: Int) + : List<MessagePrioritization>? suspend fun getExpiredMessages(group: String, expiryDate: Date, count: Int): List<MessagePrioritization>? - suspend fun getCorrelatedMessages(group: String, states: List<String>, types: List<String>?, correlationIds: String): List<MessagePrioritization>? + suspend fun getCorrelatedMessages(group: String, states: List<String>, types: List<String>?, + correlationIds: String): List<MessagePrioritization>? suspend fun updateMessagesState(ids: List<String>, state: String) @@ -51,7 +56,9 @@ interface MessagePrioritizationStateService { suspend fun setMessagesState(ids: List<String>, state: String) - suspend fun updateMessageStateAndGroupedIds(id: String, state: String, groupedIds: List<String>): MessagePrioritization + suspend fun setMessageStateANdError(id: String, state: String, error: String) + + suspend fun setMessageStateAndAggregatedIds(id: String, state: String, aggregatedIds: List<String>) suspend fun deleteMessage(id: String) @@ -64,7 +71,8 @@ interface MessagePrioritizationStateService { @Service open class MessagePrioritizationStateServiceImpl( - private val prioritizationMessageRepository: PrioritizationMessageRepository) : MessagePrioritizationStateService { + private val prioritizationMessageRepository: PrioritizationMessageRepository) + : MessagePrioritizationStateService { private val log = logger(MessagePrioritizationStateServiceImpl::class) @@ -82,6 +90,10 @@ open class MessagePrioritizationStateServiceImpl( ?: throw BluePrintProcessorException("couldn't find message for id($id)") } + override suspend fun getMessages(ids: List<String>): List<MessagePrioritization>? { + return prioritizationMessageRepository.findAllById(ids) + } + override suspend fun getExpiryEligibleMessages(count: Int): List<MessagePrioritization>? { return prioritizationMessageRepository .findByStateInAndExpiredDate(arrayListOf(MessageState.NEW.name, MessageState.WAIT.name), @@ -115,6 +127,7 @@ open class MessagePrioritizationStateServiceImpl( } } + @Transactional override suspend fun updateMessagesState(ids: List<String>, state: String) { ids.forEach { val updated = updateMessageState(it, state) @@ -124,12 +137,17 @@ open class MessagePrioritizationStateServiceImpl( @Transactional override suspend fun setMessageState(id: String, state: String) { - prioritizationMessageRepository.setStatusForMessageId(id, state) + prioritizationMessageRepository.setStateForMessageId(id, state, Date()) } @Transactional override suspend fun setMessagesState(ids: List<String>, state: String) { - prioritizationMessageRepository.setStatusForMessageIds(ids, state) + prioritizationMessageRepository.setStateForMessageIds(ids, state, Date()) + } + + @Transactional + override suspend fun setMessageStateANdError(id: String, state: String, error: String) { + prioritizationMessageRepository.setStateAndErrorForMessageId(id, state, error, Date()) } @Transactional @@ -141,16 +159,10 @@ open class MessagePrioritizationStateServiceImpl( return saveMessage(updateMessage) } - override suspend fun updateMessageStateAndGroupedIds(id: String, state: String, groupedMessageIds: List<String>) - : MessagePrioritization { - - val groupedIds = groupedMessageIds.joinToString(",") - val updateMessage = getMessage(id).apply { - this.updatedDate = Date() - this.state = state - this.aggregatedMessageIds = groupedIds - } - return saveMessage(updateMessage) + @Transactional + override suspend fun setMessageStateAndAggregatedIds(id: String, state: String, aggregatedIds: List<String>) { + val groupedIds = aggregatedIds.joinToString(",") + prioritizationMessageRepository.setStateAndAggregatedMessageIds(id, state, groupedIds, Date()) } override suspend fun deleteMessage(id: String) { diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageAggregateProcessor.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageAggregateProcessor.kt index 8dd4019dd..45f5c773d 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageAggregateProcessor.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageAggregateProcessor.kt @@ -32,14 +32,36 @@ open class MessageAggregateProcessor : AbstractMessagePrioritizeProcessor<String log.info("@@@@@ received in aggregation processor key($key), value($value)") val ids = value.split(",").map { it.trim() } if (!ids.isNullOrEmpty()) { - if (ids.size == 1) { - processorContext.forward(key, ids.first(), To.child(MessagePrioritizationConstants.PROCESSOR_OUTPUT)) - } else { - /** Implement Aggregation logic in overridden class, If necessary, - Populate New Message and Update status with Prioritized, Forward the message to next processor */ - handleAggregation(ids) - /** Update all messages to Aggregated state */ - messagePrioritizationStateService.setMessagesState(ids, MessageState.AGGREGATED.name) + try { + if (ids.size == 1) { + processorContext.forward(key, ids.first(), To.child(MessagePrioritizationConstants.PROCESSOR_OUTPUT)) + } else { + /** Implement Aggregation logic in overridden class, If necessary, + Populate New Message and Update status with Prioritized, Forward the message to next processor */ + handleAggregation(ids) + /** Update all messages to Aggregated state */ + messagePrioritizationStateService.setMessagesState(ids, MessageState.AGGREGATED.name) + } + } catch (e: Exception) { + val error = "failed in Aggregate message($ids) : ${e.message}" + log.error(error, e) + val storeMessages = messagePrioritizationStateService.getMessages(ids) + if (!storeMessages.isNullOrEmpty()) { + storeMessages.forEach { messagePrioritization -> + try { + /** Update the data store */ + messagePrioritizationStateService.setMessageStateANdError(messagePrioritization.id, + MessageState.ERROR.name, error) + /** Publish to Error topic */ + this.processorContext.forward(messagePrioritization.id, messagePrioritization, + To.child(MessagePrioritizationConstants.SINK_OUTPUT)) + } catch (sendException: Exception) { + log.error("failed to update/publish error message(${messagePrioritization.id}) : " + + "${sendException.message}", e) + } + + } + } } } } diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizeProcessor.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizeProcessor.kt index 5a5aa2575..7dde2655d 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizeProcessor.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizeProcessor.kt @@ -41,12 +41,22 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA override suspend fun processNB(key: ByteArray, value: ByteArray) { log.info("***** received in prioritize processor key(${String(key)})") - val data = JacksonUtils.readValue(String(value), MessagePrioritization::class.java) + val messagePrioritize = JacksonUtils.readValue(String(value), MessagePrioritization::class.java) ?: throw BluePrintProcessorException("failed to convert") - // Save the Message - messagePrioritizationStateService.saveMessage(data) - handleCorrelationAndNextStep(data) - + try { + // Save the Message + messagePrioritizationStateService.saveMessage(messagePrioritize) + handleCorrelationAndNextStep(messagePrioritize) + } catch (e: Exception) { + messagePrioritize.error = "failed in Prioritize message(${messagePrioritize.id}) : ${e.message}" + log.error(messagePrioritize.error) + /** Update the data store */ + messagePrioritizationStateService.setMessageStateANdError(messagePrioritize.id, MessageState.ERROR.name, + messagePrioritize.error!!) + /** Publish to Output topic */ + this.processorContext.forward(messagePrioritize.id, messagePrioritize, + To.child(MessagePrioritizationConstants.SINK_OUTPUT)) + } } override fun init(context: ProcessorContext) { diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt new file mode 100644 index 000000000..02614d821 --- /dev/null +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt @@ -0,0 +1,36 @@ +/* + * Copyright © 2018-2019 AT&T Intellectual Property. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils + +import org.apache.kafka.streams.processor.ProcessorSupplier +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.AbstractMessagePrioritizeProcessor +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.PrioritizationConfiguration +import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService + +object MessageProcessorUtils { + + fun <K, V> bluePrintProcessorSupplier(name: String, prioritizationConfiguration: PrioritizationConfiguration) + : ProcessorSupplier<K, V> { + return ProcessorSupplier<K, V> { + // Dynamically resolve the Prioritization Processor + val processorInstance = BluePrintDependencyService.instance<AbstractMessagePrioritizeProcessor<K, V>>(name) + processorInstance.prioritizationConfiguration = prioritizationConfiguration + processorInstance + } + } + +}
\ No newline at end of file |