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.List;
  23 import java.util.Properties;
  24 import java.util.function.Consumer;
  25 import com.fasterxml.jackson.core.JsonProcessingException;
  26 import com.fasterxml.jackson.databind.JsonNode;
  27 import com.fasterxml.jackson.databind.ObjectMapper;
  28 import com.fasterxml.jackson.databind.node.JsonNodeType;
  29 import org.onap.dmaap.mr.client.MRClientFactory;
  30 import org.onap.dmaap.mr.client.MRConsumer;
  31 import org.onap.dmaap.mr.client.response.MRConsumerResponse;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 public abstract class DMaaPVESMsgConsumerImpl implements DMaaPVESMsgConsumer {
  36
  37     private static final Logger LOG = LoggerFactory.getLogger(DMaaPVESMsgConsumerImpl.class);
  38     private static final String DEFAULT_SDNRUSER = "admin";
  39     private static final String DEFAULT_SDNRPASSWD = "admin";
  40
  41     private final String name = this.getClass().getSimpleName();
  42     private Properties properties = null;
  43     private MRConsumer consumer = null;
  44     private boolean running = false;
  45     private boolean ready = false;
  46     private int fetchPause = 5000; // Default pause between fetch - 5 seconds
  47     private int timeout = 15000; // Default timeout - 15 seconds
  48     protected final GeneralConfig generalConfig;
  49
  50     protected DMaaPVESMsgConsumerImpl(GeneralConfig generalConfig) {
  51         this.generalConfig = generalConfig;
  52     }
  53
  54     /*
  55      * Thread to fetch messages from the DMaaP topic. Waits for the messages to arrive on the topic until a certain timeout and returns.
  56      * If no data arrives on the topic, sleeps for a certain time period before checking again
  57      */
  58     @Override
  59     public void run() {
  60
  61         if (ready) {
  62             running = true;
  63             while (running) {
  64                 try {
  65                     boolean noData = true;
  66                     MRConsumerResponse consumerResponse = null;
  67                     consumerResponse = consumer.fetchWithReturnConsumerResponse(timeout, -1);
  68                     for (String msg : consumerResponse.getActualMessages()) {
  69                         noData = false;
  70                         LOG.debug("{} received ActualMessage from DMaaP VES Message topic {}", name,msg);
  71                         processMsg(msg);
  72                     }
  73
  74                     if (noData) {
  75                         LOG.debug("{} received ResponseCode: {}", name, consumerResponse.getResponseCode());
  76                         LOG.debug("{} received ResponseMessage: {}", name, consumerResponse.getResponseMessage());
  77                         if ((consumerResponse.getResponseCode() == null)
  78                                 && (consumerResponse.getResponseMessage().contains("SocketTimeoutException"))) {
  79                             LOG.warn("Client timeout while waiting for response from Server {}",
  80                                     consumerResponse.getResponseMessage());
  81                         }
  82                         pauseThread();
  83                     }
  84                 } catch (Exception e) {
  85                     LOG.error("Caught exception reading from DMaaP VES Message Topic", e);
  86                     running = false;
  87                 }
  88             }
  89         }
  90     }
  91
  92     /*
  93      * Create a consumer by specifying  properties containing information such as topic name, timeout, URL etc
  94      */
  95     @Override
  96     public void init(Properties properties) {
  97
  98         try {
  99
 100             String timeoutStr = properties.getProperty("timeout");
 101             LOG.debug("timeoutStr: {}", timeoutStr);
 102
 103             if ((timeoutStr != null) && (timeoutStr.length() > 0)) {
 104                 timeout = parseTimeOutValue(timeoutStr);
 105             }
 106
 107             String fetchPauseStr = properties.getProperty("fetchPause");
 108             LOG.debug("fetchPause(Str): {}",fetchPauseStr);
 109             if ((fetchPauseStr != null) && (fetchPauseStr.length() > 0)) {
 110                 fetchPause = parseFetchPause(fetchPauseStr);
 111             }
 112             LOG.debug("fetchPause: {} ",fetchPause);
 113
 114             this.consumer = MRClientFactory.createConsumer(properties);
 115             ready = true;
 116         } catch (Exception e) {
 117             LOG.error("Error initializing DMaaP VES Message consumer from file {} {}",properties, e);
 118         }
 119     }
 120
 121     private int parseTimeOutValue(String timeoutStr) {
 122         try {
 123             return Integer.parseInt(timeoutStr);
 124         } catch (NumberFormatException e) {
 125             LOG.error("Non-numeric value specified for timeout ({})",timeoutStr);
 126         }
 127         return timeout;
 128     }
 129
 130     private int parseFetchPause(String fetchPauseStr) {
 131         try {
 132             return Integer.parseInt(fetchPauseStr);
 133         } catch (NumberFormatException e) {
 134             LOG.error("Non-numeric value specified for fetchPause ({})",fetchPauseStr);
 135         }
 136         return fetchPause;
 137     }
 138
 139     private void pauseThread() throws InterruptedException {
 140         if (fetchPause > 0) {
 141             LOG.debug("No data received from fetch.  Pausing {} ms before retry", fetchPause);
 142             Thread.sleep(fetchPause);
 143         } else {
 144             LOG.debug("No data received from fetch.  No fetch pause specified - retrying immediately");
 145         }
 146     }
 147
 148     @Override
 149     public boolean isReady() {
 150         return ready;
 151     }
 152
 153     @Override
 154     public boolean isRunning() {
 155         return running;
 156     }
 157
 158     public String getProperty(String name) {
 159         return properties.getProperty(name, "");
 160     }
 161
 162     @Override
 163     public void stopConsumer() {
 164         running = false;
 165     }
 166
 167
 168     public String getBaseUrl() {
 169         return generalConfig.getBaseUrl();
 170     }
 171
 172     public String getSDNRUser() {
 173         return generalConfig.getSDNRUser() != null ? generalConfig.getSDNRUser() : DEFAULT_SDNRUSER;
 174     }
 175
 176     public String getSDNRPasswd() {
 177         return generalConfig.getSDNRPasswd() != null ? generalConfig.getSDNRPasswd() : DEFAULT_SDNRPASSWD;
 178     }
 179 }