* ================================================================================
* 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.
* ============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;
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";
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;
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;
}
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);
}
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());
+ }
}
}
}
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
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);
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);
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);
} 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;
@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));
return jsonObject;
} catch (JSONException excp) {
- log.error("DMAAP - Error reading response data.", excp);
+ logger.error("DMAAP - Error reading response data.", excp);
return null;
}
}
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
*/
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");
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;
// 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 "
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");
}
}
adds.append("filter=").append(URLEncoder.encode(fFilter, "UTF-8"));
} catch (UnsupportedEncodingException e) {
- log.error("exception at createUrlPath () : ", e);
+ logger.error("exception at createUrlPath () : ", e);
}
}
}
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);
}
Code Review / dmaap / messagerouter / dmaapclient.git / blobdiff