summaryrefslogtreecommitdiffstats
path: root/policy-endpoints/src/main/java/org
diff options
context:
space:
mode:
Diffstat (limited to 'policy-endpoints/src/main/java/org')
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactory.java20
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactory.java17
2 files changed, 26 insertions, 11 deletions
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactory.java
index 08e8dfe8..09078720 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactory.java
@@ -150,7 +150,7 @@ class IndexedDmaapTopicSinkFactory implements DmaapTopicSinkFactory {
return dmaapTopicWriters.get(busTopicParams.getTopic());
}
- DmaapTopicSink dmaapTopicSink = new InlineDmaapTopicSink(busTopicParams);
+ DmaapTopicSink dmaapTopicSink = makeSink(busTopicParams);
if (busTopicParams.isManaged()) {
dmaapTopicWriters.put(busTopicParams.getTopic(), dmaapTopicSink);
@@ -210,7 +210,7 @@ class IndexedDmaapTopicSinkFactory implements DmaapTopicSinkFactory {
final String partitionKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+ "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX);
- final String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS
+ final String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+ "." + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
/* DME2 Properties */
@@ -343,6 +343,16 @@ class IndexedDmaapTopicSinkFactory implements DmaapTopicSinkFactory {
}
}
+ /**
+ * Makes a new sink.
+ *
+ * @param busTopicParams parameters to use to configure the sink
+ * @return a new sink
+ */
+ protected DmaapTopicSink makeSink(BusTopicParams busTopicParams) {
+ return new InlineDmaapTopicSink(busTopicParams);
+ }
+
@Override
public void destroy(String topic) {
@@ -385,7 +395,7 @@ class IndexedDmaapTopicSinkFactory implements DmaapTopicSinkFactory {
if (dmaapTopicWriters.containsKey(topic)) {
return dmaapTopicWriters.get(topic);
} else {
- throw new IllegalStateException("DmaapTopicSink for " + topic + " not found");
+ throw new IllegalArgumentException("DmaapTopicSink for " + topic + " not found");
}
}
}
@@ -397,9 +407,7 @@ class IndexedDmaapTopicSinkFactory implements DmaapTopicSinkFactory {
@Override
public String toString() {
- StringBuilder builder = new StringBuilder();
- builder.append("IndexedDmaapTopicSinkFactory []");
- return builder.toString();
+ return "IndexedDmaapTopicSinkFactory []";
}
}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactory.java
index d5e04d50..f45164f8 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactory.java
@@ -152,8 +152,7 @@ class IndexedDmaapTopicSourceFactory implements DmaapTopicSourceFactory {
return dmaapTopicSources.get(busTopicParams.getTopic());
}
- DmaapTopicSource dmaapTopicSource =
- new SingleThreadedDmaapTopicSource(busTopicParams);
+ DmaapTopicSource dmaapTopicSource = makeSource(busTopicParams);
if (busTopicParams.isManaged()) {
dmaapTopicSources.put(busTopicParams.getTopic(), dmaapTopicSource);
@@ -396,6 +395,16 @@ class IndexedDmaapTopicSourceFactory implements DmaapTopicSourceFactory {
return this.build(servers, topic, null, null);
}
+ /**
+ * Makes a new source.
+ *
+ * @param busTopicParams parameters to use to configure the source
+ * @return a new source
+ */
+ protected DmaapTopicSource makeSource(BusTopicParams busTopicParams) {
+ return new SingleThreadedDmaapTopicSource(busTopicParams);
+ }
+
@Override
public void destroy(String topic) {
@@ -451,9 +460,7 @@ class IndexedDmaapTopicSourceFactory implements DmaapTopicSourceFactory {
@Override
public String toString() {
- StringBuilder builder = new StringBuilder();
- builder.append("IndexedDmaapTopicSourceFactory []");
- return builder.toString();
+ return "IndexedDmaapTopicSourceFactory []";
}
}