appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/EventHandlerImpl.java

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