gerrit.onap Code Review - appc.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * APPC
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * Copyright (C) 2017 Amdocs
   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  * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  21  */
  22
  23 package org.openecomp.appc.listener.impl;
  24 import com.att.eelf.configuration.EELFLogger;
  25 import com.att.eelf.configuration.EELFManager;
  26
  27 import org.openecomp.appc.adapter.factory.DmaapMessageAdapterFactoryImpl;
  28 import org.openecomp.appc.adapter.factory.MessageService;
  29 import org.openecomp.appc.adapter.message.Consumer;
  30 import org.openecomp.appc.adapter.message.MessageAdapterFactory;
  31 import org.openecomp.appc.adapter.message.Producer;
  32 import org.openecomp.appc.listener.EventHandler;
  33 import org.openecomp.appc.listener.ListenerProperties;
  34 import org.openecomp.appc.listener.util.Mapper;
  35 import org.openecomp.appc.logging.LoggingConstants;
  36 import org.osgi.framework.BundleContext;
  37 import org.osgi.framework.FrameworkUtil;
  38 import org.osgi.framework.ServiceReference;
  39 import org.slf4j.MDC;
  40
  41 import java.util.*;
  42
  43 /**
  44  * This class is a wrapper for the DMaaP client provided in appc-dmaap-adapter. Its aim is to ensure that only well formed
  45  * messages are sent and received on DMaaP.
  46  *
  47  */
  48 public class EventHandlerImpl implements EventHandler {
  49
  50     private final EELFLogger LOG = EELFManager.getInstance().getLogger(EventHandlerImpl.class);
  51
  52     /*
  53      * The amount of time in seconds to keep a connection to a topic open while waiting for data
  54      */
  55     private int READ_TIMEOUT = 60;
  56
  57     /*
  58      * The pool of hosts to query against
  59      */
  60     private Collection<String> pool;
  61
  62     /*
  63      * The topic to read messages from
  64      */
  65     private String readTopic;
  66
  67     /*
  68      * The topic to write messages to
  69      */
  70     private Set<String> writeTopics;
  71
  72     /*
  73      * The client (group) name to use for reading messages
  74      */
  75     private String clientName;
  76
  77     /*
  78      * The id of the client (group) that is reading messages
  79      */
  80     private String clientId;
  81
  82     /*
  83      * The api public key to use for authentication
  84      */
  85     private String apiKey;
  86
  87     /*
  88      * The api secret key to use for authentication
  89      */
  90     private String apiSecret;
  91
  92     /*
  93      * A json object containing filter arguments.
  94      */
  95     private String filter_json;
  96
  97     private MessageService messageService;
  98
  99     private Consumer reader = null;
 100     private Producer producer = null;
 101
 102     public EventHandlerImpl(ListenerProperties props) {
 103         pool = new HashSet<String>();
 104         writeTopics = new HashSet<String>();
 105
 106         if (props != null) {
 107             readTopic = props.getProperty(ListenerProperties.KEYS.TOPIC_READ);
 108             clientName = props.getProperty(ListenerProperties.KEYS.CLIENT_NAME, "APP-C");
 109             clientId = props.getProperty(ListenerProperties.KEYS.CLIENT_ID, "0");
 110             apiKey = props.getProperty(ListenerProperties.KEYS.AUTH_USER_KEY);
 111             apiSecret = props.getProperty(ListenerProperties.KEYS.AUTH_SECRET_KEY);
 112
 113             filter_json = props.getProperty(ListenerProperties.KEYS.TOPIC_READ_FILTER);
 114
 115             READ_TIMEOUT = Integer
 116                 .valueOf(props.getProperty(ListenerProperties.KEYS.TOPIC_READ_TIMEOUT, String.valueOf(READ_TIMEOUT)));
 117
 118             String hostnames = props.getProperty(ListenerProperties.KEYS.HOSTS);
 119             if (hostnames != null && !hostnames.isEmpty()) {
 120                 for (String name : hostnames.split(",")) {
 121                     pool.add(name);
 122                 }
 123             }
 124
 125             String writeTopicStr = props.getProperty(ListenerProperties.KEYS.TOPIC_WRITE);
 126             if (writeTopicStr != null) {
 127                 for (String topic : writeTopicStr.split(",")) {
 128                     writeTopics.add(topic);
 129                 }
 130             }
 131
 132             messageService = MessageService.parse(props.getProperty(ListenerProperties.KEYS.MESSAGE_SERVICE));
 133
 134             LOG.info(String.format(
 135                 "Configured to use %s client on host pool [%s]. Reading from [%s] filtered by %s. Wriring to [%s]. Authenticated using %s",
 136                 messageService, hostnames, readTopic, filter_json, writeTopics, apiKey));
 137         }
 138     }
 139
 140     @Override
 141     public List<String> getIncomingEvents() {
 142         return getIncomingEvents(1000);
 143     }
 144
 145     @Override
 146     public List<String> getIncomingEvents(int limit) {
 147         List<String> out = new ArrayList<String>();
 148         LOG.info(String.format("Getting up to %d incoming events", limit));
 149         // reuse the consumer object instead of creating a new one every time
 150         if (reader == null) {
 151                 LOG.info("Getting Consumer...");
 152                 reader = getConsumer();
 153         }
 154
 155         List<String> items = null;
 156         try{
 157             items = reader.fetch(READ_TIMEOUT * 1000, limit);
 158         }catch(Error r){
 159             LOG.error("EvenHandlerImpl.getIncomingEvents",r);
 160         }
 161
 162
 163         for (String item : items) {
 164             out.add(item);
 165         }
 166         LOG.info(String.format("Read %d messages from %s as %s/%s.", out.size(), readTopic, clientName, clientId));
 167         return out;
 168     }
 169
 170     @Override
 171     public <T> List<T> getIncomingEvents(Class<T> cls) {
 172         return getIncomingEvents(cls, 1000);
 173     }
 174
 175     @Override
 176     public <T> List<T> getIncomingEvents(Class<T> cls, int limit) {
 177         List<String> incomingStrings = getIncomingEvents(limit);
 178         return Mapper.mapList(incomingStrings, cls);
 179     }
 180
 181     @Override
 182     public void postStatus(String event) {
 183         postStatus(null, event);
 184     }
 185
 186     @Override
 187     public void postStatus(String partition, String event) {
 188         LOG.debug(String.format("Posting Message [%s]", event));
 189         if (producer == null) {
 190                 LOG.info("Getting Producer...");
 191                 producer = getProducer();
 192         }
 193         producer.post(partition, event);
 194     }
 195
 196     /**
 197      * Returns a consumer object for direct access to our Cambria consumer interface
 198      *
 199      * @return An instance of the consumer interface
 200      */
 201     protected Consumer getConsumer() {
 202         LOG.debug(String.format("Getting Consumer: %s  %s/%s/%s", pool, readTopic, clientName, clientId));
 203         if (filter_json == null && writeTopics.contains(readTopic)) {
 204             LOG.error(
 205                 "*****We will be writing and reading to the same topic without a filter. This will cause an infinite loop.*****");
 206         }
 207
 208         Consumer out=null;
 209         BundleContext ctx = FrameworkUtil.getBundle(EventHandlerImpl.class).getBundleContext();
 210         if (ctx != null) {
 211                 ServiceReference svcRef = ctx.getServiceReference(MessageAdapterFactory.class.getName());
 212                 if (svcRef != null) {
 213                         try{
 214                             out = ((MessageAdapterFactory) ctx.getService(svcRef)).createConsumer(pool, readTopic, clientName, clientId, filter_json, apiKey, apiSecret);
 215                         }catch(Error e){
 216                             //TODO:create eelf message
 217                             LOG.error("EvenHandlerImp.getConsumer calling MessageAdapterFactor.createConsumer",e);
 218                         }
 219                         for (String url : pool) {
 220                             if (url.contains("3905") || url.contains("https")) {
 221                                 out.useHttps(true);
 222                                 break;
 223                             }
 224                         }
 225                 }
 226         }
 227         return out;
 228     }
 229
 230     /**
 231      * Returns a consumer object for direct access to our Cambria producer interface
 232      *
 233      * @return An instance of the producer interface
 234      */
 235     protected Producer getProducer() {
 236         LOG.debug(String.format("Getting Producer: %s  %s", pool, readTopic));
 237
 238         Producer out = null;
 239         BundleContext ctx = FrameworkUtil.getBundle(EventHandlerImpl.class).getBundleContext();
 240         if (ctx != null) {
 241                 ServiceReference svcRef = ctx.getServiceReference(MessageAdapterFactory.class.getName());
 242                 if (svcRef != null) {
 243                         out = ((MessageAdapterFactory) ctx.getService(svcRef)).createProducer(pool, writeTopics,apiKey, apiSecret);
 244                         for (String url : pool) {
 245                             if (url.contains("3905") || url.contains("https")) {
 246                                 out.useHttps(true);
 247                                 break;
 248                             }
 249                         }
 250                 }
 251         }
 252         return out;
 253     }
 254
 255     @Override
 256     public void closeClients() {
 257         LOG.debug("Closing Consumer and Producer DMaaP clients");
 258         if (reader != null) {
 259                 reader.close();
 260         }
 261         if (producer != null) {
 262                 producer.close();
 263         }
 264     }
 265
 266     @Override
 267     public String getClientId() {
 268         return clientId;
 269     }
 270
 271     @Override
 272     public void setClientId(String clientId) {
 273         this.clientId = clientId;
 274     }
 275
 276     @Override
 277     public String getClientName() {
 278         return clientName;
 279     }
 280
 281     @Override
 282     public void setClientName(String clientName) {
 283         this.clientName = clientName;
 284         MDC.put(LoggingConstants.MDCKeys.PARTNER_NAME, clientName);
 285     }
 286
 287     @Override
 288     public void addToPool(String hostname) {
 289         pool.add(hostname);
 290     }
 291
 292     @Override
 293     public Collection<String> getPool() {
 294         return pool;
 295     }
 296
 297     @Override
 298     public void removeFromPool(String hostname) {
 299         pool.remove(hostname);
 300     }
 301
 302     @Override
 303     public String getReadTopic() {
 304         return readTopic;
 305     }
 306
 307     @Override
 308     public void setReadTopic(String readTopic) {
 309         this.readTopic = readTopic;
 310     }
 311
 312     @Override
 313     public Set<String> getWriteTopics() {
 314         return writeTopics;
 315     }
 316
 317     @Override
 318     public void setWriteTopics(Set<String> writeTopics) {
 319         this.writeTopics = writeTopics;
 320     }
 321
 322     @Override
 323     public void clearCredentials() {
 324         apiKey = null;
 325         apiSecret = null;
 326     }
 327
 328     @Override
 329     public void setCredentials(String key, String secret) {
 330         apiKey = key;
 331         apiSecret = secret;
 332     }
 333 }