2 * ============LICENSE_START=======================================================
3 * ONAP Policy Engine - Common Modules
4 * ================================================================================
5 * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.common.endpoints.event.comm.bus;
23 import java.util.ArrayList;
24 import java.util.Arrays;
25 import java.util.HashMap;
26 import java.util.List;
28 import java.util.Properties;
30 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
31 import org.onap.policy.common.endpoints.event.comm.bus.internal.SingleThreadedDmaapTopicSource;
32 import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
33 import org.slf4j.Logger;
34 import org.slf4j.LoggerFactory;
37 * Factory of DMAAP Source Topics indexed by topic name.
40 class IndexedDmaapTopicSourceFactory implements DmaapTopicSourceFactory {
41 private static final String MISSING_TOPIC = "A topic must be provided";
46 private static Logger logger = LoggerFactory.getLogger(IndexedDmaapTopicSourceFactory.class);
49 * DMaaP Topic Name Index.
51 protected HashMap<String, DmaapTopicSource> dmaapTopicSources = new HashMap<>();
54 public DmaapTopicSource build(BusTopicParams busTopicParams) {
56 if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
57 throw new IllegalArgumentException(MISSING_TOPIC);
61 if (dmaapTopicSources.containsKey(busTopicParams.getTopic())) {
62 return dmaapTopicSources.get(busTopicParams.getTopic());
65 DmaapTopicSource dmaapTopicSource = makeSource(busTopicParams);
67 if (busTopicParams.isManaged()) {
68 dmaapTopicSources.put(busTopicParams.getTopic(), dmaapTopicSource);
70 return dmaapTopicSource;
75 public List<DmaapTopicSource> build(Properties properties) {
77 String readTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS);
78 if (readTopics == null || readTopics.isEmpty()) {
79 logger.info("{}: no topic for DMaaP Source", this);
80 return new ArrayList<>();
82 List<String> readTopicList = new ArrayList<>(Arrays.asList(readTopics.split("\\s*,\\s*")));
84 List<DmaapTopicSource> dmaapTopicSourceLst = new ArrayList<>();
86 for (String topic : readTopicList) {
87 if (this.dmaapTopicSources.containsKey(topic)) {
88 dmaapTopicSourceLst.add(this.dmaapTopicSources.get(topic));
92 String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
93 + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
95 List<String> serverList;
96 if (servers != null && !servers.isEmpty()) {
97 serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
99 serverList = new ArrayList<>();
102 final String apiKey = properties.getProperty(
103 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
104 + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
106 final String apiSecret = properties.getProperty(
107 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
108 + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
110 final String aafMechId = properties.getProperty(
111 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
112 + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_MECHID_SUFFIX);
114 final String aafPassword = properties.getProperty(
115 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
116 + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_PASSWORD_SUFFIX);
118 final String consumerGroup = properties.getProperty(
119 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
120 + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX);
122 final String consumerInstance = properties.getProperty(
123 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
124 + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX);
126 final String fetchTimeoutString = properties.getProperty(
127 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
128 + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX);
130 /* DME2 Properties */
132 final String dme2Environment = properties.getProperty(
133 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
134 + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
136 final String dme2AftEnvironment = properties.getProperty(
137 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
138 + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
140 final String dme2Partner = properties.getProperty(
141 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
142 + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX);
144 final String dme2RouteOffer = properties.getProperty(
145 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
146 + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX);
148 final String dme2Latitude = properties.getProperty(
149 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
150 + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
152 final String dme2Longitude = properties.getProperty(
153 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
154 + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
156 final String dme2EpReadTimeoutMs =
157 properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
158 + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX);
160 final String dme2EpConnTimeout = properties.getProperty(
161 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
162 + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX);
164 final String dme2RoundtripTimeoutMs =
165 properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
166 + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX);
168 final String dme2Version = properties.getProperty(
169 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
170 + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_VERSION_SUFFIX);
172 final String dme2SubContextPath = properties.getProperty(
173 PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
174 + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX);
176 final String dme2SessionStickinessRequired =
177 properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
178 + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX);
180 Map<String, String> dme2AdditionalProps = new HashMap<>();
182 if (dme2EpReadTimeoutMs != null && !dme2EpReadTimeoutMs.isEmpty()) {
183 dme2AdditionalProps.put(DME2_READ_TIMEOUT_PROPERTY, dme2EpReadTimeoutMs);
185 if (dme2EpConnTimeout != null && !dme2EpConnTimeout.isEmpty()) {
186 dme2AdditionalProps.put(DME2_EP_CONN_TIMEOUT_PROPERTY, dme2EpConnTimeout);
188 if (dme2RoundtripTimeoutMs != null && !dme2RoundtripTimeoutMs.isEmpty()) {
189 dme2AdditionalProps.put(DME2_ROUNDTRIP_TIMEOUT_PROPERTY, dme2RoundtripTimeoutMs);
191 if (dme2Version != null && !dme2Version.isEmpty()) {
192 dme2AdditionalProps.put(DME2_VERSION_PROPERTY, dme2Version);
194 if (dme2RouteOffer != null && !dme2RouteOffer.isEmpty()) {
195 dme2AdditionalProps.put(DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
197 if (dme2SubContextPath != null && !dme2SubContextPath.isEmpty()) {
198 dme2AdditionalProps.put(DME2_SUBCONTEXT_PATH_PROPERTY, dme2SubContextPath);
200 if (dme2SessionStickinessRequired != null && !dme2SessionStickinessRequired.isEmpty()) {
201 dme2AdditionalProps.put(DME2_SESSION_STICKINESS_REQUIRED_PROPERTY, dme2SessionStickinessRequired);
205 if (servers == null || servers.isEmpty()) {
207 logger.error("{}: no DMaaP servers or DME2 ServiceName provided", this);
211 int fetchTimeout = DmaapTopicSource.DEFAULT_TIMEOUT_MS_FETCH;
212 if (fetchTimeoutString != null && !fetchTimeoutString.isEmpty()) {
214 fetchTimeout = Integer.parseInt(fetchTimeoutString);
215 } catch (NumberFormatException nfe) {
216 logger.warn("{}: fetch timeout {} is in invalid format for topic {} ", this, fetchTimeoutString,
221 String fetchLimitString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
222 + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX);
223 int fetchLimit = DmaapTopicSource.DEFAULT_LIMIT_FETCH;
224 if (fetchLimitString != null && !fetchLimitString.isEmpty()) {
226 fetchLimit = Integer.parseInt(fetchLimitString);
227 } catch (NumberFormatException nfe) {
228 logger.warn("{}: fetch limit {} is in invalid format for topic {} ", this, fetchLimitString,
233 String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
234 + "." + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
235 boolean managed = true;
236 if (managedString != null && !managedString.isEmpty()) {
237 managed = Boolean.parseBoolean(managedString);
240 String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
241 + "." + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
243 // default is to use HTTP if no https property exists
244 boolean useHttps = false;
245 if (useHttpsString != null && !useHttpsString.isEmpty()) {
246 useHttps = Boolean.parseBoolean(useHttpsString);
249 String allowSelfSignedCertsString =
250 properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
251 + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
253 // default is to disallow self-signed certs
254 boolean allowSelfSignedCerts = false;
255 if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
256 allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
260 DmaapTopicSource uebTopicSource = this.build(BusTopicParams.builder()
264 .apiSecret(apiSecret)
266 .password(aafPassword)
267 .consumerGroup(consumerGroup)
268 .consumerInstance(consumerInstance)
269 .fetchTimeout(fetchTimeout)
270 .fetchLimit(fetchLimit)
271 .environment(dme2Environment)
272 .aftEnvironment(dme2AftEnvironment)
273 .partner(dme2Partner)
274 .latitude(dme2Latitude)
275 .longitude(dme2Longitude)
276 .additionalProps(dme2AdditionalProps)
279 .allowSelfSignedCerts(allowSelfSignedCerts)
282 dmaapTopicSourceLst.add(uebTopicSource);
285 return dmaapTopicSourceLst;
289 public DmaapTopicSource build(List<String> servers, String topic, String apiKey, String apiSecret) {
290 return this.build(BusTopicParams.builder()
294 .apiSecret(apiSecret)
295 .fetchTimeout(DmaapTopicSource.DEFAULT_TIMEOUT_MS_FETCH)
296 .fetchLimit(DmaapTopicSource.DEFAULT_LIMIT_FETCH)
299 .allowSelfSignedCerts(false)
304 public DmaapTopicSource build(List<String> servers, String topic) {
305 return this.build(servers, topic, null, null);
309 * Makes a new source.
311 * @param busTopicParams parameters to use to configure the source
312 * @return a new source
314 protected DmaapTopicSource makeSource(BusTopicParams busTopicParams) {
315 return new SingleThreadedDmaapTopicSource(busTopicParams);
319 public void destroy(String topic) {
321 if (topic == null || topic.isEmpty()) {
322 throw new IllegalArgumentException(MISSING_TOPIC);
325 DmaapTopicSource uebTopicSource;
327 synchronized (this) {
328 if (!dmaapTopicSources.containsKey(topic)) {
332 uebTopicSource = dmaapTopicSources.remove(topic);
335 uebTopicSource.shutdown();
339 public void destroy() {
340 List<DmaapTopicSource> readers = this.inventory();
341 for (DmaapTopicSource reader : readers) {
345 synchronized (this) {
346 this.dmaapTopicSources.clear();
351 public DmaapTopicSource get(String topic) {
353 if (topic == null || topic.isEmpty()) {
354 throw new IllegalArgumentException(MISSING_TOPIC);
357 synchronized (this) {
358 if (dmaapTopicSources.containsKey(topic)) {
359 return dmaapTopicSources.get(topic);
361 throw new IllegalStateException("DmaapTopiceSource for " + topic + " not found");
367 public synchronized List<DmaapTopicSource> inventory() {
368 return new ArrayList<>(this.dmaapTopicSources.values());
372 public String toString() {
373 return "IndexedDmaapTopicSourceFactory []";