<!-- DMaaP Client -->
<dependency>
- <groupId>com.att.nsa</groupId>
+ <groupId>org.onap.dmaap.messagerouter.dmaapclient</groupId>
<artifactId>dmaapClient</artifactId>
- <version>0.2.12</version>
-<!-- <version>${dmaap.client.version}</version> -->
+ <version>${dmaap.client.version}</version>
</dependency>
<dependency>
* ============LICENSE_START=======================================================
* ONAP : APPC
* ================================================================================
- * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
* ================================================================================
* Copyright (C) 2017 Amdocs
* =============================================================================
* 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=========================================================
*/
public class DmaapConsumerImpl implements Consumer {
- private static final EELFLogger LOG = EELFManager.getInstance().getLogger(DmaapConsumerImpl.class);
- private final Configuration configuration = ConfigurationFactory.getConfiguration();
+ private static final EELFLogger LOG = EELFManager.getInstance().getLogger(DmaapConsumerImpl.class);
+ private final Configuration configuration = ConfigurationFactory.getConfiguration();
// Default values
- private static final int DEFAULT_TIMEOUT_MS = 60000;
- private static final int DEFAULT_LIMIT = 1000;
- private String topic;
- private boolean isMetricEnabled = false;
- private boolean useHttps = false;
- private MetricRegistry metricRegistry;
- private MRConsumer client = null;
- private Properties props = null;
-
+ private static final int DEFAULT_TIMEOUT_MS = 60000;
+ private static final int DEFAULT_LIMIT = 1000;
+ private String topic;
+ private boolean isMetricEnabled = false;
+ private boolean useHttps = false;
+ private MetricRegistry metricRegistry;
+ private MRConsumer client = null;
+ private Properties props = null;
public DmaapConsumerImpl(Collection<String> urls, String topicName, String consumerGroupName, String consumerId,
- String user, String password) {
+ String user, String password) {
- this(urls, topicName, consumerGroupName, consumerId,user, password,null);
+ this(urls, topicName, consumerGroupName, consumerId, user, password, null);
}
public DmaapConsumerImpl(Collection<String> urls, String topicName, String consumerGroupName, String consumerId,
- String user, String password, String filter) {
+ String user, String password, String filter) {
this.topic = topicName;
this.props = new Properties();
props.setProperty("host", urlsStr);
props.setProperty("group", consumerGroupName);
props.setProperty("id", consumerId);
- props.setProperty("username", user);
- props.setProperty("password", password);
+ if (user != null && password != null) {
+ props.setProperty("username", user);
+ props.setProperty("password", password);
+ } else {
+ props.setProperty("TransportType", "HTTPNOAUTH");
+ }
+
if (filter != null) {
props.setProperty("filter", filter);
}
metricRegistry = metricService.createRegistry("APPC");
DmaapRequestCounterMetric dmaapKpiMetric = metricRegistry.metricBuilderFactory()
- .dmaapRequestCounterBuilder()
- .withName("DMAAP_KPI").withType(MetricType.COUNTER)
- .withRecievedMessage(0)
- .withPublishedMessage(0)
- .build();
+ .dmaapRequestCounterBuilder().withName("DMAAP_KPI").withType(MetricType.COUNTER)
+ .withRecievedMessage(0).withPublishedMessage(0).build();
if (metricRegistry.register(dmaapKpiMetric)) {
- Metric[] metrics = new Metric[]{dmaapKpiMetric};
+ Metric[] metrics = new Metric[] { dmaapKpiMetric };
LogPublisher logPublisher = new LogPublisher(metricRegistry, metrics);
LogPublisher[] logPublishers = new LogPublisher[1];
logPublishers[0] = logPublisher;
PublishingPolicy manuallyScheduledPublishingPolicy = metricRegistry.policyBuilderFactory()
- .scheduledPolicyBuilder().withPublishers(logPublishers)
- .withMetrics(metrics)
- .build();
+ .scheduledPolicyBuilder().withPublishers(logPublishers).withMetrics(metrics).build();
LOG.debug("Policy getting initialized");
manuallyScheduledPublishingPolicy.init();
*/
private synchronized MRConsumer getClient(int waitMs, int limit) {
try {
- props.setProperty("timeout",String.valueOf(waitMs));
- props.setProperty("limit",String.valueOf(limit));
- String topicProducerPropFileName = DmaapUtil.createConsumerPropFile(topic,props);
+ props.setProperty("timeout", String.valueOf(waitMs));
+ props.setProperty("limit", String.valueOf(limit));
+ String topicProducerPropFileName = DmaapUtil.createConsumerPropFile(topic, props);
return MRClientFactory.createConsumer(topicProducerPropFileName);
} catch (IOException e1) {
- LOG.error("failed to createConsumer",e1);
+ LOG.error("failed to createConsumer", e1);
return null;
}
}
@Override
public synchronized void updateCredentials(String key, String secret) {
LOG.info(String.format("Setting auth to %s for %s", key, this.toString()));
- props.setProperty("user",String.valueOf(key));
- props.setProperty("password",String.valueOf(secret));
+ props.setProperty("username", String.valueOf(key));
+ props.setProperty("password", String.valueOf(secret));
client = null;
}
}
}
+ public Properties getProperties() {
+ return props;
+ }
+
+ public boolean isHttps() {
+ return useHttps;
+ }
}
public class DmaapProducerImpl implements Producer {
- private static final EELFLogger LOG = EELFManager.getInstance().getLogger(DmaapProducerImpl.class);
- private static final Configuration configuration = ConfigurationFactory.getConfiguration();
+ private static final EELFLogger LOG = EELFManager.getInstance().getLogger(DmaapProducerImpl.class);
+ private static final Configuration configuration = ConfigurationFactory.getConfiguration();
- private Set<String> topics;
+ private Set<String> topics;
- private Properties props = null;
- private MetricRegistry metricRegistry;
- private boolean useHttps = false;
- private boolean isMetricEnabled = false;
-
- private Set<MRBatchingPublisher> clients;
+ private Properties props = null;
+ private MetricRegistry metricRegistry;
+ private boolean useHttps = false;
+ private boolean isMetricEnabled = false;
+
+ private Set<MRBatchingPublisher> clients;
-
public DmaapProducerImpl(Collection<String> urls, String topicName, String user, String password) {
- this(urls, (Set<String>)null, user, password);
+ this(urls, (Set<String>) null, user, password);
this.topics = new HashSet<>();
if (topicName != null) {
Collections.addAll(topics, topicName.split(","));
public DmaapProducerImpl(Collection<String> urls, Set<String> topicNames, String user, String password) {
topics = topicNames;
- if (urls == null || user == null || password == null) {
- throw new IllegalArgumentException("one of these mandaory argument is null: urls, user, password");
+ if (urls == null) {
+ throw new IllegalArgumentException("Mandaory argument is null: urls");
}
this.props = new Properties();
String urlsStr = StringUtils.join(urls, ',');
- props.setProperty("host",urlsStr);
+ props.setProperty("host", urlsStr);
props.setProperty("id", UUID.randomUUID().toString());
- props.setProperty("username",user);
- props.setProperty("password",password);
+ if (user != null && password != null) {
+ props.setProperty("username", user);
+ props.setProperty("password", password);
+ } else {
+ props.setProperty("TransportType", "HTTPNOAUTH");
+ }
}
private void initMetric() {
metricRegistry = metricService.createRegistry("APPC");
DmaapRequestCounterMetric dmaapKpiMetric = metricRegistry.metricBuilderFactory()
- .dmaapRequestCounterBuilder()
- .withName("DMAAP_KPI").withType(MetricType.COUNTER)
- .withRecievedMessage(0)
- .withPublishedMessage(0)
- .build();
+ .dmaapRequestCounterBuilder().withName("DMAAP_KPI").withType(MetricType.COUNTER)
+ .withRecievedMessage(0).withPublishedMessage(0).build();
if (metricRegistry.register(dmaapKpiMetric)) {
- Metric[] metrics = new Metric[]{dmaapKpiMetric};
+ Metric[] metrics = new Metric[] { dmaapKpiMetric };
LogPublisher logPublisher = new LogPublisher(metricRegistry, metrics);
LogPublisher[] logPublishers = new LogPublisher[1];
logPublishers[0] = logPublisher;
PublishingPolicy manuallyScheduledPublishingPolicy = metricRegistry.policyBuilderFactory()
- .scheduledPolicyBuilder()
- .withPublishers(logPublishers)
- .withMetrics(metrics)
- .build();
+ .scheduledPolicyBuilder().withPublishers(logPublishers).withMetrics(metrics).build();
LOG.debug("Policy getting initialized");
manuallyScheduledPublishingPolicy.init();
Set<MRBatchingPublisher> out = new HashSet<>();
for (String topic : topics) {
try {
- String topicProducerPropFileName = DmaapUtil.createProducerPropFile(topic,props);
+ String topicProducerPropFileName = DmaapUtil.createProducerPropFile(topic, props);
final MRBatchingPublisher client = MRClientFactory.createBatchingPublisher(topicProducerPropFileName);
out.add(client);
} catch (Exception e) {
@Override
public synchronized void updateCredentials(String key, String secret) {
LOG.info(String.format("Setting auth to %s for %s", key, this.toString()));
- props.setProperty("user", String.valueOf(key));
+ props.setProperty("username", String.valueOf(key));
props.setProperty("password", String.valueOf(secret));
clients = null;
}
public boolean post(String partition, String data) {
boolean success = true;
Properties properties = configuration.getProperties();
- if (properties != null && properties.getProperty("metric.enabled") != null ) {
+ if (properties != null && properties.getProperty("metric.enabled") != null) {
isMetricEnabled = Boolean.valueOf(properties.getProperty("metric.enabled"));
}
if (isMetricEnabled) {
initMetric();
}
-
- // Create clients once and reuse them on subsequent posts. This is
+
+ // Create clients once and reuse them on subsequent posts. This is
// to support failover to other servers in the Dmaap cluster.
if ((clients == null) || (clients.isEmpty())) {
LOG.info("Getting CambriaBatchingPublisher Clients ...");
clients = getClients();
}
-
+
for (MRBatchingPublisher client : clients) {
try {
LOG.debug(String.format("Posting %s to %s", data, client));
for (MRBatchingPublisher client : clients) {
try {
client.close(1, TimeUnit.SECONDS);
- } catch (IOException | InterruptedException e) {
+ } catch (IOException | InterruptedException e) {
LOG.warn(String.format("Failed to cleanly close Dmaap connection for [%s]", client), e);
}
}
}
}
+ public Properties getProperties() {
+ return props;
+ }
+
+ public boolean isHttps() {
+ return useHttps;
+ }
}
* ============LICENSE_START=======================================================
* ONAP : APPC
* ================================================================================
- * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
* ================================================================================
* Copyright (C) 2017 Amdocs
* =============================================================================
* 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=========================================================
*/
package org.onap.appc.adapter.messaging.dmaap.impl;
import java.io.File;
+import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
+import org.onap.appc.configuration.Configuration;
+import org.onap.appc.configuration.ConfigurationFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
public class DmaapUtil {
- private static final char DELIMITER = '_';
+ private static final char DELIMITER = '_';
+
+ static final String DMAAP_PROPERTIES_PATH = "org.onap.appc.dmaap.profile.path";
+
+ private static final Logger log = LoggerFactory.getLogger(DmaapUtil.class);
private DmaapUtil() {
}
private static String createPreferredRouteFileIfNotExist(String topic) throws IOException {
String topicPreferredRouteFileName;
- topicPreferredRouteFileName = topic+"preferredRoute.properties";
- File fo= new File(topicPreferredRouteFileName);
- if(!fo.exists()) {
+ topicPreferredRouteFileName = topic + "preferredRoute.properties";
+ File fo = new File(topicPreferredRouteFileName);
+ if (!fo.exists()) {
ClassLoader classLoader = DmaapUtil.class.getClassLoader();
InputStream inputStream = classLoader.getResourceAsStream("preferredRoute.txt");
Properties props = new Properties();
props.load(inputStream);
- String fileName = topic != null ? topic+ DELIMITER +"MR1" : DELIMITER +"MR1";
+ String fileName = topic != null ? topic + DELIMITER + "MR1" : DELIMITER + "MR1";
props.setProperty("preferredRouteKey", fileName);
topicPreferredRouteFileName = topic + "preferredRoute.properties";
- props.store(new FileOutputStream(topicPreferredRouteFileName), "preferredRoute.properties file created on the fly for topic:" + topic + " on:" + System.currentTimeMillis());
+ props.store(new FileOutputStream(topicPreferredRouteFileName),
+ "preferredRoute.properties file created on the fly for topic:" + topic + " on:"
+ + System.currentTimeMillis());
}
return topicPreferredRouteFileName;
}
- public static String createConsumerPropFile(String topic, Properties props)throws IOException {
+ public static String createConsumerPropFile(String topic, Properties props) throws IOException {
String defaultProfFileName = "consumer.properties";
- return createConsumerProducerPropFile(topic, defaultProfFileName,props);
+
+ log.debug("Creating DMaaP Consumer Property File for topic " + topic);
+ return createConsumerProducerPropFile(topic, defaultProfFileName, props);
}
- public static String createProducerPropFile(String topic, Properties props)throws IOException {
+ public static String createProducerPropFile(String topic, Properties props) throws IOException {
String defaultProfFileName = "producer.properties";
- return createConsumerProducerPropFile(topic, defaultProfFileName,props);
+
+ log.debug("Creating DMaaP Producer Property File for topic " + topic);
+ return createConsumerProducerPropFile(topic, defaultProfFileName, props);
}
- private static String createConsumerProducerPropFile(String topic, String defaultProfFileName, Properties props) throws IOException {
- ClassLoader classLoader = DmaapUtil.class.getClassLoader();
- InputStream inputStream = classLoader.getResourceAsStream(defaultProfFileName);
- Properties defaultProps = new Properties();
- defaultProps.load(inputStream);
- defaultProps.setProperty("topic",topic);
+ private static String createConsumerProducerPropFile(String topic, String defaultProfFileName, Properties props)
+ throws IOException {
+ Properties defaultProps = getDefaultProperties(defaultProfFileName);
+
+ defaultProps.setProperty("topic", topic);
String preferredRouteFileName = DmaapUtil.createPreferredRouteFileIfNotExist(topic);
- if(props != null && !props.isEmpty()){
+ if (props != null && !props.isEmpty()) {
defaultProps.putAll(props);
}
- defaultProps.setProperty("topic",topic);
- defaultProps.setProperty("DME2preferredRouterFilePath",preferredRouteFileName);
+ defaultProps.setProperty("topic", topic);
+ defaultProps.setProperty("DME2preferredRouterFilePath", preferredRouteFileName);
String id = defaultProps.getProperty("id");
String topicConsumerPropFileName = defaultProfFileName;
- topicConsumerPropFileName = id != null ? id+ DELIMITER +topicConsumerPropFileName : DELIMITER +topicConsumerPropFileName;
- topicConsumerPropFileName = topic != null ? topic+ DELIMITER +topicConsumerPropFileName : DELIMITER +topicConsumerPropFileName;
+ topicConsumerPropFileName = id != null ? id + DELIMITER + topicConsumerPropFileName
+ : DELIMITER + topicConsumerPropFileName;
+ topicConsumerPropFileName = topic != null ? topic + DELIMITER + topicConsumerPropFileName
+ : DELIMITER + topicConsumerPropFileName;
- defaultProps.store(new FileOutputStream(topicConsumerPropFileName), defaultProfFileName+" file created on the fly for topic:"+topic+" on:"+System.currentTimeMillis());
+ defaultProps.store(new FileOutputStream(topicConsumerPropFileName), defaultProfFileName
+ + " file created on the fly for topic:" + topic + " on:" + System.currentTimeMillis());
return topicConsumerPropFileName;
}
+ private static Properties getDefaultProperties(String profileName) {
+ Properties props = new Properties();
+
+ // use appc configuration to get all properties which includes
+ // appc.properties and system properties
+ // allowing variable to be set in any location
+ Configuration config = ConfigurationFactory.getConfiguration();
+ String dmaapPropPath = config.getProperty(DMAAP_PROPERTIES_PATH);
+
+ if (dmaapPropPath != null) {
+ // load from file system
+
+ File profileFile = new File(dmaapPropPath, profileName);
+ FileInputStream inputStream = null;
+
+ log.info("Loading DMaaP Profile from " + profileFile.getAbsolutePath());
+
+ if (profileFile.exists()) {
+ try {
+ inputStream = new FileInputStream(profileFile);
+ props.load(inputStream);
+ } catch (IOException e) {
+ log.error("Exception loading DMaaP Profile from " + profileFile.getAbsolutePath(), e);
+ } finally {
+ try {
+ if (inputStream != null) {
+ inputStream.close();
+ }
+ } catch (IOException ex) {
+ log.warn("Exception closing DMaaP Profile file " + profileFile.getAbsolutePath(), ex);
+ }
+ }
+ }
+ }
+ if (props.isEmpty()) {
+ // load default Profile from class
+ log.info("Loading Default DMaaP Profile");
+
+ ClassLoader classLoader = DmaapUtil.class.getClassLoader();
+ InputStream inputStream = classLoader.getResourceAsStream(profileName);
+ try {
+ props.load(inputStream);
+ } catch (IOException e) {
+ log.error("Exception loading Default DMaaP Profile", e);
+ }
+ }
+
+ return props;
+ }
}
# ============LICENSE_START=======================================================
# ONAP : APPC
# ================================================================================
-# Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+# Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
# ================================================================================
# Copyright (C) 2017 Amdocs
# =============================================================================
# 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=========================================================
###
#TransportType-Specify which way user want to use. I.e. <HTTPAAF,DME2,HTTPAUTH >
-TransportType=HTTPAAF
+TransportType=HTTPNOAUTH
Latitude =50.000000
Longitude =-100.000000
Version =1.0
AFT_DME2_EP_READ_TIMEOUT_MS=50000
sessionstickinessrequired=NO
DME2preferredRouterFilePath=preferredRoute.txt
-
-
-
# ============LICENSE_START=======================================================
# ONAP : APPC
# ================================================================================
-# Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+# Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
# ================================================================================
# Copyright (C) 2017 Amdocs
# =============================================================================
# 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=========================================================
###
#TransportType-Specify which way user want to use. I.e. <HTTPAAF,DME2,HTTPAUTH >
-TransportType=HTTPAAF
+TransportType=HTTPNOAUTH
Latitude =50.000000
Longitude =-100.000000
Version =1.0
--- /dev/null
+package org.onap.appc.adapter.messaging.dmaap.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Properties;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+public class TestDmaapConsumerImpl {
+ String[] hostList = { "192.168.1.1" };
+ Collection<String> hosts = new HashSet<String>(Arrays.asList(hostList));
+
+ String topic = "JunitTopicOne";
+ String group = "junit-client";
+ String id = "junit-consumer-one";
+ String key = "key";
+ String secret = "secret";
+ String filter = null;
+
+ @Test
+ public void testDmaapConsumerImplNoFilter() {
+
+ DmaapConsumerImpl consumer = new DmaapConsumerImpl(hosts, topic, group, id, key, secret);
+
+ assertNotNull(consumer);
+
+ Properties props = consumer.getProperties();
+
+ assertEquals("192.168.1.1", props.getProperty("host"));
+ assertEquals("key", props.getProperty("username"));
+ assertEquals("secret", props.getProperty("password"));
+ }
+
+ @Test
+ public void testDmaapConsumerImplwithFilter() {
+
+ DmaapConsumerImpl consumer = new DmaapConsumerImpl(hosts, topic, group, id, key, secret, filter);
+
+ assertNotNull(consumer);
+
+ }
+
+ @Test
+ public void testDmaapConsumerImplNoUserPassword() {
+
+ DmaapConsumerImpl consumer = new DmaapConsumerImpl(hosts, topic, group, id, null, null);
+
+ assertNotNull(consumer);
+
+ Properties props = consumer.getProperties();
+
+ assertEquals("192.168.1.1", props.getProperty("host"));
+ assertNull(props.getProperty("username"));
+ assertNull(props.getProperty("password"));
+ assertEquals("HTTPNOAUTH", props.getProperty("TransportType"));
+ }
+
+ @Test
+ public void testUpdateCredentials() {
+ DmaapConsumerImpl consumer = new DmaapConsumerImpl(hosts, topic, group, id, null, null);
+
+ assertNotNull(consumer);
+
+ Properties props = consumer.getProperties();
+
+ assertEquals("192.168.1.1", props.getProperty("host"));
+ assertNull(props.getProperty("username"));
+ assertNull(props.getProperty("password"));
+
+ consumer.updateCredentials(key, secret);
+
+ props = consumer.getProperties();
+ assertEquals("192.168.1.1", props.getProperty("host"));
+ assertEquals("key", props.getProperty("username"));
+ assertEquals("secret", props.getProperty("password"));
+ }
+
+ @Ignore
+ @Test
+ public void testFetch() {
+ fail("Not yet implemented");
+ }
+
+ @Ignore
+ @Test
+ public void testFetchIntInt() {
+ fail("Not yet implemented");
+ }
+
+ @Test
+ public void testCloseNoClient() {
+ DmaapConsumerImpl consumer = new DmaapConsumerImpl(hosts, topic, group, id, key, secret);
+
+ assertNotNull(consumer);
+
+ consumer.close();
+ }
+
+ @Ignore
+ @Test
+ public void testCloseWithClient() {
+ fail("Not yet implemented");
+ }
+
+ @Test
+ public void testToString() {
+ DmaapConsumerImpl consumer = new DmaapConsumerImpl(hosts, topic, group, id, null, null);
+
+ assertNotNull(consumer);
+
+ assertEquals("Consumer junit-client/junit-consumer-one listening to JunitTopicOne on [192.168.1.1]",
+ consumer.toString());
+ }
+
+ @Test
+ public void testUseHttps() {
+ DmaapConsumerImpl consumer = new DmaapConsumerImpl(hosts, topic, group, id, key, secret);
+
+ assertNotNull(consumer);
+
+ assertEquals(false, consumer.isHttps());
+
+ consumer.useHttps(true);
+
+ assertEquals(true, consumer.isHttps());
+
+ }
+
+}
--- /dev/null
+package org.onap.appc.adapter.messaging.dmaap.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+public class TestDmaapProducerImpl {
+ String[] hostList = { "192.168.1.1" };
+ Collection<String> hosts = new HashSet<String>(Arrays.asList(hostList));
+
+ String topic = "JunitTopicOne";
+ String group = "junit-client";
+ String id = "junit-consumer-one";
+ String key = "key";
+ String secret = "secret";
+ String filter = null;
+
+ @Test
+ public void testDmaapProducerImplSingleTopic() {
+ DmaapProducerImpl producer = new DmaapProducerImpl(hosts, topic, key, secret);
+
+ assertNotNull(producer);
+
+ Properties props = producer.getProperties();
+
+ assertNotNull(props);
+
+ assertEquals("key", props.getProperty("username"));
+ assertEquals("secret", props.getProperty("password"));
+ }
+
+ @Test
+ public void testDmaapProducerImplMultipleTopic() {
+ String[] topicList = { "topic1", "topic2" };
+ Set<String> topicNames = new HashSet<String>(Arrays.asList(topicList));
+
+ DmaapProducerImpl producer = new DmaapProducerImpl(hosts, topicNames, key, secret);
+
+ assertNotNull(producer);
+
+ Properties props = producer.getProperties();
+
+ assertNotNull(props);
+
+ assertEquals("key", props.getProperty("username"));
+ assertEquals("secret", props.getProperty("password"));
+
+ }
+
+ @Test
+ public void testDmaapProducerImplNoUserPass() {
+ DmaapProducerImpl producer = new DmaapProducerImpl(hosts, topic, null, null);
+
+ assertNotNull(producer);
+
+ Properties props = producer.getProperties();
+
+ assertNotNull(props);
+
+ assertNull(props.getProperty("username"));
+ assertNull(props.getProperty("password"));
+ }
+
+ @Test
+ public void testUpdateCredentials() {
+ DmaapProducerImpl producer = new DmaapProducerImpl(hosts, topic, null, null);
+
+ assertNotNull(producer);
+
+ Properties props = producer.getProperties();
+
+ assertNotNull(props);
+
+ assertNull(props.getProperty("username"));
+ assertNull(props.getProperty("password"));
+
+ producer.updateCredentials(key, secret);
+
+ props = producer.getProperties();
+
+ assertNotNull(props);
+
+ assertEquals("key", props.getProperty("username"));
+ assertEquals("secret", props.getProperty("password"));
+
+ }
+
+ @Ignore
+ @Test
+ public void testPost() {
+ fail("Not yet implemented");
+ }
+
+ @Test
+ public void testCloseNoClient() {
+ DmaapProducerImpl producer = new DmaapProducerImpl(hosts, topic, key, secret);
+
+ assertNotNull(producer);
+
+ producer.close();
+ }
+
+ @Ignore
+ @Test
+ public void testCloseWithClient() {
+ fail("Not yet implemented");
+ }
+
+ @Test
+ public void testUseHttps() {
+ DmaapProducerImpl producer = new DmaapProducerImpl(hosts, topic, key, secret);
+
+ assertNotNull(producer);
+
+ assertEquals(false, producer.isHttps());
+
+ producer.useHttps(true);
+
+ assertEquals(true, producer.isHttps());
+
+ }
+
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP : APPC
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Copyright (C) 2017 Amdocs
+ * =============================================================================
+ * 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.appc.adapter.messaging.dmaap.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Field;
+import java.util.Properties;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.onap.appc.configuration.ConfigurationFactory;
+
+public class TestDmaapUtil {
+ private static Class<?> configurationFactoryClass;
+ private static Field configField;
+
+ @Test
+ public void testCreateConsumerPropFile() {
+ String topic = "JunitTopicOne";
+ Properties junitProps = new Properties();
+ junitProps.put("host", "192.168.10.10");
+ junitProps.put("group", "junit-client");
+ junitProps.put("id", "junit-consumer-one");
+ junitProps.put("filter", "none");
+
+ String junitFile = null;
+
+ // ensure file path property is not set
+ if (System.getProperty(DmaapUtil.DMAAP_PROPERTIES_PATH) != null) {
+ System.clearProperty(DmaapUtil.DMAAP_PROPERTIES_PATH);
+
+ // set configuration to null to force reloading of properties
+ try {
+ configField.set(null, null);
+ } catch (IllegalArgumentException | IllegalAccessException e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ }
+ }
+ try {
+ junitFile = DmaapUtil.createConsumerPropFile(topic, junitProps);
+ } catch (IOException e) {
+ e.printStackTrace();
+ fail("Exception creating consumer property file");
+ }
+
+ assertNotNull(junitFile);
+
+ // open file and verify properties
+ File testFile = new File(junitFile);
+ assertTrue(testFile.exists());
+
+ InputStream is = null;
+ Properties testProps = new Properties();
+ try {
+ is = new FileInputStream(testFile);
+ testProps.load(is);
+ } catch (FileNotFoundException e) {
+ e.printStackTrace();
+ fail("Exception opening consumer property file");
+ } catch (IOException e) {
+ e.printStackTrace();
+ fail("Exception opening consumer property file");
+ } finally {
+ try {
+ if (is != null) {
+ is.close();
+ }
+ } catch (IOException ex) {
+ ex.printStackTrace();
+ fail("Exception closing consumer property file");
+ }
+ }
+
+ assertFalse(testProps.isEmpty());
+
+ assertEquals(testProps.get("host"), "192.168.10.10");
+ assertEquals(testProps.get("group"), "junit-client");
+ assertEquals(testProps.get("id"), "junit-consumer-one");
+ assertEquals(testProps.get("filter"), "none");
+ assertEquals(testProps.get("TransportType"), "HTTPNOAUTH");
+ }
+
+ @Test
+ public void testCreateConsumerPropFileWithCustomProfile() {
+ String topic = "JunitTopicOne";
+ Properties junitProps = new Properties();
+ junitProps.put("host", "192.168.10.10");
+ junitProps.put("group", "junit-client");
+ junitProps.put("id", "junit-consumer-two");
+ junitProps.put("filter", "none");
+
+ String junitFile = null;
+
+ // set property for DMaaP profile
+ System.setProperty(DmaapUtil.DMAAP_PROPERTIES_PATH, "src/test/resources/org/onap/appc");
+
+ // set configuration to null to force reloading of properties
+ try {
+ configField.set(null, null);
+ } catch (IllegalArgumentException | IllegalAccessException e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ }
+
+ try {
+ junitFile = DmaapUtil.createConsumerPropFile(topic, junitProps);
+ } catch (IOException e) {
+ e.printStackTrace();
+ fail("Exception creating consumer property file");
+ }
+
+ assertNotNull(junitFile);
+
+ // open file and verify properties
+ File testFile = new File(junitFile);
+ assertTrue(testFile.exists());
+
+ InputStream is = null;
+ Properties testProps = new Properties();
+ try {
+ is = new FileInputStream(testFile);
+ testProps.load(is);
+ } catch (FileNotFoundException e) {
+ e.printStackTrace();
+ fail("Exception opening consumer property file");
+ } catch (IOException e) {
+ e.printStackTrace();
+ fail("Exception opening consumer property file");
+ } finally {
+ try {
+ if (is != null) {
+ is.close();
+ }
+ } catch (IOException ex) {
+ ex.printStackTrace();
+ fail("Exception closing consumer property file");
+ }
+ }
+
+ assertFalse(testProps.isEmpty());
+
+ assertEquals(testProps.get("host"), "192.168.10.10");
+ assertEquals(testProps.get("group"), "junit-client");
+ assertEquals(testProps.get("id"), "junit-consumer-two");
+ assertEquals(testProps.get("filter"), "none");
+ assertEquals(testProps.get("TransportType"), "HTTPAAF");
+ }
+
+ @Test
+ public void testCreateProducerPropFile() {
+ String topic = "JunitTopicOne";
+ Properties junitProps = new Properties();
+ junitProps.put("host", "192.168.10.10");
+ junitProps.put("group", "junit-client");
+ junitProps.put("id", "junit-producer-one");
+ junitProps.put("filter", "none");
+
+ String junitFile = null;
+
+ // ensure file path property is not set
+ if (System.getProperty(DmaapUtil.DMAAP_PROPERTIES_PATH) != null) {
+ System.clearProperty(DmaapUtil.DMAAP_PROPERTIES_PATH);
+
+ // set configuration to null to force reloading of properties
+ try {
+ configField.set(null, null);
+ } catch (IllegalArgumentException | IllegalAccessException e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ }
+ }
+
+ try {
+ junitFile = DmaapUtil.createProducerPropFile(topic, junitProps);
+ } catch (IOException e) {
+ e.printStackTrace();
+ fail("Exception creating consumer property file");
+ }
+
+ assertNotNull(junitFile);
+
+ // open file and verify properties
+ File testFile = new File(junitFile);
+ assertTrue(testFile.exists());
+
+ InputStream is = null;
+ Properties testProps = new Properties();
+ try {
+ is = new FileInputStream(testFile);
+ testProps.load(is);
+ } catch (FileNotFoundException e) {
+ e.printStackTrace();
+ fail("Exception opening consumer property file");
+ } catch (IOException e) {
+ e.printStackTrace();
+ fail("Exception opening consumer property file");
+ } finally {
+ try {
+ if (is != null) {
+ is.close();
+ }
+ } catch (IOException ex) {
+ ex.printStackTrace();
+ fail("Exception closing consumer property file");
+ }
+ }
+
+ assertFalse(testProps.isEmpty());
+
+ assertEquals(testProps.get("host"), "192.168.10.10");
+ assertEquals(testProps.get("group"), "junit-client");
+ assertEquals(testProps.get("id"), "junit-producer-one");
+ assertEquals(testProps.get("filter"), "none");
+ assertEquals("HTTPNOAUTH", testProps.get("TransportType"));
+ }
+
+ /**
+ * Use reflection to locate fields and methods so that they can be
+ * manipulated during the test to change the internal state accordingly.
+ *
+ * @throws NoSuchFieldException
+ * if the field(s) dont exist
+ * @throws SecurityException
+ * if reflective access is not allowed
+ * @throws NoSuchMethodException
+ * If the method(s) dont exist
+ */
+ @SuppressWarnings("nls")
+ @BeforeClass
+ public static void once() throws NoSuchFieldException, SecurityException, NoSuchMethodException {
+ configurationFactoryClass = ConfigurationFactory.class;
+
+ configField = configurationFactoryClass.getDeclaredField("config");
+ configField.setAccessible(true);
+ }
+}
--- /dev/null
+###
+# ============LICENSE_START=======================================================
+# ONAP : APPC
+# ================================================================================
+# Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+# ================================================================================
+# Copyright (C) 2017 Amdocs
+# =============================================================================
+# 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=HTTPAAF
+Latitude =50.000000
+Longitude =-100.000000
+Version =1.0
+ServiceName =dmaap-v1.dev.dmaap.dt.saat.acsi.openecomp.org/events
+Environment =TEST
+Partner=BOT_R
+routeOffer=MR1
+SubContextPath =/
+Protocol =http
+MethodType =GET
+contenttype =application/json
+#authKey=01234567890abcde:01234567890abcdefghijklmn
+#authDate=2016-02-18T13:57:37-0800
+host=127.0.0.1
+topic=org.onap.appc.UNIT-TEST
+group=jmsgrp
+id=2
+timeout=15000
+limit=1000
+filter=
+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=preferredRoute.txt
--- /dev/null
+###
+# ============LICENSE_START=======================================================
+# ONAP : APPC
+# ================================================================================
+# Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+# ================================================================================
+# Copyright (C) 2017 Amdocs
+# =============================================================================
+# 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=HTTPAAF
+Latitude =50.000000
+Longitude =-100.000000
+Version =1.0
+ServiceName =dmaap-v1.dev.dmaap.dt.saat.acsi.openecomp.org/events
+Environment =TEST
+Partner=BOT_R
+SubContextPath =/
+Protocol =http
+MethodType =POST
+contenttype = application/json
+authKey=01234567890abcde:01234567890abcdefghijklmn
+authDate=2016-07-20T11:30:56-0700
+host=127.0.0.1
+topic=org.onap.appc.UNIT-TEST
+partition=2
+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=preferredRoute.txt
+MessageSentThreadOccurance=50
s = s[0].upper() + s[1:]
return re.sub(r'(?!^)-([a-zA-Z])', lambda m: m.group(1).upper(), s)
+
leaf = ""
val = ""
li = []
if len(sys.argv) < 3:
- print 'yang2props.py <input yang> <output properties>'
+ print('yang2props.py <input yang> <output properties>')
sys.exit(2)
with open(sys.argv[1], "r") as ins:
for line in ins:
# if we see a leaf save the name for later
if "leaf " in line:
- match = re.search(r'leaf (\S+)', line)
- if match:
- leaf = match.group(1)
+ match = re.search(r'leaf (\S+)', line)
+ if match:
+ leaf = match.group(1)
# if we see enum convert the value to enum format and see if it changed
# if the value is different write a property entry
if "enum " in line:
- match = re.search(r'enum "(\S+)";', line)
- if match:
+ match = re.search(r'enum "(\S+)";', line)
+ if match:
val = match.group(1)
- enum = to_enum(val)
-
+ enum = to_enum(val)
# see if converting to enum changed the string
- if val != enum:
- property = "yang."+leaf+"."+enum+"="+val
- if property not in li:
- li.append( property)
-
-
+ if val != enum:
+ property = "yang." + leaf + "." + enum + "=" + val
+ if property not in li:
+ li.append(property)
# Open output file
-fo = open(sys.argv[2], "wb")
+fo = open(sys.argv[2], "w")
fo.write("# yang conversion properties \n")
fo.write("# used to convert Enum back to the original yang value \n")
fo.write("\n".join(li))
# Close opend file
fo.close()
-
s = s[0].upper() + s[1:]
return re.sub(r'(?!^)-([a-zA-Z])', lambda m: m.group(1).upper(), s)
+
leaf = ""
val = ""
li = []
if len(sys.argv) < 3:
- print 'yang2props.py <input yang> <output properties>'
+ print('yang2props.py <input yang> <output properties>')
sys.exit(2)
with open(sys.argv[1], "r") as ins:
for line in ins:
# if we see a leaf save the name for later
if "leaf " in line:
- match = re.search(r'leaf (\S+)', line)
- if match:
- leaf = match.group(1)
-
+ match = re.search(r'leaf (\S+)', line)
+ if match:
+ leaf = match.group(1)
+
# if we see enum convert the value to enum format and see if it changed
# if the value is different write a property entry
if "enum " in line:
- match = re.search(r'enum "(\S+)";', line)
- if match:
+ match = re.search(r'enum "(\S+)";', line)
+ if match:
val = match.group(1)
- enum = to_enum(val)
-
+ enum = to_enum(val)
# see if converting to enum changed the string
- if val != enum:
- property = "yang."+leaf+"."+enum+"="+val
- if property not in li:
- li.append( property)
-
-
+ if val != enum:
+ property = "yang." + leaf + "." + enum + "=" + val
+ if property not in li:
+ li.append(property)
# Open output file
-fo = open(sys.argv[2], "wb")
+fo = open(sys.argv[2], "w")
fo.write("# yang conversion properties \n")
fo.write("# used to convert Enum back to the original yang value \n")
fo.write("\n".join(li))
# Close opend file
fo.close()
-
s = s[0].upper() + s[1:]
return re.sub(r'(?!^)-([a-zA-Z])', lambda m: m.group(1).upper(), s)
+
leaf = ""
val = ""
li = []
if len(sys.argv) < 3:
- print 'yang2props.py <input yang> <output properties>'
+ print('yang2props.py <input yang> <output properties>')
sys.exit(2)
with open(sys.argv[1], "r") as ins:
for line in ins:
# if we see a leaf save the name for later
if "leaf " in line:
- match = re.search(r'leaf (\S+)', line)
- if match:
- leaf = match.group(1)
+ match = re.search(r'leaf (\S+)', line)
+ if match:
+ leaf = match.group(1)
# if we see enum convert the value to enum format and see if it changed
# if the value is different write a property entry
if "enum " in line:
- match = re.search(r'enum "(\S+)";', line)
- if match:
+ match = re.search(r'enum "(\S+)";', line)
+ if match:
val = match.group(1)
- enum = to_enum(val)
-
+ enum = to_enum(val)
# see if converting to enum changed the string
- if val != enum:
- property = "yang."+leaf+"."+enum+"="+val
- if property not in li:
- li.append( property)
-
-
+ if val != enum:
+ property = "yang." + leaf + "." + enum + "=" + val
+ if property not in li:
+ li.append(property)
# Open output file
-fo = open(sys.argv[2], "wb")
+fo = open(sys.argv[2], "w")
fo.write("# yang conversion properties \n")
fo.write("# used to convert Enum back to the original yang value \n")
fo.write("\n".join(li))
# Close opend file
fo.close()
-
-
+
<cdp.pal.version>1.1.25.6-oss</cdp.pal.version>
- <dmaap.client.version>0.2.12</dmaap.client.version>
+ <dmaap.client.version>1.1.3</dmaap.client.version>
<eelf.version>1.0.0</eelf.version>
<eelf.maven.plugin.version>1.0.0</eelf.maven.plugin.version>
<cadi-version>1.3.0</cadi-version>
Code Review / appc.git / commitdiff