2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
6 * Modified Copyright (C) 2018 Samsung Electronics Co., Ltd.
7 * ================================================================================
8 * Licensed under the Apache License, Version 2.0 (the "License");
9 * you may not use this file except in compliance with the License.
10 * You may obtain a copy of the License at
12 * http://www.apache.org/licenses/LICENSE-2.0
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the License is distributed on an "AS IS" BASIS,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 * See the License for the specific language governing permissions and
18 * limitations under the License.
19 * ============LICENSE_END=========================================================
22 package org.onap.policy.common.endpoints.event.comm.bus;
24 import java.util.ArrayList;
25 import java.util.Arrays;
26 import java.util.HashMap;
27 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.InlineUebTopicSink;
32 import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
33 import org.slf4j.Logger;
34 import org.slf4j.LoggerFactory;
37 * UEB Topic Sink Factory.
39 public interface UebTopicSinkFactory {
42 * Instantiates a new UEB Topic Writer.
44 * @param servers list of servers
45 * @param topic topic name
46 * @param apiKey API Key
47 * @param apiSecret API Secret
48 * @param partitionKey Consumer Group
49 * @param managed is this sink endpoint managed?
51 * @return an UEB Topic Sink
52 * @throws IllegalArgumentException if invalid parameters are present
54 public UebTopicSink build(List<String> servers, String topic, String apiKey, String apiSecret, String partitionKey,
55 boolean managed, boolean useHttps, boolean allowSelfSignedCerts);
58 * Creates an UEB Topic Writer based on properties files.
60 * @param properties Properties containing initialization values
62 * @return an UEB Topic Writer
63 * @throws IllegalArgumentException if invalid parameters are present
65 public List<UebTopicSink> build(Properties properties);
68 * Instantiates a new UEB Topic Writer.
70 * @param servers list of servers
71 * @param topic topic name
73 * @return an UEB Topic Writer
74 * @throws IllegalArgumentException if invalid parameters are present
76 public UebTopicSink build(List<String> servers, String topic);
79 * Destroys an UEB Topic Writer based on a topic.
81 * @param topic topic name
82 * @throws IllegalArgumentException if invalid parameters are present
84 public void destroy(String topic);
87 * Destroys all UEB Topic Writers.
89 public void destroy();
92 * gets an UEB Topic Writer based on topic name.
94 * @param topic the topic name
96 * @return an UEB Topic Writer with topic name
97 * @throws IllegalArgumentException if an invalid topic is provided
98 * @throws IllegalStateException if the UEB Topic Reader is an incorrect state
100 public UebTopicSink get(String topic);
103 * Provides a snapshot of the UEB Topic Writers.
105 * @return a list of the UEB Topic Writers
107 public List<UebTopicSink> inventory();
111 /* ------------- implementation ----------------- */
114 * Factory of UEB Reader Topics indexed by topic name.
116 class IndexedUebTopicSinkFactory implements UebTopicSinkFactory {
117 private static final String MISSING_TOPIC = "A topic must be provided";
122 private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSinkFactory.class);
125 * UEB Topic Name Index.
127 protected HashMap<String, UebTopicSink> uebTopicSinks = new HashMap<>();
130 public UebTopicSink build(List<String> servers, String topic, String apiKey, String apiSecret, String partitionKey,
131 boolean managed, boolean useHttps, boolean allowSelfSignedCerts) {
133 if (servers == null || servers.isEmpty()) {
134 throw new IllegalArgumentException("UEB Server(s) must be provided");
137 if (topic == null || topic.isEmpty()) {
138 throw new IllegalArgumentException(MISSING_TOPIC);
141 synchronized (this) {
142 if (uebTopicSinks.containsKey(topic)) {
143 return uebTopicSinks.get(topic);
146 UebTopicSink uebTopicWriter = new InlineUebTopicSink(BusTopicParams.builder()
150 .apiSecret(apiSecret)
151 .partitionId(partitionKey)
153 .allowSelfSignedCerts(allowSelfSignedCerts)
157 uebTopicSinks.put(topic, uebTopicWriter);
160 return uebTopicWriter;
166 public UebTopicSink build(List<String> servers, String topic) {
167 return this.build(servers, topic, null, null, null, true, false, false);
172 public List<UebTopicSink> build(Properties properties) {
174 String writeTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS);
175 if (writeTopics == null || writeTopics.isEmpty()) {
176 logger.info("{}: no topic for UEB Sink", this);
177 return new ArrayList<>();
180 List<String> writeTopicList = new ArrayList<>(Arrays.asList(writeTopics.split("\\s*,\\s*")));
181 List<UebTopicSink> newUebTopicSinks = new ArrayList<>();
182 synchronized (this) {
183 for (String topic : writeTopicList) {
184 if (this.uebTopicSinks.containsKey(topic)) {
185 newUebTopicSinks.add(this.uebTopicSinks.get(topic));
189 String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
190 + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
191 if (servers == null || servers.isEmpty()) {
192 logger.error("{}: no UEB servers configured for sink {}", this, topic);
196 List<String> serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
198 String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
199 + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
200 String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
201 + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
202 String partitionKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
203 + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX);
205 String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
206 + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
207 boolean managed = true;
208 if (managedString != null && !managedString.isEmpty()) {
209 managed = Boolean.parseBoolean(managedString);
212 String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
213 + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
215 // default is to use HTTP if no https property exists
216 boolean useHttps = false;
217 if (useHttpsString != null && !useHttpsString.isEmpty()) {
218 useHttps = Boolean.parseBoolean(useHttpsString);
222 String allowSelfSignedCertsString =
223 properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
224 + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
226 // default is to disallow self-signed certs
227 boolean allowSelfSignedCerts = false;
228 if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
229 allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
232 UebTopicSink uebTopicWriter = this.build(serverList, topic, apiKey, apiSecret, partitionKey, managed,
233 useHttps, allowSelfSignedCerts);
234 newUebTopicSinks.add(uebTopicWriter);
236 return newUebTopicSinks;
241 public void destroy(String topic) {
243 if (topic == null || topic.isEmpty()) {
244 throw new IllegalArgumentException(MISSING_TOPIC);
247 UebTopicSink uebTopicWriter;
248 synchronized (this) {
249 if (!uebTopicSinks.containsKey(topic)) {
253 uebTopicWriter = uebTopicSinks.remove(topic);
256 uebTopicWriter.shutdown();
260 public void destroy() {
261 List<UebTopicSink> writers = this.inventory();
262 for (UebTopicSink writer : writers) {
266 synchronized (this) {
267 this.uebTopicSinks.clear();
272 public UebTopicSink get(String topic) {
274 if (topic == null || topic.isEmpty()) {
275 throw new IllegalArgumentException(MISSING_TOPIC);
278 synchronized (this) {
279 if (uebTopicSinks.containsKey(topic)) {
280 return uebTopicSinks.get(topic);
282 throw new IllegalStateException("UebTopicSink for " + topic + " not found");
288 public synchronized List<UebTopicSink> inventory() {
289 return new ArrayList<>(this.uebTopicSinks.values());
294 public String toString() {
295 StringBuilder builder = new StringBuilder();
296 builder.append("IndexedUebTopicSinkFactory []");
297 return builder.toString();