aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpoint.java34
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxy.java84
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactory.java11
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseFactory.java9
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseHashedFactory.java9
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/parameters/TopicParameterGroup.java6
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.java139
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicPropertyBuilder.java29
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicFactoryTest.java26
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicPropertyBuilder.java15
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java9
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactoryTest.java9
-rw-r--r--policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java25
-rw-r--r--policy-endpoints/src/test/resources/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.json10
14 files changed, 370 insertions, 45 deletions
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpoint.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpoint.java
index 7bc7abab..bb707523 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpoint.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpoint.java
@@ -30,6 +30,8 @@ import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicSink;
import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicSource;
import org.onap.policy.common.endpoints.event.comm.bus.UebTopicSink;
import org.onap.policy.common.endpoints.event.comm.bus.UebTopicSource;
+import org.onap.policy.common.endpoints.parameters.TopicParameterGroup;
+import org.onap.policy.common.endpoints.parameters.TopicParameters;
/**
* Abstraction to managed the system's Networked Topic Endpoints, sources of all events input into
@@ -47,24 +49,52 @@ public interface TopicEndpoint extends Startable, Lockable {
List<Topic> addTopics(Properties properties);
/**
+ * Add topics configuration (sources and sinks) into a single list.
+ *
+ * @param params parameters to configure topic
+ * @return topic list
+ * @throws IllegalArgumentException when invalid arguments are provided
+ */
+ List<Topic> addTopics(TopicParameterGroup params);
+
+ /**
* Add Topic Sources to the communication infrastructure initialized per properties.
*
* @param properties properties for Topic Source construction
- * @return a generic Topic Source
+ * @return a list of generic Topic Sources
* @throws IllegalArgumentException when invalid arguments are provided
*/
List<TopicSource> addTopicSources(Properties properties);
+
+ /**
+ * Add Topic Sources to the communication infrastructure initialized per properties.
+ *
+ * @param paramList parameters for Topic Source construction
+ * @return a list of generic Topic Sources
+ * @throws IllegalArgumentException when invalid arguments are provided
+ */
+ List<TopicSource> addTopicSources(List<TopicParameters> paramList);
+
/**
* Add Topic Sinks to the communication infrastructure initialized per properties.
*
* @param properties properties for Topic Sink construction
- * @return a generic Topic Sink
+ * @return a list of generic Topic Sinks
* @throws IllegalArgumentException when invalid arguments are provided
*/
List<TopicSink> addTopicSinks(Properties properties);
/**
+ * Add Topic Sinks to the communication infrastructure initialized per properties.
+ *
+ * @param paramList parameters for Topic Sink construction
+ * @return a list of generic Topic Sinks
+ * @throws IllegalArgumentException when invalid arguments are provided
+ */
+ List<TopicSink> addTopicSinks(List<TopicParameters> paramList);
+
+ /**
* Gets all Topic Sources.
*
* @return the Topic Source List
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxy.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxy.java
index 00980fc4..c9e73152 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxy.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxy.java
@@ -34,6 +34,8 @@ import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicSource;
import org.onap.policy.common.endpoints.event.comm.bus.UebTopicFactories;
import org.onap.policy.common.endpoints.event.comm.bus.UebTopicSink;
import org.onap.policy.common.endpoints.event.comm.bus.UebTopicSource;
+import org.onap.policy.common.endpoints.parameters.TopicParameterGroup;
+import org.onap.policy.common.endpoints.parameters.TopicParameters;
import org.onap.policy.common.gson.annotation.GsonJsonIgnore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,6 +68,41 @@ class TopicEndpointProxy implements TopicEndpoint {
}
@Override
+ public List<Topic> addTopics(TopicParameterGroup params) {
+ List<Topic> topics = new ArrayList<>(params.getTopicSinks().size() + params.getTopicSources().size());
+ topics.addAll(addTopicSources(params.getTopicSources()));
+ topics.addAll(addTopicSinks(params.getTopicSinks()));
+ return topics;
+ }
+
+ @Override
+ public List<TopicSource> addTopicSources(List<TopicParameters> paramList) {
+ List<TopicSource> sources = new ArrayList<>(paramList.size());
+
+ for (TopicParameters param : paramList) {
+ switch (Topic.CommInfrastructure.valueOf(param.getTopicCommInfrastructure().toUpperCase())) {
+ case UEB:
+ sources.add(UebTopicFactories.getSourceFactory().build(param));
+ break;
+ case DMAAP:
+ sources.add(DmaapTopicFactories.getSourceFactory().build(param));
+ break;
+ case NOOP:
+ sources.add(NoopTopicFactories.getSourceFactory().build(param));
+ break;
+ default:
+ logger.debug("Unknown source type {} for topic: {}", param.getTopicCommInfrastructure(),
+ param.getTopic());
+ break;
+ }
+ }
+
+ lockSources(sources);
+
+ return sources;
+ }
+
+ @Override
public List<TopicSource> addTopicSources(Properties properties) {
// 1. Create UEB Sources
@@ -78,13 +115,42 @@ class TopicEndpointProxy implements TopicEndpoint {
sources.addAll(DmaapTopicFactories.getSourceFactory().build(properties));
sources.addAll(NoopTopicFactories.getSourceFactory().build(properties));
+ lockSources(sources);
+
+ return sources;
+ }
+
+ private void lockSources(List<TopicSource> sources) {
if (this.isLocked()) {
- for (final TopicSource source : sources) {
- source.lock();
+ sources.forEach(TopicSource::lock);
+ }
+ }
+
+ @Override
+ public List<TopicSink> addTopicSinks(List<TopicParameters> paramList) {
+ List<TopicSink> sinks = new ArrayList<>(paramList.size());
+
+ for (TopicParameters param : paramList) {
+ switch (Topic.CommInfrastructure.valueOf(param.getTopicCommInfrastructure().toUpperCase())) {
+ case UEB:
+ sinks.add(UebTopicFactories.getSinkFactory().build(param));
+ break;
+ case DMAAP:
+ sinks.add(DmaapTopicFactories.getSinkFactory().build(param));
+ break;
+ case NOOP:
+ sinks.add(NoopTopicFactories.getSinkFactory().build(param));
+ break;
+ default:
+ logger.debug("Unknown sink type {} for topic: {}", param.getTopicCommInfrastructure(),
+ param.getTopic());
+ break;
}
}
- return sources;
+ lockSinks(sinks);
+
+ return sinks;
}
@Override
@@ -99,15 +165,17 @@ class TopicEndpointProxy implements TopicEndpoint {
sinks.addAll(DmaapTopicFactories.getSinkFactory().build(properties));
sinks.addAll(NoopTopicFactories.getSinkFactory().build(properties));
- if (this.isLocked()) {
- for (final TopicSink sink : sinks) {
- sink.lock();
- }
- }
+ lockSinks(sinks);
return sinks;
}
+ private void lockSinks(List<TopicSink> sinks) {
+ if (this.isLocked()) {
+ sinks.forEach(TopicSink::lock);
+ }
+ }
+
@Override
public List<TopicSource> getTopicSources() {
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactory.java
index 9623b4fa..7164f919 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactory.java
@@ -21,6 +21,7 @@
package org.onap.policy.common.endpoints.event.comm.bus;
import java.util.List;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
/**
@@ -40,6 +41,14 @@ public class NoopTopicSourceFactory extends NoopTopicFactory<NoopTopicSource> {
* {@inheritDoc}.
*/
@Override
+ public NoopTopicSource build(BusTopicParams param) {
+ return build(param.getServers(), param.getTopic());
+ }
+
+ /**
+ * {@inheritDoc}.
+ */
+ @Override
protected NoopTopicSource build(List<String> servers, String topic) {
return new NoopTopicSource(servers, topic);
}
@@ -49,6 +58,6 @@ public class NoopTopicSourceFactory extends NoopTopicFactory<NoopTopicSource> {
*/
@Override
public String toString() {
- return "NoopTopicSourceFactory[" + super.toString() + "]";
+ return "NoopTopicSourceFactory [" + super.toString() + "]";
}
}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseFactory.java
index 897a8e19..a6b5b1d9 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseFactory.java
@@ -23,6 +23,7 @@ package org.onap.policy.common.endpoints.event.comm.bus;
import java.util.List;
import java.util.Properties;
import org.onap.policy.common.endpoints.event.comm.Topic;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
/**
* Topic Base Factory.
@@ -50,6 +51,14 @@ public interface TopicBaseFactory<T extends Topic> {
T build(List<String> servers, String topic, boolean managed);
/**
+ * Construct an instance of an endpoint.
+ *
+ * @param param parameters
+ * @return an instance of T.
+ */
+ T build(BusTopicParams param);
+
+ /**
* destroy TopicBase instance.
* @param topic topic.
*/
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseHashedFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseHashedFactory.java
index f958bd01..38ee25a4 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseHashedFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/TopicBaseHashedFactory.java
@@ -26,6 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import org.onap.policy.common.endpoints.event.comm.Topic;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
/**
* Topic Factory implementation that indexes T instances in a hash table.
@@ -103,6 +104,14 @@ public abstract class TopicBaseHashedFactory<T extends Topic> implements TopicBa
* {@inheritDoc}.
*/
@Override
+ public T build(BusTopicParams param) {
+ return this.build(param.getServers(), param.getTopic(), param.isManaged());
+ }
+
+ /**
+ * {@inheritDoc}.
+ */
+ @Override
public T build(List<String> servers, String topic, boolean managed) {
if (servers == null || servers.isEmpty()) {
throw new IllegalArgumentException(MISSING_SERVERS_MESSAGE);
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/parameters/TopicParameterGroup.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/parameters/TopicParameterGroup.java
index b89f47dc..eab9c12f 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/parameters/TopicParameterGroup.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/parameters/TopicParameterGroup.java
@@ -1,6 +1,7 @@
/*-
* ============LICENSE_START=======================================================
* Copyright (C) 2019 Nordix Foundation.
+ * Modifications Copyright (C) 2019 AT&T Intellectual Property. 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.
@@ -20,6 +21,7 @@
package org.onap.policy.common.endpoints.parameters;
+import java.util.LinkedList;
import java.util.List;
import lombok.Getter;
import lombok.Setter;
@@ -41,8 +43,8 @@ import org.onap.policy.common.parameters.annotations.NotNull;
@Setter
public class TopicParameterGroup extends ParameterGroupImpl {
- private List<TopicParameters> topicSources;
- private List<TopicParameters> topicSinks;
+ private final List<TopicParameters> topicSources = new LinkedList<>();
+ private final List<TopicParameters> topicSinks = new LinkedList<>();
public TopicParameterGroup() {
super(TopicParameterGroup.class.getSimpleName());
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.java
index ba5fe18e..22ddecd1 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.java
@@ -28,10 +28,17 @@ import static org.junit.Assert.assertTrue;
import java.util.List;
import java.util.Properties;
+import org.junit.After;
import org.junit.Test;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicFactories;
import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicPropertyBuilder;
+import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicFactories;
import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicPropertyBuilder;
+import org.onap.policy.common.endpoints.event.comm.bus.UebTopicFactories;
+import org.onap.policy.common.endpoints.event.comm.bus.UebTopicPropertyBuilder;
+import org.onap.policy.common.endpoints.parameters.TopicParameterGroup;
+import org.onap.policy.common.endpoints.parameters.TopicParameters;
import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
import org.onap.policy.common.utils.gson.GsonTestUtils;
@@ -47,41 +54,54 @@ public class TopicEndpointProxyTest {
private static final String DMAAP_SINK_TOPIC = "dmaap-sink";
private Properties configuration = new Properties();
+ private TopicParameterGroup group = new TopicParameterGroup();
/**
* Constructor.
*/
public TopicEndpointProxyTest() {
- Properties noopSourceProperties =
+ NoopTopicPropertyBuilder noopSourceBuilder =
new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_NOOP_SOURCE_TOPICS)
- .makeTopic(NOOP_SOURCE_TOPIC).build();
+ .makeTopic(NOOP_SOURCE_TOPIC);
+ configuration.putAll(noopSourceBuilder.build());
+ group.getTopicSources().add(noopSourceBuilder.getParams());
- Properties noopSinkProperties =
+ NoopTopicPropertyBuilder noopSinkBuilder =
new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_NOOP_SINK_TOPICS)
- .makeTopic(NOOP_SINK_TOPIC).build();
-
- Properties uebSourceProperties =
- new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS)
- .makeTopic(UEB_SOURCE_TOPIC).build();
-
- Properties uebSinkProperties =
- new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS)
- .makeTopic(UEB_SINK_TOPIC).build();
-
- Properties dmaapSourceProperties =
+ .makeTopic(NOOP_SINK_TOPIC);
+ configuration.putAll(noopSinkBuilder.build());
+ group.getTopicSinks().add(noopSinkBuilder.getParams());
+
+ UebTopicPropertyBuilder uebSourceBuilder =
+ new UebTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS)
+ .makeTopic(UEB_SOURCE_TOPIC);
+ configuration.putAll(uebSourceBuilder.build());
+ group.getTopicSources().add(uebSourceBuilder.getParams());
+
+ UebTopicPropertyBuilder uebSinkBuilder =
+ new UebTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS)
+ .makeTopic(UEB_SINK_TOPIC);
+ configuration.putAll(uebSinkBuilder.build());
+ group.getTopicSinks().add(uebSinkBuilder.getParams());
+
+ DmaapTopicPropertyBuilder dmaapSourceBuilder =
new DmaapTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS)
- .makeTopic(DMAAP_SOURCE_TOPIC).build();
+ .makeTopic(DMAAP_SOURCE_TOPIC);
+ configuration.putAll(dmaapSourceBuilder.build());
+ group.getTopicSources().add(dmaapSourceBuilder.getParams());
- Properties dmaapSinkProperties =
+ DmaapTopicPropertyBuilder dmaapSinkBuilder =
new DmaapTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS)
- .makeTopic(DMAAP_SINK_TOPIC).build();
-
- configuration.putAll(noopSourceProperties);
- configuration.putAll(noopSinkProperties);
- configuration.putAll(uebSourceProperties);
- configuration.putAll(uebSinkProperties);
- configuration.putAll(dmaapSourceProperties);
- configuration.putAll(dmaapSinkProperties);
+ .makeTopic(DMAAP_SINK_TOPIC);
+ configuration.putAll(dmaapSinkBuilder.build());
+ group.getTopicSinks().add(dmaapSinkBuilder.getParams());
+
+ TopicParameters invalidCommInfraParams =
+ new NoopTopicPropertyBuilder(PolicyEndPointProperties.PROPERTY_NOOP_SOURCE_TOPICS)
+ .makeTopic(NOOP_SOURCE_TOPIC).getParams();
+ invalidCommInfraParams.setTopicCommInfrastructure(Topic.CommInfrastructure.REST.name());
+ group.getTopicSources().add(invalidCommInfraParams);
+ group.getTopicSinks().add(invalidCommInfraParams);
}
private <T extends Topic> boolean exists(List<T> topics, String topicName) {
@@ -112,6 +132,21 @@ public class TopicEndpointProxyTest {
|| exists(topics, DMAAP_SINK_TOPIC);
}
+ /**
+ * Destroys all managed topics.
+ */
+ @After
+ public void tearDown() {
+ NoopTopicFactories.getSinkFactory().destroy();
+ NoopTopicFactories.getSourceFactory().destroy();
+
+ UebTopicFactories.getSinkFactory().destroy();
+ UebTopicFactories.getSourceFactory().destroy();
+
+ DmaapTopicFactories.getSinkFactory().destroy();
+ DmaapTopicFactories.getSourceFactory().destroy();
+ }
+
@Test
public void testSerialize() {
TopicEndpoint manager = new TopicEndpointProxy();
@@ -123,7 +158,18 @@ public class TopicEndpointProxyTest {
}
@Test
- public void addTopicSources() {
+ public void addTopicSourcesListOfTopicParameters() {
+ TopicEndpoint manager = new TopicEndpointProxy();
+
+ List<TopicSource> sources = manager.addTopicSources(group.getTopicSources());
+ assertSame(3, sources.size());
+
+ assertTrue(allSources(sources));
+ assertFalse(anySink(sources));
+ }
+
+ @Test
+ public void addTopicSourcesProperties() {
TopicEndpoint manager = new TopicEndpointProxy();
List<TopicSource> sources = manager.addTopicSources(configuration);
@@ -134,7 +180,18 @@ public class TopicEndpointProxyTest {
}
@Test
- public void addTopicSinks() {
+ public void addTopicSinksListOfTopicParameters() {
+ TopicEndpoint manager = new TopicEndpointProxy();
+
+ List<TopicSink> sinks = manager.addTopicSinks(group.getTopicSinks());
+ assertSame(3, sinks.size());
+
+ assertFalse(anySource(sinks));
+ assertTrue(allSinks(sinks));
+ }
+
+ @Test
+ public void addTopicSinksProperties() {
TopicEndpoint manager = new TopicEndpointProxy();
List<TopicSink> sinks = manager.addTopicSinks(configuration);
@@ -145,7 +202,7 @@ public class TopicEndpointProxyTest {
}
@Test
- public void addTopics() {
+ public void addTopicsProperties() {
TopicEndpoint manager = new TopicEndpointProxy();
List<Topic> topics = manager.addTopics(configuration);
@@ -156,6 +213,34 @@ public class TopicEndpointProxyTest {
}
@Test
+ public void addTopicsTopicParameterGroup() {
+ TopicEndpoint manager = new TopicEndpointProxy();
+
+ List<Topic> topics = manager.addTopics(group);
+ assertSame(6, topics.size());
+
+ assertTrue(allSources(topics));
+ assertTrue(allSinks(topics));
+ }
+
+ @Test
+ public void lockSinks_lockSources_locked() {
+ TopicEndpoint manager = new TopicEndpointProxy();
+ manager.lock();
+ for (Topic topic : manager.addTopics(group)) {
+ assertTrue(topic.isLocked());
+ }
+ }
+
+ @Test
+ public void lockSinks_lockSources_unlocked() {
+ TopicEndpoint manager = new TopicEndpointProxy();
+ for (Topic topic : manager.addTopics(group)) {
+ assertFalse(topic.isLocked());
+ }
+ }
+
+ @Test
public void getTopicSources() {
TopicEndpoint manager = new TopicEndpointProxy();
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicPropertyBuilder.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicPropertyBuilder.java
index dff3f788..2e9a6cd7 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicPropertyBuilder.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicPropertyBuilder.java
@@ -61,6 +61,10 @@ import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperti
import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX;
import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX;
+import java.util.Arrays;
+import lombok.Getter;
+import org.onap.policy.common.endpoints.parameters.TopicParameters;
+
public class DmaapTopicPropertyBuilder extends TopicPropertyBuilder {
public static final String SERVER = "my-server";
@@ -75,6 +79,9 @@ public class DmaapTopicPropertyBuilder extends TopicPropertyBuilder {
public static final String MY_AAF_MECHID = "my-aaf-mechid";
public static final String MY_AAF_PASS = "my-aaf-passwd";
+ @Getter
+ private TopicParameters params = new TopicParameters();
+
/**
* Constructs the object.
*
@@ -120,6 +127,28 @@ public class DmaapTopicPropertyBuilder extends TopicPropertyBuilder {
setTopicProperty(PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX, MY_SUBCONTEXT);
setTopicProperty(PROPERTY_DMAAP_DME2_VERSION_SUFFIX, MY_DME_VERSION);
+ params.setTopicCommInfrastructure("dmaap");
+ params.setTopic(topic);
+ params.setEffectiveTopic(MY_EFFECTIVE_TOPIC);
+ params.setConsumerGroup(MY_CONS_GROUP);
+ params.setConsumerInstance(MY_CONS_INST);
+ params.setManaged(true);
+ params.setUseHttps(true);
+ params.setUserName(MY_AAF_MECHID);
+ params.setPassword(MY_AAF_PASS);
+ params.setAftEnvironment(MY_AFT_ENV);
+ params.setAllowSelfSignedCerts(true);
+ params.setApiKey(MY_API_KEY);
+ params.setApiSecret(MY_API_SECRET);
+ params.setFetchLimit(MY_FETCH_LIMIT);
+ params.setFetchTimeout(MY_FETCH_TIMEOUT);
+ params.setEnvironment(MY_ENV);
+ params.setLatitude(MY_LAT);
+ params.setLongitude(MY_LONG);
+ params.setPartitionId(MY_PARTITION);
+ params.setPartner(MY_PARTNER);
+ params.setServers(Arrays.asList(SERVER));
+
return this;
}
}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicFactoryTest.java
index 1f7506f8..5cb9bcbd 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicFactoryTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicFactoryTest.java
@@ -39,6 +39,7 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
public abstract class NoopTopicFactoryTest<F extends NoopTopicFactory<T>, T extends NoopTopicEndpoint>
extends TopicFactoryTestBase<T> {
@@ -64,6 +65,17 @@ public abstract class NoopTopicFactoryTest<F extends NoopTopicFactory<T>, T exte
}
@Test
+ public void testBuildBusTopicParams() {
+ initFactory();
+
+ T item1 = buildTopic(makeParams(servers, MY_TOPIC, true));
+ assertNotNull(item1);
+
+ assertEquals(servers, item1.getServers());
+ assertEquals(MY_TOPIC, item1.getTopic());
+ }
+
+ @Test
public void testBuildListOfStringStringBoolean() {
initFactory();
@@ -196,6 +208,10 @@ public abstract class NoopTopicFactoryTest<F extends NoopTopicFactory<T>, T exte
return factory.build(properties);
}
+ protected T buildTopic(BusTopicParams param) {
+ return factory.build(param);
+ }
+
protected T buildTopic(List<String> servers, String topic, boolean managed) {
return factory.build(servers, topic, managed);
}
@@ -224,4 +240,14 @@ public abstract class NoopTopicFactoryTest<F extends NoopTopicFactory<T>, T exte
protected TopicPropertyBuilder makePropBuilder() {
return new NoopTopicPropertyBuilder(factory.getTopicsPropertyName());
}
+
+ private BusTopicParams makeParams(List<String> servers, String topic, boolean managed) {
+ BusTopicParams params = new BusTopicParams();
+
+ params.setServers(servers);
+ params.setTopic(topic);
+ params.setManaged(managed);
+
+ return params;
+ }
}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicPropertyBuilder.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicPropertyBuilder.java
index 529f8b2f..3f0553f8 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicPropertyBuilder.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicPropertyBuilder.java
@@ -27,11 +27,18 @@ import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperti
import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_EFFECTIVE_TOPIC_SUFFIX;
import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
+import java.util.Arrays;
+import lombok.Getter;
+import org.onap.policy.common.endpoints.parameters.TopicParameters;
+
public class NoopTopicPropertyBuilder extends TopicPropertyBuilder {
public static final String SERVER = "my-server";
public static final String TOPIC2 = "my-topic-2";
+ @Getter
+ private TopicParameters params = new TopicParameters();
+
/**
* Constructs the object.
*
@@ -56,6 +63,14 @@ public class NoopTopicPropertyBuilder extends TopicPropertyBuilder {
setTopicProperty(PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX, "true");
setTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX, SERVER);
+ params.setTopicCommInfrastructure("noop");
+ params.setTopic(topic);
+ params.setEffectiveTopic(MY_EFFECTIVE_TOPIC);
+ params.setManaged(true);
+ params.setUseHttps(true);
+ params.setAllowSelfSignedCerts(true);
+ params.setServers(Arrays.asList(SERVER));
+
return this;
}
}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java
index cc44716f..b4ab54fa 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java
@@ -20,10 +20,19 @@
package org.onap.policy.common.endpoints.event.comm.bus;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
public class NoopTopicSinkFactoryTest extends NoopTopicFactoryTest<NoopTopicSinkFactory, NoopTopicSink> {
@Override
protected NoopTopicSinkFactory buildFactory() {
return new NoopTopicSinkFactory();
}
+
+ @Test
+ public void testToString() {
+ assertTrue(new NoopTopicSinkFactory().toString().startsWith("NoopTopicSinkFactory ["));
+ }
}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactoryTest.java
index c8a44292..e16174fd 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactoryTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSourceFactoryTest.java
@@ -20,10 +20,19 @@
package org.onap.policy.common.endpoints.event.comm.bus;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
public class NoopTopicSourceFactoryTest extends NoopTopicFactoryTest<NoopTopicSourceFactory, NoopTopicSource> {
@Override
protected NoopTopicSourceFactory buildFactory() {
return new NoopTopicSourceFactory();
}
+
+ @Test
+ public void testToString() {
+ assertTrue(new NoopTopicSourceFactory().toString().startsWith("NoopTopicSourceFactory ["));
+ }
}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java
index 6119284f..cb5507e5 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java
@@ -45,6 +45,10 @@ import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperti
import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX;
import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX;
+import java.util.Arrays;
+import lombok.Getter;
+import org.onap.policy.common.endpoints.parameters.TopicParameters;
+
public class UebTopicPropertyBuilder extends TopicPropertyBuilder {
public static final String SERVER = "my-server";
@@ -53,6 +57,9 @@ public class UebTopicPropertyBuilder extends TopicPropertyBuilder {
public static final String MY_AAF_MECHID = "my-aaf-mechid";
public static final String MY_AAF_PASS = "my-aaf-passwd";
+ @Getter
+ private TopicParameters params = new TopicParameters();
+
/**
* Constructs the object.
*
@@ -87,6 +94,24 @@ public class UebTopicPropertyBuilder extends TopicPropertyBuilder {
setTopicProperty(PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX, MY_PARTITION);
setTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX, SERVER);
+ params.setTopicCommInfrastructure("ueb");
+ params.setTopic(topic);
+ params.setEffectiveTopic(MY_EFFECTIVE_TOPIC);
+ params.setConsumerGroup(MY_CONS_GROUP);
+ params.setConsumerInstance(MY_CONS_INST);
+ params.setManaged(true);
+ params.setUseHttps(true);
+ params.setUserName(MY_AAF_MECHID);
+ params.setPassword(MY_AAF_PASS);
+ params.setAftEnvironment(MY_AFT_ENV);
+ params.setAllowSelfSignedCerts(true);
+ params.setApiKey(MY_API_KEY);
+ params.setApiSecret(MY_API_SECRET);
+ params.setFetchLimit(MY_FETCH_LIMIT);
+ params.setFetchTimeout(MY_FETCH_TIMEOUT);
+ params.setPartitionId(MY_PARTITION);
+ params.setServers(Arrays.asList(SERVER));
+
return this;
}
}
diff --git a/policy-endpoints/src/test/resources/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.json b/policy-endpoints/src/test/resources/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.json
index 3402a524..9c8d78e5 100644
--- a/policy-endpoints/src/test/resources/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.json
+++ b/policy-endpoints/src/test/resources/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxyTest.json
@@ -8,13 +8,13 @@
"recentEvents" : [ ],
"alive" : false,
"locked" : false,
- "apiKey" : null,
- "apiSecret" : null,
+ "apiKey" : "my-api-key",
+ "apiSecret" : "my-api-secret",
"useHttps" : true,
"allowSelfSignedCerts" : true,
"consumerGroup" : "${obj.topicSources[0].consumerGroup}",
"consumerInstance" : "${obj.topicSources[0].consumerInstance}",
- "fetchTimeout" : 15000,
+ "fetchTimeout" : 101,
"fetchLimit" : 100,
"topicCommInfrastructure" : "UEB"
}, {
@@ -49,8 +49,8 @@
"recentEvents" : [ ],
"alive" : false,
"locked" : false,
- "apiKey" : null,
- "apiSecret" : null,
+ "apiKey" : "my-api-key",
+ "apiSecret" : "my-api-secret",
"useHttps" : true,
"allowSelfSignedCerts" : true,
"topicCommInfrastructure" : "UEB",