gerrit.onap Code Review - ccsdk/features.git/blob

   1 /*******************************************************************************
   2  * ============LICENSE_START========================================================================
   3  * ONAP : ccsdk feature sdnr wt mountpoint-registrar
   4  * =================================================================================================
   5  * Copyright (C) 2019 highstreet technologies GmbH Intellectual Property. All rights reserved.
   6  * =================================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   8  * in compliance with the License. You may obtain a copy of the License at
   9  *
  10  * http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software distributed under the License
  13  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  14  * or implied. See the License for the specific language governing permissions and limitations under
  15  * the License.
  16  * ============LICENSE_END==========================================================================
  17  ******************************************************************************/
  18
  19 package org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.impl;
  20
  21 import java.util.LinkedList;
  22 import java.util.List;
  23 import java.util.Map;
  24 import java.util.Properties;
  25
  26 import org.onap.ccsdk.features.sdnr.wt.common.configuration.Configuration;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29
  30 public class DMaaPVESMsgConsumerMain implements Runnable {
  31
  32         private static final Logger LOG = LoggerFactory.getLogger(DMaaPVESMsgConsumerMain.class);
  33
  34         boolean threadsRunning = false;
  35         static List<DMaaPVESMsgConsumer> consumers = new LinkedList<>();
  36         public GeneralConfig config;
  37         public PNFRegistrationConfig pnfRegistrationConfig;
  38         public FaultConfig faultConfig;
  39
  40         public DMaaPVESMsgConsumerMain(Map<String, Configuration> configMap) {
  41                 configMap.forEach((k, v) -> initialize(k, v));
  42         }
  43
  44         public void initialize(String domain, Configuration domainConfig) {
  45                 LOG.debug("In initialize method : Domain = {} and domainConfig = {}",domain,domainConfig);
  46                 String consumerClass = null;
  47                 Properties consumerProperties = new Properties();
  48                 if (domain.equalsIgnoreCase("pnfregistration")) {
  49                         this.pnfRegistrationConfig = (PNFRegistrationConfig) domainConfig;
  50
  51                         consumerClass = pnfRegistrationConfig.getConsumerClass();
  52                         LOG.debug("Consumer class = "+consumerClass);
  53
  54                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE, pnfRegistrationConfig.getTransportType());
  55                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_HOST_PORT, pnfRegistrationConfig.getHostPort());
  56                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE, pnfRegistrationConfig.getContenttype());
  57                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_GROUP, pnfRegistrationConfig.getConsumerGroup());
  58                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_ID, pnfRegistrationConfig.getConsumerId());
  59                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TOPIC, pnfRegistrationConfig.getTopic());
  60                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TIMEOUT, pnfRegistrationConfig.getTimeout());
  61                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_LIMIT, pnfRegistrationConfig.getLimit());
  62                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE, pnfRegistrationConfig.getFetchPause());
  63                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_PROTOCOL, pnfRegistrationConfig.getProtocol());
  64                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_USERNAME, pnfRegistrationConfig.getUsername());
  65                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_PASSWORD, pnfRegistrationConfig.getPassword());
  66                 } else if (domain.equalsIgnoreCase("fault")) {
  67                         this.faultConfig = (FaultConfig) domainConfig;
  68                         consumerClass = faultConfig.getConsumerClass();
  69                         LOG.debug("Consumer class = {}",consumerClass);
  70                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE, faultConfig.getTransportType());
  71                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_HOST_PORT, faultConfig.getHostPort());
  72                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE, faultConfig.getContenttype());
  73                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_GROUP, faultConfig.getConsumerGroup());
  74                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_ID, faultConfig.getConsumerId());
  75                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TOPIC, faultConfig.getTopic());
  76                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TIMEOUT, faultConfig.getTimeout());
  77                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_LIMIT, faultConfig.getLimit());
  78                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE, faultConfig.getFetchPause());
  79                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_PROTOCOL, faultConfig.getProtocol());
  80                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_USERNAME, faultConfig.getUsername());
  81                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_PASSWORD, faultConfig.getPassword());
  82                 }
  83
  84                 if (consumerClass != null) {
  85                         LOG.info("Calling createConsumer : {}",consumerClass);
  86                         threadsRunning = createConsumer(consumerClass, consumerProperties);
  87                 }
  88
  89         }
  90
  91         private static boolean updateThreadState(List<DMaaPVESMsgConsumer> consumers) {
  92                 boolean threadsRunning = false;
  93                 for (DMaaPVESMsgConsumer consumer : consumers) {
  94                         if (consumer.isRunning()) {
  95                                 threadsRunning = true;
  96                         }
  97                 }
  98                 return threadsRunning;
  99         }
 100
 101         static boolean createConsumer(String consumerClassName, Properties properties) {
 102                 Class<?> consumerClass = null;
 103
 104                 try {
 105                         consumerClass = Class.forName(consumerClassName);
 106                 } catch (Exception e) {
 107                         LOG.error("Could not find DMaap VES Message consumer class {}", consumerClassName, e);
 108                 }
 109
 110                 if (consumerClass != null) {
 111                         LOG.debug("Calling handleConsumerClass");
 112                         handleConsumerClass(consumerClass, consumerClassName, properties, consumers);
 113                 }
 114                 return !consumers.isEmpty();
 115         }
 116
 117         private static boolean handleConsumerClass(Class<?> consumerClass, String consumerClassName, Properties properties, List<DMaaPVESMsgConsumer> consumers) {
 118                 DMaaPVESMsgConsumer consumer = null;
 119
 120                 try {
 121                         consumer = (DMaaPVESMsgConsumer) consumerClass.newInstance();
 122                         LOG.debug("Successfully created an instance of consumerClass : {}",consumerClassName);
 123                 } catch (Exception e) {
 124                         LOG.error("Could not create consumer from class {}",consumerClassName, e);
 125                 }
 126
 127                 if (consumer != null) {
 128                         LOG.info("Initializing consumer {}({})", consumerClassName, properties);
 129                         consumer.init(properties);
 130
 131                         if (consumer.isReady()) {
 132                                 Thread consumerThread = new Thread(consumer);
 133                                 consumerThread.start();
 134                                 consumers.add(consumer);
 135
 136                                 LOG.info("Started consumer thread ({} : {})", consumerClassName,
 137                                                 properties);
 138                                 return true;
 139                         } else {
 140                                 LOG.debug("Consumer {} is not ready", consumerClassName);
 141                         }
 142                 }
 143                 return false;
 144         }
 145
 146         public void run() {
 147                 while (threadsRunning) {
 148                         threadsRunning = updateThreadState(consumers);
 149                         if (!threadsRunning) {
 150                                 break;
 151                         }
 152
 153                         try {
 154                                 Thread.sleep(10000);
 155                         } catch (InterruptedException e) {
 156                                 LOG.error(e.getLocalizedMessage(), e);
 157                         }
 158                 }
 159
 160                 LOG.info("No listener threads running - exiting");
 161         }
 162
 163         public static List<DMaaPVESMsgConsumer> getConsumers() {
 164                 return consumers;
 165         }
 166
 167 }