Skip to content

Commit b390502

Browse files
Juan Pablo Diaz-VazJuan Pablo Diaz-Vaz
Juan Pablo Diaz-Vaz
authored and
Juan Pablo Diaz-Vaz
committed
SUMOK-29 Added GZIP compression for POSTs
1 parent 2403a31 commit b390502

4 files changed

Lines changed: 85 additions & 38 deletions

File tree

‎build.xml‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,10 +32,12 @@
3232
<get src= "http://central.maven.org/maven2/commons-lang/commons-lang/2.6/commons-lang-2.6.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
3333
<get src= "http://central.maven.org/maven2/org/apache/lucene/lucene-core/4.8.1/lucene-core-4.8.1.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
3434
<get src= "http://central.maven.org/maven2/commons-codec/commons-codec/1.10/commons-codec-1.10.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
35-
35+
<get src= "http://central.maven.org/maven2/com/ning/async-http-client/1.9.30/async-http-client-1.9.30.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
36+
3637
<get src= "https://hamcrest.googlecode.com/files/hamcrest-core-1.3.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
3738

3839
<get src= "http://sdk-for-java.amazonwebservices.com/latest/aws-java-sdk.zip" dest="${external.dir}" usetimestamp="true" verbose="true" />
40+
3941
</target>
4042

4143
<target name="download-test">

‎src/main/java/com/mcplusa/sumologic/CloudWatchMessageModelSumologicTransformer.java‎

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,8 @@
22

33
import java.io.IOException;
44

5-
import com.amazonaws.services.kinesis.connectors.BasicJsonTransformer;
65
import com.amazonaws.services.kinesis.model.Record;
76
import com.mcplusa.sumologic.SimpleKinesisMessageModel;
8-
import com.mcplusa.sumologic.implementations.SumologicEmitter;
97
import com.mcplusa.sumologic.implementations.SumologicTransformer;
108

119
import org.apache.commons.logging.Log;
@@ -15,11 +13,6 @@
1513
import java.io.BufferedReader;
1614
import java.io.InputStreamReader;
1715
import java.util.zip.GZIPInputStream;
18-
import java.nio.charset.StandardCharsets;
19-
import java.util.Arrays;
20-
21-
import org.apache.commons.codec.binary.Base64;
22-
2316

