Added async option for sending events

- Fixes snowplow#37
- Moved enum classes to 'emitter' package

Author: jonalmeida

build.gradle

@@ -34,6 +34,7 @@ dependencies {
 
     // Apache HTTP
     compile 'org.apache.httpcomponents:httpclient:4.3.3'
+    compile 'org.apache.httpcomponents:httpasyncclient:4.0.1'
 
     // SLF4J logging API
     compile 'org.slf4j:slf4j-simple:1.7.7'

src/main/java/com/snowplowanalytics/snowplow/tracker/DevicePlatform.java

@@ -1,3 +1,16 @@
+/*
+ * Copyright (c) 2014 Snowplow Analytics Ltd. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+
 package com.snowplowanalytics.snowplow.tracker;
 
 public enum DevicePlatform {

src/main/java/com/snowplowanalytics/snowplow/tracker/Emitter.java

@@ -14,13 +14,19 @@
 package com.snowplowanalytics.snowplow.tracker;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.snowplowanalytics.snowplow.tracker.emitter.BufferOption;
+import com.snowplowanalytics.snowplow.tracker.emitter.HttpMethod;
+import com.snowplowanalytics.snowplow.tracker.emitter.HttpOption;
 
+import org.apache.http.HttpResponse;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.entity.StringEntity;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,29 +35,40 @@
 import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.Iterator;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
 public class Emitter {
 
     private URIBuilder uri;
     private BufferOption option = BufferOption.Default;
-    private EmitterHttpMethod httpMethod = EmitterHttpMethod.GET;
+    private HttpMethod httpMethod = HttpMethod.GET;
+    private HttpOption httpOption = HttpOption.Synchronous;
+    private CloseableHttpClient httpClient;
+    private CloseableHttpAsyncClient httpAsyncClient;
     private final ArrayList<Payload> buffer = new ArrayList<Payload>();
-    private final CloseableHttpClient httpClient = HttpClients.createDefault();
 
     private final Logger logger = LoggerFactory.getLogger(Emitter.class);
 
-    public Emitter(String URI, EmitterHttpMethod httpMethod) {
+    public Emitter(String URI, HttpMethod httpMethod) {
         uri = new URIBuilder()
                 .setScheme("http")
                 .setHost(URI)
                 .setPath("/i");
         this.httpMethod = httpMethod;
+        this.httpClient = HttpClients.createDefault();
     }
 
     public void setBufferOption(BufferOption option) {
         this.option = option;
     }
 
+    public void setHttpOption(HttpOption option) {
+        this.httpOption = option;
+        this.httpAsyncClient = HttpAsyncClients.createDefault();
+        this.httpAsyncClient.start();
+    }
+
     public boolean addToBuffer(Payload payload) {
         boolean ret = buffer.add(payload);
         if (buffer.size() == option.getCode())
@@ -60,11 +77,11 @@ public boolean addToBuffer(Payload payload) {
     }
 
     public void flushBuffer() {
-        if (httpMethod == EmitterHttpMethod.GET) {
+        if (httpMethod == HttpMethod.GET) {
             for (Payload payload : buffer) {
                 sendGetData(payload);
             }
-        } else if (httpMethod == EmitterHttpMethod.POST) {
+        } else if (httpMethod == HttpMethod.POST) {
             Payload postPayload = new TrackerPayload();
             postPayload.setSchema(Constants.SCHEMA_PAYLOAD_DATA);
             postPayload.setData(buffer);
@@ -79,14 +96,26 @@ private void sendPostData(Payload payload) {
 
         try {
             StringEntity params = new StringEntity(payload.toString());
+            HttpResponse httpResponse;
             httpPost.setEntity(params);
-            httpClient.execute(httpPost);
+            if (httpOption == HttpOption.Asynchronous) {
+                Future<HttpResponse> future = httpAsyncClient.execute(httpPost, null);
+                httpResponse = future.get();
+            } else {
+                httpResponse = httpClient.execute(httpPost);
+            }
+            logger.debug(httpResponse.getStatusLine().toString());
         } catch (UnsupportedEncodingException e) {
             logger.error("Encoding exception with the payload.");
             e.printStackTrace();
         } catch (IOException e) {
             logger.error("Error when sending HTTP POST.");
             e.printStackTrace();
+        } catch (InterruptedException e) {
+            logger.error("Interruption error when sending HTTP POST request.");
+            e.printStackTrace();
+        } catch (ExecutionException e) {
+            e.printStackTrace();
         }
 
     }
@@ -108,13 +137,25 @@ private void sendGetData(Payload payload) {
 
         try {
             HttpGet httpGet = new HttpGet(requestUri.build());
-            httpClient.execute(httpGet);
+            HttpResponse httpResponse;
+            if (httpOption == HttpOption.Asynchronous) {
+                Future<HttpResponse> future = httpAsyncClient.execute(httpGet, null);
+                httpResponse = future.get();
+            } else {
+                httpResponse = httpClient.execute(httpGet);
+            }
+            logger.debug(httpResponse.getStatusLine().toString());
         } catch (IOException e) {
             logger.error("Error when sending HTTP GET error.");
             e.printStackTrace();
         } catch (URISyntaxException e) {
             logger.error("Error when creating HTTP GET request. Probably parsing error..");
             e.printStackTrace();
+        } catch (InterruptedException e) {
+            logger.error("Interruption error when sending HTTP GET request.");
+            e.printStackTrace();
+        } catch (ExecutionException e) {
+            e.printStackTrace();
         }
     }
 }

src/main/java/com/snowplowanalytics/snowplow/tracker/Util.java

@@ -21,7 +21,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.io.UnsupportedEncodingException;
 import java.nio.charset.Charset;
 import java.util.Map;
 import java.util.Random;

…ytics/snowplow/tracker/BufferOption.java …owplow/tracker/emitter/BufferOption.javasrc/main/java/com/snowplowanalytics/snowplow/tracker/BufferOption.java renamed to src/main/java/com/snowplowanalytics/snowplow/tracker/emitter/BufferOption.java src/main/java/com/snowplowanalytics/snowplow/tracker/BufferOption.java renamed to src/main/java/com/snowplowanalytics/snowplow/tracker/emitter/BufferOption.java

@@ -11,7 +11,7 @@
  * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
  */
 
-package com.snowplowanalytics.snowplow.tracker;
+package com.snowplowanalytics.snowplow.tracker.emitter;
 
 public enum BufferOption {
     Instant(1),

…/snowplow/tracker/EmitterHttpMethod.java …snowplow/tracker/emitter/HttpMethod.javasrc/main/java/com/snowplowanalytics/snowplow/tracker/EmitterHttpMethod.java renamed to src/main/java/com/snowplowanalytics/snowplow/tracker/emitter/HttpMethod.java src/main/java/com/snowplowanalytics/snowplow/tracker/EmitterHttpMethod.java renamed to src/main/java/com/snowplowanalytics/snowplow/tracker/emitter/HttpMethod.java

@@ -11,9 +11,9 @@
  * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
  */
 
-package com.snowplowanalytics.snowplow.tracker;
+package com.snowplowanalytics.snowplow.tracker.emitter;
 
-public enum EmitterHttpMethod {
+public enum HttpMethod {
     GET,
     POST
 }

src/main/java/com/snowplowanalytics/snowplow/tracker/emitter/HttpOption.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright (c) 2014 Snowplow Analytics Ltd. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+
+package com.snowplowanalytics.snowplow.tracker.emitter;
+
+public enum HttpOption {
+    Synchronous,
+    Asynchronous
+}

src/test/java/com/snowplowanalytics/snowplow/tracker/EmitterTest.java

@@ -1,5 +1,7 @@
 package com.snowplowanalytics.snowplow.tracker;
 
+import com.snowplowanalytics.snowplow.tracker.emitter.HttpMethod;
+
 import junit.framework.TestCase;
 
 import org.junit.Test;
@@ -11,12 +13,12 @@ public class EmitterTest extends TestCase {
 
     @Test
     public void testEmitterConstructor() throws Exception {
-        Emitter emitter = new Emitter("segfault.ngrok.com", EmitterHttpMethod.POST);
+        Emitter emitter = new Emitter("segfault.ngrok.com", HttpMethod.POST);
     }
 
     @Test
     public void testFlushGet() throws Exception {
-        Emitter emitter = new Emitter("segfault.ngrok.com", EmitterHttpMethod.GET);
+        Emitter emitter = new Emitter("segfault.ngrok.com", HttpMethod.GET);
 
         Payload payload;
         String res;

src/test/java/com/snowplowanalytics/snowplow/tracker/TrackerTest.java

@@ -1,14 +1,15 @@
 package com.snowplowanalytics.snowplow.tracker;
 
+import com.snowplowanalytics.snowplow.tracker.emitter.HttpMethod;
+import com.snowplowanalytics.snowplow.tracker.emitter.HttpOption;
+
 import junit.framework.TestCase;
 
 import org.junit.Test;
 
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.junit.Assert.*;
-
 public class TrackerTest extends TestCase {
 
     @Test
@@ -33,8 +34,9 @@ public void testTrackPageView2() throws Exception {
 
     @Test
     public void testTrackPageView3() throws Exception {
-        Emitter emitter = new Emitter("segfault.ngrok.com", EmitterHttpMethod.GET);
+        Emitter emitter = new Emitter("segfault.ngrok.com", HttpMethod.GET);
         Tracker tracker = new Tracker(emitter, "AF003", true, "cloudfront");
+        emitter.setHttpOption(HttpOption.Asynchronous);
 
         Map<String, String> context = new HashMap<String, String>();
         context.put("some key", "some value");