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 import org.onap.ccsdk.features.sdnr.wt.common.configuration.Configuration;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29 public class DMaaPVESMsgConsumerMain implements Runnable {
  30
  31     private static final Logger LOG = LoggerFactory.getLogger(DMaaPVESMsgConsumerMain.class);
  32     private static final String _PNFREG_CLASS =
  33             "org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.impl.DMaaPPNFRegVESMsgConsumer";
  34     private static final String _FAULT_CLASS =
  35             "org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.impl.DMaaPFaultVESMsgConsumer";
  36     private static final String _PNFREG_DOMAIN = "pnfRegistration";
  37     private static final String _FAULT_DOMAIN = "fault";
  38
  39     boolean threadsRunning = false;
  40     List<DMaaPVESMsgConsumer> consumers = new LinkedList<>();
  41     private PNFRegistrationConfig pnfRegistrationConfig;
  42     private FaultConfig faultConfig;
  43     private GeneralConfig generalConfig;
  44
  45     public DMaaPVESMsgConsumerMain(Map<String, Configuration> configMap, GeneralConfig generalConfig) {
  46         this.generalConfig = generalConfig;
  47         configMap.forEach((k, v) -> initialize(k, v));
  48     }
  49
  50     public void initialize(String domain, Configuration domainConfig) {
  51         LOG.debug("In initialize method : Domain = {} and domainConfig = {}", domain, domainConfig);
  52         String consumerClass = null;
  53         Properties consumerProperties = new Properties();
  54         if (domain.equalsIgnoreCase(_PNFREG_DOMAIN)) {
  55             this.pnfRegistrationConfig = (PNFRegistrationConfig) domainConfig;
  56
  57             consumerClass = _PNFREG_CLASS;
  58             LOG.debug("Consumer class = {}",consumerClass);
  59
  60             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE,
  61                     pnfRegistrationConfig.getTransportType());
  62             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_HOST_PORT,
  63                     pnfRegistrationConfig.getHostPort());
  64             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE,
  65                     pnfRegistrationConfig.getContenttype());
  66             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_GROUP,
  67                     pnfRegistrationConfig.getConsumerGroup());
  68             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_ID,
  69                     pnfRegistrationConfig.getConsumerId());
  70             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TOPIC, pnfRegistrationConfig.getTopic());
  71             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TIMEOUT,
  72                     pnfRegistrationConfig.getTimeout());
  73             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_LIMIT, pnfRegistrationConfig.getLimit());
  74             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE,
  75                     pnfRegistrationConfig.getFetchPause());
  76             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_PROTOCOL,
  77                     pnfRegistrationConfig.getProtocol());
  78             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_USERNAME,
  79                     pnfRegistrationConfig.getUsername());
  80             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_PASSWORD,
  81                     pnfRegistrationConfig.getPassword());
  82             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_READTIMEOUT,
  83                     pnfRegistrationConfig.getClientReadTimeout());
  84             consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_CONNECTTIMEOUT,
  85                     pnfRegistrationConfig.getClientConnectTimeout());
  86             threadsRunning = createConsumer(_PNFREG_DOMAIN, consumerProperties);
  87         } else if (domain.equalsIgnoreCase(_FAULT_DOMAIN)) {
  88             this.faultConfig = (FaultConfig) domainConfig;
  89             consumerClass = _FAULT_CLASS;
  90             LOG.debug("Consumer class = {}", consumerClass);
  91             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE, faultConfig.getTransportType());
  92             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_HOST_PORT, faultConfig.getHostPort());
  93             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE, faultConfig.getContenttype());
  94             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_GROUP, faultConfig.getConsumerGroup());
  95             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_ID, faultConfig.getConsumerId());
  96             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TOPIC, faultConfig.getTopic());
  97             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TIMEOUT, faultConfig.getTimeout());
  98             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_LIMIT, faultConfig.getLimit());
  99             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE, faultConfig.getFetchPause());
 100             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_PROTOCOL, faultConfig.getProtocol());
 101             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_USERNAME, faultConfig.getUsername());
 102             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_PASSWORD, faultConfig.getPassword());
 103             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_READTIMEOUT,
 104                     faultConfig.getClientReadTimeout());
 105             consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_CONNECTTIMEOUT,
 106                     faultConfig.getClientConnectTimeout());
 107             threadsRunning = createConsumer(_FAULT_DOMAIN, consumerProperties);
 108         }
 109     }
 110
 111     private boolean updateThreadState(List<DMaaPVESMsgConsumer> consumers) {
 112         boolean threadsRunning = false;
 113         for (DMaaPVESMsgConsumer consumer : consumers) {
 114             if (consumer.isRunning()) {
 115                 threadsRunning = true;
 116             }
 117         }
 118         return threadsRunning;
 119     }
 120
 121     public boolean createConsumer(String consumerType, Properties properties) {
 122         DMaaPVESMsgConsumerImpl consumer = null;
 123
 124         if (consumerType.equalsIgnoreCase(_PNFREG_DOMAIN))
 125             consumer = new DMaaPPNFRegVESMsgConsumer(generalConfig);
 126         else if (consumerType.equalsIgnoreCase(_FAULT_DOMAIN))
 127             consumer = new DMaaPFaultVESMsgConsumer(generalConfig);
 128
 129         handleConsumer(consumer, properties, consumers);
 130         return !consumers.isEmpty();
 131     }
 132
 133     private boolean handleConsumer(DMaaPVESMsgConsumer consumer, Properties properties,
 134             List<DMaaPVESMsgConsumer> consumers) {
 135         if (consumer != null) {
 136             consumer.init(properties);
 137
 138             if (consumer.isReady()) {
 139                 Thread consumerThread = new Thread(consumer);
 140                 consumerThread.start();
 141                 consumers.add(consumer);
 142
 143                 LOG.info("Started consumer thread ({} : {})", consumer.getClass().getSimpleName(), properties);
 144                 return true;
 145             } else {
 146                 LOG.debug("Consumer {} is not ready", consumer.getClass().getSimpleName());
 147             }
 148         }
 149         return false;
 150     }
 151
 152     @Override
 153     public void run() {
 154         while (threadsRunning) {
 155             threadsRunning = updateThreadState(consumers);
 156             if (!threadsRunning) {
 157                 break;
 158             }
 159
 160             try {
 161                 Thread.sleep(10000);
 162             } catch (InterruptedException e) {
 163                 LOG.error(e.getLocalizedMessage(), e);
 164             }
 165         }
 166
 167         LOG.info("No listener threads running - exiting");
 168     }
 169
 170     public List<DMaaPVESMsgConsumer> getConsumers() {
 171         return consumers;
 172     }
 173
 174 }