fff24389371b9e410178a908dc6475d16c330353
[ccsdk/features.git] /
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 (JsonProcessingException jsonProcessingException) {
84                     LOG.warn("Failed to convert message to JsonNode: {}", jsonProcessingException.getMessage());
85                 } catch (InvalidMessageException invalidMessageException) {
86                     LOG.warn("Message is invalid because of: {}", invalidMessageException.getMessage());
87                 } catch (Exception e) {
88                     LOG.error("Caught exception reading from DMaaP VES Message Topic", e);
89                     running = false;
90                 }
91             }
92         }
93     }
94
95     @Override
96     public boolean isMessageValid(String message) {
97         return true;
98     }
99
100     protected JsonNode convertMessageToJsonNode(String message) throws JsonProcessingException {
101         return new ObjectMapper().readTree(message);
102     }
103
104     /*
105      * Create a consumer by specifying  properties containing information such as topic name, timeout, URL etc
106      */
107     @Override
108     public void init(Properties properties) {
109
110         try {
111
112             String timeoutStr = properties.getProperty("timeout");
113             LOG.debug("timeoutStr: {}", timeoutStr);
114
115             if ((timeoutStr != null) && (timeoutStr.length() > 0)) {
116                 timeout = parseTimeOutValue(timeoutStr);
117             }
118
119             String fetchPauseStr = properties.getProperty("fetchPause");
120             LOG.debug("fetchPause(Str): {}",fetchPauseStr);
121             if ((fetchPauseStr != null) && (fetchPauseStr.length() > 0)) {
122                 fetchPause = parseFetchPause(fetchPauseStr);
123             }
124             LOG.debug("fetchPause: {} ",fetchPause);
125
126             this.consumer = MRClientFactory.createConsumer(properties);
127             ready = true;
128         } catch (Exception e) {
129             LOG.error("Error initializing DMaaP VES Message consumer from file {} {}",properties, e);
130         }
131     }
132
133     private int parseTimeOutValue(String timeoutStr) {
134         try {
135             return Integer.parseInt(timeoutStr);
136         } catch (NumberFormatException e) {
137             LOG.error("Non-numeric value specified for timeout ({})",timeoutStr);
138         }
139         return timeout;
140     }
141
142     private int parseFetchPause(String fetchPauseStr) {
143         try {
144             return Integer.parseInt(fetchPauseStr);
145         } catch (NumberFormatException e) {
146             LOG.error("Non-numeric value specified for fetchPause ({})",fetchPauseStr);
147         }
148         return fetchPause;
149     }
150
151     private void pauseThread() throws InterruptedException {
152         if (fetchPause > 0) {
153             LOG.debug("No data received from fetch.  Pausing {} ms before retry", fetchPause);
154             Thread.sleep(fetchPause);
155         } else {
156             LOG.debug("No data received from fetch.  No fetch pause specified - retrying immediately");
157         }
158     }
159
160     @Override
161     public boolean isReady() {
162         return ready;
163     }
164
165     @Override
166     public boolean isRunning() {
167         return running;
168     }
169
170     public String getProperty(String name) {
171         return properties.getProperty(name, "");
172     }
173
174     @Override
175     public void stopConsumer() {
176         running = false;
177     }
178
179
180     public String getBaseUrl() {
181         return generalConfig.getBaseUrl();
182     }
183
184     public String getSDNRUser() {
185         return generalConfig.getSDNRUser() != null ? generalConfig.getSDNRUser() : DEFAULT_SDNRUSER;
186     }
187
188     public String getSDNRPasswd() {
189         return generalConfig.getSDNRPasswd() != null ? generalConfig.getSDNRPasswd() : DEFAULT_SDNRPASSWD;
190     }
191 }