Merge "Sonar fix too many method param"
[dmaap/messagerouter/dmaapclient.git] / src / main / java / org / onap / dmaap / mr / client / impl / MRSimplerBatchPublisher.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 java.io.ByteArrayOutputStream;
25 import java.io.IOException;
26 import java.io.OutputStream;
27 import java.net.MalformedURLException;
28 import java.net.URI;
29 import java.net.URISyntaxException;
30 import java.util.Collection;
31 import java.util.HashMap;
32 import java.util.LinkedList;
33 import java.util.List;
34 import java.util.Map;
35 import java.util.Properties;
36 import java.util.concurrent.LinkedBlockingQueue;
37 import java.util.concurrent.ScheduledThreadPoolExecutor;
38 import java.util.concurrent.TimeUnit;
39 import java.util.zip.GZIPOutputStream;
40
41 import javax.ws.rs.core.MultivaluedMap;
42
43 import org.slf4j.Logger;
44 import org.slf4j.LoggerFactory;
45
46 import org.apache.http.HttpException;
47 import org.apache.http.HttpStatus;
48 import org.json.JSONArray;
49 import org.json.JSONObject;
50 import org.json.JSONTokener;
51
52 import com.att.aft.dme2.api.DME2Client;
53 import com.att.aft.dme2.api.DME2Exception;
54 import org.onap.dmaap.mr.client.HostSelector;
55 import org.onap.dmaap.mr.client.MRBatchingPublisher;
56 import org.onap.dmaap.mr.client.response.MRPublisherResponse;
57 import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
58
59 public class MRSimplerBatchPublisher extends MRBaseClient implements MRBatchingPublisher {
60         private static final Logger logger = LoggerFactory.getLogger(MRSimplerBatchPublisher.class);
61
62         public static class Builder {
63                 public Builder() {
64                 }
65
66                 public Builder againstUrls(Collection<String> baseUrls) {
67                         fUrls = baseUrls;
68                         return this;
69                 }
70                 
71                 public Builder againstUrlsOrServiceName ( Collection<String> baseUrls, Collection<String> serviceName, String transportype )            
72                 {               
73                         fUrls = baseUrls;               
74                         fServiceName = serviceName;             
75                         fTransportype = transportype;           
76                         return this;            
77                 }
78
79                 public Builder onTopic(String topic) {
80                         fTopic = topic;
81                         return this;
82                 }
83
84                 public Builder batchTo(int maxBatchSize, long maxBatchAgeMs) {
85                         fMaxBatchSize = maxBatchSize;
86                         fMaxBatchAgeMs = maxBatchAgeMs;
87                         return this;
88                 }
89
90                 public Builder compress(boolean compress) {
91                         fCompress = compress;
92                         return this;
93                 }
94
95                 public Builder httpThreadTime(int threadOccuranceTime) {
96                         this.threadOccuranceTime = threadOccuranceTime;
97                         return this;
98                 }
99
100                 public Builder allowSelfSignedCertificates(boolean allowSelfSignedCerts) {
101                         fAllowSelfSignedCerts = allowSelfSignedCerts;
102                         return this;
103                 }
104
105                 public Builder withResponse(boolean withResponse) {
106                         fWithResponse = withResponse;
107                         return this;
108                 }
109
110                 public MRSimplerBatchPublisher build() {
111                         if (!fWithResponse) {
112                                 try {
113                                         return new MRSimplerBatchPublisher(fUrls, fTopic, fMaxBatchSize, fMaxBatchAgeMs, fCompress,
114                                                         fAllowSelfSignedCerts, threadOccuranceTime);
115                                 } catch (MalformedURLException e) {
116                                         throw new IllegalArgumentException(e);
117                                 }
118                         } else {
119                                 try {
120                                         return new MRSimplerBatchPublisher(fUrls, fTopic, fMaxBatchSize, fMaxBatchAgeMs, fCompress,
121                                                         fAllowSelfSignedCerts, fMaxBatchSize);
122                                 } catch (MalformedURLException e) {
123                                         throw new IllegalArgumentException(e);
124                                 }
125                         }
126
127                 }
128
129                 private Collection<String> fUrls;
130                 private Collection<String> fServiceName;                
131                 private String fTransportype;   
132                 private String fTopic;
133                 private int fMaxBatchSize = 100;
134                 private long fMaxBatchAgeMs = 1000;
135                 private boolean fCompress = false;
136                 private int threadOccuranceTime = 50;
137                 private boolean fAllowSelfSignedCerts = false;
138                 private boolean fWithResponse = false;
139
140         };
141
142         @Override
143         public int send(String partition, String msg) {
144                 return send(new message(partition, msg));
145         }
146
147         @Override
148         public int send(String msg) {
149                 return send(new message(null, msg));
150         }
151
152         @Override
153         public int send(message msg) {
154                 final LinkedList<message> list = new LinkedList<message>();
155                 list.add(msg);
156                 return send(list);
157         }
158
159         @Override
160         public synchronized int send(Collection<message> msgs) {
161                 if (fClosed) {
162                         throw new IllegalStateException("The publisher was closed.");
163                 }
164
165                 for (message userMsg : msgs) {
166                         fPending.add(new TimestampedMessage(userMsg));
167                 }
168                 return getPendingMessageCount();
169         }
170
171         @Override
172         public synchronized int getPendingMessageCount() {
173                 return fPending.size();
174         }
175
176         @Override
177         public void close() {
178                 try {
179                         final List<message> remains = close(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
180                         if (remains.isEmpty()) {
181                                 getLog().warn("Closing publisher with " + remains.size() + " messages unsent. "
182                                                 + "Consider using MRBatchingPublisher.close( long timeout, TimeUnit timeoutUnits ) to recapture unsent messages on close.");
183                         }
184                 } catch (InterruptedException e) {
185                         getLog().warn("Possible message loss. " + e.getMessage(), e);
186                         Thread.currentThread().interrupt();
187                 } catch (IOException e) {
188                         getLog().warn("Possible message loss. " + e.getMessage(), e);
189                 }
190         }
191
192         @Override
193         public List<message> close(long time, TimeUnit unit) throws IOException, InterruptedException {
194                 synchronized (this) {
195                         fClosed = true;
196
197                         // stop the background sender
198                         fExec.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
199                         fExec.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
200                         fExec.shutdown();
201                 }
202
203                 final long now = Clock.now();
204                 final long waitInMs = TimeUnit.MILLISECONDS.convert(time, unit);
205                 final long timeoutAtMs = now + waitInMs;
206
207                 while (Clock.now() < timeoutAtMs && getPendingMessageCount() > 0) {
208                         send(true);
209                         Thread.sleep(250);
210                 }
211
212                 synchronized (this) {
213                         final LinkedList<message> result = new LinkedList<message>();
214                         fPending.drainTo(result);
215                         return result;
216                 }
217         }
218
219         /**
220          * Possibly send a batch to the MR server. This is called by the background
221          * thread and the close() method
222          * 
223          * @param force
224          */
225         private synchronized void send(boolean force) {
226                 if (force || shouldSendNow()) {
227                         if (!sendBatch()) {
228                                 getLog().warn("Send failed, " + fPending.size() + " message to send.");
229
230                                 // note the time for back-off
231                                 fDontSendUntilMs = sfWaitAfterError + Clock.now();
232                         }
233                 }
234         }
235
236         private synchronized boolean shouldSendNow() {
237                 boolean shouldSend = false;
238                 if (fPending.isEmpty()) {
239                         final long nowMs = Clock.now();
240
241                         shouldSend = (fPending.size() >= fMaxBatchSize);
242                         if (!shouldSend) {
243                                 final long sendAtMs = fPending.peek().timestamp + fMaxBatchAgeMs;
244                                 shouldSend = sendAtMs <= nowMs;
245                         }
246
247                         // however, wait after an error
248                         shouldSend = shouldSend && nowMs >= fDontSendUntilMs;
249                 }
250                 return shouldSend;
251         }
252
253         /**
254          * Method to parse published JSON Objects and Arrays
255          * 
256          * @return JSONArray
257          */
258         private JSONArray parseJSON() {
259                 JSONArray jsonArray = new JSONArray();
260                 for (TimestampedMessage m : fPending) {
261                         JSONTokener jsonTokener = new JSONTokener(m.fMsg);
262                         JSONObject jsonObject = null;
263                         JSONArray tempjsonArray = null;
264                         final char firstChar = jsonTokener.next();
265                         jsonTokener.back();
266                         if ('[' == firstChar) {
267                                 tempjsonArray = new JSONArray(jsonTokener);
268                                 if (null != tempjsonArray) {
269                                         for (int i = 0; i < tempjsonArray.length(); i++) {
270                                                 jsonArray.put(tempjsonArray.getJSONObject(i));
271                                         }
272                                 }
273                         } else {
274                                 jsonObject = new JSONObject(jsonTokener);
275                                 jsonArray.put(jsonObject);
276                         }
277
278                 }
279                 return jsonArray;
280         }
281
282         private synchronized boolean sendBatch() {
283                 // it's possible for this call to be made with an empty list. in this
284                 // case, just return.
285                 if (fPending.size() < 1) {
286                         return true;
287                 }
288
289                 final long nowMs = Clock.now();
290
291                 if (this.fHostSelector != null) {
292                         host = this.fHostSelector.selectBaseHost();
293                 }
294
295                 final String httpurl = MRConstants.makeUrl(host, fTopic, props.getProperty("Protocol"),
296                                 props.getProperty("partition"));
297
298                 try {
299                         
300                         final ByteArrayOutputStream baseStream = new ByteArrayOutputStream();
301                         OutputStream os = baseStream;
302                         final String contentType = props.getProperty("contenttype");
303                         if (contentType.equalsIgnoreCase("application/json")) {
304                                 JSONArray jsonArray = parseJSON();
305                                 os.write(jsonArray.toString().getBytes());
306                                 os.close();
307
308                         } else if (contentType.equalsIgnoreCase("text/plain")) {
309                                 for (TimestampedMessage m : fPending) {
310                                         os.write(m.fMsg.getBytes());
311                                         os.write('\n');
312                                 }
313                                 os.close();
314                         } else if (contentType.equalsIgnoreCase("application/cambria")
315                                         || (contentType.equalsIgnoreCase("application/cambria-zip"))) {
316                                 if (contentType.equalsIgnoreCase("application/cambria-zip")) {
317                                         os = new GZIPOutputStream(baseStream);
318                                 }
319                                 for (TimestampedMessage m : fPending) {
320
321                                         os.write(("" + m.fPartition.length()).getBytes());
322                                         os.write('.');
323                                         os.write(("" + m.fMsg.length()).getBytes());
324                                         os.write('.');
325                                         os.write(m.fPartition.getBytes());
326                                         os.write(m.fMsg.getBytes());
327                                         os.write('\n');
328                                 }
329                                 os.close();
330                         } else {
331                                 for (TimestampedMessage m : fPending) {
332                                         os.write(m.fMsg.getBytes());
333
334                                 }
335                                 os.close();
336                         }
337
338                         final long startMs = Clock.now();
339                         if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
340
341                                 DME2Configue();
342
343                                 Thread.sleep(5);
344                                 getLog().info("sending " + fPending.size() + " msgs to " + url + subContextPath + ". Oldest: "
345                                                 + (nowMs - fPending.peek().timestamp) + " ms");
346                                 sender.setPayload(os.toString());
347                                 String dmeResponse = sender.sendAndWait(5000L);
348
349                                 final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + dmeResponse.toString();
350                                 getLog().info(logLine);
351                                 fPending.clear();
352                                 return true;
353                         }
354
355             if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
356                 getLog().info("sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: "
357                         + (nowMs - fPending.peek().timestamp) + " ms");
358                 final JSONObject result =
359                         postAuth(new PostAuthDataObject().setPath(httpurl).setData(baseStream.toByteArray())
360                                 .setContentType(contentType).setAuthKey(authKey).setAuthDate(authDate)
361                                 .setUsername(username).setPassword(password).setProtocolFlag(protocolFlag));
362                 // Here we are checking for error response. If HTTP status
363                 // code is not within the http success response code
364                 // then we consider this as error and return false
365                 if (result.getInt("status") < 200 || result.getInt("status") > 299) {
366                     return false;
367                 }
368                 final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
369                 getLog().info(logLine);
370                 fPending.clear();
371                 return true;
372             }
373
374                         if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
375                                 getLog().info("sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: "
376                                                 + (nowMs - fPending.peek().timestamp) + " ms");
377                                 final JSONObject result = post(httpurl, baseStream.toByteArray(), contentType, username, password,
378                                                 protocolFlag);
379
380                                 // Here we are checking for error response. If HTTP status
381                                 // code is not within the http success response code
382                                 // then we consider this as error and return false
383                                 if (result.getInt("status") < 200 || result.getInt("status") > 299) {
384                                         return false;
385                                 }
386                                 final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
387                                 getLog().info(logLine);
388                                 fPending.clear();
389                                 return true;
390                         }
391                         
392                         if (ProtocolTypeConstants.HTTPNOAUTH.getValue().equalsIgnoreCase(protocolFlag)) {
393                                 getLog().info("sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: "
394                                                 + (nowMs - fPending.peek().timestamp) + " ms");
395                                 final JSONObject result = postNoAuth(httpurl, baseStream.toByteArray(), contentType);
396
397                                 // Here we are checking for error response. If HTTP status
398                                 // code is not within the http success response code
399                                 // then we consider this as error and return false
400                                 if (result.getInt("status") < 200 || result.getInt("status") > 299) {
401                                         return false;
402                                 }
403                                 final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
404                                 getLog().info(logLine);
405                                 fPending.clear();
406                                 return true;
407                         }
408                 } catch (IllegalArgumentException x) {
409                         getLog().warn(x.getMessage(), x);
410                 } catch (IOException x) {
411                         getLog().warn(x.getMessage(), x);
412                 } catch (HttpException x) {
413                         getLog().warn(x.getMessage(), x);
414                 } catch (Exception x) {
415                         getLog().warn(x.getMessage(), x);
416                 }
417                 return false;
418         }
419
420         public synchronized MRPublisherResponse sendBatchWithResponse() {
421                 // it's possible for this call to be made with an empty list. in this
422                 // case, just return.
423                 if (fPending.isEmpty()) {
424                         pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
425                         pubResponse.setResponseMessage("No Messages to send");
426                         return pubResponse;
427                 }
428
429                 final long nowMs = Clock.now();
430
431                 host = this.fHostSelector.selectBaseHost();
432
433                 final String httpurl = MRConstants.makeUrl(host, fTopic, props.getProperty("Protocol"),
434                                 props.getProperty("partition"));
435                 OutputStream os = null;
436                 try {
437
438                         final ByteArrayOutputStream baseStream = new ByteArrayOutputStream();
439                         os = baseStream;
440                         final String contentType = props.getProperty("contenttype");
441                         if (contentType.equalsIgnoreCase("application/json")) {
442                                 JSONArray jsonArray = parseJSON();
443                                 os.write(jsonArray.toString().getBytes());
444                         } else if (contentType.equalsIgnoreCase("text/plain")) {
445                                 for (TimestampedMessage m : fPending) {
446                                         os.write(m.fMsg.getBytes());
447                                         os.write('\n');
448                                 }
449                         } else if (contentType.equalsIgnoreCase("application/cambria")
450                                         || (contentType.equalsIgnoreCase("application/cambria-zip"))) {
451                                 if (contentType.equalsIgnoreCase("application/cambria-zip")) {
452                                         os = new GZIPOutputStream(baseStream);
453                                 }
454                                 for (TimestampedMessage m : fPending) {
455
456                                         os.write(("" + m.fPartition.length()).getBytes());
457                                         os.write('.');
458                                         os.write(("" + m.fMsg.length()).getBytes());
459                                         os.write('.');
460                                         os.write(m.fPartition.getBytes());
461                                         os.write(m.fMsg.getBytes());
462                                         os.write('\n');
463                                 }
464                                 os.close();
465                         } else {
466                                 for (TimestampedMessage m : fPending) {
467                                         os.write(m.fMsg.getBytes());
468
469                                 }
470                         }
471
472                         final long startMs = Clock.now();
473                         if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
474
475                                 try {
476                                         DME2Configue();
477
478                                         Thread.sleep(5);
479                                         getLog().info("sending " + fPending.size() + " msgs to " + url + subContextPath + ". Oldest: "
480                                                         + (nowMs - fPending.peek().timestamp) + " ms");
481                                         sender.setPayload(os.toString());
482
483                                         String dmeResponse = sender.sendAndWait(5000L);
484
485                                         pubResponse = createMRPublisherResponse(dmeResponse, pubResponse);
486
487                                         if (Integer.valueOf(pubResponse.getResponseCode()) < 200
488                                                         || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
489
490                                                 return pubResponse;
491                                         }
492                                         final String logLine = String.valueOf((Clock.now() - startMs)) + dmeResponse.toString();
493                                         getLog().info(logLine);
494                                         fPending.clear();
495
496                                 } catch (DME2Exception x) {
497                                         getLog().warn(x.getMessage(), x);
498                                         pubResponse.setResponseCode(x.getErrorCode());
499                                         pubResponse.setResponseMessage(x.getErrorMessage());
500                                 } catch (URISyntaxException x) {
501
502                                         getLog().warn(x.getMessage(), x);
503                                         pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
504                                         pubResponse.setResponseMessage(x.getMessage());
505                                 } catch (Exception x) {
506
507                                         pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
508                                         pubResponse.setResponseMessage(x.getMessage());
509                                         logger.error("exception: ", x);
510
511                                 }
512
513                                 return pubResponse;
514                         }
515
516                         if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
517                                 getLog().info("sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: "
518                                                 + (nowMs - fPending.peek().timestamp) + " ms");
519                                 final String result = postAuthwithResponse(httpurl, baseStream.toByteArray(), contentType, authKey,
520                                                 authDate, username, password, protocolFlag);
521                                 // Here we are checking for error response. If HTTP status
522                                 // code is not within the http success response code
523                                 // then we consider this as error and return false
524
525                                 pubResponse = createMRPublisherResponse(result, pubResponse);
526
527                                 if (Integer.valueOf(pubResponse.getResponseCode()) < 200
528                                                 || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
529
530                                         return pubResponse;
531                                 }
532
533                                 final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
534                                 getLog().info(logLine);
535                                 fPending.clear();
536                                 return pubResponse;
537                         }
538
539                         if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
540                                 getLog().info("sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: "
541                                                 + (nowMs - fPending.peek().timestamp) + " ms");
542                                 final String result = postWithResponse(httpurl, baseStream.toByteArray(), contentType, username,
543                                                 password, protocolFlag);
544
545                                 // Here we are checking for error response. If HTTP status
546                                 // code is not within the http success response code
547                                 // then we consider this as error and return false
548                                 pubResponse = createMRPublisherResponse(result, pubResponse);
549
550                                 if (Integer.valueOf(pubResponse.getResponseCode()) < 200
551                                                 || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
552
553                                         return pubResponse;
554                                 }
555
556                                 final String logLine = String.valueOf((Clock.now() - startMs));
557                                 getLog().info(logLine);
558                                 fPending.clear();
559                                 return pubResponse;
560                         }
561                         
562                         if (ProtocolTypeConstants.HTTPNOAUTH.getValue().equalsIgnoreCase(protocolFlag)) {
563                                 getLog().info("sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: "
564                                                 + (nowMs - fPending.peek().timestamp) + " ms");
565                                 final String result = postNoAuthWithResponse(httpurl, baseStream.toByteArray(), contentType);
566
567                                 // Here we are checking for error response. If HTTP status
568                                 // code is not within the http success response code
569                                 // then we consider this as error and return false
570                                 pubResponse = createMRPublisherResponse(result, pubResponse);
571
572                                 if (Integer.valueOf(pubResponse.getResponseCode()) < 200
573                                                 || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
574
575                                         return pubResponse;
576                                 }
577
578                                 final String logLine = String.valueOf((Clock.now() - startMs));
579                                 getLog().info(logLine);
580                                 fPending.clear();
581                                 return pubResponse;
582                         }
583                 } catch (IllegalArgumentException x) {
584                         getLog().warn(x.getMessage(), x);
585                         pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
586                         pubResponse.setResponseMessage(x.getMessage());
587
588                 } catch (IOException x) {
589                         getLog().warn(x.getMessage(), x);
590                         pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
591                         pubResponse.setResponseMessage(x.getMessage());
592
593                 } catch (HttpException x) {
594                         getLog().warn(x.getMessage(), x);
595                         pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
596                         pubResponse.setResponseMessage(x.getMessage());
597
598                 } catch (Exception x) {
599                         getLog().warn(x.getMessage(), x);
600
601                         pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
602                         pubResponse.setResponseMessage(x.getMessage());
603
604                 }
605
606                 finally {
607                         if (fPending.size() > 0) {
608                                 getLog().warn("Send failed, " + fPending.size() + " message to send.");
609                                 pubResponse.setPendingMsgs(fPending.size());
610                         }
611                         if (os != null) {
612                                 try {
613                                         os.close();
614                                 } catch (Exception x) {
615                                         getLog().warn(x.getMessage(), x);
616                                         pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
617                                         pubResponse.setResponseMessage("Error in closing Output Stream");
618                                 }
619                         }
620                 }
621
622                 return pubResponse;
623         }
624
625         public MRPublisherResponse createMRPublisherResponse(String reply, MRPublisherResponse mrPubResponse) {
626
627                 if (reply.isEmpty()) {
628
629                         mrPubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
630                         mrPubResponse.setResponseMessage("Please verify the Producer properties");
631                 } else if (reply.startsWith("{")) {
632                         JSONObject jObject = new JSONObject(reply);
633                         if (jObject.has("message") && jObject.has("status")) {
634                                 String message = jObject.getString("message");
635                                 if (null != message) {
636                                         mrPubResponse.setResponseMessage(message);
637                                 }
638                                 mrPubResponse.setResponseCode(Integer.toString(jObject.getInt("status")));
639                         } else {
640                                 mrPubResponse.setResponseCode(String.valueOf(HttpStatus.SC_OK));
641                                 mrPubResponse.setResponseMessage(reply);
642                         }
643                 } else if (reply.startsWith("<")) {
644                         String responseCode = getHTTPErrorResponseCode(reply);
645                         if (responseCode.contains("403")) {
646                                 responseCode = "403";
647                         }
648                         mrPubResponse.setResponseCode(responseCode);
649                         mrPubResponse.setResponseMessage(getHTTPErrorResponseMessage(reply));
650                 }
651
652                 return mrPubResponse;
653         }
654
655         private final String fTopic;
656         private final int fMaxBatchSize;
657         private final long fMaxBatchAgeMs;
658         private final boolean fCompress;
659         private int threadOccuranceTime;
660         private boolean fClosed;
661         private String username;
662         private String password;
663         private String host;
664
665         // host selector
666         private HostSelector fHostSelector = null;
667
668         private final LinkedBlockingQueue<TimestampedMessage> fPending;
669         private long fDontSendUntilMs;
670         private final ScheduledThreadPoolExecutor fExec;
671
672         private String latitude;
673         private String longitude;
674         private String version;
675         private String serviceName;
676         private String env;
677         private String partner;
678         private String routeOffer;
679         private String subContextPath;
680         private String protocol;
681         private String methodType;
682         private String url;
683         private String dmeuser;
684         private String dmepassword;
685         private String contentType;
686         private static final long sfWaitAfterError = 10000;
687         private HashMap<String, String> DMETimeOuts;
688         private DME2Client sender;
689         public String protocolFlag = ProtocolTypeConstants.DME2.getValue();
690         private String authKey;
691         private String authDate;
692         private String handlers;
693         private Properties props;
694         public static String routerFilePath;
695         protected static final Map<String, String> headers = new HashMap<String, String>();
696         public static MultivaluedMap<String, Object> headersMap;
697
698         private MRPublisherResponse pubResponse;
699
700         public MRPublisherResponse getPubResponse() {
701                 return pubResponse;
702         }
703
704         public void setPubResponse(MRPublisherResponse pubResponse) {
705                 this.pubResponse = pubResponse;
706         }
707
708         public static String getRouterFilePath() {
709                 return routerFilePath;
710         }
711
712         public static void setRouterFilePath(String routerFilePath) {
713                 MRSimplerBatchPublisher.routerFilePath = routerFilePath;
714         }
715
716         public Properties getProps() {
717                 return props;
718         }
719
720         public void setProps(Properties props) {
721                 this.props = props;
722         }
723
724         public String getProtocolFlag() {
725                 return protocolFlag;
726         }
727
728         public void setProtocolFlag(String protocolFlag) {
729                 this.protocolFlag = protocolFlag;
730         }
731
732         private void DME2Configue() throws Exception {
733                 try {
734
735                         latitude = props.getProperty("Latitude");
736                         longitude = props.getProperty("Longitude");
737                         version = props.getProperty("Version");
738                         serviceName = props.getProperty("ServiceName");
739                         env = props.getProperty("Environment");
740                         partner = props.getProperty("Partner");
741                         routeOffer = props.getProperty("routeOffer");
742                         subContextPath = props.getProperty("SubContextPath") + fTopic;
743                         
744                         protocol = props.getProperty("Protocol");
745                         methodType = props.getProperty("MethodType");
746                         dmeuser = props.getProperty("username");
747                         dmepassword = props.getProperty("password");
748                         contentType = props.getProperty("contenttype");
749                         handlers = props.getProperty("sessionstickinessrequired");
750                         routerFilePath = props.getProperty("DME2preferredRouterFilePath");
751
752                         /**
753                          * Changes to DME2Client url to use Partner for auto failover
754                          * between data centers When Partner value is not provided use the
755                          * routeOffer value for auto failover within a cluster
756                          */
757
758                         String partitionKey = props.getProperty("partition");
759
760                         if (partner != null && !partner.isEmpty()) {
761                                 url = protocol + "://" + serviceName + "?version=" + version + "&envContext=" + env + "&partner="
762                                                 + partner;
763                                 if (partitionKey != null && !partitionKey.equalsIgnoreCase("")) {
764                                         url = url + "&partitionKey=" + partitionKey;
765                                 }
766                         } else if (routeOffer != null && !routeOffer.isEmpty()) {
767                                 url = protocol + "://" + serviceName + "?version=" + version + "&envContext=" + env + "&routeoffer="
768                                                 + routeOffer;
769                                 if (partitionKey != null && !partitionKey.equalsIgnoreCase("")) {
770                                         url = url + "&partitionKey=" + partitionKey;
771                                 }
772                         }
773
774                         DMETimeOuts = new HashMap<String, String>();
775                         DMETimeOuts.put("AFT_DME2_EP_READ_TIMEOUT_MS", props.getProperty("AFT_DME2_EP_READ_TIMEOUT_MS"));
776                         DMETimeOuts.put("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", props.getProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS"));
777                         DMETimeOuts.put("AFT_DME2_EP_CONN_TIMEOUT", props.getProperty("AFT_DME2_EP_CONN_TIMEOUT"));
778                         DMETimeOuts.put("Content-Type", contentType);
779                         System.setProperty("AFT_LATITUDE", latitude);
780                         System.setProperty("AFT_LONGITUDE", longitude);
781                         System.setProperty("AFT_ENVIRONMENT", props.getProperty("AFT_ENVIRONMENT"));
782                         // System.setProperty("DME2.DEBUG", "true");
783
784                         // SSL changes
785                         // System.setProperty("AFT_DME2_CLIENT_SSL_INCLUDE_PROTOCOLS",
786                         
787                         System.setProperty("AFT_DME2_CLIENT_SSL_INCLUDE_PROTOCOLS", "TLSv1.1,TLSv1.2");
788                         System.setProperty("AFT_DME2_CLIENT_IGNORE_SSL_CONFIG", "false");
789                         System.setProperty("AFT_DME2_CLIENT_KEYSTORE_PASSWORD", "changeit");
790
791                         // SSL changes
792
793                         sender = new DME2Client(new URI(url), 5000L);
794
795                         sender.setAllowAllHttpReturnCodes(true);
796                         sender.setMethod(methodType);
797                         sender.setSubContext(subContextPath);
798                         sender.setCredentials(dmeuser, dmepassword);
799                         sender.setHeaders(DMETimeOuts);
800                         if (handlers != null &&handlers.equalsIgnoreCase("yes")) {
801                                 sender.addHeader("AFT_DME2_EXCHANGE_REQUEST_HANDLERS",
802                                                 props.getProperty("AFT_DME2_EXCHANGE_REQUEST_HANDLERS"));
803                                 sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS",
804                                                 props.getProperty("AFT_DME2_EXCHANGE_REPLY_HANDLERS"));
805                                 sender.addHeader("AFT_DME2_REQ_TRACE_ON", props.getProperty("AFT_DME2_REQ_TRACE_ON"));
806                         } else {
807                                 sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", "com.att.nsa.mr.dme.client.HeaderReplyHandler");
808                         }
809                 } catch (DME2Exception x) {
810                         getLog().warn(x.getMessage(), x);
811                         throw new DME2Exception(x.getErrorCode(), x.getErrorMessage());
812                 } catch (URISyntaxException x) {
813
814                         getLog().warn(x.getMessage(), x);
815                         throw new URISyntaxException(url, x.getMessage());
816                 } catch (Exception x) {
817
818                         getLog().warn(x.getMessage(), x);
819                         throw new IllegalArgumentException(x.getMessage());
820                 }
821         }
822
823         private MRSimplerBatchPublisher(Collection<String> hosts, String topic, int maxBatchSize, long maxBatchAgeMs,
824                         boolean compress) throws MalformedURLException {
825                 super(hosts);
826
827                 if (topic == null || topic.length() < 1) {
828                         throw new IllegalArgumentException("A topic must be provided.");
829                 }
830
831                 fHostSelector = new HostSelector(hosts, null);
832                 fClosed = false;
833                 fTopic = topic;
834                 fMaxBatchSize = maxBatchSize;
835                 fMaxBatchAgeMs = maxBatchAgeMs;
836                 fCompress = compress;
837
838                 fPending = new LinkedBlockingQueue<TimestampedMessage>();
839                 fDontSendUntilMs = 0;
840                 fExec = new ScheduledThreadPoolExecutor(1);
841                 pubResponse = new MRPublisherResponse();
842
843         }
844
845         private MRSimplerBatchPublisher(Collection<String> hosts, String topic, int maxBatchSize, long maxBatchAgeMs,
846                         boolean compress, boolean allowSelfSignedCerts, int httpThreadOccurnace) throws MalformedURLException {
847                 super(hosts);
848
849                 if (topic == null || topic.length() < 1) {
850                         throw new IllegalArgumentException("A topic must be provided.");
851                 }
852
853                 fHostSelector = new HostSelector(hosts, null);
854                 fClosed = false;
855                 fTopic = topic;
856                 fMaxBatchSize = maxBatchSize;
857                 fMaxBatchAgeMs = maxBatchAgeMs;
858                 fCompress = compress;
859                 threadOccuranceTime = httpThreadOccurnace;
860                 fPending = new LinkedBlockingQueue<TimestampedMessage>();
861                 fDontSendUntilMs = 0;
862                 fExec = new ScheduledThreadPoolExecutor(1);
863                 fExec.scheduleAtFixedRate(new Runnable() {
864                         @Override
865                         public void run() {
866                                 send(false);
867                         }
868                 }, 100, threadOccuranceTime, TimeUnit.MILLISECONDS);
869                 pubResponse = new MRPublisherResponse();
870         }
871
872         private static class TimestampedMessage extends message {
873                 public TimestampedMessage(message m) {
874                         super(m);
875                         timestamp = Clock.now();
876                 }
877
878                 public final long timestamp;
879         }
880
881         public String getUsername() {
882                 return username;
883         }
884
885         public void setUsername(String username) {
886                 this.username = username;
887         }
888
889         public String getPassword() {
890                 return password;
891         }
892
893         public void setPassword(String password) {
894                 this.password = password;
895         }
896
897         public String getHost() {
898                 return host;
899         }
900
901         public void setHost(String host) {
902                 this.host = host;
903         }
904
905         public String getContentType() {
906                 return contentType;
907         }
908
909         public void setContentType(String contentType) {
910                 this.contentType = contentType;
911         }
912
913         public String getAuthKey() {
914                 return authKey;
915         }
916
917         public void setAuthKey(String authKey) {
918                 this.authKey = authKey;
919         }
920
921         public String getAuthDate() {
922                 return authDate;
923         }
924
925         public void setAuthDate(String authDate) {
926                 this.authDate = authDate;
927         }
928
929 }