2417
/**
2518
* A custom transfomer for {@link SimpleKinesisMessageModel} records in JSON format. The output is in a format

‎src/main/java/com/mcplusa/sumologic/SumologicSender.java‎

Lines changed: 64 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -1,59 +1,68 @@
11
package com.mcplusa.sumologic;
22

3+
import java.io.ByteArrayOutputStream;
34
import java.io.IOException;
5+
import java.util.concurrent.ExecutionException;
6+
import java.util.zip.GZIPOutputStream;
47

58
import org.apache.commons.logging.Log;
69
import org.apache.commons.logging.LogFactory;
7-
import org.apache.http.HttpResponse;
810
import org.apache.http.client.HttpClient;
9-
import org.apache.http.client.methods.HttpPost;
10-
import org.apache.http.entity.StringEntity;
11-
import org.apache.http.params.BasicHttpParams;
12-
import org.apache.http.params.HttpConnectionParams;
13-
import org.apache.http.params.HttpParams;
14-
import org.apache.http.protocol.HTTP;
15-
import org.apache.http.util.EntityUtils;
16-
import org.apache.http.impl.client.DefaultHttpClient;
17-
import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
11+
12+
import com.ning.http.client.AsyncHttpClient;
13+
import com.ning.http.client.AsyncHttpClientConfig;
14+
import com.ning.http.client.AsyncHttpClient.BoundRequestBuilder;
15+
import com.ning.http.client.AsyncHttpClientConfig.Builder;
16+
import com.ning.http.client.Response;
1817

1918
public class SumologicSender {
2019
private static final Log LOG = LogFactory.getLog(SumologicSender.class);
2120

2221
private String url = null;
23-
private HttpClient httpClient = null;
22+
private AsyncHttpClient client = null;
2423

25-
private int connectionTimeout = 1000;
26-
private int socketTimeout = 60000;
2724
private static final int RETRIES = 3;
2825
private static final int SLEEP_TIME = 1000;
29-
3026

3127
public SumologicSender(String url) {
3228
this.url = url;
33-
34-
HttpParams params = new BasicHttpParams();
35-
HttpConnectionParams.setConnectionTimeout(params, connectionTimeout);
36-
HttpConnectionParams.setSoTimeout(params, socketTimeout);
37-
httpClient = new DefaultHttpClient(new ThreadSafeClientConnManager(), params);
29+
30+
Builder builder = new AsyncHttpClientConfig.Builder();
31+
this.client = new AsyncHttpClient(builder.build());
3832
}
33+
34+
private BoundRequestBuilder clientPreparePost(String url){
35+
if (this.client.isClosed()){
36+
Builder builder = new AsyncHttpClientConfig.Builder();
37+
this.client = new AsyncHttpClient(builder.build());
38+
}
39+
return this.client.preparePost(url);
40+
}
3941

4042
public boolean sendToSumologic(String data) throws IOException{
4143
int retries = RETRIES;
4244
int sleep_time = SLEEP_TIME;
43-
int statusCode;
45+
int statusCode = -1;
4446

4547
do {
46-
HttpPost post = null;
47-
post = new HttpPost(url);
48-
post.setEntity(new StringEntity(data, HTTP.PLAIN_TEXT_TYPE, HTTP.UTF_8));
49-
HttpResponse response = httpClient.execute(post);
50-
statusCode = response.getStatusLine().getStatusCode();
48+
BoundRequestBuilder builder = null;
49+
builder = this.clientPreparePost(url);
50+
51+
byte[] compressedData = compressGzip(data);
52+
53+
builder.setHeader("Content-Encoding", "gzip");
54+
builder.setBody(compressedData);
55+
56+
Response response = null;
57+
try {
58+
response = builder.execute().get();
59+
statusCode = response.getStatusCode();
60+
} catch (InterruptedException e) {
61+
LOG.error("Can't send POST to Sumologic "+e.getMessage());
62+
} catch (ExecutionException e) {
63+
LOG.error("Can't send POST to Sumologic "+e.getMessage());
64+
}
5165

52-
//need to consume the body if you want to re-use the connection.
53-
EntityUtils.consume(response.getEntity());
54-
try {
55-
post.abort();
56-
} catch (Exception ignore) {}
5766
if (statusCode == 429) {
5867
LOG.warn("Got TOO MANY REQUESTS from Sumologic");
5968
retries--;
@@ -72,4 +81,29 @@ public boolean sendToSumologic(String data) throws IOException{
7281
return true;
7382
}
7483
}
84+
85+
public byte[] compressGzip(String data) {
86+
if (data == null || data.length() == 0) {
87+
return null;
88+
}
89+
90+
ByteArrayOutputStream outputStream=new ByteArrayOutputStream();
91+
GZIPOutputStream gzip;
92+
try {
93+
gzip = new GZIPOutputStream(outputStream);
94+
} catch (IOException e) {
95+
LOG.error("Cannot compress into GZIP "+e.getMessage());
96+
return null;
97+
}
98+
99+
// Put data into the GZIP buffer
100+
try {
101+
gzip.write(data.getBytes("UTF-8"));
102+
gzip.close();
103+
} catch (IOException e) {
104+
e.printStackTrace();
105+
}
106+
107+
return outputStream.toByteArray();
108+
}
75109
}

‎src/test/java/com/mcplusa/sumologic/SumologicSenderTest.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,9 @@
11
package com.mcplusa.sumologic;
22

33
import java.io.IOException;
4+
import java.nio.charset.Charset;
45
import java.util.ArrayList;
6+
import java.util.Arrays;
57
import java.util.List;
68

79
import org.junit.Assert;
@@ -14,6 +16,8 @@
1416

1517
import com.github.tomakehurst.wiremock.client.WireMock;
1618
import com.github.tomakehurst.wiremock.junit.WireMockRule;
19+
import com.mcplusa.sumologic.CloudWatchMessageModelSumologicTransformer;
20+
import com.mcplusa.sumologic.SumologicSender;
1721
import com.mcplusa.sumologic.implementations.SumologicEmitter;
1822

1923
public class SumologicSenderTest {
@@ -64,6 +68,20 @@ public void theSenderShouldReturnTrueOnSuccess () {
6468
}
6569
}
6670

71+
@Test
72+
public void decompressGzipTest() {
73+
String url = MOCKED_HOST + MOCKED_COLLECTION;
74+
75+
String data = "a string of characters";
76+
77+
SumologicSender sender = new SumologicSender(url);
78+
79+
byte[] compressData = sender.compressGzip(data);
80+
String result = CloudWatchMessageModelSumologicTransformer.decompressGzip(compressData);
81+
82+
Assert.assertTrue(data.equals(result));
83+
}
84+
6785
private void mockEmitMessages () {
6886
WireMock.stubFor(WireMock.post(WireMock.urlMatching(MOCKED_COLLECTION))
6987
.willReturn(WireMock.aResponse()

0 commit comments

Comments
 (0)