Dmaap publisher code for any component to create a session with the message-router
Issue-ID: CCSDK-693
Change-Id: Ia60079e2b013ea623f776ff7db9dea4bed4636df
Signed-off-by: janani b <janani.b@huawei.com>
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ ============LICENSE_START=======================================================
+ ~ ONAP - CDS
+ ~ ================================================================================
+ ~ Copyright (C) 2019 Huawei Technologies Co., Ltd. All rights reserved.
+ ~ ================================================================================
+ ~ 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.
+ ~ ============LICENSE_END=========================================================
+ -->
+<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.apps.blueprintsprocessor</groupId>
+ <artifactId>commons</artifactId>
+ <version>0.4.1-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>dmaap-lib</artifactId>
+ <packaging>jar</packaging>
+ <name>Blueprints Processor Dmaap Lib</name>
+ <description>Blueprints Processor Dmaap Lib</description>
+
+ <properties>
+ <dmaap.client.version>1.1.5</dmaap.client.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.onap.dmaap.messagerouter.dmaapclient</groupId>
+ <artifactId>dmaapClient</artifactId>
+ <version>${dmaap.client.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-webflux</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.glassfish.jersey.inject</groupId>
+ <artifactId>jersey-hk2</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>javax.ws.rs</groupId>
+ <artifactId>javax.ws.rs-api</artifactId>
+ <version>2.1-m07</version>
+ </dependency>
+ <dependency>
+ <groupId>org.jetbrains.kotlin</groupId>
+ <artifactId>kotlin-test-junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP - CDS
+ * ================================================================================
+ * Copyright (C) 2019 Huawei Technologies Co., Ltd. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.ccsdk.apps.blueprintsprocessor.dmaap
+
+import com.att.nsa.mr.client.MRBatchingPublisher
+import com.att.nsa.mr.client.MRClientFactory
+import com.att.nsa.mr.client.MRPublisher
+import org.slf4j.LoggerFactory
+import org.springframework.boot.context.properties.bind.Binder
+import org.springframework.boot.context.properties.source.ConfigurationPropertySources
+import org.springframework.context.annotation.Configuration
+import org.springframework.context.annotation.PropertySource
+import org.springframework.context.annotation.PropertySources
+import org.springframework.core.env.ConfigurableEnvironment
+import org.springframework.core.env.Environment
+import org.springframework.core.io.support.ResourcePropertySource
+import java.io.IOException
+import java.util.Properties
+import java.util.concurrent.TimeUnit
+
+/**
+ * Representation of DMaap event publisher, to create a session with the
+ * message router and send messages when asked for. The producer.properties
+ * is used for creating a session. In order to overwrite the parameters such
+ * as host, topic, username and password, the event.properties can be used.
+ *
+ * compName : Name of the component appended in the event.properties file
+ * to overwrite.
+ * (E.g., so.topic=cds_so : In this "so" is the component name)
+ */
+@Configuration
+@PropertySources(PropertySource("classpath:event.properties",
+ "classpath:producer.properties"))
+open class DmaapEventPublisher(compName: String = ""): EventPublisher {
+
+ /**
+ * Static variable for logging.
+ */
+ companion object {
+ var log = LoggerFactory.getLogger(DmaapEventPublisher::class.java)!!
+ }
+
+ /**
+ * The component name used in defining the event.properties file.
+ */
+ private var cName:String? = null
+
+ /**
+ * List of topics for a given message to be sent.
+ */
+ var topics = mutableListOf<String>()
+
+ /**
+ * List of clients formed for the list of topics where the messages has to
+ * be sent.
+ */
+ var clients = mutableListOf<MRBatchingPublisher>()
+
+ /**
+ * The populated values from producer.properties which are overwritten
+ * by the event.properties values according to the component information.
+ */
+ var prodProps: Properties = Properties()
+
+
+ init {
+ cName = compName
+ }
+
+ /**
+ * Loads the producer.properties file and populates all the parameters
+ * and then loads the event.properties file and populates the finalized
+ * parameters such as host, topic, username and password if available for
+ * the specified component. With this updated producer.properties, for
+ * each topic a client will be created.
+ */
+ private fun loadPropertiesInfo() {
+ if (prodProps.isEmpty) {
+ parseEventProps(cName!!)
+ addClients()
+ }
+ }
+
+ /**
+ * Adds clients for each topic into a client list.
+ */
+ private fun addClients() {
+ for (topic in topics) {
+ prodProps.setProperty("topic", topic)
+ val client = MRClientFactory.createBatchingPublisher(prodProps)
+ clients.add(client)
+ }
+ }
+
+ /**
+ * Parses the event.properties file and update it into the producer
+ * .properties, where both the files are loaded and stored.
+ */
+ private fun parseEventProps(cName: String) {
+ val env = EnvironmentContext.env as Environment
+ val propSrc = ConfigurationPropertySources.get(env)
+ val proProps = (env as ConfigurableEnvironment).propertySources.get(
+ "class path resource [producer.properties]")
+
+ if (proProps != null) {
+ val entries = (proProps as ResourcePropertySource).source.entries
+ for (e in entries) {
+ prodProps.put(e.key, e.value)
+ }
+ } else {
+ log.info("Unable to load the producer.properties file")
+ }
+
+ val eProps = Binder(propSrc).bind(cName, Properties::class.java).get()
+ val top = eProps.get("topic").toString()
+ if (top != "") {
+ topics.addAll(top.split(","))
+ }
+ prodProps.putAll(eProps)
+ }
+
+ /**
+ * Sends message to the sessions created by the information provided in
+ * the producer.properties file.
+ */
+ override fun sendMessage(partition: String , messages: Collection<String>):
+ Boolean {
+ loadPropertiesInfo()
+ var success = true
+ val dmaapMsgs = mutableListOf<MRPublisher.message>()
+ for (m in messages) {
+ dmaapMsgs.add(MRPublisher.message(partition, m))
+ }
+ for (client in clients) {
+ log.info("Sending messages to the DMaap Server")
+ try {
+ client.send(dmaapMsgs)
+ } catch (e: IOException) {
+ log.error(e.message, e)
+ success = false
+ }
+ }
+ return success
+ }
+
+ /**
+ * Closes the opened session that was used for sending messages.
+ */
+ override fun close(timeout: Long) {
+ log.debug("Closing the DMaap producer clients")
+ if (!clients.isEmpty()) {
+ for (client in clients) {
+ try {
+ client.close(timeout, TimeUnit.SECONDS)
+ } catch (e : IOException) {
+ log.warn("Unable to cleanly close the connection from " +
+ "the client $client", e)
+ }
+ }
+ }
+ }
+
+}
\ No newline at end of file
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP - CDS
+ * ================================================================================
+ * Copyright (C) 2019 Huawei Technologies Co., Ltd. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.ccsdk.apps.blueprintsprocessor.dmaap
+
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.core.env.Environment
+import org.springframework.stereotype.Component
+import javax.annotation.PostConstruct
+
+/**
+ * Abstraction of environment context information component.
+ */
+@Component
+class EnvironmentContext {
+
+ /**
+ * Environment information.
+ */
+ companion object {
+ var env: Environment? = null
+ }
+
+ /**
+ * Environment auto-wired information.
+ */
+ @Autowired
+ var environment: Environment? = null
+
+ /**
+ * Initiates the static variable after the instantiation takes place to
+ * the auto-wired variable.
+ */
+ @PostConstruct
+ private fun initStaticContext() {
+ env = environment
+ }
+
+}
\ No newline at end of file
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP - CDS
+ * ================================================================================
+ * Copyright (C) 2019 Huawei Technologies Co., Ltd. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.ccsdk.apps.blueprintsprocessor.dmaap
+
+/**
+ * Abstraction of a publisher, to send messages with the given partition in a
+ * session and closing the same.
+ */
+interface EventPublisher {
+
+ /**
+ * Sends messages through a session on a given partition.
+ */
+ fun sendMessage(partition: String, messages: Collection<String>): Boolean
+
+ /**
+ * Closes the session with the given time.
+ */
+ fun close(timeout: Long)
+
+}
--- /dev/null
+#
+# ============LICENSE_START=======================================================
+# ONAP - CDS
+# ================================================================================
+# Copyright (C) 2019 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# 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.
+# ============LICENSE_END=========================================================
+#
+
+
+so.topic=cds_so
+so.username=admin
+so.password=admin
+so.host=10.12.6.236:30226
+
--- /dev/null
+#
+# ============LICENSE_START=======================================================
+# ONAP - CDS
+# ================================================================================
+# Copyright (C) 2019 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# 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.
+# ============LICENSE_END=========================================================
+#
+
+#TransportType-Specify which way user want to use. I.e. <HTTPAAF,DME2,HTTPAUTH >
+TransportType=HTTPNOAUTH
+Latitude =50.000000
+Longitude =-100.000000
+Version =3.1
+ServiceName =dmaap-v1.dev.dmaap.dt.saat.acsi.att.com/events
+Environment =TEST
+Partner=BOT_R
+routeOffer=MR1
+SubContextPath =/
+Protocol =http
+MethodType =POST
+username =admin
+password =admin
+contenttype = application/json
+authKey=01234567890abcde:01234567890abcdefghijklmn
+authDate=2016-07-20T11:30:56-0700
+host=10.12.6.236:30227
+topic=org.onap.appc.UNIT-TEST
+partition=1
+maxBatchSize=100
+maxAgeMs=250
+AFT_DME2_EXCHANGE_REQUEST_HANDLERS=com.att.nsa.test.PreferredRouteRequestHandler
+AFT_DME2_EXCHANGE_REPLY_HANDLERS=com.att.nsa.test.PreferredRouteReplyHandler
+AFT_DME2_REQ_TRACE_ON=true
+AFT_ENVIRONMENT=AFTUAT
+AFT_DME2_EP_CONN_TIMEOUT=15000
+AFT_DME2_ROUNDTRIP_TIMEOUT_MS=240000
+AFT_DME2_EP_READ_TIMEOUT_MS=50000
+sessionstickinessrequired=NO
+DME2preferredRouterFilePath=src/test/resources/preferredRoute.txt
+MessageSentThreadOccurance=50
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP - CDS
+ * ================================================================================
+ * Copyright (C) 2019 Huawei Technologies Co., Ltd. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.ccsdk.apps.blueprintprocessor.dmaap
+
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.onap.ccsdk.apps.blueprintsprocessor.dmaap.DmaapEventPublisher
+import org.onap.ccsdk.apps.blueprintsprocessor.dmaap.EnvironmentContext
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration
+import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.http.HttpStatus
+import org.springframework.http.ResponseEntity
+import org.springframework.test.context.ContextConfiguration
+import org.springframework.test.context.TestPropertySource
+import org.springframework.test.context.junit4.SpringRunner
+import org.springframework.web.bind.annotation.PathVariable
+import org.springframework.web.bind.annotation.PostMapping
+import org.springframework.web.bind.annotation.RequestMapping
+import org.springframework.web.bind.annotation.RestController
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+
+/**
+ * Unit test cases for DMaap publisher code.
+ */
+@RunWith(SpringRunner::class)
+@EnableAutoConfiguration(exclude = [DataSourceAutoConfiguration::class])
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT)
+@ContextConfiguration(classes = [EnvironmentContext::class, TestController::class,
+ DmaapEventPublisher::class])
+@TestPropertySource(properties = ["server.port=9111","aai.topic=cds_aai",
+ "aai.username=admin","aai.password=admin","aai.host=127.0.0.1:9111",
+ "mul.topic=cds_mul_1,cds_mul_2", "mul.username=admin","mul.password=admin",
+ "mul.host=127.0.0.1:9111"])
+class TestDmaapEventPublisher {
+
+ /**
+ * Tests the event properties being set properly and sent as request.
+ */
+ @Test
+ fun testEventProperties() {
+ val strList = mutableListOf<String>()
+ val pub = DmaapEventPublisher(compName = "aai")
+ strList.add("{\n" +
+ " \"a\" : \"hello\"\n" +
+ "}")
+ pub.sendMessage("1", strList)
+ pub.close(2)
+ pub.prodProps
+ assertNotNull(pub.prodProps, "The property file updation failed")
+ assertEquals(pub.prodProps.get("topic"), "cds_aai")
+ assertEquals(pub.prodProps.get("username"), "admin")
+ assertEquals(pub.prodProps.get("password"), "admin")
+ assertEquals(pub.prodProps.get("host"), "127.0.0.1:9111")
+ }
+
+ /**
+ * Tests the event properties with multiple topics.
+ */
+ @Test
+ fun testMultiTopicProperties() {
+ val strList = mutableListOf<String>()
+ val pub = DmaapEventPublisher(compName = "mul")
+ strList.add("{\n" +
+ " \"a\" : \"hello\"\n" +
+ "}")
+ pub.sendMessage("1", strList)
+ pub.close(2)
+ var tops = pub.topics
+ assertNotNull(pub.prodProps, "The property file updation failed")
+ assertEquals(tops[0], "cds_mul_1")
+ assertEquals(tops[1], "cds_mul_2")
+ //assertEquals(pub.topics.contains("cds_mul_2`"), true)
+ assertEquals(pub.prodProps.get("username"), "admin")
+ assertEquals(pub.prodProps.get("password"), "admin")
+ assertEquals(pub.prodProps.get("host"), "127.0.0.1:9111")
+ }
+}
+
+/**
+ * Rest controller for testing the client request that is sent.
+ */
+@RestController
+@RequestMapping(path = ["/events"])
+open class TestController {
+
+ /**
+ * Accepts request for a topic and sends a message as response.
+ */
+ @PostMapping(path = ["/{topic}"])
+ fun postTopic(@PathVariable(value = "topic") topic : String):
+ ResponseEntity<Any> {
+ var a = "{\n" +
+ " \"message\" : \"The message is published into $topic " +
+ "topic\"\n" +
+ "}"
+ return ResponseEntity(a, HttpStatus.OK)
+ }
+}
--- /dev/null
+<!--\r
+ ~ ============LICENSE_START=======================================================\r
+ ~ ONAP - CDS\r
+ ~ ================================================================================\r
+ ~ Copyright (C) 2019 Huawei Technologies Co., Ltd. All rights reserved.\r
+ ~ ================================================================================\r
+ ~ Licensed under the Apache License, Version 2.0 (the "License");\r
+ ~ you may not use this file except in compliance with the License.\r
+ ~ You may obtain a copy of the License at\r
+ ~\r
+ ~ http://www.apache.org/licenses/LICENSE-2.0\r
+ ~\r
+ ~ Unless required by applicable law or agreed to in writing, software\r
+ ~ distributed under the License is distributed on an "AS IS" BASIS,\r
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+ ~ See the License for the specific language governing permissions and\r
+ ~ limitations under the License.\r
+ ~ ============LICENSE_END=========================================================\r
+ -->\r
+\r
+\r
+<configuration>\r
+ <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">\r
+ <!-- encoders are assigned the type\r
+ ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->\r
+ <encoder>\r
+ <pattern>%d{HH:mm:ss.SSS} %-5level %logger{100} - %msg%n</pattern>\r
+ </encoder>\r
+ </appender>\r
+\r
+ <logger name="org.springframework.test" level="warn"/>\r
+ <logger name="org.springframework" level="warn"/>\r
+ <logger name="org.hibernate" level="info"/>\r
+ <logger name="org.onap.ccsdk.apps.blueprintsprocessor" level="info"/>\r
+\r
+ <root level="warn">\r
+ <appender-ref ref="STDOUT"/>\r
+ </root>\r
+\r
+</configuration>\r
--- /dev/null
+# ============LICENSE_START==========================================
+# ONAP : APPC
+# ===================================================================
+# Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the License);
+# you may not use this software 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.
+#
+# ECOMP is a trademark and service mark of AT&T Intellectual Property.
+# ============LICENSE_END============================================
+preferredRouteKey=MR1
<module>db-lib</module>
<module>rest-lib</module>
<module>core</module>
+ <module>dmaap-lib</module>
</modules>
<dependencies>
<dependency>
Code Review / ccsdk / apps.git / commitdiff