gerrit.onap Code Review - policy/common.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * policy-endpoints
   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
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  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=========================================================
  20  */
  21
  22 package org.onap.policy.common.endpoints.event.comm.bus;
  23
  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;
  29
  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;
  35
  36 /**
  37  * UEB Topic Sink Factory.
  38  */
  39 public interface UebTopicSinkFactory {
  40
  41     /**
  42      * Instantiates a new UEB Topic Writer.
  43      *
  44      * @param busTopicParams parameters object
  45      * @return an UEB Topic Sink
  46      */
  47     UebTopicSink build(BusTopicParams busTopicParams);
  48
  49     /**
  50      * Creates an UEB Topic Writer based on properties files.
  51      *
  52      * @param properties Properties containing initialization values
  53      *
  54      * @return an UEB Topic Writer
  55      * @throws IllegalArgumentException if invalid parameters are present
  56      */
  57     List<UebTopicSink> build(Properties properties);
  58
  59     /**
  60      * Instantiates a new UEB Topic Writer.
  61      *
  62      * @param servers list of servers
  63      * @param topic topic name
  64      *
  65      * @return an UEB Topic Writer
  66      * @throws IllegalArgumentException if invalid parameters are present
  67      */
  68     UebTopicSink build(List<String> servers, String topic);
  69
  70     /**
  71      * Destroys an UEB Topic Writer based on a topic.
  72      *
  73      * @param topic topic name
  74      * @throws IllegalArgumentException if invalid parameters are present
  75      */
  76     void destroy(String topic);
  77
  78     /**
  79      * Destroys all UEB Topic Writers.
  80      */
  81     void destroy();
  82
  83     /**
  84      * gets an UEB Topic Writer based on topic name.
  85      *
  86      * @param topic the topic name
  87      *
  88      * @return an UEB Topic Writer with topic name
  89      * @throws IllegalArgumentException if an invalid topic is provided
  90      * @throws IllegalStateException if the UEB Topic Reader is an incorrect state
  91      */
  92     UebTopicSink get(String topic);
  93
  94     /**
  95      * Provides a snapshot of the UEB Topic Writers.
  96      *
  97      * @return a list of the UEB Topic Writers
  98      */
  99     List<UebTopicSink> inventory();
 100 }
 101
 102
 103 /* ------------- implementation ----------------- */
 104
 105 /**
 106  * Factory of UEB Reader Topics indexed by topic name.
 107  */
 108 class IndexedUebTopicSinkFactory implements UebTopicSinkFactory {
 109     private static final String MISSING_TOPIC = "A topic must be provided";
 110
 111     /**
 112      * Logger.
 113      */
 114     private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSinkFactory.class);
 115
 116     /**
 117      * UEB Topic Name Index.
 118      */
 119     protected HashMap<String, UebTopicSink> uebTopicSinks = new HashMap<>();
 120
 121     @Override
 122     public UebTopicSink build(BusTopicParams busTopicParams) {
 123
 124         if (busTopicParams.getServers() == null || busTopicParams.getServers().isEmpty()) {
 125             throw new IllegalArgumentException("UEB Server(s) must be provided");
 126         }
 127
 128         if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
 129             throw new IllegalArgumentException(MISSING_TOPIC);
 130         }
 131
 132         synchronized (this) {
 133             if (uebTopicSinks.containsKey(busTopicParams.getTopic())) {
 134                 return uebTopicSinks.get(busTopicParams.getTopic());
 135             }
 136
 137             UebTopicSink uebTopicWriter = makeSink(busTopicParams);
 138
 139             if (busTopicParams.isManaged()) {
 140                 uebTopicSinks.put(busTopicParams.getTopic(), uebTopicWriter);
 141             }
 142
 143             return uebTopicWriter;
 144         }
 145     }
 146
 147
 148     @Override
 149     public UebTopicSink build(List<String> servers, String topic) {
 150         return this.build(BusTopicParams.builder()
 151                 .servers(servers)
 152                 .topic(topic)
 153                 .managed(true)
 154                 .useHttps(false)
 155                 .allowSelfSignedCerts(false)
 156                 .build());
 157     }
 158
 159
 160     @Override
 161     public List<UebTopicSink> build(Properties properties) {
 162
 163         String writeTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS);
 164         if (writeTopics == null || writeTopics.isEmpty()) {
 165             logger.info("{}: no topic for UEB Sink", this);
 166             return new ArrayList<>();
 167         }
 168
 169         List<String> writeTopicList = new ArrayList<>(Arrays.asList(writeTopics.split("\\s*,\\s*")));
 170         List<UebTopicSink> newUebTopicSinks = new ArrayList<>();
 171         synchronized (this) {
 172             for (String topic : writeTopicList) {
 173                 if (this.uebTopicSinks.containsKey(topic)) {
 174                     newUebTopicSinks.add(this.uebTopicSinks.get(topic));
 175                     continue;
 176                 }
 177
 178                 String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
 179                         + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
 180                 if (servers == null || servers.isEmpty()) {
 181                     logger.error("{}: no UEB servers configured for sink {}", this, topic);
 182                     continue;
 183                 }
 184
 185                 final List<String> serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
 186
 187                 final String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS
 188                                 + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
 189                 final String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS
 190                                 + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
 191                 final String partitionKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS
 192                                 + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX);
 193
 194                 String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
 195                         + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
 196                 boolean managed = true;
 197                 if (managedString != null && !managedString.isEmpty()) {
 198                     managed = Boolean.parseBoolean(managedString);
 199                 }
 200
 201                 String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
 202                         + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
 203
 204                 // default is to use HTTP if no https property exists
 205                 boolean useHttps = false;
 206                 if (useHttpsString != null && !useHttpsString.isEmpty()) {
 207                     useHttps = Boolean.parseBoolean(useHttpsString);
 208                 }
 209
 210
 211                 String allowSelfSignedCertsString =
 212                         properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
 213                                 + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
 214
 215                 // default is to disallow self-signed certs
 216                 boolean allowSelfSignedCerts = false;
 217                 if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
 218                     allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
 219                 }
 220
 221                 UebTopicSink uebTopicWriter = this.build(BusTopicParams.builder()
 222                         .servers(serverList)
 223                         .topic(topic)
 224                         .apiKey(apiKey)
 225                         .apiSecret(apiSecret)
 226                         .partitionId(partitionKey)
 227                         .managed(managed)
 228                         .useHttps(useHttps)
 229                         .allowSelfSignedCerts(allowSelfSignedCerts)
 230                         .build());
 231                 newUebTopicSinks.add(uebTopicWriter);
 232             }
 233             return newUebTopicSinks;
 234         }
 235     }
 236
 237     @Override
 238     public void destroy(String topic) {
 239
 240         if (topic == null || topic.isEmpty()) {
 241             throw new IllegalArgumentException(MISSING_TOPIC);
 242         }
 243
 244         UebTopicSink uebTopicWriter;
 245         synchronized (this) {
 246             if (!uebTopicSinks.containsKey(topic)) {
 247                 return;
 248             }
 249
 250             uebTopicWriter = uebTopicSinks.remove(topic);
 251         }
 252
 253         uebTopicWriter.shutdown();
 254     }
 255
 256     @Override
 257     public void destroy() {
 258         List<UebTopicSink> writers = this.inventory();
 259         for (UebTopicSink writer : writers) {
 260             writer.shutdown();
 261         }
 262
 263         synchronized (this) {
 264             this.uebTopicSinks.clear();
 265         }
 266     }
 267
 268     @Override
 269     public UebTopicSink get(String topic) {
 270
 271         if (topic == null || topic.isEmpty()) {
 272             throw new IllegalArgumentException(MISSING_TOPIC);
 273         }
 274
 275         synchronized (this) {
 276             if (uebTopicSinks.containsKey(topic)) {
 277                 return uebTopicSinks.get(topic);
 278             } else {
 279                 throw new IllegalStateException("UebTopicSink for " + topic + " not found");
 280             }
 281         }
 282     }
 283
 284     @Override
 285     public synchronized List<UebTopicSink> inventory() {
 286         return new ArrayList<>(this.uebTopicSinks.values());
 287     }
 288
 289     /**
 290      * Makes a new sink.
 291      *
 292      * @param busTopicParams parameters to use to configure the sink
 293      * @return a new sink
 294      */
 295     protected UebTopicSink makeSink(BusTopicParams busTopicParams) {
 296         return new InlineUebTopicSink(busTopicParams);
 297     }
 298
 299
 300     @Override
 301     public String toString() {
 302         StringBuilder builder = new StringBuilder();
 303         builder.append("IndexedUebTopicSinkFactory []");
 304         return builder.toString();
 305     }
 306
 307 }