src/main/java/org/onap/dmaap/mr/client/impl/MRConsumerImpl.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  *  ================================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *  You may obtain a copy of the License at
  10  *        http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License.
  17  *  ============LICENSE_END=========================================================
  18  *
  19  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20  *
  21  *******************************************************************************/
  22 package org.onap.dmaap.mr.client.impl;
  23
  24 import com.att.aft.dme2.api.DME2Client;
  25 import com.att.aft.dme2.api.DME2Exception;
  26 import org.onap.dmaap.mr.client.HostSelector;
  27 import org.onap.dmaap.mr.client.MRClientFactory;
  28 import org.onap.dmaap.mr.client.MRConsumer;
  29 import org.onap.dmaap.mr.client.response.MRConsumerResponse;
  30 import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
  31 import java.io.File;
  32 import java.io.FileReader;
  33 import java.io.IOException;
  34 import java.io.UnsupportedEncodingException;
  35 import java.net.MalformedURLException;
  36 import java.net.URI;
  37 import java.net.URISyntaxException;
  38 import java.net.URLEncoder;
  39 import java.util.Collection;
  40 import java.util.HashMap;
  41 import java.util.LinkedList;
  42 import java.util.List;
  43 import java.util.Properties;
  44 import java.util.concurrent.TimeUnit;
  45 import org.apache.http.HttpException;
  46 import org.apache.http.HttpStatus;
  47 import org.json.JSONArray;
  48 import org.json.JSONException;
  49 import org.json.JSONObject;
  50 import org.json.JSONTokener;
  51 import org.slf4j.Logger;
  52 import org.slf4j.LoggerFactory;
  53
  54 public class MRConsumerImpl extends MRBaseClient implements MRConsumer {
  55
  56     private Logger log = LoggerFactory.getLogger(this.getClass().getName());
  57
  58     public static final String routerFilePath = null;
  59
  60     public String protocolFlag = ProtocolTypeConstants.DME2.getValue();
  61     public String consumerFilePath;
  62
  63     private static final String SUCCESS_MESSAGE = "Success";
  64     private static final long DEFAULT_DME2_PER_ENDPOINT_TIMEOUT_MS = 10000L;
  65     private static final long DEFAULT_DME2_REPLY_HANDLER_TIMEOUT_MS = 10000L;
  66
  67     private final String fTopic;
  68     private final String fGroup;
  69     private final String fId;
  70     private final int fTimeoutMs;
  71     private final int fLimit;
  72     private String fFilter;
  73     private String username;
  74     private String password;
  75     private String host;
  76     private HostSelector fHostSelector = null;
  77     private String url;
  78     private DME2Client sender;
  79     private String authKey;
  80     private String authDate;
  81     private Properties props;
  82     private HashMap<String, String> DMETimeOuts;
  83     private long dme2ReplyHandlerTimeoutMs;
  84     private long longPollingMs;
  85
  86     public MRConsumerImpl(Collection<String> hostPart, final String topic, final String consumerGroup,
  87             final String consumerId, int timeoutMs, int limit, String filter, String apiKey_username,
  88             String apiSecret_password) throws MalformedURLException {
  89         this(hostPart, topic, consumerGroup, consumerId, timeoutMs, limit, filter, apiKey_username, apiSecret_password,
  90                 false);
  91     }
  92
  93     public MRConsumerImpl(Collection<String> hostPart, final String topic, final String consumerGroup,
  94             final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret,
  95             boolean allowSelfSignedCerts) throws MalformedURLException {
  96         super(hostPart, topic + "::" + consumerGroup + "::" + consumerId);
  97
  98         fTopic = topic;
  99         fGroup = consumerGroup;
 100         fId = consumerId;
 101         fTimeoutMs = timeoutMs;
 102         fLimit = limit;
 103         fFilter = filter;
 104
 105         fHostSelector = new HostSelector(hostPart);
 106     }
 107
 108     @Override
 109     public Iterable<String> fetch() throws IOException, Exception {
 110         // fetch with the timeout and limit set in constructor
 111         return fetch(fTimeoutMs, fLimit);
 112     }
 113
 114     @Override
 115     public Iterable<String> fetch(int timeoutMs, int limit) throws Exception {
 116         final LinkedList<String> msgs = new LinkedList<>();
 117
 118         try {
 119             if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
 120                 dmeConfigure(timeoutMs, limit);
 121                 try {
 122                     String reply = sender.sendAndWait(timeoutMs + 10000L);
 123                     final JSONObject o = getResponseDataInJson(reply);
 124                     if (o != null) {
 125                         final JSONArray a = o.getJSONArray("result");
 126                         if (a != null) {
 127                             for (int i = 0; i < a.length(); i++) {
 128                                 if (a.get(i) instanceof String)
 129                                     msgs.add(a.getString(i));
 130                                 else
 131                                     msgs.add(a.getJSONObject(i).toString());
 132                             }
 133                         }
 134                     }
 135                 } catch (JSONException e) {
 136                     // unexpected response
 137                     reportProblemWithResponse();
 138                     log.error("exception: ", e);
 139                 } catch (HttpException e) {
 140                     throw new IOException(e);
 141                 }
 142             }
 143
 144             if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
 145                 final String urlPath = createUrlPath(MRConstants.makeConsumerUrl(fHostSelector.selectBaseHost(), fTopic,
 146                         fGroup, fId, props.getProperty("Protocol")), timeoutMs, limit);
 147
 148                 try {
 149                     final JSONObject o = get(urlPath, username, password, protocolFlag);
 150
 151                     if (o != null) {
 152                         final JSONArray a = o.getJSONArray("result");
 153                         if (a != null) {
 154                             for (int i = 0; i < a.length(); i++) {
 155                                 if (a.get(i) instanceof String)
 156                                     msgs.add(a.getString(i));
 157                                 else
 158                                     msgs.add(a.getJSONObject(i).toString());
 159                             }
 160                         }
 161                     }
 162                 } catch (JSONException e) {
 163                     // unexpected response
 164                     reportProblemWithResponse();
 165                     log.error("exception: ", e);
 166                 } catch (HttpException e) {
 167                     throw new IOException(e);
 168                 }
 169             }
 170
 171             if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
 172                 final String urlPath = createUrlPath(
 173                         MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId, props.getProperty("Protocol")),
 174                         timeoutMs, limit);
 175
 176                 try {
 177                     final JSONObject o = getAuth(urlPath, authKey, authDate, username, password, protocolFlag);
 178                     if (o != null) {
 179                         final JSONArray a = o.getJSONArray("result");
 180                         if (a != null) {
 181                             for (int i = 0; i < a.length(); i++) {
 182                                 if (a.get(i) instanceof String)
 183                                     msgs.add(a.getString(i));
 184                                 else
 185                                     msgs.add(a.getJSONObject(i).toString());
 186                             }
 187                         }
 188                     }
 189                 } catch (JSONException e) {
 190                     // unexpected response
 191                     reportProblemWithResponse();
 192                     log.error("exception: ", e);
 193                 } catch (HttpException e) {
 194                     throw new IOException(e);
 195                 }
 196             }
 197
 198             if (ProtocolTypeConstants.HTTPNOAUTH.getValue().equalsIgnoreCase(protocolFlag)) {
 199                 final String urlPath = createUrlPath(MRConstants.makeConsumerUrl(fHostSelector.selectBaseHost(), fTopic,
 200                         fGroup, fId, props.getProperty("Protocol")), timeoutMs, limit);
 201
 202                 try {
 203                     final JSONObject o = getNoAuth(urlPath);
 204                     if (o != null) {
 205                         final JSONArray a = o.getJSONArray("result");
 206                         if (a != null) {
 207                             for (int i = 0; i < a.length(); i++) {
 208                                 if (a.get(i) instanceof String)
 209                                     msgs.add(a.getString(i));
 210                                 else
 211                                     msgs.add(a.getJSONObject(i).toString());
 212                             }
 213                         }
 214                     }
 215                 } catch (JSONException e) {
 216                     // unexpected response
 217                     reportProblemWithResponse();
 218                     log.error("exception: ", e);
 219                 } catch (HttpException e) {
 220                     throw new IOException(e);
 221                 }
 222             }
 223         } catch (JSONException e) {
 224             // unexpected response
 225             reportProblemWithResponse();
 226             log.error("exception: ", e);
 227         } catch (HttpException e) {
 228             throw new IOException(e);
 229         } catch (Exception e) {
 230             throw e;
 231         }
 232
 233         return msgs;
 234     }
 235
 236     @Override
 237     public MRConsumerResponse fetchWithReturnConsumerResponse() {
 238         // fetch with the timeout and limit set in constructor
 239         return fetchWithReturnConsumerResponse(fTimeoutMs, fLimit);
 240     }
 241
 242     @Override
 243     public MRConsumerResponse fetchWithReturnConsumerResponse(int timeoutMs, int limit) {
 244         final LinkedList<String> msgs = new LinkedList<String>();
 245         MRConsumerResponse mrConsumerResponse = new MRConsumerResponse();
 246         try {
 247             if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
 248                 dmeConfigure(timeoutMs, limit);
 249
 250                 long timeout = (dme2ReplyHandlerTimeoutMs > 0 && longPollingMs == timeoutMs) ? dme2ReplyHandlerTimeoutMs
 251                         : (timeoutMs + DEFAULT_DME2_REPLY_HANDLER_TIMEOUT_MS);
 252                 String reply = sender.sendAndWait(timeout);
 253
 254                 final JSONObject o = getResponseDataInJsonWithResponseReturned(reply);
 255
 256                 if (o != null) {
 257                     final JSONArray a = o.getJSONArray("result");
 258
 259                     if (a != null) {
 260                         for (int i = 0; i < a.length(); i++) {
 261                             if (a.get(i) instanceof String)
 262                                 msgs.add(a.getString(i));
 263                             else
 264                                 msgs.add(a.getJSONObject(i).toString());
 265                         }
 266                     }
 267                 }
 268                 createMRConsumerResponse(reply, mrConsumerResponse);
 269             }
 270
 271             if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
 272                 final String urlPath = createUrlPath(MRConstants.makeConsumerUrl(fHostSelector.selectBaseHost(), fTopic,
 273                         fGroup, fId, props.getProperty("Protocol")), timeoutMs, limit);
 274
 275                 String response = getResponse(urlPath, username, password, protocolFlag);
 276                 final JSONObject o = getResponseDataInJsonWithResponseReturned(response);
 277                 if (o != null) {
 278                     final JSONArray a = o.getJSONArray("result");
 279
 280                     if (a != null) {
 281                         for (int i = 0; i < a.length(); i++) {
 282                             if (a.get(i) instanceof String)
 283                                 msgs.add(a.getString(i));
 284                             else
 285                                 msgs.add(a.getJSONObject(i).toString());
 286                         }
 287                     }
 288                 }
 289                 createMRConsumerResponse(response, mrConsumerResponse);
 290             }
 291
 292             if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
 293                 final String urlPath = createUrlPath(
 294                         MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId, props.getProperty("Protocol")),
 295                         timeoutMs, limit);
 296
 297                 String response = getAuthResponse(urlPath, authKey, authDate, username, password, protocolFlag);
 298                 final JSONObject o = getResponseDataInJsonWithResponseReturned(response);
 299                 if (o != null) {
 300                     final JSONArray a = o.getJSONArray("result");
 301
 302                     if (a != null) {
 303                         for (int i = 0; i < a.length(); i++) {
 304                             if (a.get(i) instanceof String)
 305                                 msgs.add(a.getString(i));
 306                             else
 307                                 msgs.add(a.getJSONObject(i).toString());
 308                         }
 309                     }
 310                 }
 311                 createMRConsumerResponse(response, mrConsumerResponse);
 312             }
 313
 314             if (ProtocolTypeConstants.HTTPNOAUTH.getValue().equalsIgnoreCase(protocolFlag)) {
 315                 final String urlPath = createUrlPath(MRConstants.makeConsumerUrl(fHostSelector.selectBaseHost(), fTopic,
 316                         fGroup, fId, props.getProperty("Protocol")), timeoutMs, limit);
 317
 318                 String response = getNoAuthResponse(urlPath, username, password, protocolFlag);
 319                 final JSONObject o = getResponseDataInJsonWithResponseReturned(response);
 320                 if (o != null) {
 321                     final JSONArray a = o.getJSONArray("result");
 322
 323                     if (a != null) {
 324                         for (int i = 0; i < a.length(); i++) {
 325                             if (a.get(i) instanceof String)
 326                                 msgs.add(a.getString(i));
 327                             else
 328                                 msgs.add(a.getJSONObject(i).toString());
 329                         }
 330                     }
 331                 }
 332                 createMRConsumerResponse(response, mrConsumerResponse);
 333             }
 334
 335         } catch (JSONException e) {
 336             mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
 337             mrConsumerResponse.setResponseMessage(e.getMessage());
 338             log.error("json exception: ", e);
 339         } catch (HttpException e) {
 340             mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
 341             mrConsumerResponse.setResponseMessage(e.getMessage());
 342             log.error("http exception: ", e);
 343         } catch (DME2Exception e) {
 344             mrConsumerResponse.setResponseCode(e.getErrorCode());
 345             mrConsumerResponse.setResponseMessage(e.getErrorMessage());
 346             log.error("DME2 exception: ", e);
 347         } catch (Exception e) {
 348             mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
 349             mrConsumerResponse.setResponseMessage(e.getMessage());
 350             log.error("exception: ", e);
 351         }
 352         mrConsumerResponse.setActualMessages(msgs);
 353         return mrConsumerResponse;
 354     }
 355
 356     @Override
 357     protected void reportProblemWithResponse() {
 358         log.warn("There was a problem with the server response. Blacklisting for 3 minutes.");
 359         super.reportProblemWithResponse();
 360         fHostSelector.reportReachabilityProblem(3, TimeUnit.MINUTES);
 361     }
 362
 363     private void createMRConsumerResponse(String reply, MRConsumerResponse mrConsumerResponse) {
 364         if (reply.startsWith("{")) {
 365             JSONObject jObject = new JSONObject(reply);
 366             String message = jObject.getString("message");
 367             int status = jObject.getInt("status");
 368
 369             mrConsumerResponse.setResponseCode(Integer.toString(status));
 370
 371             if (null != message) {
 372                 mrConsumerResponse.setResponseMessage(message);
 373             }
 374         } else if (reply.startsWith("<")) {
 375             mrConsumerResponse.setResponseCode(getHTTPErrorResponseCode(reply));
 376             mrConsumerResponse.setResponseMessage(getHTTPErrorResponseMessage(reply));
 377         } else {
 378             mrConsumerResponse.setResponseCode(String.valueOf(HttpStatus.SC_OK));
 379             mrConsumerResponse.setResponseMessage(SUCCESS_MESSAGE);
 380         }
 381     }
 382
 383     private JSONObject getResponseDataInJson(String response) {
 384         try {
 385             JSONTokener jsonTokener = new JSONTokener(response);
 386             JSONObject jsonObject = null;
 387             final char firstChar = jsonTokener.next();
 388             jsonTokener.back();
 389             if ('[' == firstChar) {
 390                 JSONArray jsonArray = new JSONArray(jsonTokener);
 391                 jsonObject = new JSONObject();
 392                 jsonObject.put("result", jsonArray);
 393             } else {
 394                 jsonObject = new JSONObject(jsonTokener);
 395             }
 396
 397             return jsonObject;
 398         } catch (JSONException excp) {
 399             log.error("DMAAP - Error reading response data.", excp);
 400             return null;
 401         }
 402     }
 403
 404     private JSONObject getResponseDataInJsonWithResponseReturned(String response) {
 405         JSONTokener jsonTokener = new JSONTokener(response);
 406         JSONObject jsonObject = null;
 407         final char firstChar = jsonTokener.next();
 408         jsonTokener.back();
 409         if (null != response && response.length() == 0) {
 410             return null;
 411         }
 412
 413         if ('[' == firstChar) {
 414             JSONArray jsonArray = new JSONArray(jsonTokener);
 415             jsonObject = new JSONObject();
 416             jsonObject.put("result", jsonArray);
 417         } else if ('{' == firstChar) {
 418             return null;
 419         } else if ('<' == firstChar) {
 420             return null;
 421         } else {
 422             jsonObject = new JSONObject(jsonTokener);
 423         }
 424
 425         return jsonObject;
 426     }
 427
 428     private void dmeConfigure(int timeoutMs, int limit) throws IOException, DME2Exception, URISyntaxException {
 429         this.longPollingMs = timeoutMs;
 430         String latitude = props.getProperty("Latitude");
 431         String longitude = props.getProperty("Longitude");
 432         String version = props.getProperty("Version");
 433         String serviceName = props.getProperty("ServiceName");
 434         String env = props.getProperty("Environment");
 435         String partner = props.getProperty("Partner");
 436         String routeOffer = props.getProperty("routeOffer");
 437         String subContextPath = props.getProperty("SubContextPath") + fTopic + "/" + fGroup + "/" + fId;
 438         String protocol = props.getProperty("Protocol");
 439         String methodType = props.getProperty("MethodType");
 440         String dmeuser = props.getProperty("username");
 441         String dmepassword = props.getProperty("password");
 442         String contenttype = props.getProperty("contenttype");
 443         String handlers = props.getProperty("sessionstickinessrequired");
 444
 445         /**
 446          * Changes to DME2Client url to use Partner for auto failover between data centers When Partner value is not
 447          * provided use the routeOffer value for auto failover within a cluster
 448          */
 449
 450         String preferredRouteKey = readRoute("preferredRouteKey");
 451
 452         if (partner != null && !partner.isEmpty() && preferredRouteKey != null && !preferredRouteKey.isEmpty()) {
 453             url = protocol + "://" + serviceName + "?version=" + version + "&envContext=" + env + "&partner=" + partner
 454                     + "&routeoffer=" + preferredRouteKey;
 455         } else if (partner != null && !partner.isEmpty()) {
 456             url = protocol + "://" + serviceName + "?version=" + version + "&envContext=" + env + "&partner=" + partner;
 457         } else if (routeOffer != null && !routeOffer.isEmpty()) {
 458             url = protocol + "://" + serviceName + "?version=" + version + "&envContext=" + env + "&routeoffer="
 459                     + routeOffer;
 460         }
 461
 462         if (timeoutMs != -1)
 463             url = url + "&timeout=" + timeoutMs;
 464         if (limit != -1)
 465             url = url + "&limit=" + limit;
 466
 467         // Add filter to DME2 Url
 468         if (fFilter != null && fFilter.length() > 0)
 469             url = url + "&filter=" + URLEncoder.encode(fFilter, "UTF-8");
 470
 471         DMETimeOuts = new HashMap<>();
 472         DMETimeOuts.put("AFT_DME2_EP_READ_TIMEOUT_MS", props.getProperty("AFT_DME2_EP_READ_TIMEOUT_MS"));
 473         DMETimeOuts.put("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", props.getProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS"));
 474         DMETimeOuts.put("AFT_DME2_EP_CONN_TIMEOUT", props.getProperty("AFT_DME2_EP_CONN_TIMEOUT"));
 475         DMETimeOuts.put("Content-Type", contenttype);
 476         System.setProperty("AFT_LATITUDE", latitude);
 477         System.setProperty("AFT_LONGITUDE", longitude);
 478         System.setProperty("AFT_ENVIRONMENT", props.getProperty("AFT_ENVIRONMENT"));
 479
 480         // SSL changes
 481         System.setProperty("AFT_DME2_CLIENT_SSL_INCLUDE_PROTOCOLS", "TLSv1.1,TLSv1.2");
 482         System.setProperty("AFT_DME2_CLIENT_IGNORE_SSL_CONFIG", "false");
 483         System.setProperty("AFT_DME2_CLIENT_KEYSTORE_PASSWORD", "changeit");
 484         // SSL changes
 485
 486         long dme2PerEndPointTimeoutMs;
 487         try {
 488             dme2PerEndPointTimeoutMs = Long.parseLong(props.getProperty("DME2_PER_HANDLER_TIMEOUT_MS"));
 489             // backward compatibility
 490             if (dme2PerEndPointTimeoutMs <= 0) {
 491                 dme2PerEndPointTimeoutMs = timeoutMs + DEFAULT_DME2_PER_ENDPOINT_TIMEOUT_MS;
 492             }
 493         } catch (NumberFormatException nfe) {
 494             // backward compatibility
 495             dme2PerEndPointTimeoutMs = timeoutMs + DEFAULT_DME2_PER_ENDPOINT_TIMEOUT_MS;
 496             getLog().debug(
 497                     "DME2_PER_HANDLER_TIMEOUT_MS not set and using default " + DEFAULT_DME2_PER_ENDPOINT_TIMEOUT_MS);
 498         }
 499
 500         try {
 501             dme2ReplyHandlerTimeoutMs = Long.parseLong(props.getProperty("DME2_REPLY_HANDLER_TIMEOUT_MS"));
 502         } catch (NumberFormatException nfe) {
 503             try {
 504                 long dme2EpReadTimeoutMs = Long.parseLong(props.getProperty("AFT_DME2_EP_READ_TIMEOUT_MS"));
 505                 long dme2EpConnTimeoutMs = Long.parseLong(props.getProperty("AFT_DME2_EP_CONN_TIMEOUT"));
 506                 dme2ReplyHandlerTimeoutMs = timeoutMs + dme2EpReadTimeoutMs + dme2EpConnTimeoutMs;
 507                 getLog().debug(
 508                         "DME2_REPLY_HANDLER_TIMEOUT_MS not set and using default from timeoutMs, AFT_DME2_EP_READ_TIMEOUT_MS and AFT_DME2_EP_CONN_TIMEOUT "
 509                                 + dme2ReplyHandlerTimeoutMs);
 510             } catch (NumberFormatException e) {
 511                 // backward compatibility
 512                 dme2ReplyHandlerTimeoutMs = timeoutMs + DEFAULT_DME2_REPLY_HANDLER_TIMEOUT_MS;
 513                 getLog().debug("DME2_REPLY_HANDLER_TIMEOUT_MS not set and using default " + dme2ReplyHandlerTimeoutMs);
 514             }
 515         }
 516         // backward compatibility
 517         if (dme2ReplyHandlerTimeoutMs <= 0) {
 518             dme2ReplyHandlerTimeoutMs = timeoutMs + DEFAULT_DME2_REPLY_HANDLER_TIMEOUT_MS;
 519         }
 520
 521         sender = new DME2Client(new URI(url), dme2PerEndPointTimeoutMs);
 522         sender.setAllowAllHttpReturnCodes(true);
 523         sender.setMethod(methodType);
 524         sender.setSubContext(subContextPath);
 525         if (dmeuser != null && dmepassword != null) {
 526             sender.setCredentials(dmeuser, dmepassword);
 527         }
 528         sender.setHeaders(DMETimeOuts);
 529         sender.setPayload("");
 530         if (handlers != null && handlers.equalsIgnoreCase("yes")) {
 531             sender.addHeader("AFT_DME2_EXCHANGE_REQUEST_HANDLERS",
 532                     props.getProperty("AFT_DME2_EXCHANGE_REQUEST_HANDLERS"));
 533             sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REPLY_HANDLERS"));
 534             sender.addHeader("AFT_DME2_REQ_TRACE_ON", props.getProperty("AFT_DME2_REQ_TRACE_ON"));
 535         } else {
 536             sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", "com.att.nsa.mr.dme.client.HeaderReplyHandler");
 537         }
 538     }
 539
 540     protected String createUrlPath(String url, int timeoutMs, int limit) throws IOException {
 541         final StringBuilder contexturl = new StringBuilder(url);
 542         final StringBuilder adds = new StringBuilder();
 543
 544         if (timeoutMs > -1) {
 545             adds.append("timeout=").append(timeoutMs);
 546         }
 547
 548         if (limit > -1) {
 549             if (adds.length() > 0) {
 550                 adds.append("&");
 551             }
 552             adds.append("limit=").append(limit);
 553         }
 554
 555         if (fFilter != null && fFilter.length() > 0) {
 556             try {
 557                 if (adds.length() > 0) {
 558                     adds.append("&");
 559                 }
 560                 adds.append("filter=").append(URLEncoder.encode(fFilter, "UTF-8"));
 561             } catch (UnsupportedEncodingException e) {
 562                 log.error("exception at createUrlPath ()  :  ", e);
 563             }
 564         }
 565
 566         if (adds.length() > 0) {
 567             contexturl.append("?").append(adds.toString());
 568         }
 569
 570         return contexturl.toString();
 571     }
 572
 573     private String readRoute(String routeKey) {
 574         try {
 575             MRClientFactory.prop.load(new FileReader(new File(MRClientFactory.routeFilePath)));
 576         } catch (Exception ex) {
 577             log.error("Reply Router Error " + ex);
 578         }
 579         return MRClientFactory.prop.getProperty(routeKey);
 580     }
 581
 582     public static List<String> stringToList(String str) {
 583         final LinkedList<String> set = new LinkedList<>();
 584         if (str != null) {
 585             final String[] parts = str.trim().split(",");
 586             for (String part : parts) {
 587                 final String trimmed = part.trim();
 588                 if (trimmed.length() > 0) {
 589                     set.add(trimmed);
 590                 }
 591             }
 592         }
 593         return set;
 594     }
 595
 596     public static String getRouterFilePath() {
 597         return routerFilePath;
 598     }
 599
 600     public static void setRouterFilePath(String routerFilePath) {
 601         MRSimplerBatchPublisher.routerFilePath = routerFilePath;
 602     }
 603
 604     public String getConsumerFilePath() {
 605         return consumerFilePath;
 606     }
 607
 608     public void setConsumerFilePath(String consumerFilePath) {
 609         this.consumerFilePath = consumerFilePath;
 610     }
 611
 612     public String getProtocolFlag() {
 613         return protocolFlag;
 614     }
 615
 616     public void setProtocolFlag(String protocolFlag) {
 617         this.protocolFlag = protocolFlag;
 618     }
 619
 620     public Properties getProps() {
 621         return props;
 622     }
 623
 624     public void setProps(Properties props) {
 625         this.props = props;
 626     }
 627
 628     public String getUsername() {
 629         return username;
 630     }
 631
 632     public void setUsername(String username) {
 633         this.username = username;
 634     }
 635
 636     public String getPassword() {
 637         return password;
 638     }
 639
 640     public void setPassword(String password) {
 641         this.password = password;
 642     }
 643
 644     public String getHost() {
 645         return host;
 646     }
 647
 648     public void setHost(String host) {
 649         this.host = host;
 650     }
 651
 652     public String getAuthKey() {
 653         return authKey;
 654     }
 655
 656     public void setAuthKey(String authKey) {
 657         this.authKey = authKey;
 658     }
 659
 660     public String getAuthDate() {
 661         return authDate;
 662     }
 663
 664     public void setAuthDate(String authDate) {
 665         this.authDate = authDate;
 666     }
 667
 668     public String getfFilter() {
 669         return fFilter;
 670     }
 671
 672     public void setfFilter(String fFilter) {
 673         this.fFilter = fFilter;
 674     }
 675 }