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

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