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 * servers list of servers
47 * apiSecret API Secret
48 * partitionKey Consumer Group
49 * managed is this sink endpoint managed?
50 * @param busTopicParams parameter object
51 * @return an UEB Topic Sink
52 * @throws IllegalArgumentException if invalid parameters are present
54 UebTopicSink build(BusTopicParams busTopicParams);
57 * Creates an UEB Topic Writer based on properties files.
59 * @param properties Properties containing initialization values
61 * @return an UEB Topic Writer
62 * @throws IllegalArgumentException if invalid parameters are present
64 List<UebTopicSink> build(Properties properties);
67 * Instantiates a new UEB Topic Writer.
69 * @param servers list of servers
70 * @param topic topic name
72 * @return an UEB Topic Writer
73 * @throws IllegalArgumentException if invalid parameters are present
75 UebTopicSink build(List<String> servers, String topic);
78 * Destroys an UEB Topic Writer based on a topic.
80 * @param topic topic name
81 * @throws IllegalArgumentException if invalid parameters are present
83 void destroy(String topic);
86 * Destroys all UEB Topic Writers.
91 * gets an UEB Topic Writer based on topic name.
93 * @param topic the topic name
95 * @return an UEB Topic Writer with topic name
96 * @throws IllegalArgumentException if an invalid topic is provided
97 * @throws IllegalStateException if the UEB Topic Reader is an incorrect state
99 UebTopicSink get(String topic);
102 * Provides a snapshot of the UEB Topic Writers.
104 * @return a list of the UEB Topic Writers
106 List<UebTopicSink> inventory();
110 /* ------------- implementation ----------------- */
113 * Factory of UEB Reader Topics indexed by topic name.
115 class IndexedUebTopicSinkFactory implements UebTopicSinkFactory {
116 private static final String MISSING_TOPIC = "A topic must be provided";
121 private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSinkFactory.class);
124 * UEB Topic Name Index.
126 protected HashMap<String, UebTopicSink> uebTopicSinks = new HashMap<>();
129 public UebTopicSink build(BusTopicParams busTopicParams) {
131 if (busTopicParams.getServers() == null || busTopicParams.getServers().isEmpty()) {
132 throw new IllegalArgumentException("UEB Server(s) must be provided");
135 if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
136 throw new IllegalArgumentException(MISSING_TOPIC);
139 synchronized (this) {
140 if (uebTopicSinks.containsKey(busTopicParams.getTopic())) {
141 return uebTopicSinks.get(busTopicParams.getTopic());
144 UebTopicSink uebTopicWriter = new InlineUebTopicSink(busTopicParams);
146 if (busTopicParams.isManaged()) {
147 uebTopicSinks.put(busTopicParams.getTopic(), uebTopicWriter);
150 return uebTopicWriter;
156 public UebTopicSink build(List<String> servers, String topic) {
157 return this.build(BusTopicParams.builder()
162 .allowSelfSignedCerts(false)
168 public List<UebTopicSink> build(Properties properties) {
170 String writeTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS);
171 if (writeTopics == null || writeTopics.isEmpty()) {
172 logger.info("{}: no topic for UEB Sink", this);
173 return new ArrayList<>();
176 List<String> writeTopicList = new ArrayList<>(Arrays.asList(writeTopics.split("\\s*,\\s*")));
177 List<UebTopicSink> newUebTopicSinks = new ArrayList<>();
178 synchronized (this) {
179 for (String topic : writeTopicList) {
180 if (this.uebTopicSinks.containsKey(topic)) {
181 newUebTopicSinks.add(this.uebTopicSinks.get(topic));
185 String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
186 + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
187 if (servers == null || servers.isEmpty()) {
188 logger.error("{}: no UEB servers configured for sink {}", this, topic);
192 List<String> serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
194 String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
195 + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
196 String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
197 + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
198 String partitionKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
199 + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX);
201 String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
202 + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
203 boolean managed = true;
204 if (managedString != null && !managedString.isEmpty()) {
205 managed = Boolean.parseBoolean(managedString);
208 String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
209 + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
211 // default is to use HTTP if no https property exists
212 boolean useHttps = false;
213 if (useHttpsString != null && !useHttpsString.isEmpty()) {
214 useHttps = Boolean.parseBoolean(useHttpsString);
218 String allowSelfSignedCertsString =
219 properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
220 + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
222 // default is to disallow self-signed certs
223 boolean allowSelfSignedCerts = false;
224 if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
225 allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
228 UebTopicSink uebTopicWriter = this.build(BusTopicParams.builder()
232 .apiSecret(apiSecret)
233 .partitionId(partitionKey)
236 .allowSelfSignedCerts(allowSelfSignedCerts)
238 newUebTopicSinks.add(uebTopicWriter);
240 return newUebTopicSinks;
245 public void destroy(String topic) {
247 if (topic == null || topic.isEmpty()) {
248 throw new IllegalArgumentException(MISSING_TOPIC);
251 UebTopicSink uebTopicWriter;
252 synchronized (this) {
253 if (!uebTopicSinks.containsKey(topic)) {
257 uebTopicWriter = uebTopicSinks.remove(topic);
260 uebTopicWriter.shutdown();
264 public void destroy() {
265 List<UebTopicSink> writers = this.inventory();
266 for (UebTopicSink writer : writers) {
270 synchronized (this) {
271 this.uebTopicSinks.clear();
276 public UebTopicSink get(String topic) {
278 if (topic == null || topic.isEmpty()) {
279 throw new IllegalArgumentException(MISSING_TOPIC);
282 synchronized (this) {
283 if (uebTopicSinks.containsKey(topic)) {
284 return uebTopicSinks.get(topic);
286 throw new IllegalStateException("UebTopicSink for " + topic + " not found");
292 public synchronized List<UebTopicSink> inventory() {
293 return new ArrayList<>(this.uebTopicSinks.values());
298 public String toString() {
299 StringBuilder builder = new StringBuilder();
300 builder.append("IndexedUebTopicSinkFactory []");
301 return builder.toString();