policy-endpoints/src/main/java/org/onap/policy/drools/event/comm/bus/NoopTopicSinkFactory.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * policy-endpoints
   4  * ================================================================================
   5  * Copyright (C) 2017 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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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=========================================================
  19  */
  20 package org.onap.policy.drools.event.comm.bus;
  21
  22 import java.util.ArrayList;
  23 import java.util.Arrays;
  24 import java.util.HashMap;
  25 import java.util.List;
  26 import java.util.Properties;
  27
  28 import org.onap.policy.drools.properties.PolicyProperties;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 /**
  33  * Noop Topic Sink Factory
  34  */
  35 public interface NoopTopicSinkFactory {
  36
  37   /**
  38    * Creates noop topic sinks based on properties files
  39    *
  40    * @param properties Properties containing initialization values
  41    *
  42    * @return a noop topic sink
  43    * @throws IllegalArgumentException if invalid parameters are present
  44    */
  45   public List<NoopTopicSink> build(Properties properties);
  46
  47   /**
  48    * builds a noop sink
  49    *
  50    * @param servers list of servers
  51    * @param topic topic name
  52    * @param managed is this sink endpoint managed?
  53    * @return a noop topic sink
  54    * @throws IllegalArgumentException if invalid parameters are present
  55    */
  56   public NoopTopicSink build(List<String> servers, String topic, boolean managed);
  57
  58   /**
  59    * Destroys a sink based on the topic
  60    *
  61    * @param topic topic name
  62    * @throws IllegalArgumentException if invalid parameters are present
  63    */
  64   public void destroy(String topic);
  65
  66   /**
  67    * gets a sink based on topic name
  68    *
  69    * @param topic the topic name
  70    *
  71    * @return a sink with topic name
  72    * @throws IllegalArgumentException if an invalid topic is provided
  73    * @throws IllegalStateException if the sink is in an incorrect state
  74    */
  75   public NoopTopicSink get(String topic);
  76
  77   /**
  78    * Provides a snapshot of the UEB Topic Writers
  79    *
  80    * @return a list of the UEB Topic Writers
  81    */
  82   public List<NoopTopicSink> inventory();
  83
  84   /**
  85    * Destroys all sinks
  86    */
  87   public void destroy();
  88 }
  89
  90
  91 /* ------------- implementation ----------------- */
  92
  93 /**
  94  * Factory of noop sinks
  95  */
  96 class IndexedNoopTopicSinkFactory implements NoopTopicSinkFactory {
  97   /**
  98    * Logger
  99    */
 100   private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSinkFactory.class);
 101
 102   /**
 103    * noop topic sinks map
 104    */
 105   protected HashMap<String, NoopTopicSink> noopTopicSinks = new HashMap<>();
 106
 107   @Override
 108   public List<NoopTopicSink> build(Properties properties) {
 109
 110     final String sinkTopics = properties.getProperty(PolicyProperties.PROPERTY_NOOP_SINK_TOPICS);
 111     if (sinkTopics == null || sinkTopics.isEmpty()) {
 112       logger.info("{}: no topic for noop sink", this);
 113       return new ArrayList<>();
 114     }
 115
 116     final List<String> sinkTopicList =
 117         new ArrayList<>(Arrays.asList(sinkTopics.split("\\s*,\\s*")));
 118     final List<NoopTopicSink> newSinks = new ArrayList<NoopTopicSink>();
 119     synchronized (this) {
 120       for (final String topic : sinkTopicList) {
 121         if (this.noopTopicSinks.containsKey(topic)) {
 122           newSinks.add(this.noopTopicSinks.get(topic));
 123           continue;
 124         }
 125
 126         String servers = properties.getProperty(PolicyProperties.PROPERTY_NOOP_SINK_TOPICS + "."
 127             + topic + PolicyProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
 128
 129         if (servers == null || servers.isEmpty())
 130           servers = "noop";
 131
 132         final List<String> serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
 133
 134         final String managedString =
 135             properties.getProperty(PolicyProperties.PROPERTY_NOOP_SINK_TOPICS + "." + topic
 136                 + PolicyProperties.PROPERTY_MANAGED_SUFFIX);
 137         boolean managed = true;
 138         if (managedString != null && !managedString.isEmpty()) {
 139           managed = Boolean.parseBoolean(managedString);
 140         }
 141
 142         final NoopTopicSink noopSink = this.build(serverList, topic, managed);
 143         newSinks.add(noopSink);
 144       }
 145       return newSinks;
 146     }
 147   }
 148
 149   @Override
 150   public NoopTopicSink build(List<String> servers, String topic, boolean managed) {
 151     if (servers == null) {
 152       servers = new ArrayList<>();
 153     }
 154
 155     if (servers.isEmpty()) {
 156       servers.add("noop");
 157     }
 158
 159     if (topic == null || topic.isEmpty()) {
 160       throw new IllegalArgumentException("A topic must be provided");
 161     }
 162
 163     synchronized (this) {
 164       if (this.noopTopicSinks.containsKey(topic)) {
 165         return this.noopTopicSinks.get(topic);
 166       }
 167
 168       final NoopTopicSink sink = new NoopTopicSink(servers, topic);
 169
 170       if (managed)
 171         this.noopTopicSinks.put(topic, sink);
 172
 173       return sink;
 174     }
 175   }
 176
 177   @Override
 178   public void destroy(String topic) {
 179     if (topic == null || topic.isEmpty()) {
 180       throw new IllegalArgumentException("A topic must be provided");
 181     }
 182
 183     NoopTopicSink noopSink;
 184     synchronized (this) {
 185       if (!this.noopTopicSinks.containsKey(topic)) {
 186         return;
 187       }
 188
 189       noopSink = this.noopTopicSinks.remove(topic);
 190     }
 191
 192     noopSink.shutdown();
 193   }
 194
 195   @Override
 196   public void destroy() {
 197     final List<NoopTopicSink> sinks = this.inventory();
 198     for (final NoopTopicSink sink : sinks) {
 199       sink.shutdown();
 200     }
 201
 202     synchronized (this) {
 203       this.noopTopicSinks.clear();
 204     }
 205   }
 206
 207   @Override
 208   public NoopTopicSink get(String topic) {
 209     if (topic == null || topic.isEmpty()) {
 210       throw new IllegalArgumentException("A topic must be provided");
 211     }
 212
 213     synchronized (this) {
 214       if (this.noopTopicSinks.containsKey(topic)) {
 215         return this.noopTopicSinks.get(topic);
 216       } else {
 217         throw new IllegalStateException("DmaapTopicSink for " + topic + " not found");
 218       }
 219     }
 220   }
 221
 222   @Override
 223   public List<NoopTopicSink> inventory() {
 224     return new ArrayList<>(this.noopTopicSinks.values());
 225   }
 226 }