diff options
Diffstat (limited to 'policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal')
7 files changed, 67 insertions, 23 deletions
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBase.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBase.java index 3e4f48e7..ccf25753 100644 --- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBase.java +++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBase.java @@ -1,8 +1,8 @@ /* * ============LICENSE_START======================================================= - * policy-endpoints + * ONAP * ================================================================================ - * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved. + * Copyright (C) 2017-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. @@ -60,7 +60,7 @@ public abstract class BusTopicBase extends TopicBase implements ApiKeyEnabled { * @throws IllegalArgumentException if invalid parameters are present */ public BusTopicBase(BusTopicParams busTopicParams) { - super(busTopicParams.getServers(), busTopicParams.getTopic()); + super(busTopicParams.getServers(), busTopicParams.getTopic(), busTopicParams.getEffectiveTopic()); this.apiKey = busTopicParams.getApiKey(); this.apiSecret = busTopicParams.getApiSecret(); this.useHttps = busTopicParams.isUseHttps(); diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParams.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParams.java index 6e3c2632..b9817abf 100644 --- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParams.java +++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParams.java @@ -1,6 +1,6 @@ /* * ============LICENSE_START======================================================= - * policy-endpoints + * ONAP * ================================================================================ * Copyright (C) 2018 Samsung Electronics Co., Ltd. All rights reserved. * Modifications Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved. @@ -51,6 +51,7 @@ public class BusTopicParams { private List<String> servers; private Map<String, String> additionalProps; private String topic; + private String effectiveTopic; private String apiKey; private String apiSecret; private String consumerGroup; @@ -122,6 +123,10 @@ public class BusTopicParams { return topic; } + public String getEffectiveTopic() { + return effectiveTopic; + } + public String getApiKey() { return apiKey; } @@ -276,6 +281,11 @@ public class BusTopicParams { return this; } + public TopicParamsBuilder effectiveTopic(String effectiveTopic) { + this.params.effectiveTopic = effectiveTopic; + return this; + } + public TopicParamsBuilder apiKey(String apiKey) { this.params.apiKey = apiKey; return this; diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSink.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSink.java index 01531055..ba556bb8 100644 --- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSink.java +++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSink.java @@ -1,8 +1,8 @@ /*- * ============LICENSE_START======================================================= - * policy-endpoints + * ONAP * ================================================================================ - * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved. + * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved. * Modified Copyright (C) 2018 Samsung Electronics Co., Ltd. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); @@ -87,7 +87,7 @@ public class InlineDmaapTopicSink extends InlineBusTopicSink implements DmaapTop if (allNullOrEmpty(this.environment, this.aftEnvironment, this.latitude, this.longitude, this.partner)) { this.publisher = new BusPublisher.CambriaPublisherWrapper(BusTopicParams.builder() .servers(this.servers) - .topic(this.topic) + .topic(this.effectiveTopic) .apiKey(this.apiKey) .apiSecret(this.apiSecret) .userName(this.userName) @@ -98,7 +98,7 @@ public class InlineDmaapTopicSink extends InlineBusTopicSink implements DmaapTop } else { this.publisher = new BusPublisher.DmaapDmePublisherWrapper(BusTopicParams.builder() .servers(this.servers) - .topic(this.topic) + .topic(this.effectiveTopic) .userName(this.userName) .password(this.password) .environment(this.environment) diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSink.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSink.java index f989f808..f258d5d9 100644 --- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSink.java +++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSink.java @@ -1,8 +1,8 @@ /* * ============LICENSE_START======================================================= - * policy-endpoints + * ONAP * ================================================================================ - * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved. + * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved. * Modified Copyright (C) 2018 Samsung Electronics Co., Ltd. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); @@ -63,7 +63,7 @@ public class InlineUebTopicSink extends InlineBusTopicSink implements UebTopicSi this.publisher = new BusPublisher.CambriaPublisherWrapper(BusTopicParams.builder() .servers(this.servers) - .topic(this.topic) + .topic(this.effectiveTopic) .apiKey(this.apiKey) .apiSecret(this.apiSecret) .useHttps(this.useHttps) diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSource.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSource.java index e4064c5d..e5d08a2a 100644 --- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSource.java +++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSource.java @@ -1,8 +1,8 @@ /*- * ============LICENSE_START======================================================= - * policy-endpoints + * ONAP * ================================================================================ - * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved. + * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved. * Modifications Copyright (C) 2018-2019 Samsung Electronics Co., Ltd. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); @@ -89,7 +89,7 @@ public class SingleThreadedDmaapTopicSource extends SingleThreadedBusTopicSource public void init() throws MalformedURLException { BusTopicParams.TopicParamsBuilder builder = BusTopicParams.builder() .servers(this.servers) - .topic(this.topic) + .topic(this.effectiveTopic) .apiKey(this.apiKey) .apiSecret(this.apiSecret) .consumerGroup(this.consumerGroup) diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSource.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSource.java index f6d4b531..e210762d 100644 --- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSource.java +++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSource.java @@ -1,8 +1,8 @@ /* * ============LICENSE_START======================================================= - * policy-endpoints + * ONAP * ================================================================================ - * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved. + * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved. * Modifications Copyright (C) 2018-2019 Samsung Electronics Co., Ltd. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); @@ -38,7 +38,6 @@ public class SingleThreadedUebTopicSource extends SingleThreadedBusTopicSource i */ public SingleThreadedUebTopicSource(BusTopicParams busTopicParams) { super(busTopicParams); - this.init(); } @@ -49,7 +48,7 @@ public class SingleThreadedUebTopicSource extends SingleThreadedBusTopicSource i public void init() { this.consumer = new BusConsumer.CambriaConsumerWrapper(BusTopicParams.builder() .servers(this.servers) - .topic(this.topic) + .topic(this.effectiveTopic) .apiKey(this.apiKey) .apiSecret(this.apiSecret) .consumerGroup(this.consumerGroup) diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBase.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBase.java index 80664554..6f07df1b 100644 --- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBase.java +++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBase.java @@ -1,8 +1,8 @@ /* * ============LICENSE_START======================================================= - * policy-endpoints + * ONAP * ================================================================================ - * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved. + * Copyright (C) 2017-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. @@ -44,7 +44,12 @@ public abstract class TopicBase implements Topic { /** * Topic. */ - protected String topic; + protected final String topic; + + /** + * Topic Alias. + */ + protected final String effectiveTopic; /** * Event cache. @@ -78,6 +83,18 @@ public abstract class TopicBase implements Topic { * @throws IllegalArgumentException if invalid parameters are present */ public TopicBase(List<String> servers, String topic) { + this(servers, topic, topic); + } + + /** + * Instantiates a new Topic Base. + * + * @param servers list of servers + * @param topic topic name + * + * @throws IllegalArgumentException if invalid parameters are present + */ + public TopicBase(List<String> servers, String topic, String effectiveTopic) { if (servers == null || servers.isEmpty()) { throw new IllegalArgumentException("Server(s) must be provided"); @@ -87,8 +104,16 @@ public abstract class TopicBase implements Topic { throw new IllegalArgumentException("A Topic must be provided"); } + String effectiveTopicCopy; + if (effectiveTopic == null || effectiveTopic.isEmpty()) { + effectiveTopicCopy = topic; + } else { + effectiveTopicCopy = effectiveTopic; + } + this.servers = servers; this.topic = topic; + this.effectiveTopic = effectiveTopicCopy; } @Override @@ -204,6 +229,11 @@ public abstract class TopicBase implements Topic { } @Override + public String getEffectiveTopic() { + return effectiveTopic; + } + + @Override public boolean isAlive() { return this.alive; } @@ -222,7 +252,12 @@ public abstract class TopicBase implements Topic { @Override public String toString() { - return "TopicBase [servers=" + servers + ", topic=" + topic + ", #recentEvents=" + recentEvents.size() - + ", locked=" + locked + ", #topicListeners=" + topicListeners.size() + "]"; + return "TopicBase [servers=" + servers + + ", topic=" + topic + + ", effectiveTopic=" + effectiveTopic + + ", #recentEvents=" + recentEvents.size() + + ", locked=" + locked + + ", #topicListeners=" + topicListeners.size() + + "]"; } } |