diff options
author | 2018-06-14 18:05:40 -0400 | |
---|---|---|
committer | 2018-06-14 18:07:36 -0400 | |
commit | 2721925ea824afb932b09aa9575fdb35311325d7 (patch) | |
tree | ff0f4ee805254ea05873ece12b9b928872d7f2a5 /src/main/java/com/att/nsa/mr/client | |
parent | 5519f8372acf8d4f573739b8e819f7a1e569a019 (diff) |
fixes for null pointer exceptions
Issue-ID: DMAAP-519
Change-Id: Ia32d0bd58c5f438b50e361221a96b988b00a1120
Signed-off-by: sunil unnava <su622b@att.com>
Diffstat (limited to 'src/main/java/com/att/nsa/mr/client')
3 files changed, 69 insertions, 12 deletions
diff --git a/src/main/java/com/att/nsa/mr/client/MRClientFactory.java b/src/main/java/com/att/nsa/mr/client/MRClientFactory.java index 32b44e4..f237776 100644 --- a/src/main/java/com/att/nsa/mr/client/MRClientFactory.java +++ b/src/main/java/com/att/nsa/mr/client/MRClientFactory.java @@ -39,6 +39,7 @@ import com.att.nsa.mr.client.impl.MRConsumerImpl; import com.att.nsa.mr.client.impl.MRMetaClient; import com.att.nsa.mr.client.impl.MRSimplerBatchPublisher; import com.att.nsa.mr.test.clients.ProtocolTypeConstants; +import com.att.nsa.mr.tools.ValidatorUtil; /** * A factory for MR clients.<br/> @@ -483,7 +484,7 @@ public class MRClientFactory { MRSimplerBatchPublisher pub; if (withResponse) { pub = new MRSimplerBatchPublisher.Builder() - .againstUrls(MRConsumerImpl.stringToList(props.getProperty("host"))) + .againstUrlsOrServiceName(MRConsumerImpl.stringToList(props.getProperty("host")),MRConsumerImpl.stringToList(props.getProperty("ServiceName")), props.getProperty("TransportType")) .onTopic(props.getProperty("topic")) .batchTo(Integer.parseInt(props.getProperty("maxBatchSize")), Integer.parseInt(props.getProperty("maxAgeMs").toString())) @@ -492,7 +493,7 @@ public class MRClientFactory { .withResponse(withResponse).build(); } else { pub = new MRSimplerBatchPublisher.Builder() - .againstUrls(MRConsumerImpl.stringToList(props.getProperty("host"))) + .againstUrlsOrServiceName(MRConsumerImpl.stringToList(props.getProperty("host")), MRConsumerImpl.stringToList(props.getProperty("ServiceName")), props.getProperty("TransportType")) .onTopic(props.getProperty("topic")) .batchTo(Integer.parseInt(props.getProperty("maxBatchSize")), Integer.parseInt(props.getProperty("maxAgeMs").toString())) @@ -512,13 +513,16 @@ public class MRClientFactory { } pub.setProtocolFlag(props.getProperty("TransportType")); pub.setProps(props); - routeFilePath = props.getProperty("DME2preferredRouterFilePath"); - routeReader = new FileReader(new File(routeFilePath)); prop = new Properties(); - File fo = new File(routeFilePath); - if (!fo.exists()) { - routeWriter = new FileWriter(new File(routeFilePath)); + if (props.getProperty("TransportType").equalsIgnoreCase(ProtocolTypeConstants.DME2.getValue())) { + routeFilePath = props.getProperty("DME2preferredRouterFilePath"); + routeReader = new FileReader(new File(routeFilePath)); + File fo = new File(routeFilePath); + if (!fo.exists()) { + routeWriter = new FileWriter(new File(routeFilePath)); + } } + // pub.setContentType(contentType); return pub; } @@ -623,6 +627,7 @@ public class MRClientFactory { public static MRConsumer createConsumer(Properties props) throws FileNotFoundException, IOException { int timeout; + ValidatorUtil.validateSubscriber(props); if (props.getProperty("timeout") != null) timeout = Integer.parseInt(props.getProperty("timeout")); else diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java b/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java index b54fedb..faa81ce 100644 --- a/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java +++ b/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java @@ -127,6 +127,8 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { try { // getLog().info ( "Receiving msgs from: " + // url+subContextPath ); + long timeout = (dme2ReplyHandlerTimeoutMs > 0 && longPollingMs == timeoutMs) ? dme2ReplyHandlerTimeoutMs : (timeoutMs+ defaultDme2ReplyHandlerTimeoutMs); + //String reply = sender.sendAndWait(timeout); String reply = sender.sendAndWait(timeoutMs + 10000L); final JSONObject o = getResponseDataInJson(reply); // msgs.add(reply); @@ -362,6 +364,11 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { private HashMap<String, String> DMETimeOuts; private String handlers; public static final String routerFilePath = null; + private long dme2ReplyHandlerTimeoutMs; + private long longPollingMs; + private static final long defaultDme2PerEndPointTimeoutMs = 10000L; + private static final long defaultDme2ReplyHandlerTimeoutMs = 10000L; + public static String getRouterFilePath() { return routerFilePath; @@ -388,6 +395,7 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { } private void DMEConfigure(int timeoutMs, int limit) throws IOException, DME2Exception, URISyntaxException { + this.longPollingMs = timeoutMs; latitude = props.getProperty("Latitude"); longitude = props.getProperty("Longitude"); version = props.getProperty("Version"); @@ -459,7 +467,38 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { System.setProperty("AFT_DME2_CLIENT_KEYSTORE_PASSWORD", "changeit"); // SSL changes - sender = new DME2Client(new URI(url), timeoutMs + 10000L); + long dme2PerEndPointTimeoutMs; + try { + dme2PerEndPointTimeoutMs = Long.parseLong(props.getProperty("DME2_PER_HANDLER_TIMEOUT_MS")); + //backward compatibility + if ( dme2PerEndPointTimeoutMs <= 0) { + dme2PerEndPointTimeoutMs = timeoutMs + defaultDme2PerEndPointTimeoutMs; + } + } catch (NumberFormatException nfe) { + //backward compatibility + dme2PerEndPointTimeoutMs = timeoutMs + defaultDme2PerEndPointTimeoutMs; + getLog().debug("DME2_PER_HANDLER_TIMEOUT_MS not set and using default " + defaultDme2PerEndPointTimeoutMs); + } + + try { + 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")); + 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 " + dme2ReplyHandlerTimeoutMs); + }catch (NumberFormatException e) { + //backward compatibility + dme2ReplyHandlerTimeoutMs = timeoutMs + defaultDme2ReplyHandlerTimeoutMs; + getLog().debug("DME2_REPLY_HANDLER_TIMEOUT_MS not set and using default " + dme2ReplyHandlerTimeoutMs); + } + } + //backward compatibility + if ( dme2ReplyHandlerTimeoutMs <= 0) { + dme2ReplyHandlerTimeoutMs = timeoutMs + defaultDme2ReplyHandlerTimeoutMs; + } + sender = new DME2Client(new URI(url),dme2PerEndPointTimeoutMs); sender.setAllowAllHttpReturnCodes(true); sender.setMethod(methodType); sender.setSubContext(subContextPath); @@ -469,7 +508,7 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { sender.setHeaders(DMETimeOuts); sender.setPayload(""); - if (handlers.equalsIgnoreCase("yes")) { + 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")); @@ -594,7 +633,8 @@ public class MRConsumerImpl extends MRBaseClient implements MRConsumer { if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) { DMEConfigure(timeoutMs, limit); - String reply = sender.sendAndWait(timeoutMs + 10000L); + long timeout = (dme2ReplyHandlerTimeoutMs > 0 && longPollingMs == timeoutMs) ? dme2ReplyHandlerTimeoutMs : (timeoutMs+ defaultDme2ReplyHandlerTimeoutMs); + String reply = sender.sendAndWait(timeout); final JSONObject o = getResponseDataInJsonWithResponseReturned(reply); diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java b/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java index 2f7680b..5bb5087 100644 --- a/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java +++ b/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java @@ -67,6 +67,14 @@ public class MRSimplerBatchPublisher extends MRBaseClient implements MRBatchingP fUrls = baseUrls; return this; } + + public Builder againstUrlsOrServiceName ( Collection<String> baseUrls, Collection<String> serviceName, String transportype ) + { + fUrls = baseUrls; + fServiceName = serviceName; + fTransportype = transportype; + return this; + } public Builder onTopic(String topic) { fTopic = topic; @@ -119,6 +127,8 @@ public class MRSimplerBatchPublisher extends MRBaseClient implements MRBatchingP } private Collection<String> fUrls; + private Collection<String> fServiceName; + private String fTransportype; private String fTopic; private int fMaxBatchSize = 100; private long fMaxBatchAgeMs = 1000; @@ -278,7 +288,9 @@ public class MRSimplerBatchPublisher extends MRBaseClient implements MRBatchingP final long nowMs = Clock.now(); - host = this.fHostSelector.selectBaseHost(); + if (this.fHostSelector != null) { + host = this.fHostSelector.selectBaseHost(); + } final String httpurl = MRConstants.makeUrl(host, fTopic, props.getProperty("Protocol"), props.getProperty("partition")); @@ -796,7 +808,7 @@ public class MRSimplerBatchPublisher extends MRBaseClient implements MRBatchingP sender.setSubContext(subContextPath); sender.setCredentials(dmeuser, dmepassword); sender.setHeaders(DMETimeOuts); - if (handlers.equalsIgnoreCase("yes")) { + 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", |