diff options
author | 2019-03-19 19:37:13 +0530 | |
---|---|---|
committer | 2019-03-19 21:05:46 +0530 | |
commit | 11b7e74df31f88aab0c2d1a406fc502de30d4f37 (patch) | |
tree | 92cd1d1d6504f8304d15b040e0e7ea61e563c776 /ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin | |
parent | b9859c7a69aece893b6cacce4dd0d4ce2d0badcb (diff) |
Dmaap Publisher base code
Dmaap publisher code for any component to create a session with the message-router
Issue-ID: CCSDK-693
Change-Id: I4e992f8e0d797a9a5ba169c47b81a54cabc907cb
Signed-off-by: janani b <janani.b@huawei.com>
Diffstat (limited to 'ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin')
7 files changed, 450 insertions, 220 deletions
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 000000000..dd884092d --- /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 000000000..21d7128e4 --- /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/EnvironmentContext.kt b/ms/blueprintsprocessor/modules/commons/dmaap-lib/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/dmaap/BluePrintDmaapLibConfiguration.kt index 1d2a28ce8..ca1a07ba3 100644 --- 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/BluePrintDmaapLibConfiguration.kt @@ -20,37 +20,27 @@ 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 +import org.springframework.boot.context.properties.EnableConfigurationProperties +import org.springframework.context.annotation.ComponentScan +import org.springframework.context.annotation.Configuration /** - * Abstraction of environment context information component. + * Representation of DMAAP lib configuration to load the required property + * files into the application context. */ -@Component -class EnvironmentContext { +@Configuration +@ComponentScan +@EnableConfigurationProperties +open class BluePrintDmaapLibConfiguration - /** - * Environment information. - */ +/** + * Util constants required for DMAAP library to use. + */ +class DmaapLibConstants { 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 + 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 000000000..7abbefbca --- /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 index 7d02e806c..97df2723b 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 { - - /** - * Sends messages through a session on a given partition. - */ - fun sendMessage(partition: String, messages: Collection<String>): Boolean +open class DmaapClientProperties { + lateinit var props: Properties + lateinit var type: String + lateinit var host: String + lateinit var topic: String +} - /** - * Closes the session with the given time. - */ - fun close(timeout: Long) +/** + * Representation of data required for HTTP no auth DMAAP client. + */ +open class HttpNoAuthDmaapClientProperties : DmaapClientProperties() +/** + * 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 7c686f089..000000000 --- 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/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 000000000..ad023b962 --- /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 |