import kotlinx.coroutines.runBlocking
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ActionIdentifiers
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.CommonHeader
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
@ContextConfiguration(classes = [CliExecutorConfiguration::class,
ExecutionServiceConfiguration::class,
BluePrintSshLibConfiguration::class, BluePrintScriptsServiceImpl::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class, BluePrintDependencyService::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class, BluePrintDependencyService::class])
@DirtiesContext
@TestPropertySource(properties = [], locations = ["classpath:application-test.properties"])
class ComponentCliExecutorTest {
import org.junit.Before
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.functions.config.snapshots.ComponentConfigSnapshotsExecutor.Companion.DIFF_JSON
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [ResourceConfigSnapshotService::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class,
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class,
BluePrintDBLibConfiguration::class, BluePrintLoadConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
@ComponentScan(basePackages = ["org.onap.ccsdk.cds.blueprintsprocessor", "org.onap.ccsdk.cds.controllerblueprints"])
--- /dev/null
+
+To Delete Topics
+------------------
+kafka-topics --zookeeper localhost:2181 --delete --topic prioritize-input-topic
+kafka-topics --zookeeper localhost:2181 --delete --topic prioritize-output-topic
+kafka-topics --zookeeper localhost:2181 --delete --topic prioritize-expired-topic
+kafka-topics --zookeeper localhost:2181 --delete --topic test-prioritize-application-PriorityMessage-changelog
+
+Create Topics
+--------------
+
+kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic prioritize-input-topic
+kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic prioritize-output-topic
+kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic prioritize-expired-topic
+
+To List topics
+----------------
+kafka-topics --list --bootstrap-server localhost:9092
+
+
+To Listen for Output
+----------------------
+kafka-console-consumer --bootstrap-server localhost:9092 --topic prioritize-output-topic --from-beginning
+
+kafka-console-consumer --bootstrap-server localhost:9092 --topic prioritize-input-topic --from-beginning
+
+kafka-console-consumer --bootstrap-server localhost:9092 --topic prioritize-expired-topic --from-beginning
+
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
+ <artifactId>functions</artifactId>
+ <version>0.7.0-SNAPSHOT</version>
+ </parent>
+
+ <groupId>org.onap.ccsdk.cds.blueprintsprocessor.functions</groupId>
+ <artifactId>message-prioritizaion</artifactId>
+
+ <name>Blueprints Processor Function - Message Prioritization</name>
+ <description>Blueprints Processor Function - Message Prioritization</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
+ <artifactId>message-lib</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.h2database</groupId>
+ <artifactId>h2</artifactId>
+ </dependency>
+ </dependencies>
+</project>
--- /dev/null
+/*
+ * 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
+
+import org.apache.kafka.streams.processor.ProcessorContext
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService
+import org.onap.ccsdk.cds.blueprintsprocessor.message.kafka.AbstractBluePrintMessageProcessor
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
+
+/** CDS Message Prioritazation Kafka Stream Processor abstract class to implement */
+abstract class AbstractMessagePrioritizeProcessor<K, V> : AbstractBluePrintMessageProcessor<K, V>() {
+
+ private val log = logger(AbstractMessagePrioritizeProcessor::class)
+
+ lateinit var prioritizationConfiguration: PrioritizationConfiguration
+ lateinit var messagePrioritizationStateService: MessagePrioritizationStateService
+
+ override fun init(context: ProcessorContext) {
+ this.processorContext = context
+ /** Get the State service to update in store */
+ this.messagePrioritizationStateService = BluePrintDependencyService
+ .messagePrioritizationStateService()
+
+ }
+
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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
+
+import org.springframework.context.annotation.ComponentScan
+import org.springframework.context.annotation.Configuration
+
+@Configuration
+@ComponentScan
+open class MessagePrioritizationConfiguration
+
+
+object MessagePrioritizationConstants {
+
+ const val SOURCE_INPUT = "source-prioritization-input"
+
+ const val PROCESSOR_PRIORITIZE = "processor-prioritization-prioritize"
+ const val PROCESSOR_AGGREGATE = "processor-prioritization-aggregate"
+ const val PROCESSOR_OUTPUT = "processor-prioritization-output"
+
+ const val SINK_OUTPUT = "sink-prioritization-output"
+ const val SINK_EXPIRED = "sink-prioritization-expired"
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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
+
+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
+import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BlueprintMessageConsumerService
+import org.onap.ccsdk.cds.blueprintsprocessor.message.service.KafkaStreamConsumerFunction
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+
+open class MessagePrioritizationConsumer(
+ private val bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService) {
+
+ private val log = logger(MessagePrioritizationConsumer::class)
+
+ lateinit var streamingConsumerService: BlueprintMessageConsumerService
+
+ open fun consumerService(selector: String): BlueprintMessageConsumerService {
+ return bluePrintMessageLibPropertyService
+ .blueprintMessageConsumerService(selector)
+ }
+
+ open fun kafkaStreamConsumerFunction(prioritizationConfiguration: PrioritizationConfiguration)
+ : KafkaStreamConsumerFunction {
+ return object : KafkaStreamConsumerFunction {
+
+ override suspend fun createTopology(messageConsumerProperties: MessageConsumerProperties,
+ additionalConfig: Map<String, Any>?): Topology {
+
+ val topology = Topology()
+ val kafkaStreamsBasicAuthConsumerProperties = messageConsumerProperties
+ as KafkaStreamsBasicAuthConsumerProperties
+
+ val topics = kafkaStreamsBasicAuthConsumerProperties.topic.split(",")
+ log.info("Consuming prioritization topics($topics)")
+
+ topology.addSource(MessagePrioritizationConstants.SOURCE_INPUT, *topics.toTypedArray())
+
+ topology.addProcessor(MessagePrioritizationConstants.PROCESSOR_PRIORITIZE,
+ bluePrintProcessorSupplier<ByteArray, ByteArray>(MessagePrioritizationConstants.PROCESSOR_PRIORITIZE,
+ prioritizationConfiguration),
+ MessagePrioritizationConstants.SOURCE_INPUT)
+
+ topology.addProcessor(MessagePrioritizationConstants.PROCESSOR_AGGREGATE,
+ bluePrintProcessorSupplier<String, String>(MessagePrioritizationConstants.PROCESSOR_AGGREGATE,
+ prioritizationConfiguration),
+ MessagePrioritizationConstants.PROCESSOR_PRIORITIZE)
+
+ topology.addProcessor(MessagePrioritizationConstants.PROCESSOR_OUTPUT,
+ bluePrintProcessorSupplier<String, String>(MessagePrioritizationConstants.PROCESSOR_OUTPUT,
+ prioritizationConfiguration),
+ MessagePrioritizationConstants.PROCESSOR_AGGREGATE)
+
+ topology.addSink(MessagePrioritizationConstants.SINK_EXPIRED,
+ prioritizationConfiguration.expiredTopic,
+ 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
+ return topology
+ }
+ }
+ }
+
+ suspend fun startConsuming(prioritizationConfiguration: PrioritizationConfiguration) {
+ streamingConsumerService = consumerService(prioritizationConfiguration.inputTopicSelector)
+
+ // Dynamic Consumer Function to create Topology
+ val consumerFunction = kafkaStreamConsumerFunction(prioritizationConfiguration)
+ streamingConsumerService.consume(null, consumerFunction)
+ }
+
+ suspend fun shutDown() {
+ if (streamingConsumerService != null) {
+ streamingConsumerService.shutDown()
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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
+
+import java.io.Serializable
+
+object MessageActionConstants {
+ const val PRIORITIZE = "prioritize"
+}
+
+enum class MessageState(val id: String) {
+ NEW("new"),
+ WAIT("wait"),
+ EXPIRED("expired"),
+ PRIORITIZED("prioritized"),
+ AGGREGATED("aggregated"),
+ COMPLETED("completed"),
+ ERROR("error")
+}
+
+open class PrioritizationConfiguration : Serializable {
+ lateinit var expiryConfiguration: ExpiryConfiguration
+ lateinit var shutDownConfiguration: ShutDownConfiguration
+ lateinit var cleanConfiguration: CleanConfiguration
+ lateinit var inputTopicSelector: String // Consumer Configuration Selector
+ lateinit var expiredTopic: String // Publish Configuration Selector
+ lateinit var outputTopic: String // Publish Configuration Selector
+}
+
+open class ExpiryConfiguration : Serializable {
+ var frequencyMilli: Long = 30000L
+ var maxPollRecord: Int = 1000
+}
+
+open class ShutDownConfiguration : Serializable {
+ var waitMill: Long = 30000L
+}
+
+open class CleanConfiguration : Serializable {
+ var frequencyMilli: Long = 30000L
+ var expiredRecordsHoldDays: Int = 5
+}
+
+open class UpdateStateRequest : Serializable {
+ lateinit var id: String
+ var group: String? = null
+ var state: String? = null
+}
+
+data class CorrelationCheckResponse(var message: String? = null,
+ var correlated: Boolean = false)
+
+data class TypeCorrelationKey(val type: String, val correlationId: String)
+
--- /dev/null
+/*
+ * 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
+
+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
+
+
+/**
+ * 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(",")
+ return ascendingKey
+}
+
+fun MessagePrioritization.toTypeNCorrelation(): TypeCorrelationKey {
+ val ascendingKey = this.correlationId!!.split(",")
+ .map { it.trim() }.sorted().joinToString(",")
+ return TypeCorrelationKey(this.type, ascendingKey)
+}
--- /dev/null
+/*
+ * 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
--- /dev/null
+/*
+ * 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.db
+
+import org.springframework.data.domain.Pageable
+import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Modifying
+import org.springframework.data.jpa.repository.Query
+import org.springframework.stereotype.Repository
+import org.springframework.transaction.annotation.Transactional
+import java.util.*
+
+@Repository
+@Transactional(readOnly = true)
+interface PrioritizationMessageRepository : JpaRepository<MessagePrioritization, String> {
+
+ @Query("FROM MessagePrioritization pm WHERE pm.group = :group ORDER BY pm.createdDate asc")
+ fun findByGroup(group: String, count: Pageable): List<MessagePrioritization>?
+
+ @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+ "ORDER BY pm.createdDate asc")
+ fun findByGroupAndStateIn(group: String, states: List<String>, count: Pageable): List<MessagePrioritization>?
+
+ @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+ "ORDER BY pm.updatedDate asc")
+ fun findByGroupAndStateInOrderByUpdatedDate(group: String, states: List<String>, count: Pageable)
+ : List<MessagePrioritization>?
+
+ @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+ "AND pm.expiryDate > :expiryCheckDate ORDER BY pm.createdDate asc")
+ fun findByGroupAndStateInAndNotExpiredDate(group: String, states: List<String>, expiryCheckDate: Date,
+ count: Pageable): List<MessagePrioritization>?
+
+ @Query("FROM MessagePrioritization pm WHERE pm.state in :states " +
+ "AND pm.expiryDate < :expiryCheckDate ORDER BY pm.createdDate asc")
+ fun findByStateInAndExpiredDate(states: List<String>, expiryCheckDate: Date,
+ count: Pageable): List<MessagePrioritization>?
+
+ @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+ "AND pm.expiryDate < :expiryCheckDate ORDER BY pm.createdDate asc")
+ fun findByGroupAndStateInAndExpiredDate(group: String, states: List<String>, expiryCheckDate: Date,
+ count: Pageable): List<MessagePrioritization>?
+
+ @Query("FROM MessagePrioritization pm WHERE pm.group = :group " +
+ "AND pm.expiryDate < :expiryCheckDate ORDER BY pm.createdDate asc")
+ fun findByByGroupAndExpiredDate(group: String, expiryCheckDate: Date, count: Pageable): List<MessagePrioritization>?
+
+ @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+ "AND pm.correlationId = :correlationId ORDER BY pm.createdDate asc")
+ fun findByGroupAndCorrelationId(group: String, states: List<String>, correlationId: String)
+ : List<MessagePrioritization>?
+
+ @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+ "AND pm.type in :types AND pm.correlationId = :correlationId ORDER BY pm.createdDate asc")
+ fun findByGroupAndTypesAndCorrelationId(group: String, states: List<String>, types: List<String>,
+ correlationId: String): List<MessagePrioritization>?
+
+ @Modifying
+ @Transactional
+ @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 SET state = :state, updatedDate = :currentDate " +
+ "WHERE id IN :ids")
+ fun setStateForMessageIds(ids: List<String>, state: String, currentDate: Date): Int
+
+ @Modifying
+ @Transactional
+ @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
+ @Query("DELETE FROM MessagePrioritization pm WHERE pm.group = :group")
+ fun deleteGroup(group: String)
+
+ @Modifying
+ @Transactional
+ @Query("DELETE FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state IN :states")
+ fun deleteGroupAndStateIn(group: String, states: List<String>)
+}
+
--- /dev/null
+/*
+ * 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.db
+
+import com.fasterxml.jackson.annotation.JsonFormat
+import org.hibernate.annotations.Proxy
+import org.springframework.data.annotation.LastModifiedDate
+import org.springframework.data.jpa.domain.support.AuditingEntityListener
+import org.springframework.data.jpa.repository.config.EnableJpaAuditing
+import java.util.*
+import javax.persistence.*
+
+@EnableJpaAuditing
+@EntityListeners(AuditingEntityListener::class)
+@Entity
+@Table(name = "MESSAGE_PRIORITIZATION")
+@Proxy(lazy = false)
+open class MessagePrioritization {
+ @Id
+ @Column(name = "message_id", length = 50)
+ lateinit var id: String
+
+ @Column(name = "message_group", length = 50, nullable = false)
+ lateinit var group: String
+
+ @Column(name = "message_type", length = 50, nullable = false)
+ lateinit var type: String
+
+ /** States Defined by MessageState */
+ @Column(name = "message_state", length = 20, nullable = false)
+ lateinit var state: String
+
+ @Lob
+ @Column(name = "message", nullable = false)
+ 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
+
+ @Lob
+ @Column(name = "correlation_id", nullable = true)
+ var correlationId: String? = null
+
+ @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
+ @Temporal(TemporalType.TIMESTAMP)
+ @Column(name = "created_date", nullable = false)
+ var createdDate = Date()
+
+ @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
+ @LastModifiedDate
+ @Temporal(TemporalType.TIMESTAMP)
+ @Column(name = "updated_date", nullable = false)
+ var updatedDate: Date? = null
+
+ @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
+ @Temporal(TemporalType.TIMESTAMP)
+ @Column(name = "expiry_date", nullable = false)
+ var expiryDate: Date? = null
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.service
+
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.PrioritizationMessageRepository
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.toFormatedCorrelation
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.springframework.data.domain.PageRequest
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+import java.util.*
+
+interface MessagePrioritizationStateService {
+
+ suspend fun saveMessage(message: MessagePrioritization): MessagePrioritization
+
+ 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 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 updateMessagesState(ids: List<String>, state: String)
+
+ suspend fun updateMessageState(id: String, state: String): MessagePrioritization
+
+ suspend fun setMessageState(id: String, state: String)
+
+ suspend fun setMessagesState(ids: List<String>, state: String)
+
+ suspend fun setMessageStateANdError(id: String, state: String, error: String)
+
+ suspend fun setMessageStateAndAggregatedIds(id: String, state: String, aggregatedIds: List<String>)
+
+ suspend fun deleteMessage(id: String)
+
+ suspend fun deleteMessageByGroup(group: String)
+
+ suspend fun deleteMessageStates(group: String, states: List<String>)
+
+ suspend fun deleteExpiredMessage(group: String, retentionDays: Int)
+}
+
+@Service
+open class MessagePrioritizationStateServiceImpl(
+ private val prioritizationMessageRepository: PrioritizationMessageRepository)
+ : MessagePrioritizationStateService {
+
+ private val log = logger(MessagePrioritizationStateServiceImpl::class)
+
+ @Transactional
+ override suspend fun saveMessage(message: MessagePrioritization): MessagePrioritization {
+ if (!message.correlationId.isNullOrBlank()) {
+ message.correlationId = message.toFormatedCorrelation()
+ }
+ message.updatedDate = Date()
+ return prioritizationMessageRepository.save(message)
+ }
+
+ override suspend fun getMessage(id: String): MessagePrioritization {
+ return prioritizationMessageRepository.findById(id).orElseGet(null)
+ ?: 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),
+ Date(), PageRequest.of(0, count))
+ }
+
+ override suspend fun getMessageForStatesNotExpiredIn(group: String, states: List<String>, count: Int)
+ : List<MessagePrioritization>? {
+ return prioritizationMessageRepository.findByGroupAndStateInAndNotExpiredDate(group,
+ states, Date(), PageRequest.of(0, count))
+ }
+
+ override suspend fun getMessageForStatesExpired(group: String, states: List<String>, count: Int)
+ : List<MessagePrioritization>? {
+ return prioritizationMessageRepository.findByGroupAndStateInAndExpiredDate(group,
+ states, Date(), PageRequest.of(0, count))
+ }
+
+ override suspend fun getExpiredMessages(group: String, expiryDate: Date, count: Int)
+ : List<MessagePrioritization>? {
+ return prioritizationMessageRepository.findByByGroupAndExpiredDate(group,
+ expiryDate, PageRequest.of(0, count))
+ }
+
+ override suspend fun getCorrelatedMessages(group: String, states: List<String>, types: List<String>?,
+ correlationIds: String): List<MessagePrioritization>? {
+ return if (!types.isNullOrEmpty()) {
+ prioritizationMessageRepository.findByGroupAndTypesAndCorrelationId(group, states, types, correlationIds)
+ } else {
+ prioritizationMessageRepository.findByGroupAndCorrelationId(group, states, correlationIds)
+ }
+ }
+
+ @Transactional
+ override suspend fun updateMessagesState(ids: List<String>, state: String) {
+ ids.forEach {
+ val updated = updateMessageState(it, state)
+ log.info("message($it) update to state(${updated.state})")
+ }
+ }
+
+ @Transactional
+ override suspend fun setMessageState(id: String, state: String) {
+ prioritizationMessageRepository.setStateForMessageId(id, state, Date())
+ }
+
+ @Transactional
+ override suspend fun setMessagesState(ids: List<String>, state: String) {
+ prioritizationMessageRepository.setStateForMessageIds(ids, state, Date())
+ }
+
+ @Transactional
+ override suspend fun setMessageStateANdError(id: String, state: String, error: String) {
+ prioritizationMessageRepository.setStateAndErrorForMessageId(id, state, error, Date())
+ }
+
+ @Transactional
+ override suspend fun updateMessageState(id: String, state: String): MessagePrioritization {
+ val updateMessage = getMessage(id).apply {
+ this.updatedDate = Date()
+ this.state = state
+ }
+ 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) {
+ return prioritizationMessageRepository.deleteById(id)
+ }
+
+ override suspend fun deleteMessageByGroup(group: String) {
+ return prioritizationMessageRepository.deleteGroup(group)
+ }
+
+ override suspend fun deleteMessageStates(group: String, states: List<String>) {
+ return prioritizationMessageRepository.deleteGroupAndStateIn(group, states)
+ }
+
+ override suspend fun deleteExpiredMessage(group: String, retentionDays: Int) {
+ return prioritizationMessageRepository.deleteGroupAndStateIn(group,
+ arrayListOf(MessageState.EXPIRED.name))
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.topology
+
+import org.apache.kafka.streams.processor.To
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.AbstractMessagePrioritizeProcessor
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+
+
+open class MessageAggregateProcessor : AbstractMessagePrioritizeProcessor<String, String>() {
+
+ private val log = logger(MessageAggregateProcessor::class)
+
+ override suspend fun processNB(key: String, value: String) {
+
+ log.info("@@@@@ received in aggregation processor key($key), value($value)")
+ val ids = value.split(",").map { it.trim() }
+ if (!ids.isNullOrEmpty()) {
+ 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)
+ }
+
+ }
+ }
+ }
+ }
+ }
+
+ /** Child will override this implementation , if necessary */
+ open suspend fun handleAggregation(messageIds: List<String>) {
+ log.info("messages($messageIds) aggregated")
+ messageIds.forEach { id ->
+ processorContext.forward(id, id, To.child(MessagePrioritizationConstants.PROCESSOR_OUTPUT))
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.topology
+
+import org.apache.kafka.streams.processor.To
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.AbstractMessagePrioritizeProcessor
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+
+
+open class MessageOutputProcessor : AbstractMessagePrioritizeProcessor<String, String>() {
+
+ private val log = logger(MessageOutputProcessor::class)
+
+ override suspend fun processNB(key: String, value: String) {
+ log.info("$$$$$ received in output processor key($key), value($value)")
+ val message = messagePrioritizationStateService.updateMessageState(value, MessageState.COMPLETED.name)
+ processorContext.forward(message.id, message, To.child(MessagePrioritizationConstants.SINK_OUTPUT))
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.topology
+
+import org.apache.kafka.streams.processor.To
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.PrioritizationConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService
+import org.onap.ccsdk.cds.blueprintsprocessor.message.kafka.AbstractBluePrintMessagePunctuator
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+
+
+class MessagePriorityExpiryPunctuator(private val messagePrioritizationStateService: MessagePrioritizationStateService)
+ : AbstractBluePrintMessagePunctuator() {
+
+ private val log = logger(MessagePriorityExpiryPunctuator::class)
+ lateinit var configuration: PrioritizationConfiguration
+
+ override suspend fun punctuateNB(timestamp: Long) {
+
+ log.info("**** executing expiry punctuator applicationId(${processorContext.applicationId()}), " +
+ "taskId(${processorContext.taskId()})")
+ val expiryConfiguration = configuration.expiryConfiguration
+ val fetchMessages = messagePrioritizationStateService
+ .getExpiryEligibleMessages(expiryConfiguration.maxPollRecord)
+
+ val expiredIds = fetchMessages?.map { it.id }
+ if (expiredIds != null && expiredIds.isNotEmpty()) {
+ messagePrioritizationStateService.updateMessagesState(expiredIds, MessageState.EXPIRED.name)
+ fetchMessages.forEach { expired ->
+ processorContext.forward(expired.id, expired,
+ To.child(MessagePrioritizationConstants.SINK_EXPIRED))
+ }
+ }
+ }
+}
+
+class MessagePriorityCleanPunctuator(private val messagePrioritizationStateService: MessagePrioritizationStateService)
+ : AbstractBluePrintMessagePunctuator() {
+
+ private val log = logger(MessagePriorityCleanPunctuator::class)
+ lateinit var configuration: PrioritizationConfiguration
+
+ override suspend fun punctuateNB(timestamp: Long) {
+ log.info("**** executing clean punctuator applicationId(${processorContext.applicationId()}), " +
+ "taskId(${processorContext.taskId()})")
+ //TODO
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.topology
+
+import org.apache.kafka.common.serialization.Deserializer
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.common.serialization.Serializer
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
+import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
+import java.nio.charset.Charset
+
+open class MessagePrioritizationSerde : Serde<MessagePrioritization> {
+
+ override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) {
+ }
+
+ override fun close() {
+ }
+
+ override fun deserializer(): Deserializer<MessagePrioritization> {
+ return object : Deserializer<MessagePrioritization> {
+ override fun deserialize(topic: String, data: ByteArray): MessagePrioritization {
+ return JacksonUtils.readValue(String(data), MessagePrioritization::class.java)
+ ?: throw BluePrintProcessorException("failed to convert")
+ }
+
+ override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) {
+ }
+
+ override fun close() {
+ }
+ }
+ }
+
+ override fun serializer(): Serializer<MessagePrioritization> {
+ return object : Serializer<MessagePrioritization> {
+ override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) {
+ }
+
+ override fun serialize(topic: String?, data: MessagePrioritization): ByteArray {
+ return data.asJsonString().toByteArray(Charset.defaultCharset())
+ }
+
+ override fun close() {
+ }
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.topology
+
+import org.apache.kafka.streams.processor.Cancellable
+import org.apache.kafka.streams.processor.ProcessorContext
+import org.apache.kafka.streams.processor.PunctuationType
+import org.apache.kafka.streams.processor.To
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.AbstractMessagePrioritizeProcessor
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessageCorrelationUtils
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
+import java.time.Duration
+import java.util.*
+
+
+open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteArray, ByteArray>() {
+
+ private val log = logger(MessagePrioritizeProcessor::class)
+
+ lateinit var expiryCancellable: Cancellable
+ lateinit var cleanCancellable: Cancellable
+
+ override suspend fun processNB(key: ByteArray, value: ByteArray) {
+ log.info("***** received in prioritize processor key(${String(key)})")
+ val messagePrioritize = JacksonUtils.readValue(String(value), MessagePrioritization::class.java)
+ ?: throw BluePrintProcessorException("failed to convert")
+ 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) {
+ super.init(context)
+ /** set up expiry marking cron */
+ initializeExpiryPunctuator()
+ /** Set up cleaning records cron */
+ initializeCleanPunctuator()
+ }
+
+ override fun close() {
+ log.info("closing prioritization processor applicationId(${processorContext.applicationId()}), " +
+ "taskId(${processorContext.taskId()})")
+ expiryCancellable.cancel()
+ cleanCancellable.cancel()
+ }
+
+ open fun initializeExpiryPunctuator() {
+ val expiryPunctuator = MessagePriorityExpiryPunctuator(messagePrioritizationStateService)
+ expiryPunctuator.processorContext = processorContext
+ expiryPunctuator.configuration = prioritizationConfiguration
+ val expiryConfiguration = prioritizationConfiguration.expiryConfiguration
+ expiryCancellable = processorContext.schedule(Duration.ofMillis(expiryConfiguration.frequencyMilli),
+ PunctuationType.WALL_CLOCK_TIME, expiryPunctuator)
+ log.info("Expiry punctuator setup complete with frequency(${expiryConfiguration.frequencyMilli})mSec")
+ }
+
+ open fun initializeCleanPunctuator() {
+ val cleanPunctuator = MessagePriorityCleanPunctuator(messagePrioritizationStateService)
+ cleanPunctuator.processorContext = processorContext
+ cleanPunctuator.configuration = prioritizationConfiguration
+ val cleanConfiguration = prioritizationConfiguration.cleanConfiguration
+ cleanCancellable = processorContext.schedule(Duration.ofDays(cleanConfiguration.expiredRecordsHoldDays.toLong()),
+ PunctuationType.WALL_CLOCK_TIME, cleanPunctuator)
+ log.info("Clean punctuator setup complete with expiry " +
+ "hold(${cleanConfiguration.expiredRecordsHoldDays})days")
+ }
+
+ open suspend fun handleCorrelationAndNextStep(messagePrioritization: MessagePrioritization) {
+ /** Check correlation enabled and correlation field has populated */
+ if (!messagePrioritization.correlationId.isNullOrBlank()) {
+ val id = messagePrioritization.id
+ val group = messagePrioritization.group
+ val correlationId = messagePrioritization.correlationId!!
+ val types = getGroupCorrelationTypes(messagePrioritization)
+ log.info("checking correlation for message($id), group($group), types($types), " +
+ "correlation id($correlationId)")
+
+ /** Get all previously received messages from database for group and optional types and correlation Id */
+ val waitingCorrelatedStoreMessages = messagePrioritizationStateService.getCorrelatedMessages(group,
+ arrayListOf(MessageState.NEW.name, MessageState.WAIT.name), types, correlationId)
+
+ /** If multiple records found, then check correlation */
+ if (!waitingCorrelatedStoreMessages.isNullOrEmpty() && waitingCorrelatedStoreMessages.size > 1) {
+ /** Check all correlation satisfies */
+ val correlationResults = MessageCorrelationUtils
+ .correlatedMessagesWithTypes(waitingCorrelatedStoreMessages, types)
+
+ if (correlationResults.correlated) {
+ /** Correlation satisfied */
+ val correlatedIds = waitingCorrelatedStoreMessages.map { it.id }.joinToString(",")
+ /** Send only correlated ids to next processor */
+ this.processorContext.forward(UUID.randomUUID().toString(), correlatedIds,
+ To.child(MessagePrioritizationConstants.PROCESSOR_AGGREGATE))
+ } else {
+ /** Correlation not satisfied */
+ log.trace("correlation not matched : ${correlationResults.message}")
+ val waitMessageIds = waitingCorrelatedStoreMessages.map { it.id }
+ // Update the Message state to Wait
+ messagePrioritizationStateService.setMessagesState(waitMessageIds, MessageState.WAIT.name)
+ }
+ } else {
+ /** received first message of group and correlation Id, update the message with wait state */
+ messagePrioritizationStateService.setMessageState(messagePrioritization.id, MessageState.WAIT.name)
+ }
+ } else {
+ // No Correlation check needed, simply forward to next processor.
+ messagePrioritizationStateService.setMessageState(messagePrioritization.id, MessageState.PRIORITIZED.name)
+ this.processorContext.forward(messagePrioritization.id, messagePrioritization.id,
+ To.child(MessagePrioritizationConstants.PROCESSOR_AGGREGATE))
+ }
+ }
+
+ /** If consumer wants specific correlation with respect to group and types, then populate the specific types,
+ * otherwise correlation happens with group and correlationId */
+ open fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? {
+ return null
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.CorrelationCheckResponse
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.toFormatedCorrelation
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.toTypeNCorrelation
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+
+object MessageCorrelationUtils {
+
+ /** Assumption is message is of same group **/
+ fun correlatedMessages(collectedMessages: List<MessagePrioritization>): CorrelationCheckResponse {
+ val correlationCheckResponse = CorrelationCheckResponse(message = "not correlated")
+ if (collectedMessages.size > 1) {
+ val filteredMessage = collectedMessages.filter { !it.correlationId.isNullOrBlank() }
+ if (filteredMessage.isNotEmpty()) {
+ val groupedMessage = filteredMessage.groupBy { it.toFormatedCorrelation() }
+ if (groupedMessage.size == 1) {
+ correlationCheckResponse.correlated = true
+ correlationCheckResponse.message = null
+ }
+ }
+ } else {
+ correlationCheckResponse.message = "received only one message for that group"
+ }
+ return correlationCheckResponse
+ }
+
+ /** Assumption is message is of same group and checking for required types **/
+ fun correlatedMessagesWithTypes(collectedMessages: List<MessagePrioritization>, types: List<String>?)
+ : CorrelationCheckResponse {
+
+ return if (!types.isNullOrEmpty() && collectedMessages.size > 1) {
+
+ val unknownMessageTypes = collectedMessages.filter { !types.contains(it.type) }.map { it.id }
+ if (!unknownMessageTypes.isNullOrEmpty()) {
+ throw BluePrintProcessorException("Messages($unknownMessageTypes) is not in type of($types)")
+ }
+
+ val copyTypes = types.toTypedArray().copyOf().toMutableList()
+
+ val filteredMessage = collectedMessages.filter {
+ !it.correlationId.isNullOrBlank()
+ && types.contains(it.type)
+ }
+ var correlatedKeys: MutableSet<String> = mutableSetOf()
+ if (filteredMessage.isNotEmpty()) {
+ val correlatedMap = filteredMessage.groupBy { it.toTypeNCorrelation() }
+ val foundType = correlatedMap.keys.map { it.type }
+ copyTypes.removeAll(foundType)
+ correlatedKeys = correlatedMap.keys.map {
+ it.correlationId
+ }.toMutableSet()
+ }
+ /** Check if any Types missing and same correlation id for all types */
+ return if (copyTypes.isEmpty()) {
+ if (correlatedKeys.size == 1) CorrelationCheckResponse(correlated = true)
+ else CorrelationCheckResponse(message = "not matching correlation keys($correlatedKeys)")
+ } else {
+ CorrelationCheckResponse(message = "couldn't find types($copyTypes)")
+ }
+ } else {
+ return correlatedMessages(collectedMessages)
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.CleanConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.ExpiryConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.PrioritizationConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.ShutDownConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import java.util.*
+
+object MessagePrioritizationSample {
+
+ fun samplePrioritizationConfiguration(): PrioritizationConfiguration {
+ return PrioritizationConfiguration().apply {
+ inputTopicSelector = "prioritize-input"
+ outputTopic = "prioritize-output-topic"
+ expiredTopic = "prioritize-expired-topic"
+ expiryConfiguration = ExpiryConfiguration().apply {
+ frequencyMilli = 10000L
+ maxPollRecord = 2000
+ }
+ shutDownConfiguration = ShutDownConfiguration().apply {
+ waitMill = 2000L
+ }
+ cleanConfiguration = CleanConfiguration().apply {
+ frequencyMilli = 10000L
+ expiredRecordsHoldDays = 5
+ }
+ }
+ }
+
+ private fun currentDatePlusDays(days: Int): Date {
+ val calender = Calendar.getInstance()
+ calender.add(Calendar.DATE, days)
+ return calender.time
+ }
+
+ fun sampleMessages(messageState: String, count: Int): List<MessagePrioritization> {
+ return sampleMessages("sample-group", messageState, count)
+ }
+
+ fun sampleMessages(groupName: String, messageState: String, count: Int): List<MessagePrioritization> {
+ val messages: MutableList<MessagePrioritization> = arrayListOf()
+ repeat(count) {
+ val backPressureMessage = createMessage(groupName, messageState,
+ "sample-type", null)
+ messages.add(backPressureMessage)
+ }
+ return messages
+ }
+
+ fun sampleMessageWithSameCorrelation(groupName: String, messageState: String, count: Int): List<MessagePrioritization> {
+ val messages: MutableList<MessagePrioritization> = arrayListOf()
+ repeat(count) {
+ val backPressureMessage = createMessage(groupName, messageState, "sample-type",
+ "key1=value1,key2=value2")
+ messages.add(backPressureMessage)
+ }
+ return messages
+ }
+
+ fun sampleMessageWithDifferentTypeSameCorrelation(groupName: String, messageState: String,
+ count: Int): List<MessagePrioritization> {
+ val messages: MutableList<MessagePrioritization> = arrayListOf()
+ repeat(count) {
+ val backPressureMessage = createMessage(groupName, messageState, "type-$it",
+ "key1=value1,key2=value2")
+ messages.add(backPressureMessage)
+ }
+ return messages
+ }
+
+ fun createMessage(groupName: String, messageState: String, messageType: String,
+ messageCorrelationId: String?): MessagePrioritization {
+
+ return MessagePrioritization().apply {
+ id = UUID.randomUUID().toString()
+ group = groupName
+ type = messageType
+ state = messageState
+ correlationId = messageCorrelationId
+ message = "I am the Message"
+ createdDate = Date()
+ updatedDate = Date()
+ expiryDate = currentDatePlusDays(3)
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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
--- /dev/null
+/*
+ * 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
+
+import io.mockk.coEvery
+import io.mockk.every
+import io.mockk.spyk
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import org.junit.Before
+import org.junit.runner.RunWith
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.PrioritizationMessageRepository
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessagePrioritizationSample
+import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService
+import org.onap.ccsdk.cds.blueprintsprocessor.message.service.KafkaBasicAuthMessageProducerService
+import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
+import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest
+import org.springframework.context.ApplicationContext
+import org.springframework.test.annotation.DirtiesContext
+import org.springframework.test.context.ContextConfiguration
+import org.springframework.test.context.TestPropertySource
+import org.springframework.test.context.junit4.SpringRunner
+import kotlin.test.Test
+import kotlin.test.assertNotNull
+
+
+@RunWith(SpringRunner::class)
+@DataJpaTest
+@DirtiesContext
+@ContextConfiguration(classes = [BluePrintMessageLibConfiguration::class,
+ BlueprintPropertyConfiguration::class, BluePrintProperties::class,
+ MessagePrioritizationConfiguration::class, TestDatabaseConfiguration::class])
+@TestPropertySource(properties =
+[
+ "spring.jpa.show-sql=true",
+ "spring.jpa.properties.hibernate.show_sql=true",
+ "spring.jpa.hibernate.naming.physical-strategy=org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl",
+
+ "blueprintsprocessor.messageconsumer.prioritize-input.type=kafka-streams-basic-auth",
+ "blueprintsprocessor.messageconsumer.prioritize-input.bootstrapServers=127.0.0.1:9092",
+ "blueprintsprocessor.messageconsumer.prioritize-input.applicationId=test-prioritize-application",
+ "blueprintsprocessor.messageconsumer.prioritize-input.topic=prioritize-input-topic",
+
+ // To send initial test message
+ "blueprintsprocessor.messageproducer.prioritize-input.type=kafka-basic-auth",
+ "blueprintsprocessor.messageproducer.prioritize-input.bootstrapServers=127.0.0.1:9092",
+ "blueprintsprocessor.messageproducer.prioritize-input.topic=prioritize-input-topic"
+])
+open class MessagePrioritizationConsumerTest {
+
+ @Autowired
+ lateinit var applicationContext: ApplicationContext
+
+ @Autowired
+ lateinit var prioritizationMessageRepository: PrioritizationMessageRepository
+
+ @Autowired
+ lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
+
+ @Before
+ fun setup() {
+ BluePrintDependencyService.inject(applicationContext)
+ }
+
+ @Test
+ fun testBluePrintKafkaJDBCKeyStore() {
+ runBlocking {
+ assertNotNull(prioritizationMessageRepository, "failed to get prioritizationMessageRepository")
+
+ val messagePrioritizationService: MessagePrioritizationStateService = BluePrintDependencyService
+ .instance(MessagePrioritizationStateService::class)
+ assertNotNull(messagePrioritizationService, "failed to get messagePrioritizationService")
+
+ MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 1).forEach {
+ val message = messagePrioritizationService.saveMessage(it)
+ val repoResult = messagePrioritizationService.getMessage(message.id)
+ assertNotNull(repoResult, "failed to get inserted message.")
+ }
+ }
+ }
+
+ @Test
+ fun testStartConsuming() {
+ runBlocking {
+ val configuration = MessagePrioritizationSample.samplePrioritizationConfiguration()
+
+ val streamingConsumerService = bluePrintMessageLibPropertyService
+ .blueprintMessageConsumerService(configuration.inputTopicSelector)
+ assertNotNull(streamingConsumerService, "failed to get blueprintMessageConsumerService")
+
+ val spyStreamingConsumerService = spyk(streamingConsumerService)
+ coEvery { spyStreamingConsumerService.consume(any(), any()) } returns Unit
+ coEvery { spyStreamingConsumerService.shutDown() } returns Unit
+ val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
+ val spyMessagePrioritizationConsumer = spyk(messagePrioritizationConsumer)
+
+
+ // Test Topology
+ val kafkaStreamConsumerFunction = spyMessagePrioritizationConsumer.kafkaStreamConsumerFunction(configuration)
+ val messageConsumerProperties = bluePrintMessageLibPropertyService
+ .messageConsumerProperties("blueprintsprocessor.messageconsumer.prioritize-input")
+ val topology = kafkaStreamConsumerFunction.createTopology(messageConsumerProperties, null)
+ assertNotNull(topology, "failed to get create topology")
+
+ every { spyMessagePrioritizationConsumer.consumerService(any()) } returns spyStreamingConsumerService
+ spyMessagePrioritizationConsumer.startConsuming(configuration)
+ spyMessagePrioritizationConsumer.shutDown()
+ }
+ }
+
+ /** Integration Kafka Testing, Enable and use this test case only for local desktop testing with real kafka broker */
+ //@Test
+ fun testMessagePrioritizationConsumer() {
+ runBlocking {
+ val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
+ messagePrioritizationConsumer.startConsuming(MessagePrioritizationSample.samplePrioritizationConfiguration())
+
+ /** Send sample message with every 1 sec */
+ val blueprintMessageProducerService = bluePrintMessageLibPropertyService
+ .blueprintMessageProducerService("prioritize-input") as KafkaBasicAuthMessageProducerService
+ launch {
+ MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 2).forEach {
+ delay(100)
+ val headers: MutableMap<String, String> = hashMapOf()
+ headers["id"] = it.id
+ blueprintMessageProducerService.sendMessageNB(message = it.asJsonString(false),
+ headers = headers)
+ }
+
+ MessagePrioritizationSample
+ .sampleMessageWithSameCorrelation("same-group", MessageState.NEW.name, 2)
+ .forEach {
+ delay(100)
+ val headers: MutableMap<String, String> = hashMapOf()
+ headers["id"] = it.id
+ blueprintMessageProducerService.sendMessageNB(message = it.asJsonString(false),
+ headers = headers)
+ }
+
+ MessagePrioritizationSample
+ .sampleMessageWithDifferentTypeSameCorrelation("group-typed", MessageState.NEW.name, 3)
+ .forEach {
+ delay(2000)
+ val headers: MutableMap<String, String> = hashMapOf()
+ headers["id"] = it.id
+ blueprintMessageProducerService.sendMessageNB(message = it.asJsonString(false),
+ headers = headers)
+ }
+ }
+ delay(10000)
+ messagePrioritizationConsumer.shutDown()
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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
+
+import org.onap.ccsdk.cds.blueprintsprocessor.db.primary.PrimaryDBLibGenericService
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessageAggregateProcessor
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessageOutputProcessor
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessagePrioritizeProcessor
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.ComponentScan
+import org.springframework.context.annotation.Configuration
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate
+import org.springframework.stereotype.Service
+import javax.sql.DataSource
+
+@Configuration
+@ComponentScan(basePackages = ["org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db"])
+@EnableAutoConfiguration
+open class TestDatabaseConfiguration {
+
+ @Bean("primaryDBLibGenericService")
+ open fun primaryDBLibGenericService(dataSource: DataSource): PrimaryDBLibGenericService {
+ return PrimaryDBLibGenericService(NamedParameterJdbcTemplate(dataSource))
+ }
+}
+
+@Service(MessagePrioritizationConstants.PROCESSOR_PRIORITIZE)
+open class TestMessagePrioritizeProcessor : MessagePrioritizeProcessor() {
+ override fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? {
+ return when (messagePrioritization.group) {
+ "group-typed" -> arrayListOf("type-0", "type-1", "type-2")
+ else -> null
+ }
+ }
+}
+
+@Service(MessagePrioritizationConstants.PROCESSOR_AGGREGATE)
+open class DefaultMessageAggregateProcessor() : MessageAggregateProcessor()
+
+@Service(MessagePrioritizationConstants.PROCESSOR_OUTPUT)
+open class DefaultMessageOutputProcessor : MessageOutputProcessor()
\ No newline at end of file
--- /dev/null
+/*
+ * 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.junit.Test
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import kotlin.test.assertTrue
+
+class MessageCorrelationUtilsTest {
+
+ @Test
+ fun testCorrelationKeysReordered() {
+
+ val message1 = MessagePrioritizationSample.createMessage("sample-group", MessageState.NEW.name,
+ "type-0", "key1=value1,key2=value2")
+ val message2 = MessagePrioritizationSample.createMessage("sample-group", MessageState.NEW.name,
+ "type-0", "key2=value2,key1=value1")
+
+ val multipleMessages: MutableList<MessagePrioritization> = arrayListOf()
+ multipleMessages.add(message1)
+ multipleMessages.add(message2)
+ val multipleMessagesResponse = MessageCorrelationUtils.correlatedMessages(multipleMessages)
+ assertTrue(multipleMessagesResponse.correlated, "failed in multipleMessages correlated keys reordered")
+ }
+
+ @Test
+ fun differentTypesWithSameCorrelationMessages() {
+ /** With Types **/
+ /* Assumption is Same group with different types */
+ val differentTypesWithSameCorrelationMessages = MessagePrioritizationSample
+ .sampleMessageWithDifferentTypeSameCorrelation("sample-group", MessageState.NEW.name, 3)
+ val differentTypesWithSameCorrelationMessagesResponse = MessageCorrelationUtils.correlatedMessagesWithTypes(
+ differentTypesWithSameCorrelationMessages,
+ arrayListOf("type-0", "type-1", "type-2"))
+ assertTrue(differentTypesWithSameCorrelationMessagesResponse.correlated,
+ "failed to correlate differentTypesWithSameCorrelationMessagesResponse")
+
+ /* Assumption is Same group with different types and one missing expected types,
+ In this case type-3 message is missing */
+ val differentTypesWithSameCorrelationMessagesResWithMissingType = MessageCorrelationUtils.correlatedMessagesWithTypes(
+ differentTypesWithSameCorrelationMessages,
+ arrayListOf("type-0", "type-1", "type-2", "type-3"))
+ assertTrue(!differentTypesWithSameCorrelationMessagesResWithMissingType.correlated,
+ "failed to correlate differentTypesWithSameCorrelationMessagesResWithMissingType")
+ }
+
+ @Test
+ fun withSameCorrelationMessagesWithIgnoredTypes() {
+ /** With ignoring Types */
+ /** Assumption is only one message received */
+ val withSameCorrelationOneMessages = MessagePrioritizationSample
+ .sampleMessageWithSameCorrelation("sample-group", MessageState.NEW.name, 1)
+ val withSameCorrelationOneMessagesResp = MessageCorrelationUtils.correlatedMessagesWithTypes(
+ withSameCorrelationOneMessages, null)
+ assertTrue(!withSameCorrelationOneMessagesResp.correlated,
+ "failed to correlate withSameCorrelationMessagesResp")
+
+ /** Assumption is two message received for same group with same correlation */
+ val withSameCorrelationMessages = MessagePrioritizationSample
+ .sampleMessageWithSameCorrelation("sample-group", MessageState.NEW.name, 2)
+ val withSameCorrelationMessagesResp = MessageCorrelationUtils.correlatedMessagesWithTypes(
+ withSameCorrelationMessages, null)
+ assertTrue(withSameCorrelationMessagesResp.correlated,
+ "failed to correlate withSameCorrelationMessagesResp")
+ }
+
+ @Test
+ fun differentTypesWithDifferentCorrelationMessage() {
+ /** Assumption is two message received for same group with different expected types and different correlation */
+ val message1 = MessagePrioritizationSample.createMessage("sample-group", MessageState.NEW.name,
+ "type-0", "key1=value1,key2=value2")
+ val message2 = MessagePrioritizationSample.createMessage("sample-group", MessageState.NEW.name,
+ "type-1", "key1=value1,key2=value3")
+ val differentTypesWithDifferentCorrelationMessage: MutableList<MessagePrioritization> = arrayListOf()
+ differentTypesWithDifferentCorrelationMessage.add(message1)
+ differentTypesWithDifferentCorrelationMessage.add(message2)
+ val differentTypesWithDifferentCorrelationMessageResp = MessageCorrelationUtils.correlatedMessagesWithTypes(
+ differentTypesWithDifferentCorrelationMessage,
+ arrayListOf("type-0", "type-1"))
+ assertTrue(!differentTypesWithDifferentCorrelationMessageResp.correlated,
+ "failed to correlate differentTypesWithDifferentCorrelationMessageResp")
+ }
+}
\ No newline at end of file
--- /dev/null
+<!--
+ ~ 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.
+ -->
+
+<configuration>
+
+ <property name="localPattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n"/>
+ <property name="defaultPattern"
+ value="%date{ISO8601,UTC}|%X{RequestID}|%X{InvocationID}|%thread|%X{ServiceName}|%X{ClientIPAddress}|%logger{50}| %msg%n"/>
+ <property name="testing"
+ value="%X{RequestID}|%X{InvocationID}|%logger{50}| %msg%n"/>
+
+ <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+ <!-- encoders are assigned the type
+ ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+ <encoder>
+ <pattern>${localPattern}</pattern>
+ </encoder>
+ </appender>
+
+ <logger name="org.springframework.test" level="warn"/>
+ <logger name="org.springframework" level="warn"/>
+ <logger name="org.hibernate.type.descriptor.sql" level="warn"/>
+ <logger name="org.onap.ccsdk.cds.blueprintsprocessor" level="info"/>
+
+ <root level="warn">
+ <appender-ref ref="STDOUT"/>
+ </root>
+
+</configuration>
package org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor
import com.fasterxml.jackson.databind.JsonNode
+import io.mockk.coEvery
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.spyk
import kotlinx.coroutines.runBlocking
import org.junit.Test
-import org.junit.runner.RunWith
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.StepData
+import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.AbstractScriptComponentFunction
+import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.ComponentFunctionScriptingService
+import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.scripts.BlueprintJythonService
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
import org.onap.ccsdk.cds.controllerblueprints.core.asJsonType
import org.onap.ccsdk.cds.controllerblueprints.core.putJsonElement
import org.onap.ccsdk.cds.controllerblueprints.core.utils.BluePrintMetadataUtils
import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
-import org.springframework.beans.factory.annotation.Autowired
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration
-import org.springframework.context.annotation.ComponentScan
-import org.springframework.test.annotation.DirtiesContext
-import org.springframework.test.context.TestPropertySource
-import org.springframework.test.context.junit4.SpringRunner
-
-@RunWith(SpringRunner::class)
-@EnableAutoConfiguration
-@ComponentScan(basePackages = ["org.onap.ccsdk.cds.blueprintsprocessor", "org.onap.ccsdk.cds.controllerblueprints"])
-@DirtiesContext
-@TestPropertySource(properties =
-["blueprints.processor.functions.python.executor.modulePaths=./../../../../components/scripts/python/ccsdk_netconf,./../../../../components/scripts/python/ccsdk_blueprints",
- "blueprints.processor.functions.python.executor.executionPath=./../../../../components/scripts/python/ccsdk_netconf"],
- locations = ["classpath:application-test.properties"])
-class ComponentNetconfExecutorTest {
-
- @Autowired
- lateinit var componentNetconfExecutor: ComponentNetconfExecutor
+import org.springframework.context.ApplicationContext
+class ComponentNetconfExecutorTest {
@Test
fun testComponentNetconfExecutor() {
runBlocking {
+
+ val applicationContext = mockk<ApplicationContext>()
+ every { applicationContext.getBean(any()) } returns mockk()
+
+ val blueprintJythonService = mockk<BlueprintJythonService>()
+ val mockAbstractScriptComponentFunction = spyk<AbstractScriptComponentFunction>()
+ coEvery { mockAbstractScriptComponentFunction.executeScript(any()) } returns mockk()
+
+ coEvery { blueprintJythonService.jythonComponentInstance(any(), any()) } returns mockAbstractScriptComponentFunction
+
+ val componentFunctionScriptingService = ComponentFunctionScriptingService(applicationContext,
+ blueprintJythonService)
+
+ val componentNetconfExecutor = ComponentNetconfExecutor(componentFunctionScriptingService)
+
val executionServiceInput = JacksonUtils.readValueFromClassPathFile("requests/sample-activate-request.json",
ExecutionServiceInput::class.java)!!
val bluePrintRuntimeService = BluePrintMetadataUtils.getBluePrintRuntime("1234",
"./../../../../components/model-catalog/blueprint-model/test-blueprint/baseconfiguration")
- val assignmentParams = "{\n" +
- " \"ipAddress\": \"127.0.0.1\",\n" +
- " \"hostName\": \"vnf-host\"\n" +
- " }"
+ val assignmentParams = """{
+ "ipAddress" : "127.0.0.1",
+ "hostName" : "vnf-host"
+ }
+ """.trimIndent()
val json = """{
"hostname" : "127.0.0.1"
bluePrintRuntimeService.setNodeTemplateAttributeValue("resource-assignment", "assignment-params",
JacksonUtils.jsonNode(assignmentParams))
- val executionContext = bluePrintRuntimeService.getExecutionContext()
-
componentNetconfExecutor.bluePrintRuntimeService = bluePrintRuntimeService
//TODO("Set Attribute properties")
executionServiceInput.stepData = stepInputData
componentNetconfExecutor.applyNB(executionServiceInput)
}
-
}
}
<module>restconf-executor</module>
<module>cli-executor</module>
<module>config-snapshots</module>
+ <module>message-prioritizaion</module>
</modules>
<dependencies>
import org.junit.Before
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
import org.onap.ccsdk.cds.blueprintsprocessor.core.utils.PayloadUtils
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
InputResourceResolutionProcessor::class, DefaultResourceResolutionProcessor::class,
DatabaseResourceAssignmentProcessor::class, RestResourceResolutionProcessor::class,
CapabilityResourceResolutionProcessor::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class,
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class,
BluePrintDBLibConfiguration::class, BluePrintLoadConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
@ComponentScan(basePackages = ["org.onap.ccsdk.cds.blueprintsprocessor", "org.onap.ccsdk.cds.controllerblueprints"])
*/
package org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.mock
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.rest.RestClientProperties
import org.onap.ccsdk.cds.blueprintsprocessor.rest.service.BluePrintRestLibPropertyService
-class MockBluePrintRestLibPropertyService(bluePrintProperties: BluePrintProperties) :
+class MockBluePrintRestLibPropertyService(bluePrintProperties: BluePrintPropertiesService) :
BluePrintRestLibPropertyService(bluePrintProperties) {
fun mockBlueprintWebClientService (selector: String):
import org.junit.Test
import org.junit.runner.RunWith
import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintCoreConfiguration
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertiesService
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.primary.BluePrintDBLibPropertySevice
import org.onap.ccsdk.cds.blueprintsprocessor.db.primary.PrimaryDatabaseConfiguration
import kotlin.test.assertNotNull
@RunWith(SpringRunner::class)
-@ContextConfiguration(classes = [DatabaseResourceAssignmentProcessor::class, BlueprintPropertyConfiguration::class,
- BluePrintProperties::class, BluePrintDBLibPropertySevice::class, BluePrintDBLibConfiguration::class,
+@ContextConfiguration(classes = [DatabaseResourceAssignmentProcessor::class, BluePrintPropertyConfiguration::class,
+ BluePrintPropertiesService::class, BluePrintDBLibPropertySevice::class, BluePrintDBLibConfiguration::class,
BluePrintCoreConfiguration::class, MockDatabaseConfiguration::class, MockBlueprintProcessorCatalogServiceImpl::class,
- BlueprintPropertiesService::class, PrimaryDatabaseConfiguration::class])
+ BluePrintPropertiesService::class, PrimaryDatabaseConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
class DatabaseResourceResolutionProcessorTest {
import kotlinx.coroutines.runBlocking
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.ResourceAssignmentRuntimeService
import org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.mock.MockBluePrintRestLibPropertyService
import org.onap.ccsdk.cds.blueprintsprocessor.functions.resource.resolution.mock.MockRestResourceResolutionProcessor
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [MockRestResourceResolutionProcessor::class, MockBluePrintRestLibPropertyService::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class, RestClientProperties::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class, RestClientProperties::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
class RestResourceResolutionProcessorTest {
@Autowired
import io.mockk.mockk
import kotlinx.coroutines.runBlocking
import org.junit.Test
-import org.junit.runner.RunWith
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ActionIdentifiers
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.CommonHeader
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.StepData
+import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.ComponentFunctionScriptingService
import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.ComponentScriptExecutor
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintContext
import org.onap.ccsdk.cds.controllerblueprints.core.service.DefaultBluePrintRuntimeService
import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
-import org.springframework.beans.factory.annotation.Autowired
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration
-import org.springframework.context.annotation.ComponentScan
-import org.springframework.test.annotation.DirtiesContext
-import org.springframework.test.context.TestPropertySource
-import org.springframework.test.context.junit4.SpringRunner
+import org.springframework.context.ApplicationContext
import kotlin.test.assertNotNull
-@RunWith(SpringRunner::class)
-@EnableAutoConfiguration
-@ComponentScan(basePackages = ["org.onap.ccsdk.cds.blueprintsprocessor", "org.onap.ccsdk.cds.controllerblueprints"])
-@DirtiesContext
-@TestPropertySource(properties =
-["server.port=9111",
- "blueprintsprocessor.restconfEnabled=true",
- "blueprintsprocessor.restclient.odlPrimary.type=basic-auth",
- "blueprintsprocessor.restclient.odlPrimary.url=http://127.0.0.1:9111",
- "blueprintsprocessor.restclient.odlPrimary.userId=sampleuser",
- "blueprintsprocessor.restclient.odlPrimary.token=sampletoken"],
- locations = ["classpath:application-test.properties"])
class ComponentRestconfExecutorTest {
- @Autowired
- lateinit var componentScriptExecutor: ComponentScriptExecutor
-
@Test
fun `test Restconf Component Instance`() {
runBlocking {
+
+ val applicationContext = mockk<ApplicationContext>()
+ every { applicationContext.getBean(any()) } returns mockk()
+ val componentFunctionScriptingService = ComponentFunctionScriptingService(applicationContext, mockk())
+ val componentScriptExecutor = ComponentScriptExecutor(componentFunctionScriptingService)
+
assertNotNull(componentScriptExecutor, "failed to get ComponentRestconfExecutor instance")
val executionServiceInput = ExecutionServiceInput().apply {
commonHeader = CommonHeader().apply {
package org.onap.ccsdk.cds.blueprintsprocessor.db
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.db.primary.BluePrintDBLibPropertySevice
import org.onap.ccsdk.cds.blueprintsprocessor.db.primary.PrimaryDBLibGenericService
import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
@Configuration
@EnableConfigurationProperties
-open class BluePrintDBLibConfiguration(private var bluePrintProperties: BluePrintProperties) {
+open class BluePrintDBLibConfiguration(private var bluePrintPropertiesService: BluePrintPropertiesService) {
@Bean("primary-database-properties")
open fun getPrimaryProperties(): PrimaryDataSourceProperties {
- return bluePrintProperties.propertyBeanType(DBLibConstants.PREFIX_DB,
+ return bluePrintPropertiesService.propertyBeanType(DBLibConstants.PREFIX_DB,
PrimaryDataSourceProperties::class.java)
}
}
package org.onap.ccsdk.cds.blueprintsprocessor.db.primary
import com.fasterxml.jackson.databind.JsonNode
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.db.*
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
import org.springframework.stereotype.Service
@Service
-class BluePrintDBLibPropertySevice(private var bluePrintProperties: BluePrintProperties) {
+class BluePrintDBLibPropertySevice(private var bluePrintPropertiesService: BluePrintPropertiesService) {
fun JdbcTemplate(jsonNode: JsonNode): BluePrintDBLibGenericService {
val dBConnetionProperties = dBDataSourceProperties(jsonNode)
}
private fun dBDataSourceProperties(prefix: String): DBDataSourceProperties {
- val type = bluePrintProperties.propertyBeanType("$prefix.type", String::class.java)
+ val type = bluePrintPropertiesService.propertyBeanType("$prefix.type", String::class.java)
return when (type) {
DBLibConstants.MARIA_DB -> {
mariaDBConnectionProperties(prefix)
}
private fun mySqlDBConnectionProperties(prefix: String): MySqlDataSourceProperties {
- return bluePrintProperties.propertyBeanType(prefix, MySqlDataSourceProperties::class.java)
+ return bluePrintPropertiesService.propertyBeanType(prefix, MySqlDataSourceProperties::class.java)
}
private fun mariaDBConnectionProperties(prefix: String): MariaDataSourceProperties {
- return bluePrintProperties.propertyBeanType(prefix, MariaDataSourceProperties::class.java)
+ return bluePrintPropertiesService.propertyBeanType(prefix, MariaDataSourceProperties::class.java)
}
private fun primaryDBConnectionProperties(prefix: String): PrimaryDataSourceProperties {
- return bluePrintProperties.propertyBeanType(prefix, PrimaryDataSourceProperties::class.java)
+ return bluePrintPropertiesService.propertyBeanType(prefix, PrimaryDataSourceProperties::class.java)
}
}
\ No newline at end of file
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.ComponentScan
import org.springframework.context.annotation.Configuration
-import org.springframework.context.annotation.Primary
import org.springframework.data.jpa.repository.config.EnableJpaAuditing
import org.springframework.data.jpa.repository.config.EnableJpaRepositories
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate
open class PrimaryDatabaseConfiguration(private val primaryDataSourceProperties: PrimaryDataSourceProperties) {
private val log = LoggerFactory.getLogger(PrimaryDatabaseConfiguration::class.java)!!
- @Primary
@Bean("primaryEntityManager")
open fun primaryEntityManager(): LocalContainerEntityManagerFactoryBean {
val em = LocalContainerEntityManagerFactoryBean()
return em
}
- @Primary
@Bean("primaryDataSource")
open fun primaryDataSource(): DataSource {
val dataSource = DriverManagerDataSource()
return dataSource
}
- @Primary
@Bean("primaryTransactionManager")
open fun primaryTransactionManager(): PlatformTransactionManager {
val transactionManager = JpaTransactionManager()
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.onap.ccsdk.cds.controllerblueprints.core.config.BluePrintLoadConfiguration
import org.springframework.beans.factory.annotation.Autowired
import kotlin.test.assertNotNull
@RunWith(SpringRunner::class)
-@ContextConfiguration(classes = [BlueprintPropertyConfiguration::class, BluePrintProperties::class,
+@ContextConfiguration(classes = [BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class,
BluePrintDBLibConfiguration::class, BluePrintLoadConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
@ComponentScan(basePackages = ["org.onap.ccsdk.cds.blueprintsprocessor", "org.onap.ccsdk.cds.controllerblueprints"])
package org.onap.ccsdk.cds.blueprintsprocessor.dmaap
import com.fasterxml.jackson.databind.JsonNode
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.dmaap.DmaapLibConstants.Companion.SERVICE_BLUEPRINT_DMAAP_LIB_PROPERTY
import org.onap.ccsdk.cds.blueprintsprocessor.dmaap.DmaapLibConstants.Companion.TYPE_HTTP_AAF_AUTH
import org.onap.ccsdk.cds.blueprintsprocessor.dmaap.DmaapLibConstants.Companion.TYPE_HTTP_NO_AUTH
import org.springframework.core.env.Environment
import org.springframework.core.io.support.ResourcePropertySource
import org.springframework.stereotype.Service
-import java.util.Properties
+import java.util.*
/**
* Representation of DMAAP lib property service to load the properties
@Service(SERVICE_BLUEPRINT_DMAAP_LIB_PROPERTY)
@Configuration
@PropertySources(PropertySource("classpath:event.properties"))
-open class BluePrintDmaapLibPropertyService(private var bluePrintProperties:
- BluePrintProperties) {
+open class BluePrintDmaapLibPropertyService(private var bluePrintPropertiesService: BluePrintPropertiesService) {
/**
* Static variable for logging.
*/
companion object {
var log = LoggerFactory.getLogger(
- BluePrintDmaapLibPropertyService::class.java)!!
+ BluePrintDmaapLibPropertyService::class.java)!!
}
/**
* requires.
*/
fun dmaapClientProperties(prefix: String): DmaapClientProperties {
- val type = bluePrintProperties.propertyBeanType(
- "$prefix.type", String::class.java)
- val clientProps : DmaapClientProperties
+ val type = bluePrintPropertiesService.propertyBeanType(
+ "$prefix.type", String::class.java)
+ val clientProps: DmaapClientProperties
when (type) {
TYPE_HTTP_NO_AUTH -> {
- clientProps = bluePrintProperties.propertyBeanType(
- prefix, HttpNoAuthDmaapClientProperties::class.java)
+ clientProps = bluePrintPropertiesService.propertyBeanType(
+ prefix, HttpNoAuthDmaapClientProperties::class.java)
clientProps.props = parseEventProps()
}
TYPE_HTTP_AAF_AUTH -> {
- clientProps = bluePrintProperties.propertyBeanType(
- prefix, AafAuthDmaapClientProperties::class.java)
+ clientProps = bluePrintPropertiesService.propertyBeanType(
+ prefix, AafAuthDmaapClientProperties::class.java)
clientProps.props = parseEventProps()
}
*/
fun dmaapClientProperties(jsonNode: JsonNode): DmaapClientProperties {
val type = jsonNode.get("type").textValue()
- val clientProps : DmaapClientProperties
+ val clientProps: DmaapClientProperties
when (type) {
TYPE_HTTP_NO_AUTH -> {
clientProps = JacksonUtils.readValue(jsonNode,
- HttpNoAuthDmaapClientProperties::class.java)!!
+ HttpNoAuthDmaapClientProperties::class.java)!!
clientProps.props = parseEventProps()
}
TYPE_HTTP_AAF_AUTH -> {
clientProps = JacksonUtils.readValue(jsonNode,
- AafAuthDmaapClientProperties::class.java)!!
+ AafAuthDmaapClientProperties::class.java)!!
clientProps.props = parseEventProps()
}
private fun parseEventProps(): Properties {
val prodProps = Properties()
val proProps = (env as ConfigurableEnvironment).propertySources.get(
- "class path resource [event.properties]")
+ "class path resource [event.properties]")
if (proProps != null) {
val entries = (proProps as ResourcePropertySource).source.entries
import com.fasterxml.jackson.databind.ObjectMapper
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.dmaap.BluePrintDmaapLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.dmaap.BluePrintDmaapLibPropertyService
import org.springframework.beans.factory.annotation.Autowired
@EnableAutoConfiguration(exclude = [DataSourceAutoConfiguration::class])
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT)
@ContextConfiguration(classes = [BluePrintDmaapLibConfiguration::class, TestController::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class])
@TestPropertySource(properties = ["server.port=9111",
"blueprintsprocessor.dmaapclient.aai.topic=cds_aai",
"blueprintsprocessor.dmaapclient.aai.type=HTTPNOAUTH",
package org.onap.ccsdk.cds.blueprintsprocessor.grpc.service
import com.fasterxml.jackson.databind.JsonNode
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.grpc.*
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
import org.onap.ccsdk.cds.controllerblueprints.core.returnNullIfMissing
import org.springframework.stereotype.Service
@Service(GRPCLibConstants.SERVICE_BLUEPRINT_GRPC_LIB_PROPERTY)
-open class BluePrintGrpcLibPropertyService(private var bluePrintProperties: BluePrintProperties) {
+open class BluePrintGrpcLibPropertyService(private var bluePrintPropertiesService: BluePrintPropertiesService) {
/** GRPC Server Lib Property Service */
fun grpcServerProperties(jsonNode: JsonNode): GrpcServerProperties {
}
fun grpcServerProperties(prefix: String): GrpcServerProperties {
- val type = bluePrintProperties.propertyBeanType(
+ val type = bluePrintPropertiesService.propertyBeanType(
"$prefix.type", String::class.java)
return when (type) {
GRPCLibConstants.TYPE_TOKEN_AUTH -> {
}
private fun tokenAuthGrpcServerProperties(prefix: String): TokenAuthGrpcServerProperties {
- return bluePrintProperties.propertyBeanType(prefix, TokenAuthGrpcServerProperties::class.java)
+ return bluePrintPropertiesService.propertyBeanType(prefix, TokenAuthGrpcServerProperties::class.java)
}
private fun tlsAuthGrpcServerProperties(prefix: String): TLSAuthGrpcServerProperties {
- return bluePrintProperties.propertyBeanType(prefix, TLSAuthGrpcServerProperties::class.java)
+ return bluePrintPropertiesService.propertyBeanType(prefix, TLSAuthGrpcServerProperties::class.java)
}
/** GRPC Client Lib Property Service */
}
fun grpcClientProperties(prefix: String): GrpcClientProperties {
- val type = bluePrintProperties.propertyBeanType(
+ val type = bluePrintPropertiesService.propertyBeanType(
"$prefix.type", String::class.java)
return when (type) {
GRPCLibConstants.TYPE_TOKEN_AUTH -> {
}
private fun tokenAuthGrpcClientProperties(prefix: String): TokenAuthGrpcClientProperties {
- return bluePrintProperties.propertyBeanType(prefix, TokenAuthGrpcClientProperties::class.java)
+ return bluePrintPropertiesService.propertyBeanType(prefix, TokenAuthGrpcClientProperties::class.java)
}
private fun tlsAuthGrpcClientProperties(prefix: String): TLSAuthGrpcClientProperties {
- return bluePrintProperties.propertyBeanType(prefix, TLSAuthGrpcClientProperties::class.java)
+ return bluePrintPropertiesService.propertyBeanType(prefix, TLSAuthGrpcClientProperties::class.java)
}
private fun basicAuthGrpcClientProperties(prefix: String): BasicAuthGrpcClientProperties {
- return bluePrintProperties.propertyBeanType(prefix, BasicAuthGrpcClientProperties::class.java)
+ return bluePrintPropertiesService.propertyBeanType(prefix, BasicAuthGrpcClientProperties::class.java)
}
}
\ No newline at end of file
import com.fasterxml.jackson.databind.ObjectMapper
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.grpc.*
import org.onap.ccsdk.cds.controllerblueprints.core.jsonAsJsonType
import org.springframework.beans.factory.annotation.Autowired
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [BluePrintGrpcLibConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class])
@TestPropertySource(properties =
["blueprintsprocessor.grpcclient.sample.type=basic-auth",
"blueprintsprocessor.grpcclient.sample.host=127.0.0.1",
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-streams</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-streams-test-utils</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
/*
* Copyright © 2019 IBM.
+ * Modifications 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.
const val PROPERTY_MESSAGE_CONSUMER_PREFIX = "blueprintsprocessor.messageconsumer."
const val PROPERTY_MESSAGE_PRODUCER_PREFIX = "blueprintsprocessor.messageproducer."
const val TYPE_KAFKA_BASIC_AUTH = "kafka-basic-auth"
+ const val TYPE_KAFKA_STREAMS_BASIC_AUTH = "kafka-streams-basic-auth"
}
}
\ No newline at end of file
package org.onap.ccsdk.cds.blueprintsprocessor.message
+import org.apache.kafka.streams.StreamsConfig
+
/** Producer Properties **/
open class MessageProducerProperties
lateinit var bootstrapServers: String
var topic: String? = null
var clientId: String? = null
+ // strongest producing guarantee
+ var acks: String = "all"
+ var retries: Int = 0
+ // ensure we don't push duplicates
+ var enableIdempotence: Boolean = true
}
/** Consumer Properties **/
open class MessageConsumerProperties
+open class KafkaStreamsConsumerProperties : MessageConsumerProperties() {
+ lateinit var bootstrapServers: String
+ lateinit var applicationId: String
+ lateinit var topic: String
+ var autoOffsetReset: String = "latest"
+ var processingGuarantee: String = StreamsConfig.EXACTLY_ONCE
+}
+
+open class KafkaStreamsBasicAuthConsumerProperties : KafkaStreamsConsumerProperties()
+
open class KafkaMessageConsumerProperties : MessageConsumerProperties() {
lateinit var bootstrapServers: String
lateinit var groupId: String
--- /dev/null
+/*
+ * 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.message.kafka
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.runBlocking
+import org.apache.kafka.streams.processor.Processor
+import org.apache.kafka.streams.processor.ProcessorContext
+import org.apache.kafka.streams.processor.Punctuator
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+
+/** CDS Kafka Stream Processor abstract class to implement */
+abstract class AbstractBluePrintMessageProcessor<K, V> : Processor<K, V> {
+
+ private val log = logger(AbstractBluePrintMessageProcessor::class)
+
+ lateinit var processorContext: ProcessorContext
+
+
+ override fun process(key: K, value: V) = runBlocking(Dispatchers.IO) {
+ try {
+ processNB(key, value)
+ } catch (e: Exception) {
+ log.error("failed in processor(${this.javaClass.simpleName}) message(${this.javaClass.simpleName} :", e)
+ }
+ }
+
+ override fun init(context: ProcessorContext) {
+ log.info("initializing processor (${this.javaClass.simpleName})")
+ this.processorContext = context
+
+ }
+
+ override fun close() {
+ log.info("closing processor (${this.javaClass.simpleName})")
+ }
+
+ abstract suspend fun processNB(key: K, value: V)
+}
+
+/** CDS Kafka Stream Punctuator abstract class to implement */
+abstract class AbstractBluePrintMessagePunctuator : Punctuator {
+ lateinit var processorContext: ProcessorContext
+
+
+ override fun punctuate(timestamp: Long) = runBlocking(Dispatchers.IO) {
+ punctuateNB(timestamp)
+ }
+
+ abstract suspend fun punctuateNB(timestamp: Long)
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.message.kafka
+
+/*
+import org.apache.kafka.streams.processor.ProcessorContext
+import org.apache.kafka.streams.processor.StateStore
+import org.apache.kafka.streams.state.StoreBuilder
+import org.apache.kafka.streams.state.StoreSupplier
+import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibGenericService
+import org.onap.ccsdk.cds.blueprintsprocessor.db.primaryDBLibGenericService
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
+import java.util.*
+
+
+class KafkaJDBCKeyStoreSupplier(private val name: String) : StoreSupplier<KafkaJDBCStore> {
+
+ override fun get(): KafkaJDBCStore {
+ // Get the DBLibGenericService Instance
+ val bluePrintDBLibGenericService = BluePrintDependencyService.primaryDBLibGenericService()
+ return KafkaJDBCStoreImpl(name, bluePrintDBLibGenericService)
+ }
+
+ override fun name(): String {
+ return name
+ }
+
+ override fun metricsScope(): String {
+ return "jdbc-state"
+ }
+}
+
+class KafkaJDBCKeyStoreBuilder(private val storeSupplier: KafkaJDBCKeyStoreSupplier)
+ : StoreBuilder<KafkaJDBCStore> {
+
+ private var logConfig: MutableMap<String, String> = HashMap()
+ private var enableCaching: Boolean = false
+ private var enableLogging = true
+
+ override fun logConfig(): MutableMap<String, String> {
+ return logConfig
+ }
+
+ override fun withCachingDisabled(): StoreBuilder<KafkaJDBCStore> {
+ enableCaching = false
+ return this
+ }
+
+ override fun loggingEnabled(): Boolean {
+ return enableLogging
+ }
+
+ override fun withLoggingDisabled(): StoreBuilder<KafkaJDBCStore> {
+ enableLogging = false
+ return this
+ }
+
+ override fun withCachingEnabled(): StoreBuilder<KafkaJDBCStore> {
+ enableCaching = true
+ return this
+ }
+
+ override fun withLoggingEnabled(config: MutableMap<String, String>?): StoreBuilder<KafkaJDBCStore> {
+ enableLogging = true
+ return this
+ }
+
+ override fun name(): String {
+ return "KafkaJDBCKeyStoreBuilder"
+ }
+
+ override fun build(): KafkaJDBCStore {
+ return storeSupplier.get()
+ }
+}
+
+interface KafkaJDBCStore : StateStore {
+
+ suspend fun query(sql: String, params: Map<String, Any>): List<Map<String, Any>>
+
+ suspend fun update(sql: String, params: Map<String, Any>): Int
+}
+
+
+class KafkaJDBCStoreImpl(private val name: String,
+ private val bluePrintDBLibGenericService: BluePrintDBLibGenericService)
+ : KafkaJDBCStore {
+
+ private val log = logger(KafkaJDBCStoreImpl::class)
+
+ override fun isOpen(): Boolean {
+ log.info("isOpen...")
+ return true
+ }
+
+ override fun init(context: ProcessorContext, root: StateStore) {
+ log.info("init...")
+ }
+
+ override fun flush() {
+ log.info("flush...")
+ }
+
+ override fun close() {
+ log.info("Close...")
+ }
+
+ override fun name(): String {
+ return name
+ }
+
+ override fun persistent(): Boolean {
+ return true
+ }
+
+ override suspend fun query(sql: String, params: Map<String, Any>): List<Map<String, Any>> {
+ log.info("Query : $sql")
+ log.info("Params : $params")
+ return bluePrintDBLibGenericService.query(sql, params)
+ }
+
+ override suspend fun update(sql: String, params: Map<String, Any>): Int {
+ log.info("Query : $sql")
+ log.info("Params : $params")
+ return bluePrintDBLibGenericService.update(sql, params)
+ }
+}
+*/
/*
* Copyright © 2019 IBM.
+ * Modifications 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.
package org.onap.ccsdk.cds.blueprintsprocessor.message.service
import com.fasterxml.jackson.databind.JsonNode
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.message.*
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
import org.springframework.stereotype.Service
@Service(MessageLibConstants.SERVICE_BLUEPRINT_MESSAGE_LIB_PROPERTY)
-open class BluePrintMessageLibPropertyService(private var bluePrintProperties: BluePrintProperties) {
+open class BluePrintMessageLibPropertyService(private var bluePrintPropertiesService: BluePrintPropertiesService) {
fun blueprintMessageProducerService(jsonNode: JsonNode): BlueprintMessageProducerService {
val messageClientProperties = messageProducerProperties(jsonNode)
}
fun messageProducerProperties(prefix: String): MessageProducerProperties {
- val type = bluePrintProperties.propertyBeanType("$prefix.type", String::class.java)
+ val type = bluePrintPropertiesService.propertyBeanType("$prefix.type", String::class.java)
return when (type) {
MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
kafkaBasicAuthMessageProducerProperties(prefix)
}
private fun kafkaBasicAuthMessageProducerProperties(prefix: String): KafkaBasicAuthMessageProducerProperties {
- return bluePrintProperties.propertyBeanType(
+ return bluePrintPropertiesService.propertyBeanType(
prefix, KafkaBasicAuthMessageProducerProperties::class.java)
}
/** Return Message Consumer Properties for [prefix] definitions. */
fun messageConsumerProperties(prefix: String): MessageConsumerProperties {
- val type = bluePrintProperties.propertyBeanType("$prefix.type", String::class.java)
+ val type = bluePrintPropertiesService.propertyBeanType("$prefix.type", String::class.java)
return when (type) {
MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
kafkaBasicAuthMessageConsumerProperties(prefix)
}
+ MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> {
+ kafkaStreamsBasicAuthMessageConsumerProperties(prefix)
+ }
else -> {
throw BluePrintProcessorException("Message adaptor($type) is not supported")
}
MessageLibConstants.TYPE_KAFKA_BASIC_AUTH -> {
JacksonUtils.readValue(jsonNode, KafkaBasicAuthMessageConsumerProperties::class.java)!!
}
+ MessageLibConstants.TYPE_KAFKA_STREAMS_BASIC_AUTH -> {
+ JacksonUtils.readValue(jsonNode, KafkaStreamsBasicAuthConsumerProperties::class.java)!!
+ }
else -> {
throw BluePrintProcessorException("Message adaptor($type) is not supported")
}
is KafkaBasicAuthMessageConsumerProperties -> {
return KafkaBasicAuthMessageConsumerService(messageConsumerProperties)
}
+ is KafkaStreamsBasicAuthConsumerProperties -> {
+ return KafkaStreamsBasicAuthConsumerService(messageConsumerProperties)
+ }
else -> {
throw BluePrintProcessorException("couldn't get Message client service for")
}
}
private fun kafkaBasicAuthMessageConsumerProperties(prefix: String): KafkaBasicAuthMessageConsumerProperties {
- return bluePrintProperties.propertyBeanType(
+ return bluePrintPropertiesService.propertyBeanType(
prefix, KafkaBasicAuthMessageConsumerProperties::class.java)
}
+ private fun kafkaStreamsBasicAuthMessageConsumerProperties(prefix: String): KafkaStreamsBasicAuthConsumerProperties {
+ return bluePrintProperties.propertyBeanType(
+ prefix, KafkaStreamsBasicAuthConsumerProperties::class.java)
+ }
+
}
import kotlinx.coroutines.channels.Channel
import org.apache.kafka.clients.consumer.Consumer
import org.apache.kafka.clients.consumer.ConsumerRecords
+import org.apache.kafka.streams.Topology
import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
interface KafkaConsumerRecordsFunction : ConsumerFunction {
suspend fun invoke(messageConsumerProperties: MessageConsumerProperties, consumer: Consumer<*, *>,
consumerRecords: ConsumerRecords<*, *>)
+}
+
+interface KafkaStreamConsumerFunction : ConsumerFunction {
+ suspend fun createTopology(messageConsumerProperties: MessageConsumerProperties,
+ additionalConfig: Map<String, Any>?): Topology
}
\ No newline at end of file
headers.forEach { (key, value) -> recordHeaders.add(RecordHeader(key, value.toByteArray())) }
}
val callback = Callback { metadata, exception ->
- log.info("message published offset(${metadata.offset()}, headers :$headers )")
+ log.trace("message published to(${metadata.topic()}), offset(${metadata.offset()}), headers :$headers")
}
- messageTemplate().send(record, callback).get()
+ messageTemplate().send(record, callback)
return true
}
configProps[BOOTSTRAP_SERVERS_CONFIG] = messageProducerProperties.bootstrapServers
configProps[KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java
configProps[VALUE_SERIALIZER_CLASS_CONFIG] = ByteArraySerializer::class.java
+ configProps[ACKS_CONFIG] = messageProducerProperties.acks
+ configProps[ENABLE_IDEMPOTENCE_CONFIG] = messageProducerProperties.enableIdempotence
if (messageProducerProperties.clientId != null) {
configProps[CLIENT_ID_CONFIG] = messageProducerProperties.clientId!!
}
--- /dev/null
+/*
+ * 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.message.service
+
+import kotlinx.coroutines.channels.Channel
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.streams.KafkaStreams
+import org.apache.kafka.streams.StreamsConfig
+import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsBasicAuthConsumerProperties
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import java.util.*
+
+open class KafkaStreamsBasicAuthConsumerService(private val messageConsumerProperties: KafkaStreamsBasicAuthConsumerProperties)
+ : BlueprintMessageConsumerService {
+
+ val log = logger(KafkaStreamsBasicAuthConsumerService::class)
+ lateinit var kafkaStreams: KafkaStreams
+
+ private fun streamsConfig(additionalConfig: Map<String, Any>? = null): Properties {
+ val configProperties = Properties()
+ configProperties[StreamsConfig.APPLICATION_ID_CONFIG] = messageConsumerProperties.applicationId
+ configProperties[StreamsConfig.BOOTSTRAP_SERVERS_CONFIG] = messageConsumerProperties.bootstrapServers
+ configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = messageConsumerProperties.autoOffsetReset
+ configProperties[StreamsConfig.PROCESSING_GUARANTEE_CONFIG] = messageConsumerProperties.processingGuarantee
+ // TODO("Security Implementation based on type")
+ /** add or override already set properties */
+ additionalConfig?.let { configProperties.putAll(it) }
+ /** Create Kafka consumer */
+ return configProperties
+ }
+
+ override suspend fun subscribe(additionalConfig: Map<String, Any>?): Channel<String> {
+ throw BluePrintProcessorException("not implemented")
+ }
+
+ override suspend fun subscribe(topics: List<String>, additionalConfig: Map<String, Any>?): Channel<String> {
+ throw BluePrintProcessorException("not implemented")
+ }
+
+ override suspend fun consume(additionalConfig: Map<String, Any>?, consumerFunction: ConsumerFunction) {
+ val streamsConfig = streamsConfig(additionalConfig)
+ val kafkaStreamConsumerFunction = consumerFunction as KafkaStreamConsumerFunction
+ val topology = kafkaStreamConsumerFunction.createTopology(messageConsumerProperties, additionalConfig)
+ log.info("Kafka streams topology : ${topology.describe()}")
+ kafkaStreams = KafkaStreams(topology, streamsConfig)
+ kafkaStreams.cleanUp()
+ kafkaStreams.start()
+ kafkaStreams.localThreadsMetadata().forEach { data -> log.info("Topology : $data") }
+ }
+
+ override suspend fun shutDown() {
+ if (kafkaStreams != null) {
+ kafkaStreams.close()
+ }
+ }
+}
\ No newline at end of file
import org.apache.kafka.common.TopicPartition
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties
import org.onap.ccsdk.cds.controllerblueprints.core.logger
@RunWith(SpringRunner::class)
@DirtiesContext
@ContextConfiguration(classes = [BluePrintMessageLibConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class])
@TestPropertySource(properties =
["blueprintsprocessor.messageconsumer.sample.type=kafka-basic-auth",
"blueprintsprocessor.messageconsumer.sample.bootstrapServers=127.0.0.1:9092",
import org.apache.kafka.clients.producer.KafkaProducer
import org.apache.kafka.clients.producer.RecordMetadata
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.test.annotation.DirtiesContext
@RunWith(SpringRunner::class)
@DirtiesContext
@ContextConfiguration(classes = [BluePrintMessageLibConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class])
@TestPropertySource(properties =
["blueprintsprocessor.messageproducer.sample.type=kafka-basic-auth",
"blueprintsprocessor.messageproducer.sample.bootstrapServers=127.0.0.1:9092",
--- /dev/null
+/*
+ * 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.message.service
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.streams.Topology
+import org.apache.kafka.streams.processor.Processor
+import org.apache.kafka.streams.processor.ProcessorSupplier
+import org.apache.kafka.streams.state.Stores
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsBasicAuthConsumerProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.test.annotation.DirtiesContext
+import org.springframework.test.context.ContextConfiguration
+import org.springframework.test.context.TestPropertySource
+import org.springframework.test.context.junit4.SpringRunner
+import kotlin.test.assertNotNull
+
+
+@RunWith(SpringRunner::class)
+@DirtiesContext
+@ContextConfiguration(classes = [BluePrintMessageLibConfiguration::class,
+ BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+@TestPropertySource(properties =
+[
+ "blueprintsprocessor.messageproducer.sample.type=kafka-basic-auth",
+ "blueprintsprocessor.messageproducer.sample.bootstrapServers=127.0.0.1:9092",
+ "blueprintsprocessor.messageproducer.sample.topic=default-stream-topic",
+ "blueprintsprocessor.messageproducer.sample.clientId=default-client-id",
+
+ "blueprintsprocessor.messageconsumer.stream-consumer.type=kafka-streams-basic-auth",
+ "blueprintsprocessor.messageconsumer.stream-consumer.bootstrapServers=127.0.0.1:9092",
+ "blueprintsprocessor.messageconsumer.stream-consumer.applicationId=test-streams-application",
+ "blueprintsprocessor.messageconsumer.stream-consumer.topic=default-stream-topic"
+
+])
+class KafkaStreamsBasicAuthConsumerServiceTest {
+ @Autowired
+ lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
+
+ @Test
+ fun testProperties() {
+ val blueprintMessageConsumerService = bluePrintMessageLibPropertyService.blueprintMessageConsumerService("stream-consumer")
+ assertNotNull(blueprintMessageConsumerService, "failed to get blueprintMessageProducerService")
+ }
+
+ /** Integration Kafka Testing, Enable and use this test case only for local desktop testing with real kafka broker */
+ //@Test
+ fun testKafkaStreamingMessageConsumer() {
+ runBlocking {
+ val streamingConsumerService = bluePrintMessageLibPropertyService.blueprintMessageConsumerService("stream-consumer")
+
+ // Dynamic Consumer Function to create Topology
+ val consumerFunction = object : KafkaStreamConsumerFunction {
+ override suspend fun createTopology(messageConsumerProperties: MessageConsumerProperties,
+ additionalConfig: Map<String, Any>?): Topology {
+ val topology = Topology()
+ val kafkaStreamsBasicAuthConsumerProperties = messageConsumerProperties
+ as KafkaStreamsBasicAuthConsumerProperties
+
+ val topics = kafkaStreamsBasicAuthConsumerProperties.topic.split(",")
+ topology.addSource("Source", *topics.toTypedArray())
+ // Processor Supplier
+ val firstProcessorSupplier = object : ProcessorSupplier<ByteArray, ByteArray> {
+ override fun get(): Processor<ByteArray, ByteArray> {
+ return FirstProcessor()
+ }
+ }
+ val changelogConfig: MutableMap<String, String> = hashMapOf()
+ changelogConfig.put("min.insync.replicas", "1")
+
+ // Store Buolder
+ val countStoreSupplier = Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore("PriorityMessageState"),
+ Serdes.String(),
+ PriorityMessageSerde())
+ .withLoggingEnabled(changelogConfig)
+
+ topology.addProcessor("FirstProcessor", firstProcessorSupplier, "Source")
+ topology.addStateStore(countStoreSupplier, "FirstProcessor")
+ topology.addSink("SINK", "default-stream-topic-out", Serdes.String().serializer(),
+ PriorityMessageSerde().serializer(), "FirstProcessor")
+ return topology
+ }
+ }
+
+ /** Send message with every 1 sec */
+ val blueprintMessageProducerService = bluePrintMessageLibPropertyService
+ .blueprintMessageProducerService("sample") as KafkaBasicAuthMessageProducerService
+ launch {
+ repeat(5) {
+ delay(1000)
+ val headers: MutableMap<String, String> = hashMapOf()
+ headers["id"] = it.toString()
+ blueprintMessageProducerService.sendMessageNB(message = "this is my message($it)",
+ headers = headers)
+ }
+ }
+ streamingConsumerService.consume(null, consumerFunction)
+ delay(10000)
+ streamingConsumerService.shutDown()
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * 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.message.service
+
+import org.apache.kafka.common.serialization.Deserializer
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.common.serialization.Serializer
+import org.apache.kafka.streams.processor.Processor
+import org.apache.kafka.streams.processor.ProcessorContext
+import org.apache.kafka.streams.state.KeyValueStore
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
+import java.io.Serializable
+import java.nio.charset.Charset
+import java.util.*
+
+class PriorityMessage : Serializable {
+ lateinit var id: String
+ lateinit var requestMessage: String
+}
+
+open class PriorityMessageSerde : Serde<PriorityMessage> {
+
+ override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) {
+ }
+
+ override fun close() {
+ }
+
+ override fun deserializer(): Deserializer<PriorityMessage> {
+ return object : Deserializer<PriorityMessage> {
+ override fun deserialize(topic: String, data: ByteArray): PriorityMessage {
+ return JacksonUtils.readValue(String(data), PriorityMessage::class.java)
+ ?: throw BluePrintProcessorException("failed to convert")
+ }
+
+ override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) {
+ }
+
+ override fun close() {
+ }
+ }
+ }
+
+ override fun serializer(): Serializer<PriorityMessage> {
+ return object : Serializer<PriorityMessage> {
+ override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) {
+ }
+
+ override fun serialize(topic: String?, data: PriorityMessage): ByteArray {
+ return data.asJsonString().toByteArray(Charset.defaultCharset())
+ }
+
+ override fun close() {
+ }
+ }
+ }
+}
+
+
+class FirstProcessor : Processor<ByteArray, ByteArray> {
+
+ private val log = logger(FirstProcessor::class)
+
+ private lateinit var context: ProcessorContext
+ private lateinit var kvStore: KeyValueStore<String, PriorityMessage>
+
+ override fun process(key: ByteArray, value: ByteArray) {
+ log.info("First Processor key(${String(key)} : value(${String(value)})")
+ val newMessage = PriorityMessage().apply {
+ id = UUID.randomUUID().toString()
+ requestMessage = String(value)
+ }
+ kvStore.put(newMessage.id, newMessage)
+ this.context.forward(newMessage.id, newMessage)
+ }
+
+ override fun init(context: ProcessorContext) {
+ log.info("init... ${context.keySerde()}, ${context.valueSerde()}")
+ this.context = context
+ this.kvStore = context.getStateStore("PriorityMessageState") as KeyValueStore<String, PriorityMessage>
+ }
+
+ override fun close() {
+ log.info("Close...")
+ }
+}
\ No newline at end of file
import org.springframework.stereotype.Service
@Configuration
-open class BluePrintCoreConfiguration(private val bluePrintPropertiesService: BlueprintPropertiesService) {
+open class BluePrintCoreConfiguration(private val bluePrintPropertiesService: BluePrintPropertiesService) {
companion object {
const val PREFIX_BLUEPRINT_PROCESSOR = "blueprintsprocessor"
}
@Configuration
-open class BlueprintPropertyConfiguration {
+open class BluePrintPropertyConfiguration {
@Autowired
lateinit var environment: Environment
}
@Service
-open class BlueprintPropertiesService(private var bluePrintPropertyBinder: Binder) {
+open class BluePrintPropertiesService(private var bluePrintPropertyBinder: Binder) {
fun <T> propertyBeanType(prefix: String, type: Class<T>): T {
return bluePrintPropertyBinder.bind(prefix, Bindable.of(type)).get()
}
+++ /dev/null
-/*
- * Copyright © 2017-2018 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.core
-
-import org.springframework.boot.context.properties.bind.Bindable
-import org.springframework.boot.context.properties.bind.Binder
-import org.springframework.stereotype.Service
-
-@Service
-open class BluePrintProperties(var bluePrintPropertyBinder: Binder) {
-
- fun <T> propertyBeanType(prefix: String, type: Class<T>): T {
- return bluePrintPropertyBinder.bind(prefix, Bindable.of(type)).get()
- }
-}
\ No newline at end of file
package org.onap.ccsdk.cds.blueprintsprocessor.rest.service
import com.fasterxml.jackson.databind.JsonNode
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.rest.*
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
import org.springframework.stereotype.Service
@Service(RestLibConstants.SERVICE_BLUEPRINT_REST_LIB_PROPERTY)
-open class BluePrintRestLibPropertyService(private var bluePrintProperties:
- BluePrintProperties) {
+open class BluePrintRestLibPropertyService(private var bluePrintPropertiesService: BluePrintPropertiesService) {
private var preInterceptor: PreInterceptor? = null
private var postInterceptor: PostInterceptor? = null
}
fun restClientProperties(prefix: String): RestClientProperties {
- val type = bluePrintProperties.propertyBeanType(
- "$prefix.type", String::class.java)
+ val type = bluePrintPropertiesService.propertyBeanType(
+ "$prefix.type", String::class.java)
return when (type) {
RestLibConstants.TYPE_BASIC_AUTH -> {
basicAuthRestClientProperties(prefix)
}
else -> {
throw BluePrintProcessorException("Rest adaptor($type) is" +
- " not supported")
+ " not supported")
}
}
}
}
else -> {
throw BluePrintProcessorException(
- "Rest adaptor($type) is not supported")
+ "Rest adaptor($type) is not supported")
}
}
}
private fun blueprintWebClientService(restClientProperties: RestClientProperties):
- BlueprintWebClientService {
+ BlueprintWebClientService {
when (restClientProperties) {
is SSLRestClientProperties -> {
}
private fun tokenRestClientProperties(prefix: String):
- TokenAuthRestClientProperties {
- return bluePrintProperties.propertyBeanType(
- prefix, TokenAuthRestClientProperties::class.java)
+ TokenAuthRestClientProperties {
+ return bluePrintPropertiesService.propertyBeanType(
+ prefix, TokenAuthRestClientProperties::class.java)
}
private fun basicAuthRestClientProperties(prefix: String):
- BasicAuthRestClientProperties {
- return bluePrintProperties.propertyBeanType(
- prefix, BasicAuthRestClientProperties::class.java)
+ BasicAuthRestClientProperties {
+ return bluePrintPropertiesService.propertyBeanType(
+ prefix, BasicAuthRestClientProperties::class.java)
}
private fun sslBasicAuthRestClientProperties(prefix: String):
- SSLRestClientProperties {
+ SSLRestClientProperties {
val sslProps: SSLBasicAuthRestClientProperties =
- bluePrintProperties.propertyBeanType(
- prefix, SSLBasicAuthRestClientProperties::class.java)
+ bluePrintPropertiesService.propertyBeanType(
+ prefix, SSLBasicAuthRestClientProperties::class.java)
val basicProps: BasicAuthRestClientProperties =
- bluePrintProperties.propertyBeanType(
- prefix, BasicAuthRestClientProperties::class.java)
+ bluePrintPropertiesService.propertyBeanType(
+ prefix, BasicAuthRestClientProperties::class.java)
sslProps.basicAuth = basicProps
return sslProps
}
private fun sslTokenAuthRestClientProperties(prefix: String):
- SSLRestClientProperties {
+ SSLRestClientProperties {
val sslProps: SSLTokenAuthRestClientProperties =
- bluePrintProperties.propertyBeanType(prefix,
- SSLTokenAuthRestClientProperties::class.java)
+ bluePrintPropertiesService.propertyBeanType(prefix,
+ SSLTokenAuthRestClientProperties::class.java)
val basicProps: TokenAuthRestClientProperties =
- bluePrintProperties.propertyBeanType(prefix,
- TokenAuthRestClientProperties::class.java)
+ bluePrintPropertiesService.propertyBeanType(prefix,
+ TokenAuthRestClientProperties::class.java)
sslProps.tokenAuth = basicProps
return sslProps
}
private fun sslNoAuthRestClientProperties(prefix: String):
- SSLRestClientProperties {
- return bluePrintProperties.propertyBeanType(
- prefix, SSLRestClientProperties::class.java)
+ SSLRestClientProperties {
+ return bluePrintPropertiesService.propertyBeanType(
+ prefix, SSLRestClientProperties::class.java)
}
private fun policyManagerRestClientProperties(prefix: String):
- PolicyManagerRestClientProperties {
- return bluePrintProperties.propertyBeanType(
- prefix, PolicyManagerRestClientProperties::class.java)
+ PolicyManagerRestClientProperties {
+ return bluePrintPropertiesService.propertyBeanType(
+ prefix, PolicyManagerRestClientProperties::class.java)
}
interface PreInterceptor {
import com.fasterxml.jackson.databind.ObjectMapper
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.rest.BluePrintRestLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.rest.SSLBasicAuthRestClientProperties
import org.onap.ccsdk.cds.blueprintsprocessor.rest.SSLRestClientProperties
import kotlin.test.assertNotNull
@RunWith(SpringRunner::class)
-@ContextConfiguration(classes = [BluePrintRestLibConfiguration::class, BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+@ContextConfiguration(classes = [BluePrintRestLibConfiguration::class, BluePrintPropertyConfiguration::class,
+ BluePrintPropertiesService::class])
@TestPropertySource(properties =
["blueprintsprocessor.restclient.sample.type=basic-auth",
"blueprintsprocessor.restclient.sample.url=http://localhost:8080",
import org.junit.Before
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.rest.BluePrintRestLibConfiguration
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.autoconfigure.EnableAutoConfiguration
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT)
@ContextConfiguration(classes = [BluePrintRestLibConfiguration::class, SampleController::class,
SecurityConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class])
@TestPropertySource(properties =
[
"server.port=8443",
package org.onap.ccsdk.cds.blueprintsprocessor.ssh.service
import com.fasterxml.jackson.databind.JsonNode
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
import org.onap.ccsdk.cds.blueprintsprocessor.ssh.BasicAuthSshClientProperties
import org.onap.ccsdk.cds.blueprintsprocessor.ssh.SshClientProperties
import org.onap.ccsdk.cds.blueprintsprocessor.ssh.SshLibConstants
import org.springframework.stereotype.Service
@Service(SshLibConstants.SERVICE_BLUEPRINT_SSH_LIB_PROPERTY)
-open class BluePrintSshLibPropertyService(private var bluePrintProperties: BluePrintProperties) {
+open class BluePrintSshLibPropertyService(private var bluePrintProperties: BluePrintPropertiesService) {
fun blueprintSshClientService(jsonNode: JsonNode): BlueprintSshClientService {
val restClientProperties = sshClientProperties(jsonNode)
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.ssh.BasicAuthSshClientProperties
import org.onap.ccsdk.cds.blueprintsprocessor.ssh.BluePrintSshLibConfiguration
import org.springframework.beans.factory.annotation.Autowired
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [BluePrintSshLibConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class])
@TestPropertySource(properties =
["blueprintsprocessor.sshclient.sample.type=basic-auth",
"blueprintsprocessor.sshclient.sample.host=127.0.0.1",
import org.apache.sshd.server.session.ServerSession
import org.apache.sshd.server.shell.ProcessShellCommandFactory
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.ssh.BluePrintSshLibConfiguration
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.test.context.ContextConfiguration
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [BluePrintSshLibConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class])
@TestPropertySource(properties =
["blueprintsprocessor.sshclient.sample.type=basic-auth",
"blueprintsprocessor.sshclient.sample.host=localhost",
import org.junit.Test
import org.junit.runner.RunWith
import org.junit.runners.MethodSorters
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.primary.domain.BlueprintModelSearch
import org.onap.ccsdk.cds.controllerblueprints.core.*
@RunWith(SpringRunner::class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@ContextConfiguration(classes = [DesignerApiTestConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class, BluePrintDBLibConfiguration::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class, BluePrintDBLibConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
class BlueprintModelControllerTest {
import org.junit.Test
import org.junit.runner.RunWith
import org.junit.runners.MethodSorters
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [DesignerApiTestConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class, BluePrintDBLibConfiguration::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class, BluePrintDBLibConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
class ModelTypeControllerTest {
import org.junit.Test
import org.junit.runner.RunWith
import org.junit.runners.MethodSorters
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
@RunWith(SpringRunner::class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@ContextConfiguration(classes = [DesignerApiTestConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class, BluePrintDBLibConfiguration::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class, BluePrintDBLibConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
class ResourceDictionaryControllerTest {
import org.junit.Assert
import org.junit.Test
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.designer.api.DesignerApiTestConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.designer.api.load.ModelTypeLoadService
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [DesignerApiTestConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class, BluePrintDBLibConfiguration::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class, BluePrintDBLibConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
class BluePrintEnhancerServiceImplTest {
import org.junit.Test
import org.junit.runner.RunWith
import org.junit.runners.MethodSorters
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.designer.api.DesignerApiTestConfiguration
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [DesignerApiTestConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class, BluePrintDBLibConfiguration::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class, BluePrintDBLibConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
class ModelTypeServiceTest {
import org.junit.Test
import org.junit.runner.RunWith
import org.junit.runners.MethodSorters
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.designer.api.DesignerApiTestConfiguration
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [DesignerApiTestConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class, BluePrintDBLibConfiguration::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class, BluePrintDBLibConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
class ModelTypeReactRepositoryTest {
import org.junit.Test
import org.junit.runner.RunWith
import org.junit.runners.MethodSorters
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.db.BluePrintDBLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.designer.api.DesignerApiTestConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.designer.api.domain.ResourceDictionary
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [DesignerApiTestConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class, BluePrintDBLibConfiguration::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class, BluePrintDBLibConfiguration::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
class ResourceDictionaryReactRepositoryTest {
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.junit.runner.RunWith
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
-import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService
import org.springframework.beans.factory.annotation.Autowired
@RunWith(SpringRunner::class)
@ContextConfiguration(classes = [BluePrintMessageLibConfiguration::class,
- BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+ BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class])
@TestPropertySource(locations = ["classpath:application-test.properties"])
class BluePrintProcessingKafkaConsumerTest {
Code Review / ccsdk / cds.git / commitdiff