aboutsummaryrefslogtreecommitdiffstats
path: root/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxy.java
diff options
context:
space:
mode:
authorJim Hahn <jrh3@att.com>2019-07-27 08:20:17 -0400
committerJim Hahn <jrh3@att.com>2019-07-29 15:30:06 -0400
commitf14578b9c5d9a98cceb5342d69b0326289b1690c (patch)
tree74af4f4e2de4e4b64b01d31b914ace84431dc8df /policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxy.java
parent4513f428543dbd09f855fdd392d90d628f5d403e (diff)
Add addTopic(List<TopicParams>)
New services are typically configured using TopicParameters. These are converted into Properties before being passed to TopicEndpoint.addTopic(), which then converts them into BusTopicParams before configuring a given topic. Added TopicEndpoint.addTopic(List<TopicParams>), which makes it possible to configure the topics without going into the intermediate Properties. Furthermore, because TopicParams is a subclass of BusTopicParams, no conversion is needed for that either, so the TopicParams can be passed directly into the configuration classes. Incorporated changes from review. Change-Id: Id87e2c6812e36ae1a3ac680e6b35208667971782 Issue-ID: POLICY-1953 Signed-off-by: Jim Hahn <jrh3@att.com>
Diffstat (limited to 'policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxy.java')
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/TopicEndpointProxy.java84
1 files changed, 76 insertions, 8 deletions
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() {