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  * Copyright (C) 2021 Samsung Electronics Intellectual Property. All rights reserved.
   7  * =================================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   9  * in compliance with the License. You may obtain a copy of the License at
  10  *
  11  * http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software distributed under the License
  14  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  15  * or implied. See the License for the specific language governing permissions and limitations under
  16  * the License.
  17  * ============LICENSE_END==========================================================================
  18  */
  19
  20 package org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.impl;
  21
  22 import java.util.LinkedList;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.Properties;
  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 = "org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.impl.DMaaPPNFRegVESMsgConsumer";
  33         private static final String _FAULT_CLASS = "org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.impl.DMaaPFaultVESMsgConsumer";
  34         private static final String _PNFREG_DOMAIN = "pnfRegistration";
  35         private static final String _FAULT_DOMAIN = "fault";
  36
  37         boolean threadsRunning = false;
  38         List<DMaaPVESMsgConsumer> consumers = new LinkedList<>();
  39         private PNFRegistrationConfig pnfRegistrationConfig;
  40         private FaultConfig faultConfig;
  41         private GeneralConfig generalConfig;
  42
  43         public DMaaPVESMsgConsumerMain(Map<String, MessageConfig> configMap, GeneralConfig generalConfig) {
  44                 this.generalConfig = generalConfig;
  45                 configMap.forEach(this::initialize);
  46         }
  47
  48         public void initialize(String domain, MessageConfig domainConfig) {
  49                 LOG.debug("In initialize method : Domain = {} and domainConfig = {}", domain, domainConfig);
  50                 String consumerClass;
  51                 Properties consumerProperties = new Properties();
  52                 if (domain.equalsIgnoreCase(_PNFREG_DOMAIN)) {
  53                         this.pnfRegistrationConfig = (PNFRegistrationConfig) domainConfig;
  54                         consumerClass = _PNFREG_CLASS;
  55                         LOG.debug("Consumer class = {}", consumerClass);
  56
  57                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE,
  58                                         pnfRegistrationConfig.getTransportType());
  59                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_HOST_PORT,
  60                                         pnfRegistrationConfig.getHostPort());
  61                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE,
  62                                         pnfRegistrationConfig.getContenttype());
  63                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_GROUP,
  64                                         pnfRegistrationConfig.getConsumerGroup());
  65                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_ID,
  66                                         pnfRegistrationConfig.getConsumerId());
  67                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TOPIC, pnfRegistrationConfig.getTopic());
  68                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_TIMEOUT,
  69                                         pnfRegistrationConfig.getTimeout());
  70                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_LIMIT, pnfRegistrationConfig.getLimit());
  71                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE,
  72                                         pnfRegistrationConfig.getFetchPause());
  73                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_PROTOCOL,
  74                                         pnfRegistrationConfig.getProtocol());
  75                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_USERNAME,
  76                                         pnfRegistrationConfig.getUsername());
  77                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_PASSWORD,
  78                                         pnfRegistrationConfig.getPassword());
  79                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_READTIMEOUT,
  80                                         pnfRegistrationConfig.getClientReadTimeout());
  81                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_CONNECTTIMEOUT,
  82                                         pnfRegistrationConfig.getClientConnectTimeout());
  83                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_URI,
  84                                         pnfRegistrationConfig.getHTTPProxyURI());
  85                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_USER,
  86                                         pnfRegistrationConfig.getHTTPProxyUsername());
  87                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_PASSWORD,
  88                                         pnfRegistrationConfig.getHTTPProxyPassword());
  89
  90                         threadsRunning = createConsumer(_PNFREG_DOMAIN, consumerProperties);
  91                 } else if (domain.equalsIgnoreCase(_FAULT_DOMAIN)) {
  92                         this.faultConfig = (FaultConfig) domainConfig;
  93                         consumerClass = _FAULT_CLASS;
  94                         LOG.debug("Consumer class = {}", consumerClass);
  95                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE, faultConfig.getTransportType());
  96                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_HOST_PORT, faultConfig.getHostPort());
  97                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE, faultConfig.getContenttype());
  98                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_GROUP, faultConfig.getConsumerGroup());
  99                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_ID, faultConfig.getConsumerId());
 100                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TOPIC, faultConfig.getTopic());
 101                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TIMEOUT, faultConfig.getTimeout());
 102                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_LIMIT, faultConfig.getLimit());
 103                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE, faultConfig.getFetchPause());
 104                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_PROTOCOL, faultConfig.getProtocol());
 105                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_USERNAME, faultConfig.getUsername());
 106                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_PASSWORD, faultConfig.getPassword());
 107                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_READTIMEOUT,
 108                                         faultConfig.getClientReadTimeout());
 109                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_CONNECTTIMEOUT,
 110                                         faultConfig.getClientConnectTimeout());
 111                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_URI,
 112                                         faultConfig.getHTTPProxyURI());
 113                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_USER,
 114                                         faultConfig.getHTTPProxyUsername());
 115                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_PASSWORD,
 116                                         faultConfig.getHTTPProxyPassword());
 117                         threadsRunning = createConsumer(_FAULT_DOMAIN, consumerProperties);
 118                 }
 119         }
 120
 121         private boolean updateThreadState(List<DMaaPVESMsgConsumer> consumers) {
 122                 boolean threadsRunning = false;
 123                 for (DMaaPVESMsgConsumer consumer : consumers) {
 124                         if (consumer.isRunning()) {
 125                                 threadsRunning = true;
 126                         }
 127                 }
 128                 return threadsRunning;
 129         }
 130
 131         public boolean createConsumer(String consumerType, Properties properties) {
 132                 DMaaPVESMsgConsumerImpl consumer = null;
 133
 134                 if (consumerType.equalsIgnoreCase(_PNFREG_DOMAIN))
 135                         consumer = new DMaaPPNFRegVESMsgConsumer(generalConfig);
 136                 else if (consumerType.equalsIgnoreCase(_FAULT_DOMAIN))
 137                         consumer = new DMaaPFaultVESMsgConsumer(generalConfig);
 138
 139                 handleConsumer(consumer, properties, consumers);
 140                 return !consumers.isEmpty();
 141         }
 142
 143         private boolean handleConsumer(DMaaPVESMsgConsumer consumer, Properties properties,
 144                         List<DMaaPVESMsgConsumer> consumers) {
 145                 if (consumer != null) {
 146                         consumer.init(properties);
 147
 148                         if (consumer.isReady()) {
 149                                 Thread consumerThread = new Thread(consumer);
 150                                 consumerThread.start();
 151                                 consumers.add(consumer);
 152
 153                                 LOG.info("Started consumer thread ({} : {})", consumer.getClass().getSimpleName(), properties);
 154                                 return true;
 155                         } else {
 156                                 LOG.debug("Consumer {} is not ready", consumer.getClass().getSimpleName());
 157                         }
 158                 }
 159                 return false;
 160         }
 161
 162         @Override
 163         public void run() {
 164                 while (threadsRunning) {
 165                         threadsRunning = updateThreadState(consumers);
 166                         if (!threadsRunning) {
 167                                 break;
 168                         }
 169
 170                         try {
 171                                 Thread.sleep(10000);
 172                         } catch (InterruptedException e) {
 173                                 LOG.error(e.getLocalizedMessage(), e);
 174                         }
 175                 }
 176
 177                 LOG.info("No listener threads running - exiting");
 178         }
 179
 180         public List<DMaaPVESMsgConsumer> getConsumers() {
 181                 return consumers;
 182         }
 183
 184 }