2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017 AT&T Intellectual Property. All rights
7 * ================================================================================
8 * Licensed under the Apache License, Version 2.0 (the "License");
9 * you may not use this file except in compliance with the License.
10 * You may obtain a copy of the License at
12 * http://www.apache.org/licenses/LICENSE-2.0
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the License is distributed on an "AS IS" BASIS,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 * See the License for the specific language governing permissions and
18 * limitations under the License.
19 * ============LICENSE_END=========================================================
22 package org.openecomp.sdnc.dmaapclient;
25 import java.io.FileInputStream;
26 import java.util.Properties;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
31 import com.att.nsa.mr.client.MRClientFactory;
32 import com.att.nsa.mr.client.MRConsumer;
33 import com.att.nsa.mr.client.response.MRConsumerResponse;
35 public abstract class SdncDmaapConsumer implements Runnable {
37 private static final Logger LOG = LoggerFactory
38 .getLogger(SdncDmaapConsumer.class);
40 private String propertiesPath = "";
41 private Properties properties = null;
42 MRConsumer consumer = null;
43 MRConsumerResponse consumerResponse = null;
44 boolean running = false;
45 boolean ready = false;
46 int fetchPause = 5000; // Default pause between fetchs - 5 seconds
48 public boolean isReady() {
52 int timeout = 15000; // Default timeout - 15 seconds
54 public boolean isRunning() {
58 public SdncDmaapConsumer() {
62 public SdncDmaapConsumer(Properties properties, String propertiesPath) {
63 init(properties, propertiesPath);
66 public String getProperty(String name) {
67 return(properties.getProperty(name, ""));
70 public void init(Properties properties, String propertiesPath) {
72 this.propertiesPath = propertiesPath;
76 this.properties = (Properties) properties.clone();
78 this.properties.load(new FileInputStream(new File(propertiesPath)));
80 String timeoutStr = properties.getProperty("timeout");
82 if ((timeoutStr != null) && (timeoutStr.length() > 0)) {
84 timeout = Integer.parseInt(timeoutStr);
85 } catch (NumberFormatException e) {
86 LOG.error("Non-numeric value specified for timeout ("+timeoutStr+")");
90 String fetchPauseStr = properties.getProperty("fetchPause");
91 if ((fetchPauseStr != null) && (fetchPauseStr.length() > 0)) {
93 fetchPause = Integer.parseInt(fetchPauseStr);
94 } catch (NumberFormatException e) {
95 LOG.error("Non-numeric valud specified for fetchPause ("+fetchPauseStr+")");
99 this.consumer = MRClientFactory.createConsumer(propertiesPath);
101 } catch (Exception e) {
102 LOG.error("Error initializing DMaaP consumer from file "+propertiesPath, e);
116 boolean noData = true;
117 consumerResponse = consumer.fetchWithReturnConsumerResponse(timeout, -1);
118 for (String msg : consumerResponse.getActualMessages()) {
120 LOG.info("Received message from DMaaP:\n"+msg);
125 if (fetchPause > 0) {
127 LOG.info("No data received from fetch. Pausing "+fetchPause+" ms before retry");
128 Thread.sleep(fetchPause);
131 LOG.info("No data received from fetch. No fetch pause specified - retrying immediately");
134 } catch (Exception e) {
135 LOG.error("Caught exception reading from DMaaP", e);
145 abstract public void processMsg(String msg) throws InvalidMessageException;