2 * Copyright © 2019 IBM.
3 * Modifications Copyright © 2018-2019 AT&T Intellectual Property.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
18 package org.onap.ccsdk.cds.blueprintsprocessor.message.service
20 import com.fasterxml.jackson.databind.node.ObjectNode
21 import org.apache.commons.lang.builder.ToStringBuilder
22 import org.apache.kafka.clients.producer.Callback
23 import org.apache.kafka.clients.producer.KafkaProducer
24 import org.apache.kafka.clients.producer.ProducerRecord
25 import org.apache.kafka.common.header.internals.RecordHeader
26 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
27 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceOutput
28 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.Status
29 import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageProducerProperties
30 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
31 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
32 import org.slf4j.LoggerFactory
33 import java.nio.charset.Charset
35 class KafkaMessageProducerService(
36 private val messageProducerProperties: MessageProducerProperties
38 BlueprintMessageProducerService {
40 private val log = LoggerFactory.getLogger(KafkaMessageProducerService::class.java)!!
42 private var kafkaProducer: KafkaProducer<String, ByteArray>? = null
44 private val messageLoggerService = MessageLoggerService()
48 const val MAX_ERR_MSG_LEN = 128
51 override suspend fun sendMessageNB(key: String, message: Any, headers: MutableMap<String, String>?): Boolean {
52 checkNotNull(messageProducerProperties.topic) { "default topic is not configured" }
53 return sendMessageNB(key, messageProducerProperties.topic!!, message, headers)
56 override suspend fun sendMessageNB(
60 headers: MutableMap<String, String>?
62 var clonedMessage = message
63 if (clonedMessage is ExecutionServiceOutput) {
64 clonedMessage = truncateResponse(clonedMessage)
67 val byteArrayMessage = when (clonedMessage) {
68 is String -> clonedMessage.toByteArray(Charset.defaultCharset())
69 else -> clonedMessage.asJsonString().toByteArray(Charset.defaultCharset())
72 val record = ProducerRecord<String, ByteArray>(topic, key, byteArrayMessage)
73 val recordHeaders = record.headers()
74 messageLoggerService.messageProducing(recordHeaders)
76 headers.forEach { (key, value) -> recordHeaders.add(RecordHeader(key, value.toByteArray())) }
78 val callback = Callback { metadata, exception ->
79 if (exception != null)
80 log.error("ERROR : ${exception.message}")
82 var logMessage = when (clonedMessage) {
83 is ExecutionServiceInput ->
84 "Request published to ${metadata.topic()} for CBA: ${clonedMessage.actionIdentifiers.blueprintName} version: ${clonedMessage.actionIdentifiers.blueprintVersion}"
85 is ExecutionServiceOutput ->
86 "Response published to ${metadata.topic()} for CBA: ${clonedMessage.actionIdentifiers.blueprintName} version: ${clonedMessage.actionIdentifiers.blueprintVersion}"
87 else -> "Message published to(${metadata.topic()}), offset(${metadata.offset()}), headers :$headers"
92 messageTemplate().send(record, callback)
96 fun messageTemplate(additionalConfig: Map<String, ByteArray>? = null): KafkaProducer<String, ByteArray> {
97 log.trace("Producer client properties : ${ToStringBuilder.reflectionToString(messageProducerProperties)}")
98 val configProps = messageProducerProperties.getConfig()
100 /** Add additional Properties */
101 if (additionalConfig != null)
102 configProps.putAll(additionalConfig)
104 if (kafkaProducer == null)
105 kafkaProducer = KafkaProducer(configProps)
107 return kafkaProducer!!
111 * Truncation of BP responses
113 private fun truncateResponse(executionServiceOutput: ExecutionServiceOutput): ExecutionServiceOutput {
114 /** Truncation of error messages */
115 var truncErrMsg = executionServiceOutput.status.errorMessage
116 if (truncErrMsg != null && truncErrMsg.length > MAX_ERR_MSG_LEN) {
117 truncErrMsg = "${truncErrMsg.substring(0, MAX_ERR_MSG_LEN)}" +
118 " [...]. Check Blueprint Processor logs for more information."
120 /** Truncation for Command Executor responses */
121 var truncPayload = executionServiceOutput.payload.deepCopy()
122 val workflowName = executionServiceOutput.actionIdentifiers.actionName
123 if (truncPayload.path("$workflowName-response").has("execute-command-logs")) {
124 var cmdExecLogNode = truncPayload.path("$workflowName-response") as ObjectNode
125 cmdExecLogNode.replace("execute-command-logs", "Check Command Executor logs for more information.".asJsonPrimitive())
127 return ExecutionServiceOutput().apply {
128 correlationUUID = executionServiceOutput.correlationUUID
129 commonHeader = executionServiceOutput.commonHeader
130 actionIdentifiers = executionServiceOutput.actionIdentifiers
131 status = Status().apply {
132 code = executionServiceOutput.status.code
133 eventType = executionServiceOutput.status.eventType
134 timestamp = executionServiceOutput.status.timestamp
135 errorMessage = truncErrMsg
136 message = executionServiceOutput.status.message
138 payload = truncPayload
139 stepData = executionServiceOutput.stepData