dmaap-listener/src/main/java/org/onap/ccsdk/sli/northbound/dmaapclient/SdncDmaapConsumerImpl.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * openECOMP : SDN-C
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights
   6  *                      reserved.
   7  * Modifications Copyright © 2018 IBM.
   8  * ================================================================================
   9  * Licensed under the Apache License, Version 2.0 (the "License");
  10  * you may not use this file except in compliance with the License.
  11  * You may obtain a copy of the License at
  12  *
  13  *      http://www.apache.org/licenses/LICENSE-2.0
  14  *
  15  * Unless required by applicable law or agreed to in writing, software
  16  * distributed under the License is distributed on an "AS IS" BASIS,
  17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18  * See the License for the specific language governing permissions and
  19  * limitations under the License.
  20  * ============LICENSE_END=========================================================
  21  */
  22
  23 package org.onap.ccsdk.sli.northbound.dmaapclient;
  24
  25 import java.io.File;
  26 import java.io.FileInputStream;
  27 import java.util.Properties;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30 import com.att.nsa.mr.client.MRClientFactory;
  31 import com.att.nsa.mr.client.MRConsumer;
  32 import com.att.nsa.mr.client.response.MRConsumerResponse;
  33
  34 public abstract class SdncDmaapConsumerImpl implements SdncDmaapConsumer {
  35
  36         private static final Logger LOG = LoggerFactory
  37                         .getLogger(SdncDmaapConsumer.class);
  38
  39     private final String name = this.getClass().getSimpleName();
  40         private Properties properties = null;
  41     private MRConsumer consumer = null;
  42     private MRConsumerResponse consumerResponse = null;
  43     private boolean running = false;
  44     private boolean ready = false;
  45     private int fetchPause = 5000; // Default pause between fetch - 5 seconds
  46     private int timeout = 15000; // Default timeout - 15 seconds
  47
  48         public SdncDmaapConsumerImpl() {
  49
  50         }
  51
  52         public SdncDmaapConsumerImpl(Properties properties, String propertiesPath) {
  53                 init(properties, propertiesPath);
  54         }
  55
  56     public boolean isReady() {
  57         return ready;
  58     }
  59
  60     public boolean isRunning() {
  61         return running;
  62     }
  63
  64         public String getProperty(String name) {
  65         return properties.getProperty(name, "");
  66         }
  67
  68     public void init(Properties properties, String propertiesPath) {
  69
  70         try (FileInputStream in = new FileInputStream(new File(propertiesPath))) {
  71
  72             LOG.debug("propertiesPath: " + propertiesPath);
  73             this.properties = (Properties) properties.clone();
  74             this.properties.load(in);
  75
  76
  77                         String timeoutStr = this.properties.getProperty("timeout");
  78             LOG.debug("timeoutStr: " + timeoutStr);
  79
  80                         if ((timeoutStr != null) && (timeoutStr.length() > 0)) {
  81                 timeout = parseTimeOutValue(timeoutStr);
  82                         }
  83
  84                         String fetchPauseStr = this.properties.getProperty("fetchPause");
  85             LOG.debug("fetchPause(Str): " + fetchPauseStr);
  86                         if ((fetchPauseStr != null) && (fetchPauseStr.length() > 0)) {
  87                 fetchPause = parseFetchPause(fetchPauseStr);
  88             }
  89             LOG.debug("fetchPause: " + fetchPause);
  90
  91
  92             this.consumer = MRClientFactory.createConsumer(propertiesPath);
  93             ready = true;
  94         } catch (Exception e) {
  95             LOG.error("Error initializing DMaaP consumer from file " + propertiesPath, e);
  96         }
  97     }
  98
  99     private int parseTimeOutValue(String timeoutStr) {
 100                                 try {
 101             return Integer.parseInt(timeoutStr);
 102                                 } catch (NumberFormatException e) {
 103             LOG.error("Non-numeric value specified for timeout (" + timeoutStr + ")");
 104                                 }
 105         return timeout;
 106                         }
 107
 108     private int parseFetchPause(String fetchPauseStr) {
 109         try {
 110             return Integer.parseInt(fetchPauseStr);
 111         } catch (NumberFormatException e) {
 112             LOG.error("Non-numeric value specified for fetchPause (" + fetchPauseStr + ")");
 113                 }
 114         return fetchPause;
 115         }
 116
 117
 118         @Override
 119         public void run() {
 120                 if (ready) {
 121
 122                         running = true;
 123
 124                         while (running) {
 125
 126                                 try {
 127                                         boolean noData = true;
 128                                         consumerResponse = consumer.fetchWithReturnConsumerResponse(timeout, -1);
 129                                         for (String msg : consumerResponse.getActualMessages()) {
 130                                                 noData = false;
 131                                                 LOG.info(name + " received ActualMessage from DMaaP:\n"+msg);
 132                                                 processMsg(msg);
 133                                         }
 134
 135                                         if (noData) {
 136                         LOG.info(name + " received ResponseCode: " + consumerResponse.getResponseCode());
 137                                             LOG.info(name + " received ResponseMessage: " + consumerResponse.getResponseMessage());
 138                         pauseThread();
 139                                         }
 140                                 } catch (Exception e) {
 141                                         LOG.error("Caught exception reading from DMaaP", e);
 142                                         running = false;
 143                                 }
 144
 145
 146                         }
 147                 }
 148     }
 149
 150     private void pauseThread() throws InterruptedException {
 151         if (fetchPause > 0) {
 152             LOG.info(String.format("No data received from fetch.  Pausing %d ms before retry", fetchPause));
 153             Thread.sleep(fetchPause);
 154         } else {
 155             LOG.info("No data received from fetch.  No fetch pause specified - retrying immediately");
 156         }
 157         }
 158
 159         public abstract void processMsg(String msg) throws InvalidMessageException;
 160 }