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                         if( out != null ) {
 222                                 for (String url : pool) {
 223                                         if (url.contains("3905") || url.contains("https")) {
 224                                                 out.useHttps(true);
 225                                                 break;
 226                                         }
 227                             }
 228                         }
 229                 }
 230         }
 231         return out;
 232     }
 233
 234     /**
 235      * Returns a consumer object for direct access to our Cambria producer interface
 236      *
 237      * @return An instance of the producer interface
 238      */
 239     protected Producer getProducer() {
 240         LOG.debug(String.format("Getting Producer: %s  %s", pool, readTopic));
 241
 242         Producer out = null;
 243         BundleContext ctx = FrameworkUtil.getBundle(EventHandlerImpl.class).getBundleContext();
 244         if (ctx != null) {
 245                 ServiceReference svcRef = ctx.getServiceReference(MessageAdapterFactory.class.getName());
 246                 if (svcRef != null) {
 247                         out = ((MessageAdapterFactory) ctx.getService(svcRef)).createProducer(pool, writeTopics,apiKey, apiSecret);
 248                         if( out != null ) {
 249                                 for (String url : pool) {
 250                                 if (url.contains("3905") || url.contains("https")) {
 251                                     out.useHttps(true);
 252                                     break;
 253                                 }
 254                             }
 255                         }
 256                 }
 257         }
 258         return out;
 259     }
 260
 261     @Override
 262     public void closeClients() {
 263         LOG.debug("Closing Consumer and Producer DMaaP clients");
 264         if (reader != null) {
 265                 reader.close();
 266         }
 267         if (producer != null) {
 268                 producer.close();
 269         }
 270     }
 271
 272     @Override
 273     public String getClientId() {
 274         return clientId;
 275     }
 276
 277     @Override
 278     public void setClientId(String clientId) {
 279         this.clientId = clientId;
 280     }
 281
 282     @Override
 283     public String getClientName() {
 284         return clientName;
 285     }
 286
 287     @Override
 288     public void setClientName(String clientName) {
 289         this.clientName = clientName;
 290         MDC.put(LoggingConstants.MDCKeys.PARTNER_NAME, clientName);
 291     }
 292
 293     @Override
 294     public void addToPool(String hostname) {
 295         pool.add(hostname);
 296     }
 297
 298     @Override
 299     public Collection<String> getPool() {
 300         return pool;
 301     }
 302
 303     @Override
 304     public void removeFromPool(String hostname) {
 305         pool.remove(hostname);
 306     }
 307
 308     @Override
 309     public String getReadTopic() {
 310         return readTopic;
 311     }
 312
 313     @Override
 314     public void setReadTopic(String readTopic) {
 315         this.readTopic = readTopic;
 316     }
 317
 318     @Override
 319     public Set<String> getWriteTopics() {
 320         return writeTopics;
 321     }
 322
 323     @Override
 324     public void setWriteTopics(Set<String> writeTopics) {
 325         this.writeTopics = writeTopics;
 326     }
 327
 328     @Override
 329     public void clearCredentials() {
 330         apiKey = null;
 331         apiSecret = null;
 332     }
 333
 334     @Override
 335     public void setCredentials(String key, String secret) {
 336         apiKey = key;
 337         apiSecret = secret;
 338     }
 339 }