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 _CM_CLASS = "org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.impl.DMaaPCMVESMsgConsumer";
  35         private static final String _PNFREG_DOMAIN = "pnfRegistration";
  36         private static final String _FAULT_DOMAIN = "fault";
  37         private static final String _CM_DOMAIN = "provisioning";
  38
  39         boolean threadsRunning = false;
  40         List<DMaaPVESMsgConsumer> consumers = new LinkedList<>();
  41         private PNFRegistrationConfig pnfRegistrationConfig;
  42         private FaultConfig faultConfig;
  43         private GeneralConfig generalConfig;
  44         private ProvisioningConfig provisioningConfig;
  45
  46         public DMaaPVESMsgConsumerMain(Map<String, MessageConfig> configMap, GeneralConfig generalConfig) {
  47                 this.generalConfig = generalConfig;
  48                 configMap.forEach(this::initialize);
  49         }
  50
  51         public void initialize(String domain, MessageConfig domainConfig) {
  52                 LOG.debug("In initialize method : Domain = {} and domainConfig = {}", domain, domainConfig);
  53                 String consumerClass;
  54                 Properties consumerProperties = new Properties();
  55                 if (domain.equalsIgnoreCase(_PNFREG_DOMAIN)) {
  56                         this.pnfRegistrationConfig = (PNFRegistrationConfig) domainConfig;
  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                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_URI,
  87                                         pnfRegistrationConfig.getHTTPProxyURI());
  88                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_USER,
  89                                         pnfRegistrationConfig.getHTTPProxyUsername());
  90                         consumerProperties.put(PNFRegistrationConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_PASSWORD,
  91                                         pnfRegistrationConfig.getHTTPProxyPassword());
  92
  93                         threadsRunning = createConsumer(_PNFREG_DOMAIN, consumerProperties);
  94                 } else if (domain.equalsIgnoreCase(_FAULT_DOMAIN)) {
  95                         this.faultConfig = (FaultConfig) domainConfig;
  96                         consumerClass = _FAULT_CLASS;
  97                         LOG.debug("Consumer class = {}", consumerClass);
  98                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE, faultConfig.getTransportType());
  99                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_HOST_PORT, faultConfig.getHostPort());
 100                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE, faultConfig.getContenttype());
 101                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_GROUP, faultConfig.getConsumerGroup());
 102                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_ID, faultConfig.getConsumerId());
 103                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TOPIC, faultConfig.getTopic());
 104                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_TIMEOUT, faultConfig.getTimeout());
 105                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_LIMIT, faultConfig.getLimit());
 106                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE, faultConfig.getFetchPause());
 107                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_PROTOCOL, faultConfig.getProtocol());
 108                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_USERNAME, faultConfig.getUsername());
 109                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_PASSWORD, faultConfig.getPassword());
 110                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_READTIMEOUT,
 111                                         faultConfig.getClientReadTimeout());
 112                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_CONNECTTIMEOUT,
 113                                         faultConfig.getClientConnectTimeout());
 114                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_URI,
 115                                         faultConfig.getHTTPProxyURI());
 116                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_USER,
 117                                         faultConfig.getHTTPProxyUsername());
 118                         consumerProperties.put(FaultConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_PASSWORD,
 119                                         faultConfig.getHTTPProxyPassword());
 120                         threadsRunning = createConsumer(_FAULT_DOMAIN, consumerProperties);
 121                 } else if (domain.equalsIgnoreCase(_CM_DOMAIN)) {
 122                         this.provisioningConfig = (ProvisioningConfig) domainConfig;
 123                         consumerClass = _CM_CLASS;
 124                         LOG.debug("Consumer class = {}", consumerClass);
 125                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_TRANSPORTTYPE, provisioningConfig.getTransportType());
 126                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_HOST_PORT, provisioningConfig.getHostPort());
 127                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_CONTENTTYPE, provisioningConfig.getContenttype());
 128                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_GROUP, provisioningConfig.getConsumerGroup());
 129                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_ID, provisioningConfig.getConsumerId());
 130                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_TOPIC, provisioningConfig.getTopic());
 131                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_TIMEOUT, provisioningConfig.getTimeout());
 132                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_LIMIT, provisioningConfig.getLimit());
 133                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_FETCHPAUSE, provisioningConfig.getFetchPause());
 134                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_PROTOCOL, provisioningConfig.getProtocol());
 135                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_USERNAME, provisioningConfig.getUsername());
 136                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_PASSWORD, provisioningConfig.getPassword());
 137                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_CLIENT_READTIMEOUT,
 138                                         provisioningConfig.getClientReadTimeout());
 139                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_CLIENT_CONNECTTIMEOUT,
 140                                         provisioningConfig.getClientConnectTimeout());
 141                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_URI,
 142                                         provisioningConfig.getHTTPProxyURI());
 143                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_USER,
 144                                         provisioningConfig.getHTTPProxyUsername());
 145                         consumerProperties.put(ProvisioningConfig.PROPERTY_KEY_CONSUMER_CLIENT_HTTPPROXY_AUTH_PASSWORD,
 146                                         provisioningConfig.getHTTPProxyPassword());
 147                         threadsRunning = createConsumer(_CM_DOMAIN, consumerProperties);
 148                 }
 149         }
 150
 151         private boolean updateThreadState(List<DMaaPVESMsgConsumer> consumers) {
 152                 boolean threadsRunning = false;
 153                 for (DMaaPVESMsgConsumer consumer : consumers) {
 154                         if (consumer.isRunning()) {
 155                                 threadsRunning = true;
 156                         }
 157                 }
 158                 return threadsRunning;
 159         }
 160
 161         public boolean createConsumer(String consumerType, Properties properties) {
 162                 DMaaPVESMsgConsumerImpl consumer = null;
 163
 164                 if (consumerType.equalsIgnoreCase(_PNFREG_DOMAIN))
 165                         consumer = new DMaaPPNFRegVESMsgConsumer(generalConfig);
 166                 else if (consumerType.equalsIgnoreCase(_FAULT_DOMAIN))
 167                         consumer = new DMaaPFaultVESMsgConsumer(generalConfig);
 168                 else if (consumerType.equalsIgnoreCase(_CM_DOMAIN))
 169                         consumer = new DMaaPCMVESMsgConsumer(generalConfig);
 170
 171                 handleConsumer(consumer, properties, consumers);
 172                 return !consumers.isEmpty();
 173         }
 174
 175         private boolean handleConsumer(DMaaPVESMsgConsumer consumer, Properties properties,
 176                         List<DMaaPVESMsgConsumer> consumers) {
 177                 if (consumer != null) {
 178                         consumer.init(properties);
 179
 180                         if (consumer.isReady()) {
 181                                 Thread consumerThread = new Thread(consumer);
 182                                 consumerThread.start();
 183                                 consumers.add(consumer);
 184
 185                                 LOG.info("Started consumer thread ({} : {})", consumer.getClass().getSimpleName(), properties);
 186                                 return true;
 187                         } else {
 188                                 LOG.debug("Consumer {} is not ready", consumer.getClass().getSimpleName());
 189                         }
 190                 }
 191                 return false;
 192         }
 193
 194         @Override
 195         public void run() {
 196                 while (threadsRunning) {
 197                         threadsRunning = updateThreadState(consumers);
 198                         if (!threadsRunning) {
 199                                 break;
 200                         }
 201
 202                         try {
 203                                 Thread.sleep(10000);
 204                         } catch (InterruptedException e) {
 205                                 LOG.error(e.getLocalizedMessage(), e);
 206                         }
 207                 }
 208
 209                 LOG.info("No listener threads running - exiting");
 210         }
 211
 212         public List<DMaaPVESMsgConsumer> getConsumers() {
 213                 return consumers;
 214         }
 215
 216 }