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.io.File;
  22 import java.io.FileInputStream;
  23 import java.util.LinkedList;
  24 import java.util.List;
  25 import java.util.Map;
  26 import java.util.Properties;
  27
  28 import org.onap.ccsdk.features.sdnr.wt.common.configuration.Configuration;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 public class DMaaPVESMsgConsumerMain implements Runnable {
  33
  34         private static final Logger LOG = LoggerFactory.getLogger(DMaaPVESMsgConsumerMain.class);
  35
  36         boolean threadsRunning = false;
  37         static List<DMaaPVESMsgConsumer> consumers = new LinkedList<>();
  38         public GeneralConfig config;
  39         public PNFRegistrationConfig pnfRegistrationConfig;
  40         public FaultConfig faultConfig;
  41
  42         public DMaaPVESMsgConsumerMain(Map<String, Configuration> configMap) {
  43                 configMap.forEach((k, v) -> initialize(k, v));
  44         }
  45
  46         public void initialize(String domain, Configuration domainConfig) {
  47                 LOG.debug("In initialize method : Domain = {} and domainConfig = {}",domain,domainConfig);
  48                 String consumerClass = null;
  49                 Properties consumerProperties = new Properties();
  50                 if (domain.equalsIgnoreCase("pnfregistration")) {
  51                         this.pnfRegistrationConfig = (PNFRegistrationConfig) domainConfig;
  52
  53                         consumerClass = pnfRegistrationConfig.getConsumerClass();
  54                         LOG.debug("Consumer class = "+consumerClass);
  55
  56                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE, pnfRegistrationConfig.getTransportType());
  57                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_HOST_PORT, pnfRegistrationConfig.getHostPort());
  58                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE, pnfRegistrationConfig.getContenttype());
  59                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_GROUP, pnfRegistrationConfig.getConsumerGroup());
  60                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_ID, pnfRegistrationConfig.getConsumerId());
  61                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TOPIC, pnfRegistrationConfig.getTopic());
  62                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TIMEOUT, pnfRegistrationConfig.getTimeout());
  63                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_LIMIT, pnfRegistrationConfig.getLimit());
  64                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE, pnfRegistrationConfig.getFetchPause());
  65                 } else if (domain.equalsIgnoreCase("fault")) {
  66                         this.faultConfig = (FaultConfig) domainConfig;
  67                         consumerClass = faultConfig.getConsumerClass();
  68                         LOG.debug("Consumer class = {}",consumerClass);
  69                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE, faultConfig.getTransportType());
  70                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_HOST_PORT, faultConfig.getHostPort());
  71                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE, faultConfig.getContenttype());
  72                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_GROUP, faultConfig.getConsumerGroup());
  73                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_ID, faultConfig.getConsumerId());
  74                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TOPIC, faultConfig.getTopic());
  75                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TIMEOUT, faultConfig.getTimeout());
  76                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_LIMIT, faultConfig.getLimit());
  77                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE, faultConfig.getFetchPause());
  78                 }
  79
  80                 if (consumerClass != null) {
  81                         LOG.info("Calling createConsumer : {}",consumerClass);
  82                         threadsRunning = createConsumer(consumerClass, consumerProperties);
  83                 }
  84
  85         }
  86
  87         private static boolean updateThreadState(List<DMaaPVESMsgConsumer> consumers) {
  88                 boolean threadsRunning = false;
  89                 for (DMaaPVESMsgConsumer consumer : consumers) {
  90                         if (consumer.isRunning()) {
  91                                 threadsRunning = true;
  92                         }
  93                 }
  94                 return threadsRunning;
  95         }
  96
  97         static boolean createConsumer(String consumerClassName, Properties properties) {
  98                 Class<?> consumerClass = null;
  99
 100                 try {
 101                         consumerClass = Class.forName(consumerClassName);
 102                 } catch (Exception e) {
 103                         LOG.error("Could not find DMaap VES Message consumer class {}", consumerClassName, e);
 104                 }
 105
 106                 if (consumerClass != null) {
 107                         LOG.debug("Calling handleConsumerClass");
 108                         handleConsumerClass(consumerClass, consumerClassName, properties, consumers);
 109                 }
 110                 return !consumers.isEmpty();
 111         }
 112
 113         private static boolean handleConsumerClass(Class<?> consumerClass, String consumerClassName, Properties properties, List<DMaaPVESMsgConsumer> consumers) {
 114                 DMaaPVESMsgConsumer consumer = null;
 115
 116                 try {
 117                         consumer = (DMaaPVESMsgConsumer) consumerClass.newInstance();
 118                         LOG.debug("Successfully created an instance of consumerClass : {}",consumerClassName);
 119                 } catch (Exception e) {
 120                         LOG.error("Could not create consumer from class {}",consumerClassName, e);
 121                 }
 122
 123                 if (consumer != null) {
 124                         LOG.info("Initializing consumer {}({})", consumerClassName, properties);
 125                         consumer.init(properties);
 126
 127                         if (consumer.isReady()) {
 128                                 Thread consumerThread = new Thread(consumer);
 129                                 consumerThread.start();
 130                                 consumers.add(consumer);
 131
 132                                 LOG.info("Started consumer thread ({} : {})", consumerClassName,
 133                                                 properties);
 134                                 return true;
 135                         } else {
 136                                 LOG.debug("Consumer {} is not ready", consumerClassName);
 137                         }
 138                 }
 139                 return false;
 140         }
 141
 142         public void run() {
 143                 while (threadsRunning) {
 144                         threadsRunning = updateThreadState(consumers);
 145                         if (!threadsRunning) {
 146                                 break;
 147                         }
 148
 149                         try {
 150                                 Thread.sleep(10000);
 151                         } catch (InterruptedException e) {
 152                                 LOG.error(e.getLocalizedMessage(), e);
 153                         }
 154                 }
 155
 156                 LOG.info("No listener threads running - exiting");
 157         }
 158
 159         public static List<DMaaPVESMsgConsumer> getConsumers() {
 160                 return consumers;
 161         }
 162
 163 }