diff options
Diffstat (limited to 'src/main/java/org/onap/dmaap/mr/client/impl/MRConsumerImpl.java')
| -rw-r--r-- | src/main/java/org/onap/dmaap/mr/client/impl/MRConsumerImpl.java | 168 |
1 files changed, 100 insertions, 68 deletions
diff --git a/src/main/java/org/onap/dmaap/mr/client/impl/MRConsumerImpl.java b/src/main/java/org/onap/dmaap/mr/client/impl/MRConsumerImpl.java index 6c67313..57ae3ee 100644 --- a/src/main/java/org/onap/dmaap/mr/client/impl/MRConsumerImpl.java +++ b/src/main/java/org/onap/dmaap/mr/client/impl/MRConsumerImpl.java @@ -4,11 +4,13 @@ * ================================================================================ * Copyright © 2017 AT&T Intellectual Property. All rights reserved. * ================================================================================ + * Modifications Copyright © 2021 Orange. + * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * http://www.apache.org/licenses/LICENSE-2.0 - * + * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. @@ -17,24 +19,26 @@ * ============LICENSE_END========================================================= * * ECOMP is a trademark and service mark of AT&T Intellectual Property. - * + * *******************************************************************************/ + package org.onap.dmaap.mr.client.impl; import com.att.aft.dme2.api.DME2Client; import com.att.aft.dme2.api.DME2Exception; -import org.onap.dmaap.mr.client.HostSelector; -import org.onap.dmaap.mr.client.MRClientFactory; -import org.onap.dmaap.mr.client.MRConsumer; -import org.onap.dmaap.mr.client.response.MRConsumerResponse; -import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants; - -import java.io.*; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.UnsupportedEncodingException; import java.net.MalformedURLException; import java.net.URI; import java.net.URISyntaxException; import java.net.URLEncoder; -import java.util.*; +import java.util.Collection; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Properties; import java.util.concurrent.TimeUnit; import org.apache.http.HttpException; import org.apache.http.HttpStatus; @@ -42,20 +46,24 @@ import org.json.JSONArray; import org.json.JSONException; import org.json.JSONObject; import org.json.JSONTokener; +import org.onap.dmaap.mr.client.HostSelector; +import org.onap.dmaap.mr.client.MRClientFactory; +import org.onap.dmaap.mr.client.MRConsumer; +import org.onap.dmaap.mr.client.ProtocolType; +import org.onap.dmaap.mr.client.response.MRConsumerResponse; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class MRConsumerImpl extends MRBaseClient implements MRConsumer { - private Logger log = LoggerFactory.getLogger(this.getClass().getName()); + private static final Logger logger = LoggerFactory.getLogger(MRConsumerImpl.class); public static final String ROUTER_FILE_PATH = null; - public String protocolFlag = ProtocolTypeConstants.DME2.getValue(); + public String protocolFlag = ProtocolType.DME2.getValue(); public String consumerFilePath; private static final String JSON_RESULT = "result"; - private static final String PROPS_PROTOCOL = "Protocol"; private static final String EXECPTION_MESSAGE = "exception: "; private static final String SUCCESS_MESSAGE = "Success"; @@ -70,6 +78,29 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { private static final String URL_PARAM_LIMIT = "limit"; private static final String URL_PARAM_TIMEOUT = "timeout"; + private static final String USERNAME = "username"; + private static final String SERVICE_NAME = "ServiceName"; + private static final String PARTNER = "Partner"; + private static final String ROUTE_OFFER = "routeOffer"; + private static final String PROTOCOL = "Protocol"; + private static final String METHOD_TYPE = "MethodType"; + private static final String CONTENT_TYPE = "contenttype"; + private static final String LATITUDE = "Latitude"; + private static final String LONGITUDE = "Longitude"; + private static final String AFT_ENVIRONMENT = "AFT_ENVIRONMENT"; + private static final String VERSION = "Version"; + private static final String ENVIRONMENT = "Environment"; + private static final String SUB_CONTEXT_PATH = "SubContextPath"; + private static final String SESSION_STICKINESS_REQUIRED = "sessionstickinessrequired"; + private static final String AFT_DME2_EP_READ_TIMEOUT_MS = "AFT_DME2_EP_READ_TIMEOUT_MS"; + private static final String AFT_DME2_ROUNDTRIP_TIMEOUT_MS = "AFT_DME2_ROUNDTRIP_TIMEOUT_MS"; + private static final String AFT_DME2_EP_CONN_TIMEOUT = "AFT_DME2_EP_CONN_TIMEOUT"; + private static final String AFT_DME2_EXCHANGE_REQUEST_HANDLERS = "AFT_DME2_EXCHANGE_REQUEST_HANDLERS"; + private static final String AFT_DME2_EXCHANGE_REPLY_HANDLERS = "AFT_DME2_EXCHANGE_REPLY_HANDLERS"; + private static final String AFT_DME2_REQ_TRACE_ON = "AFT_DME2_REQ_TRACE_ON"; + private static final String DME2_PER_HANDLER_TIMEOUT_MS = "DME2_PER_HANDLER_TIMEOUT_MS"; + private static final String DME2_REPLY_HANDLER_TIMEOUT_MS = "DME2_REPLY_HANDLER_TIMEOUT_MS"; + private final String fTopic; private final String fGroup; private final String fId; @@ -192,8 +223,8 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { public Iterable<String> fetch(int timeoutMs, int limit) throws Exception { final LinkedList<String> msgs = new LinkedList<>(); - ProtocolTypeConstants protocolFlagEnum = null; - for(ProtocolTypeConstants type : ProtocolTypeConstants.values()) { + ProtocolType protocolFlagEnum = null; + for (ProtocolType type : ProtocolType.values()) { if (type.getValue().equalsIgnoreCase(protocolFlag)) { protocolFlagEnum = type; } @@ -211,27 +242,27 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { break; case AAF_AUTH: String urlAuthPath = createUrlPath(MRConstants.makeConsumerUrl(fHostSelector.selectBaseHost(), fTopic, - fGroup, fId, props.getProperty(PROPS_PROTOCOL)), timeoutMs, limit); + fGroup, fId, props.getProperty(PROTOCOL)), timeoutMs, limit); final JSONObject o = get(urlAuthPath, username, password, protocolFlag); readJsonData(msgs, o); break; case AUTH_KEY: final String urlKeyPath = createUrlPath( - MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId, props.getProperty(PROPS_PROTOCOL)), + MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId, props.getProperty(PROTOCOL)), timeoutMs, limit); final JSONObject authObject = getAuth(urlKeyPath, authKey, authDate, username, password, protocolFlag); readJsonData(msgs, authObject); break; case HTTPNOAUTH: final String urlNoAuthPath = createUrlPath(MRConstants.makeConsumerUrl(fHostSelector.selectBaseHost(), fTopic, - fGroup, fId, props.getProperty(PROPS_PROTOCOL)), timeoutMs, limit); + fGroup, fId, props.getProperty(PROTOCOL)), timeoutMs, limit); readJsonData(msgs, getNoAuth(urlNoAuthPath)); break; } } catch (JSONException e) { // unexpected response reportProblemWithResponse(); - log.error(EXECPTION_MESSAGE, e); + logger.error(EXECPTION_MESSAGE, e); } catch (HttpException e) { throw new IOException(e); } @@ -244,10 +275,11 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { final JSONArray a = o.getJSONArray(JSON_RESULT); if (a != null) { for (int i = 0; i < a.length(); i++) { - if (a.get(i) instanceof String) + if (a.get(i) instanceof String) { msgs.add(a.getString(i)); - else + } else { msgs.add(a.getJSONObject(i).toString()); + } } } } @@ -264,7 +296,7 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { final LinkedList<String> msgs = new LinkedList<>(); MRConsumerResponse mrConsumerResponse = new MRConsumerResponse(); try { - if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) { + if (ProtocolType.DME2.getValue().equalsIgnoreCase(protocolFlag)) { dmeConfigure(timeoutMs, limit); long timeout = (dme2ReplyHandlerTimeoutMs > 0 && longPollingMs == timeoutMs) ? dme2ReplyHandlerTimeoutMs @@ -277,9 +309,9 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { createMRConsumerResponse(reply, mrConsumerResponse); } - if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) { + if (ProtocolType.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) { final String urlPath = createUrlPath(MRConstants.makeConsumerUrl(fHostSelector.selectBaseHost(), fTopic, - fGroup, fId, props.getProperty(PROPS_PROTOCOL)), timeoutMs, limit); + fGroup, fId, props.getProperty(PROTOCOL)), timeoutMs, limit); String response = getResponse(urlPath, username, password, protocolFlag); final JSONObject o = getResponseDataInJsonWithResponseReturned(response); @@ -287,9 +319,9 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { createMRConsumerResponse(response, mrConsumerResponse); } - if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) { + if (ProtocolType.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) { final String urlPath = createUrlPath( - MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId, props.getProperty(PROPS_PROTOCOL)), + MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId, props.getProperty(PROTOCOL)), timeoutMs, limit); String response = getAuthResponse(urlPath, authKey, authDate, username, password, protocolFlag); @@ -298,9 +330,9 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { createMRConsumerResponse(response, mrConsumerResponse); } - if (ProtocolTypeConstants.HTTPNOAUTH.getValue().equalsIgnoreCase(protocolFlag)) { + if (ProtocolType.HTTPNOAUTH.getValue().equalsIgnoreCase(protocolFlag)) { final String urlPath = createUrlPath(MRConstants.makeConsumerUrl(fHostSelector.selectBaseHost(), fTopic, - fGroup, fId, props.getProperty(PROPS_PROTOCOL)), timeoutMs, limit); + fGroup, fId, props.getProperty(PROTOCOL)), timeoutMs, limit); String response = getNoAuthResponse(urlPath, username, password, protocolFlag); final JSONObject o = getResponseDataInJsonWithResponseReturned(response); @@ -311,19 +343,19 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { } catch (JSONException e) { mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR)); mrConsumerResponse.setResponseMessage(e.getMessage()); - log.error("json exception: ", e); + logger.error("json exception: ", e); } catch (HttpException e) { mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR)); mrConsumerResponse.setResponseMessage(e.getMessage()); - log.error("http exception: ", e); + logger.error("http exception: ", e); } catch (DME2Exception e) { mrConsumerResponse.setResponseCode(e.getErrorCode()); mrConsumerResponse.setResponseMessage(e.getErrorMessage()); - log.error("DME2 exception: ", e); + logger.error("DME2 exception: ", e); } catch (Exception e) { mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR)); mrConsumerResponse.setResponseMessage(e.getMessage()); - log.error(EXECPTION_MESSAGE, e); + logger.error(EXECPTION_MESSAGE, e); } mrConsumerResponse.setActualMessages(msgs); return mrConsumerResponse; @@ -331,16 +363,16 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { @Override protected void reportProblemWithResponse() { - log.warn("There was a problem with the server response. Blacklisting for 3 minutes."); + logger.warn("There was a problem with the server response. Blacklisting for 3 minutes."); super.reportProblemWithResponse(); fHostSelector.reportReachabilityProblem(3, TimeUnit.MINUTES); } private void createMRConsumerResponse(String reply, MRConsumerResponse mrConsumerResponse) { if (reply.startsWith("{")) { - JSONObject jObject = new JSONObject(reply); - String message = jObject.getString("message"); - int status = jObject.getInt("status"); + JSONObject jsonObject = new JSONObject(reply); + String message = jsonObject.getString("message"); + int status = jsonObject.getInt("status"); mrConsumerResponse.setResponseCode(Integer.toString(status)); @@ -372,7 +404,7 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { return jsonObject; } catch (JSONException excp) { - log.error("DMAAP - Error reading response data.", excp); + logger.error("DMAAP - Error reading response data.", excp); return null; } } @@ -403,22 +435,22 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { private void dmeConfigure(int timeoutMs, int limit) throws IOException, DME2Exception, URISyntaxException { this.longPollingMs = timeoutMs; - String latitude = props.getProperty("Latitude"); - String longitude = props.getProperty("Longitude"); - String version = props.getProperty("Version"); - String serviceName = props.getProperty("ServiceName"); - String env = props.getProperty("Environment"); - String partner = props.getProperty("Partner"); - String routeOffer = props.getProperty("routeOffer"); - String subContextPath = props.getProperty("SubContextPath") + fTopic + "/" + fGroup + "/" + fId; - String protocol = props.getProperty(PROPS_PROTOCOL); - String methodType = props.getProperty("MethodType"); - String dmeuser = props.getProperty("username"); - String dmepassword = props.getProperty("password"); - String contenttype = props.getProperty("contenttype"); - String handlers = props.getProperty("sessionstickinessrequired"); - - /** + String latitude = props.getProperty(LATITUDE); + String longitude = props.getProperty(LONGITUDE); + String version = props.getProperty(VERSION); + String serviceName = props.getProperty(SERVICE_NAME); + String env = props.getProperty(ENVIRONMENT); + String partner = props.getProperty(PARTNER); + String routeOffer = props.getProperty(ROUTE_OFFER); + String subContextPath = props.getProperty(SUB_CONTEXT_PATH) + fTopic + "/" + fGroup + "/" + fId; + String protocol = props.getProperty(PROTOCOL); + String methodType = props.getProperty(METHOD_TYPE); + String dmeuser = props.getProperty(USERNAME); + String dmepassword = props.getProperty(USERNAME); + String contenttype = props.getProperty(CONTENT_TYPE); + String handlers = props.getProperty(SESSION_STICKINESS_REQUIRED); + + /* * Changes to DME2Client url to use Partner for auto failover between data centers When Partner value is not * provided use the routeOffer value for auto failover within a cluster */ @@ -458,13 +490,13 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { url = contextUrl.toString(); DMETimeOuts = new HashMap<>(); - DMETimeOuts.put("AFT_DME2_EP_READ_TIMEOUT_MS", props.getProperty("AFT_DME2_EP_READ_TIMEOUT_MS")); - DMETimeOuts.put("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", props.getProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS")); - DMETimeOuts.put("AFT_DME2_EP_CONN_TIMEOUT", props.getProperty("AFT_DME2_EP_CONN_TIMEOUT")); + DMETimeOuts.put("AFT_DME2_EP_READ_TIMEOUT_MS", props.getProperty(AFT_DME2_EP_READ_TIMEOUT_MS)); + DMETimeOuts.put("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", props.getProperty(AFT_DME2_ROUNDTRIP_TIMEOUT_MS)); + DMETimeOuts.put("AFT_DME2_EP_CONN_TIMEOUT", props.getProperty(AFT_DME2_EP_CONN_TIMEOUT)); DMETimeOuts.put("Content-Type", contenttype); System.setProperty("AFT_LATITUDE", latitude); System.setProperty("AFT_LONGITUDE", longitude); - System.setProperty("AFT_ENVIRONMENT", props.getProperty("AFT_ENVIRONMENT")); + System.setProperty("AFT_ENVIRONMENT", props.getProperty(AFT_ENVIRONMENT)); // SSL changes System.setProperty("AFT_DME2_CLIENT_SSL_INCLUDE_PROTOCOLS", "TLSv1.1,TLSv1.2"); @@ -474,7 +506,7 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { long dme2PerEndPointTimeoutMs; try { - dme2PerEndPointTimeoutMs = Long.parseLong(props.getProperty("DME2_PER_HANDLER_TIMEOUT_MS")); + dme2PerEndPointTimeoutMs = Long.parseLong(props.getProperty(DME2_PER_HANDLER_TIMEOUT_MS)); // backward compatibility if (dme2PerEndPointTimeoutMs <= 0) { dme2PerEndPointTimeoutMs = timeoutMs + DEFAULT_DME2_PER_ENDPOINT_TIMEOUT_MS; @@ -483,15 +515,15 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { // backward compatibility dme2PerEndPointTimeoutMs = timeoutMs + DEFAULT_DME2_PER_ENDPOINT_TIMEOUT_MS; getLog().debug( - "DME2_PER_HANDLER_TIMEOUT_MS not set and using default " + DEFAULT_DME2_PER_ENDPOINT_TIMEOUT_MS); + DME2_PER_HANDLER_TIMEOUT_MS + " not set and using default " + DEFAULT_DME2_PER_ENDPOINT_TIMEOUT_MS); } try { - dme2ReplyHandlerTimeoutMs = Long.parseLong(props.getProperty("DME2_REPLY_HANDLER_TIMEOUT_MS")); + dme2ReplyHandlerTimeoutMs = Long.parseLong(props.getProperty(DME2_REPLY_HANDLER_TIMEOUT_MS)); } catch (NumberFormatException nfe) { try { - long dme2EpReadTimeoutMs = Long.parseLong(props.getProperty("AFT_DME2_EP_READ_TIMEOUT_MS")); - long dme2EpConnTimeoutMs = Long.parseLong(props.getProperty("AFT_DME2_EP_CONN_TIMEOUT")); + long dme2EpReadTimeoutMs = Long.parseLong(props.getProperty(AFT_DME2_EP_READ_TIMEOUT_MS)); + long dme2EpConnTimeoutMs = Long.parseLong(props.getProperty(AFT_DME2_EP_CONN_TIMEOUT)); dme2ReplyHandlerTimeoutMs = timeoutMs + dme2EpReadTimeoutMs + dme2EpConnTimeoutMs; getLog().debug( "DME2_REPLY_HANDLER_TIMEOUT_MS not set and using default from timeoutMs, AFT_DME2_EP_READ_TIMEOUT_MS and AFT_DME2_EP_CONN_TIMEOUT " @@ -518,9 +550,9 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { sender.setPayload(""); if (handlers != null && handlers.equalsIgnoreCase("yes")) { sender.addHeader("AFT_DME2_EXCHANGE_REQUEST_HANDLERS", - props.getProperty("AFT_DME2_EXCHANGE_REQUEST_HANDLERS")); - sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REPLY_HANDLERS")); - sender.addHeader("AFT_DME2_REQ_TRACE_ON", props.getProperty("AFT_DME2_REQ_TRACE_ON")); + props.getProperty(AFT_DME2_EXCHANGE_REQUEST_HANDLERS)); + sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", props.getProperty(AFT_DME2_EXCHANGE_REPLY_HANDLERS)); + sender.addHeader("AFT_DME2_REQ_TRACE_ON", props.getProperty(AFT_DME2_REQ_TRACE_ON)); } else { sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", "com.att.nsa.mr.dme.client.HeaderReplyHandler"); } @@ -548,7 +580,7 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { } adds.append("filter=").append(URLEncoder.encode(fFilter, "UTF-8")); } catch (UnsupportedEncodingException e) { - log.error("exception at createUrlPath () : ", e); + logger.error("exception at createUrlPath () : ", e); } } @@ -560,10 +592,10 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { } private String readRoute(String routeKey) { - try(InputStream input = new FileInputStream(MRClientFactory.routeFilePath)) { + try (InputStream input = new FileInputStream(MRClientFactory.routeFilePath)) { MRClientFactory.prop.load(input); } catch (Exception ex) { - log.error("Reply Router Error " + ex); + logger.error("Reply Router Error " + ex); } return MRClientFactory.prop.getProperty(routeKey); } |