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 com.fasterxml.jackson.core.JsonProcessingException;
  23 import com.fasterxml.jackson.databind.JsonNode;
  24 import com.fasterxml.jackson.databind.ObjectMapper;
  25 import java.util.List;
  26 import java.util.Properties;
  27 import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.config.GeneralConfig;
  28 import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.kafka.VESMsgKafkaConsumer;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 public abstract class StrimziKafkaVESMsgConsumerImpl
  33         implements StrimziKafkaVESMsgConsumer, StrimziKafkaVESMsgValidator {
  34
  35     private static final Logger LOG = LoggerFactory.getLogger(StrimziKafkaVESMsgConsumerImpl.class);
  36     private static final String DEFAULT_SDNRUSER = "admin";
  37     private static final String DEFAULT_SDNRPASSWD = "admin";
  38
  39     private final String name = this.getClass().getSimpleName();
  40     private VESMsgKafkaConsumer consumer = null;
  41     private boolean running = false;
  42     private boolean ready = false;
  43     private int fetchPause = 5000; // Default pause between fetch - 5 seconds
  44     protected final GeneralConfig generalConfig;
  45
  46     protected StrimziKafkaVESMsgConsumerImpl(GeneralConfig generalConfig) {
  47         this.generalConfig = generalConfig;
  48     }
  49
  50     /*
  51      * Thread to fetch messages from the Kafka topic. Waits for the messages to
  52      * arrive on the topic until a certain timeout and returns. If no data arrives
  53      * 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                     List<String> consumerResponse = null;
  64                     consumerResponse = consumer.poll();
  65                     for (String msg : consumerResponse) {
  66                         noData = false;
  67                         LOG.debug("{} received ActualMessage from Kafka VES Message topic {}", name, msg);
  68                         if (isMessageValid(msg)) {
  69                             processMsg(msg);
  70                         }
  71                     }
  72
  73                     if (noData) {
  74                         pauseThread();
  75                     }
  76                 } catch (InterruptedException e) {
  77                     LOG.warn("Caught exception reading from Kafka Message Topic", e);
  78                     Thread.currentThread().interrupt();
  79                 } catch (JsonProcessingException jsonProcessingException) {
  80                     LOG.warn("Failed to convert message to JsonNode: {}", jsonProcessingException.getMessage());
  81                 } catch (InvalidMessageException invalidMessageException) {
  82                     LOG.warn("Message is invalid because of: {}", invalidMessageException.getMessage());
  83                 } catch (Exception e) {
  84                     LOG.error("Caught exception reading from Kafka Message Topic", e);
  85                     running = false;
  86                 }
  87             }
  88         }
  89     }
  90
  91     @Override
  92     public boolean isMessageValid(String message) {
  93         return true;
  94     }
  95
  96     protected JsonNode convertMessageToJsonNode(String message) throws JsonProcessingException {
  97         return new ObjectMapper().readTree(message);
  98     }
  99
 100     /*
 101      * Create a Kafka consumer by specifying properties containing information such as
 102      * topic name, timeout, URL etc
 103      */
 104     @Override
 105     public void init(Properties strimziKafkaProperties, Properties consumerProperties) {
 106
 107         try {
 108             this.consumer = new VESMsgKafkaConsumer(strimziKafkaProperties, consumerProperties);
 109             this.consumer.subscribe(consumerProperties.getProperty("topic"));
 110             ready = true;
 111         } catch (Exception e) {
 112             LOG.error("Error initializing Kafka Message consumer from file {} {}", consumerProperties, e);
 113         }
 114     }
 115
 116     private void pauseThread() throws InterruptedException {
 117         if (fetchPause > 0) {
 118             LOG.debug("No data received from fetch.  Pausing {} ms before retry", fetchPause);
 119             Thread.sleep(fetchPause);
 120         } else {
 121             LOG.debug("No data received from fetch.  No fetch pause specified - retrying immediately");
 122         }
 123     }
 124
 125     @Override
 126     public boolean isReady() {
 127         return ready;
 128     }
 129
 130     @Override
 131     public boolean isRunning() {
 132         return running;
 133     }
 134
 135     /*
 136      * public String getProperty(String name) { return properties.getProperty(name,
 137      * ""); }
 138      */
 139     @Override
 140     public void stopConsumer() {
 141         running = false;
 142     }
 143
 144     public String getBaseUrl() {
 145         return generalConfig.getBaseUrl();
 146     }
 147
 148     public String getSDNRUser() {
 149         return generalConfig.getSDNRUser() != null ? generalConfig.getSDNRUser() : DEFAULT_SDNRUSER;
 150     }
 151
 152     public String getSDNRPasswd() {
 153         return generalConfig.getSDNRPasswd() != null ? generalConfig.getSDNRPasswd() : DEFAULT_SDNRPASSWD;
 154     }
 155 }