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.ExecutionServiceOutput
27 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.Status
28 import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageProducerProperties
29 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
30 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
31 import org.onap.ccsdk.cds.controllerblueprints.core.defaultToUUID
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()
47 const val MAX_ERR_MSG_LEN = 128
50 override suspend fun sendMessageNB(message: Any): Boolean {
51 checkNotNull(messageProducerProperties.topic) { "default topic is not configured" }
52 return sendMessageNB(messageProducerProperties.topic!!, message)
55 override suspend fun sendMessageNB(message: Any, headers: MutableMap<String, String>?): Boolean {
56 checkNotNull(messageProducerProperties.topic) { "default topic is not configured" }
57 return sendMessageNB(messageProducerProperties.topic!!, message, headers)
60 override suspend fun sendMessageNB(
63 headers: MutableMap<String, String>?
65 var clonedMessage = message
66 if (clonedMessage is ExecutionServiceOutput) {
67 clonedMessage = truncateResponse(clonedMessage)
70 val byteArrayMessage = when (clonedMessage) {
71 is String -> clonedMessage.toByteArray(Charset.defaultCharset())
72 else -> clonedMessage.asJsonString().toByteArray(Charset.defaultCharset())
75 val record = ProducerRecord<String, ByteArray>(topic, defaultToUUID(), byteArrayMessage)
76 val recordHeaders = record.headers()
77 messageLoggerService.messageProducing(recordHeaders)
79 headers.forEach { (key, value) -> recordHeaders.add(RecordHeader(key, value.toByteArray())) }
81 val callback = Callback { metadata, exception ->
82 if (exception == null) log.trace("message published to(${metadata.topic()}), offset(${metadata.offset()}), headers :$headers")
83 else log.error("ERROR : ${exception.message}")
85 messageTemplate().send(record, callback)
89 fun messageTemplate(additionalConfig: Map<String, ByteArray>? = null): KafkaProducer<String, ByteArray> {
90 log.trace("Producer client properties : ${ToStringBuilder.reflectionToString(messageProducerProperties)}")
91 val configProps = messageProducerProperties.getConfig()
93 /** Add additional Properties */
94 if (additionalConfig != null)
95 configProps.putAll(additionalConfig)
97 if (kafkaProducer == null)
98 kafkaProducer = KafkaProducer(configProps)
100 return kafkaProducer!!
104 * Truncation of BP responses
106 private fun truncateResponse(executionServiceOutput: ExecutionServiceOutput): ExecutionServiceOutput {
107 /** Truncation of error messages */
108 var truncErrMsg = executionServiceOutput.status.errorMessage
109 if (truncErrMsg != null && truncErrMsg.length > MAX_ERR_MSG_LEN) {
110 truncErrMsg = "${truncErrMsg.substring(0,MAX_ERR_MSG_LEN)}" +
111 " [...]. Check Blueprint Processor logs for more information."
113 /** Truncation for Command Executor responses */
114 var truncPayload = executionServiceOutput.payload.deepCopy()
115 val workflowName = executionServiceOutput.actionIdentifiers.actionName
116 if (truncPayload.path("$workflowName-response").has("execute-command-logs")) {
117 var cmdExecLogNode = truncPayload.path("$workflowName-response") as ObjectNode
118 cmdExecLogNode.replace("execute-command-logs", "Check Command Executor logs for more information.".asJsonPrimitive())
120 return ExecutionServiceOutput().apply {
121 correlationUUID = executionServiceOutput.correlationUUID
122 commonHeader = executionServiceOutput.commonHeader
123 actionIdentifiers = executionServiceOutput.actionIdentifiers
124 status = Status().apply {
125 code = executionServiceOutput.status.code
126 eventType = executionServiceOutput.status.eventType
127 timestamp = executionServiceOutput.status.timestamp
128 errorMessage = truncErrMsg
129 message = executionServiceOutput.status.message
131 payload = truncPayload
132 stepData = executionServiceOutput.stepData