Merge "Dmaap Publisher base code"
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/pom.xml b/ms/blueprintsprocessor/modules/commons/dmaap-lib/pom.xml
index 6ad7d84..30f83d3 100644
--- a/ms/blueprintsprocessor/modules/commons/dmaap-lib/pom.xml
+++ b/ms/blueprintsprocessor/modules/commons/dmaap-lib/pom.xml
@@ -31,9 +31,6 @@
<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>
@@ -71,6 +68,10 @@
<artifactId>kotlin-test-junit</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.onap.ccsdk.apps.blueprintsprocessor</groupId>
+ <artifactId>processor-core</artifactId>
+ </dependency>
</dependencies>
</project>
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/AafAuthDmaapClientService.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/AafAuthDmaapClientService.kt
new file mode 100644
index 0000000..dd88409
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/AafAuthDmaapClientService.kt
@@ -0,0 +1,63 @@
+/*-
+ * ============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.impl.MRSimplerBatchPublisher
+
+/**
+ * Representation of DMAAP client service for AAF auth type.
+ */
+class AafAuthDmaapClientService(private val clientProps:
+ AafAuthDmaapClientProperties)
+ : BluePrintDmaapClientService {
+
+ /**
+ * The constructed DMAAP client.
+ */
+ var clients: MutableList<MRBatchingPublisher> = mutableListOf()
+
+
+ /**
+ * Returns the DMAAP client after constructing it properly with the data
+ * that is required for AAF auth connection.
+ */
+ override fun getDmaapClient(): MutableList<MRBatchingPublisher> {
+ if (!clients.isEmpty()) {
+ return clients
+ }
+ val topics = mutableListOf<String>()
+ topics.addAll(clientProps.topic.split(","))
+
+ for (t in topics) {
+ val client = MRClientFactory.createBatchingPublisher(
+ clientProps.host, t, clientProps.username,
+ clientProps.password, 1, 1, false,
+ clientProps.type, "")
+ val batchPublisher = client as MRSimplerBatchPublisher
+ batchPublisher.props = clientProps.props
+ clients.add(client)
+ }
+ return clients
+ }
+
+}
\ No newline at end of file
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapClientService.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapClientService.kt
new file mode 100644
index 0000000..21d7128
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapClientService.kt
@@ -0,0 +1,102 @@
+/*-
+ * ============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.MRPublisher
+import org.slf4j.LoggerFactory
+import java.io.IOException
+import java.util.concurrent.TimeUnit
+
+
+/**
+ * Abstraction of DMAAP client services that could form DMAAP client from the
+ * properties provided. This abstraction also provides a mechanism to send
+ * messages with the given partition in a session and closing the same.
+ */
+interface BluePrintDmaapClientService {
+
+ /**
+ * Static variable for logging.
+ */
+ companion object {
+ var log = LoggerFactory.getLogger(
+ BluePrintDmaapClientService::class.java)!!
+ }
+
+ /**
+ * Returns the properly constructed DMAAP client with the type.
+ */
+ fun getDmaapClient(): MutableList<MRBatchingPublisher>
+
+ /**
+ * Sends messages to the sessions created by the information provided from
+ * application.properties and event.properties file
+ */
+ fun sendMessage(msgs: Collection<String>): Boolean {
+ var success = true
+ val clients = getDmaapClient()
+ val dmaapMsgs = mutableListOf<MRPublisher.message>()
+ for (m in msgs) {
+ dmaapMsgs.add(MRPublisher.message("1", m))
+ }
+ log.info("Sending messages to the DMAAP Server")
+ for (client in clients) {
+ try {
+ client.send(dmaapMsgs)
+ } catch (e: IOException) {
+ success = false
+ log.error(e.message, e)
+ }
+ }
+ return success
+ }
+
+ /**
+ * Sends message to the sessions created by the information provided from
+ * application.properties and event.properties file
+ */
+ fun sendMessage(msg: String): Boolean {
+ val msgs = mutableListOf<String>()
+ msgs.add(msg)
+ return sendMessage(msgs)
+ }
+
+ /**
+ * Closes the opened session that was used for sending messages.
+ */
+ fun close(timeout: Long): MutableList<MutableList<MRPublisher.message>>? {
+ log.debug("Closing the DMAAP producer clients")
+ var msgs: MutableList<MutableList<MRPublisher.message>> =
+ mutableListOf()
+ val clients = getDmaapClient()
+ for (client in clients) {
+ try {
+ var ms = client.close(timeout, TimeUnit.SECONDS)
+ msgs.add(ms)
+ } catch (e: IOException) {
+ log.warn("Unable to cleanly close the connection from the " +
+ "client $client", e)
+ }
+ }
+ return msgs
+ }
+}
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapLibConfiguration.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapLibConfiguration.kt
new file mode 100644
index 0000000..ca1a07b
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapLibConfiguration.kt
@@ -0,0 +1,46 @@
+/*-
+ * ============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.boot.context.properties.EnableConfigurationProperties
+import org.springframework.context.annotation.ComponentScan
+import org.springframework.context.annotation.Configuration
+
+/**
+ * Representation of DMAAP lib configuration to load the required property
+ * files into the application context.
+ */
+@Configuration
+@ComponentScan
+@EnableConfigurationProperties
+open class BluePrintDmaapLibConfiguration
+
+/**
+ * Util constants required for DMAAP library to use.
+ */
+class DmaapLibConstants {
+ companion object {
+ const val SERVICE_BLUEPRINT_DMAAP_LIB_PROPERTY = "blueprint" +
+ "-dmaap-lib-property-service"
+ const val TYPE_HTTP_NO_AUTH = "HTTPNOAUTH"
+ const val TYPE_HTTP_AAF_AUTH = "HTTPAAF"
+ }
+}
\ No newline at end of file
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapLibPropertyService.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapLibPropertyService.kt
new file mode 100644
index 0000000..7abbefb
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapLibPropertyService.kt
@@ -0,0 +1,187 @@
+/*-
+ * ============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.fasterxml.jackson.databind.JsonNode
+import org.onap.ccsdk.apps.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.apps.blueprintsprocessor.dmaap.DmaapLibConstants.Companion.SERVICE_BLUEPRINT_DMAAP_LIB_PROPERTY
+import org.onap.ccsdk.apps.blueprintsprocessor.dmaap.DmaapLibConstants.Companion.TYPE_HTTP_AAF_AUTH
+import org.onap.ccsdk.apps.blueprintsprocessor.dmaap.DmaapLibConstants.Companion.TYPE_HTTP_NO_AUTH
+import org.onap.ccsdk.apps.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.apps.controllerblueprints.core.utils.JacksonUtils
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Autowired
+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 org.springframework.stereotype.Service
+import java.util.Properties
+
+/**
+ * Representation of DMAAP lib property service to load the properties
+ * according to the connection type to the DMAAP server and returning back
+ * the appropriate DMAAP client to send messages DMAAP client.
+ */
+@Service(SERVICE_BLUEPRINT_DMAAP_LIB_PROPERTY)
+@Configuration
+@PropertySources(PropertySource("classpath:event.properties"))
+open class BluePrintDmaapLibPropertyService(private var bluePrintProperties:
+ BluePrintProperties) {
+
+ /**
+ * Static variable for logging.
+ */
+ companion object {
+ var log = LoggerFactory.getLogger(
+ BluePrintDmaapLibPropertyService::class.java)!!
+ }
+
+ /**
+ * Environment entity to derive it from the system to load a specific
+ * property file.
+ */
+ @Autowired
+ lateinit var env: Environment
+
+ /**
+ * Returns the DMAAP client by providing the input properties as a JSON
+ * node.
+ */
+ fun blueprintDmaapClientService(jsonNode: JsonNode):
+ BluePrintDmaapClientService {
+ val dmaapProps = dmaapClientProperties(jsonNode)
+ return blueprintDmaapClientService(dmaapProps)
+ }
+
+ /**
+ * Returns the DMAAP client by providing the input properties as a
+ * selector string.
+ */
+ fun blueprintDmaapClientService(selector: String):
+ BluePrintDmaapClientService {
+ val prefix = "blueprintsprocessor.dmaapclient.$selector"
+ val dmaapProps = dmaapClientProperties(prefix)
+ return blueprintDmaapClientService(dmaapProps)
+ }
+
+ /**
+ * Returns the DMAAP client properties from the type of connection it
+ * requires.
+ */
+ fun dmaapClientProperties(prefix: String): DmaapClientProperties {
+ val type = bluePrintProperties.propertyBeanType(
+ "$prefix.type", String::class.java)
+ val clientProps : DmaapClientProperties
+
+ when (type) {
+ TYPE_HTTP_NO_AUTH -> {
+ clientProps = bluePrintProperties.propertyBeanType(
+ prefix, HttpNoAuthDmaapClientProperties::class.java)
+ clientProps.props = parseEventProps()
+ }
+
+ TYPE_HTTP_AAF_AUTH -> {
+ clientProps = bluePrintProperties.propertyBeanType(
+ prefix, AafAuthDmaapClientProperties::class.java)
+ clientProps.props = parseEventProps()
+ }
+
+ else -> {
+ throw BluePrintProcessorException("DMAAP adaptor($type) is " +
+ "not supported")
+ }
+ }
+ return clientProps
+ }
+
+ /**
+ * Returns the DMAAP client properties from the type of connection it
+ * requires.
+ */
+ fun dmaapClientProperties(jsonNode: JsonNode): DmaapClientProperties {
+ val type = jsonNode.get("type").textValue()
+ val clientProps : DmaapClientProperties
+
+ when (type) {
+ TYPE_HTTP_NO_AUTH -> {
+ clientProps = JacksonUtils.readValue(jsonNode,
+ HttpNoAuthDmaapClientProperties::class.java)!!
+ clientProps.props = parseEventProps()
+ }
+
+ TYPE_HTTP_AAF_AUTH -> {
+ clientProps = JacksonUtils.readValue(jsonNode,
+ AafAuthDmaapClientProperties::class.java)!!
+ clientProps.props = parseEventProps()
+ }
+
+ else -> {
+ throw BluePrintProcessorException("DMAAP adaptor($type) is " +
+ "not supported")
+ }
+ }
+ return clientProps
+ }
+
+ /**
+ * Returns DMAAP client service according to the type of client properties.
+ */
+ private fun blueprintDmaapClientService(clientProps: DmaapClientProperties):
+ BluePrintDmaapClientService {
+ when (clientProps) {
+ is HttpNoAuthDmaapClientProperties -> {
+ return HttpNoAuthDmaapClientService(clientProps)
+ }
+
+ is AafAuthDmaapClientProperties -> {
+ return AafAuthDmaapClientService(clientProps)
+ }
+
+ else -> {
+ throw BluePrintProcessorException("Unable to get the DMAAP " +
+ "client")
+ }
+ }
+ }
+
+ /**
+ * Parses the event.properties file which contains the default values for
+ * the connection required.
+ */
+ private fun parseEventProps(): Properties {
+ val prodProps = Properties()
+ val proProps = (env as ConfigurableEnvironment).propertySources.get(
+ "class path resource [event.properties]")
+
+ if (proProps != null) {
+ val entries = (proProps as ResourcePropertySource).source.entries
+ for (e in entries) {
+ prodProps.put(e.key, e.value)
+ }
+ } else {
+ log.error("Unable to load the event.properties file")
+ }
+ return prodProps
+ }
+}
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/EventPublisher.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/DmaapClientProperties.kt
similarity index 63%
rename from ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/EventPublisher.kt
rename to ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/DmaapClientProperties.kt
index 7d02e80..97df272 100644
--- a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/EventPublisher.kt
+++ b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/DmaapClientProperties.kt
@@ -20,20 +20,27 @@
package org.onap.ccsdk.apps.blueprintsprocessor.dmaap
+import java.util.Properties
+
/**
- * Abstraction of a publisher, to send messages with the given partition in a
- * session and closing the same.
+ * Representation of data required for all DMAAP client.
*/
-interface EventPublisher {
+open class DmaapClientProperties {
+ lateinit var props: Properties
+ lateinit var type: String
+ lateinit var host: String
+ lateinit var topic: String
+}
- /**
- * Sends messages through a session on a given partition.
- */
- fun sendMessage(partition: String, messages: Collection<String>): Boolean
+/**
+ * Representation of data required for HTTP no auth DMAAP client.
+ */
+open class HttpNoAuthDmaapClientProperties : DmaapClientProperties()
- /**
- * Closes the session with the given time.
- */
- fun close(timeout: Long)
-
+/**
+ * Representation of data required for AAF auth DMAAP client.
+ */
+open class AafAuthDmaapClientProperties : DmaapClientProperties() {
+ lateinit var username: String
+ lateinit var password: String
}
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/DmaapEventPublisher.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/DmaapEventPublisher.kt
deleted file mode 100644
index 7c686f0..0000000
--- a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/DmaapEventPublisher.kt
+++ /dev/null
@@ -1,181 +0,0 @@
-/*-
- * ============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
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/EnvironmentContext.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/EnvironmentContext.kt
deleted file mode 100644
index 1d2a28c..0000000
--- a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/EnvironmentContext.kt
+++ /dev/null
@@ -1,56 +0,0 @@
-/*-
- * ============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
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/HttpNoAuthDmaapClientService.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/HttpNoAuthDmaapClientService.kt
new file mode 100644
index 0000000..ad023b9
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/HttpNoAuthDmaapClientService.kt
@@ -0,0 +1,62 @@
+/*-
+ * ============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.createSimplePublisher
+import com.att.nsa.mr.client.impl.MRSimplerBatchPublisher
+
+/**
+ * Representation of DMAAP client service for HTTP no auth type.
+ */
+class HttpNoAuthDmaapClientService(private val clientProps:
+ HttpNoAuthDmaapClientProperties)
+ : BluePrintDmaapClientService {
+
+ /**
+ * The constructed DMAAP client.
+ */
+ var clients: MutableList<MRBatchingPublisher> = mutableListOf()
+
+
+ /**
+ * Returns the DMAAP client after constructing it properly with the data
+ * that is required for HTTP no auth connection.
+ */
+ override fun getDmaapClient(): MutableList<MRBatchingPublisher> {
+ if (!clients.isEmpty()) {
+ return clients
+ }
+ val topics = mutableListOf<String>()
+ topics.addAll(clientProps.topic.split(","))
+
+ for (t in topics) {
+ val client = createSimplePublisher(clientProps.host, t)
+ val batchPublisher = client as MRSimplerBatchPublisher
+ batchPublisher.setProtocolFlag(clientProps.type)
+ batchPublisher.props = clientProps.props
+ clients.add(client)
+ }
+
+ return clients
+ }
+
+}
\ No newline at end of file
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/resources/event.properties b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/resources/event.properties
index be764d8..320b08e 100644
--- a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/resources/event.properties
+++ b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/resources/event.properties
@@ -18,9 +18,9 @@
# ============LICENSE_END=========================================================
#
+#TransportType-Specify which way user want to use. I.e. <HTTPAAF,DME2,HTTPAUTH >
+Protocol =http
+partition=1
+contenttype = application/json
-so.topic=cds_so
-so.username=admin
-so.password=admin
-so.host=10.12.6.236:30226
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/resources/producer.properties b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/resources/producer.properties
deleted file mode 100644
index c3c228b..0000000
--- a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/resources/producer.properties
+++ /dev/null
@@ -1,52 +0,0 @@
-#
-# ============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
diff --git a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/test/kotlin/org/ccsdk/apps/blueprintprocessor/dmaap/TestDmaapEventPublisher.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/test/kotlin/org/ccsdk/apps/blueprintprocessor/dmaap/TestDmaapEventPublisher.kt
index ac88821..9b3aae5 100644
--- a/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/test/kotlin/org/ccsdk/apps/blueprintprocessor/dmaap/TestDmaapEventPublisher.kt
+++ b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/test/kotlin/org/ccsdk/apps/blueprintprocessor/dmaap/TestDmaapEventPublisher.kt
@@ -20,10 +20,14 @@
package org.ccsdk.apps.blueprintprocessor.dmaap
+import com.fasterxml.jackson.databind.ObjectMapper
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.onap.ccsdk.apps.blueprintsprocessor.core.BluePrintProperties
+import org.onap.ccsdk.apps.blueprintsprocessor.core.BlueprintPropertyConfiguration
+import org.onap.ccsdk.apps.blueprintsprocessor.dmaap.BluePrintDmaapLibConfiguration
+import org.onap.ccsdk.apps.blueprintsprocessor.dmaap.BluePrintDmaapLibPropertyService
+import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.autoconfigure.EnableAutoConfiguration
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration
import org.springframework.boot.test.context.SpringBootTest
@@ -45,32 +49,53 @@
@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"])
+@ContextConfiguration(classes = [BluePrintDmaapLibConfiguration::class, TestController::class,
+ BlueprintPropertyConfiguration::class, BluePrintProperties::class])
+@TestPropertySource(properties = ["server.port=9111",
+ "blueprintsprocessor.dmaapclient.aai.topic=cds_aai",
+ "blueprintsprocessor.dmaapclient.aai.type=HTTPNOAUTH",
+ "blueprintsprocessor.dmaapclient.aai.host=127.0.0.1:9111",
+ "blueprintsprocessor.dmaapclient.multi.topic=cds_multi1,cds_multi2",
+ "blueprintsprocessor.dmaapclient.multi.type=HTTPNOAUTH",
+ "blueprintsprocessor.dmaapclient.multi.host=127.0.0.1:9111"])
class TestDmaapEventPublisher {
+ @Autowired
+ lateinit var dmaapService : BluePrintDmaapLibPropertyService
+
/**
* Tests the event properties being set properly and sent as request.
*/
@Test
fun testEventProperties() {
val strList = mutableListOf<String>()
- val pub = DmaapEventPublisher(compName = "aai")
+ val dmaapClient = dmaapService.blueprintDmaapClientService("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")
+ dmaapClient.sendMessage(strList)
+ val msgs = dmaapClient.close(2)
+ assertEquals(msgs!!.size, 1)
+ val topic1 = msgs.get(0)
+ assertEquals(topic1!!.size, 0)
+ }
+
+ /**
+ * Tests the event properties being set properly and sent as request with
+ * single message.
+ */
+ @Test
+ fun testEventPropertiesWithSingleMsg() {
+ val dmaapClient = dmaapService.blueprintDmaapClientService("aai")
+ val str : String = "{\n" +
+ " \"a\" : \"hello\"\n" +
+ "}"
+ dmaapClient.sendMessage(str)
+ val msgs = dmaapClient.close(2)
+ assertEquals(msgs!!.size, 1)
+ val topic1 = msgs.get(0)
+ assertEquals(topic1!!.size, 0)
}
/**
@@ -79,21 +104,93 @@
@Test
fun testMultiTopicProperties() {
val strList = mutableListOf<String>()
- val pub = DmaapEventPublisher(compName = "mul")
+ val dmaapClient = dmaapService.blueprintDmaapClientService("multi")
+
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")
+ dmaapClient.sendMessage(strList)
+ val msgs = dmaapClient.close(2)
+ assertEquals(msgs!!.size, 2)
+ val topic1 = msgs.get(0)
+ assertEquals(topic1!!.size, 0)
+ val topic2 = msgs.get(1)
+ assertEquals(topic2!!.size, 0)
}
+
+
+ /**
+ * Tests the event properties with multiple topics with JSON node as input.
+ */
+ @Test
+ fun testMultiTopicPropertiesWithJsonInput() {
+ val jsonString = "{\n" +
+ " \"topic\" : \"cds_json1,cds_json2\",\n" +
+ " \"type\" : \"HTTPNOAUTH\",\n" +
+ " \"host\" : \"127.0.0.1:9111\"\n" +
+ "}"
+ val mapper = ObjectMapper()
+ val node = mapper.readTree(jsonString)
+ val strList = mutableListOf<String>()
+ val dmaapClient = dmaapService.blueprintDmaapClientService(node)
+
+ strList.add("{\n" +
+ " \"a\" : \"hello\"\n" +
+ "}")
+ dmaapClient.sendMessage(strList)
+ val msgs = dmaapClient.close(2)
+ assertEquals(msgs!!.size, 2)
+ val topic1 = msgs.get(0)
+ assertEquals(topic1!!.size, 0)
+ val topic2 = msgs.get(1)
+ assertEquals(topic2!!.size, 0)
+ }
+
+
+ /**
+ * Tests the event properties with multiple messages.
+ */
+ @Test
+ fun testMultiMsgsProperties() {
+ val strList = mutableListOf<String>()
+ val dmaapClient = dmaapService.blueprintDmaapClientService("aai")
+
+ strList.add("{\n" +
+ " \"a\" : \"hello\"\n" +
+ "}")
+ strList.add("{\n" +
+ " \"a\" : \"second\"\n" +
+ "}")
+ dmaapClient.sendMessage(strList)
+ val msgs = dmaapClient.close(2)
+ assertEquals(msgs!!.size, 1)
+ val topic1 = msgs.get(0)
+ assertEquals(topic1!!.size, 0)
+ }
+
+ /**
+ * Tests the DMAAP client properties generated with the complete prefix.
+ */
+ @Test
+ fun testDmaapClientProperties() {
+ val properties = dmaapService.dmaapClientProperties(
+ "blueprintsprocessor.dmaapclient.aai")
+ assertNotNull(properties, "failed to create property bean")
+ assertNotNull(properties.host, "failed to get url property" +
+ " in property bean")
+ }
+
+ /**
+ * Tests the blueprint DMAAP client service with only selector prefix.
+ */
+ @Test
+ fun testBlueprintDmaapClientService() {
+ val blueprintDmaapClientService =
+ dmaapService.blueprintDmaapClientService("aai")
+ assertNotNull(blueprintDmaapClientService,
+ "failed to create blueprintDmaapClientService")
+ }
+
}
/**
diff --git a/ms/blueprintsprocessor/parent/pom.xml b/ms/blueprintsprocessor/parent/pom.xml
index 46eba5a..f1c601f 100755
--- a/ms/blueprintsprocessor/parent/pom.xml
+++ b/ms/blueprintsprocessor/parent/pom.xml
@@ -50,6 +50,7 @@
<onap.logger.slf4j>1.2.2</onap.logger.slf4j>
<powermock.version>1.7.4</powermock.version>
<mockk.version>1.9</mockk.version>
+ <dmaap.client.version>1.1.5</dmaap.client.version>
</properties>
<dependencyManagement>
<dependencies>