sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/impl/DMaaPVESMsgConsumerImpl.java

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