Support for bulk user and event processing

Author: dehora

.gitignore

@@ -1,4 +1,4 @@
-*/**/SpikeMain.java
+*/**/Spike*.java
 .gradle
 *.class
 out/*

README.md

@@ -138,6 +138,20 @@ UserCollection users = User.list();
 while(users.hasNext()) {
     System.out.println(users.next().getUserId());
 }
+
+// Bulk submit users
+final List<JobItem<User>> items = Lists.newArrayList();
+items.add(new JobItem<User>("post", user1));
+items.add(new JobItem<User>("post", user2));
+items.add(new JobItem<User>("delete", user3));
+final Job job = User.submit(items);
+System.out.println(job.getID());
+
+// Bulk submit, add to an existing job
+final List<JobItem<User>> moreItems = Lists.newArrayList();
+items.add(new JobItem<User>("post", user4));
+items.add(new JobItem<User>("delete", user5));
+User.submit(moreItems, job);
 ```
 
 ### Contacts
@@ -244,6 +258,20 @@ Event event = new Event().setEventName("bought-hat")
     .putMetadata("found_date", System.currentTimeMillis())
     .putMetadata("new_signup", true);
 Event.create(event);
+
+// Bulk submit events
+final List<JobItem<Event>> items = Lists.newArrayList();
+items.add(new JobItem<Event>("post", event1));
+items.add(new JobItem<Event>("post", event2));
+items.add(new JobItem<Event>("post", event3));
+final Job job = Event.submit(items);
+System.out.println(job.getID());
+
+// Bulk submit, add to an existing job
+final List<JobItem<Event>> moreItems = Lists.newArrayList();
+items.add(new JobItem<Event>("post", event4));
+items.add(new JobItem<Event>("delete", event5));
+Event.submit(moreItems, job);
 ```
 
 

intercom-java/src/main/java/io/intercom/api/Conditions.java

@@ -0,0 +1,26 @@
+package io.intercom.api;
+
+import com.google.common.collect.Lists;
+
+class Conditions {
+
+    /**
+     * Ensures that an object reference passed as a parameter to the calling
+     * method is not null. Variant of Guava's Preconditions that returns an
+     * InvalidException containing an ErrorCollection
+     *
+     * @param reference an object reference
+     * @param errorMessage the exception message to use if the check fails
+     * @return the non-null reference that was validated
+     * @throws InvalidException if {@code reference} is null
+     */
+    public static <T> T checkNotNull(T reference, String errorMessage) {
+        if (reference == null) {
+            throw new InvalidException(
+                new ErrorCollection(
+                    Lists.newArrayList(
+                        new Error("invalid", "item method must be supplied"))));
+        }
+        return reference;
+    }
+}

intercom-java/src/main/java/io/intercom/api/Contact.java

@@ -140,7 +140,7 @@ void checkValidConvertUser(HashMap<String, String> convertContact) {
     @SuppressWarnings("UnusedDeclaration")
     @JsonIgnoreProperties(ignoreUnknown = true)
     @JsonInclude(JsonInclude.Include.NON_DEFAULT)
-    static class ContactConvert extends TypedData {
+    static class ContactConvert {
 
         @JsonProperty("contact")
         private Map<String, String> contact;

intercom-java/src/main/java/io/intercom/api/DataResource.java

@@ -1,6 +1,7 @@
 package io.intercom.api;
 
 import java.net.URI;
+import java.util.List;
 import java.util.Map;
 
 abstract class DataResource {
@@ -21,6 +22,11 @@ public static <T, R> R create(T entity, String collectionPath, Class<R> response
         return resource.post(response, entity);
     }
 
+    public static <T, R> R create(T entity, List<String> paths, Class<R> response) {
+        final HttpClient resource = new HttpClient(UriBuilder.newBuilder().path(paths).build());
+        return resource.post(response, entity);
+    }
+
     public static <T, R> R update(T entity, String collectionPath, Class<R> response) {
         final HttpClient resource = new HttpClient(UriBuilder.newBuilder().path(collectionPath).build());
         return resource.post(response, entity);

intercom-java/src/main/java/io/intercom/api/Event.java

@@ -8,13 +8,23 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
 @SuppressWarnings("UnusedDeclaration")
 @JsonIgnoreProperties(ignoreUnknown = true)
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public class Event extends TypedData {
 
+    private static final List<String> BULK_METHODS = Lists.newArrayList("post");
+    private static final ArrayList<String> BULK_PATHS = Lists.newArrayListWithExpectedSize(2);
+
+    static {
+        BULK_PATHS.add("bulk");
+        BULK_PATHS.add("events");
+    }
+
     public static void create(Event event) throws InvalidException, AuthorizationException {
 
         validateCreateEvent(event);
@@ -25,6 +35,32 @@ public static void create(Event event) throws InvalidException, AuthorizationExc
         DataResource.create(event, "events", Void.class);
     }
 
+    public static Job submit(List<JobItem<Event>> items)
+        throws AuthorizationException, ClientException, ServerException, InvalidException, RateLimitException {
+        return submit(items, null);
+    }
+
+    public static Job submit(List<JobItem<Event>> items, Job job)
+        throws AuthorizationException, ClientException, ServerException, InvalidException, RateLimitException {
+        return Job.submit(validateJobItems(items), job, BULK_PATHS);
+    }
+
+    public static JobItemCollection<Event> listJobErrorFeed(String jobID)
+        throws AuthorizationException, ClientException, ServerException, InvalidException, RateLimitException {
+        return Job.listJobErrorFeed(jobID, Event.class);
+    }
+
+    @VisibleForTesting
+    static List<JobItem<Event>> validateJobItems(List<JobItem<Event>> items) {
+        final JobSupport jobSupport = new JobSupport();
+        for (JobItem<Event> item : items) {
+            jobSupport.validateJobItem(item, BULK_METHODS);
+            validateCreateEvent(item.getData());
+        }
+
+        return items;
+    }
+
     private static final ErrorCollection INVALID_NAME = new ErrorCollection(
         Lists.newArrayList(
             new Error("invalid", "an event must supply an event name")));

intercom-java/src/main/java/io/intercom/api/HttpClient.java

@@ -1,5 +1,6 @@
 package io.intercom.api;
 
+import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.google.common.collect.Lists;
@@ -73,57 +74,46 @@ private HttpClient(URI uri, Map<String, String> headers) {
     }
 
     public <T> T get(Class<T> reqres) throws IntercomException {
-        HttpURLConnection conn = null;
-        try {
-            conn = initializeConnection(uri, "GET");
-            return runRequest(uri, reqres, conn);
-        } catch (IOException e) {
-            return throwLocalException(e);
-        } finally {
-            IOUtils.disconnectQuietly(conn);
-        }
+        return get(getJavaType(reqres));
+    }
+
+    <T> T get(JavaType responseType) throws IntercomException {
+        return executeHttpMethod("GET", null, responseType);
+    }
+
+    public <T> T delete(Class<T> reqres) {
+        return executeHttpMethod("DELETE", null, getJavaType(reqres));
     }
 
     public <T, E> T put(Class<T> reqres, E entity) {
         headers.put("Content-Type", APPLICATION_JSON);
-        HttpURLConnection conn = null;
-        try {
-            conn = initializeConnection(uri, "PUT");
-            prepareRequestEntity(entity, conn);
-            return runRequest(uri, reqres, conn);
-        } catch (IOException e) {
-            return throwLocalException(e);
-        } finally {
-            IOUtils.disconnectQuietly(conn);
-        }
+        return executeHttpMethod("PUT", (E) entity, getJavaType(reqres));
     }
 
     public <T, E> T post(Class<T> reqres, E entity) {
         headers.put("Content-Type", APPLICATION_JSON);
-        HttpURLConnection conn = null;
-        try {
-            conn = initializeConnection(uri, "POST");
-            prepareRequestEntity(entity, conn);
-            return runRequest(uri, reqres, conn);
-        } catch (IOException e) {
-            return throwLocalException(e);
-        } finally {
-            IOUtils.disconnectQuietly(conn);
-        }
+        return executeHttpMethod("POST", entity, getJavaType(reqres));
     }
 
-    public <T> T delete(Class<T> reqres) {
+    private <T, E> T executeHttpMethod(String method, E entity, JavaType responseType) {
         HttpURLConnection conn = null;
         try {
-            conn = initializeConnection(uri, "DELETE");
-            return runRequest(uri, reqres, conn);
+            conn = initializeConnection(uri, method);
+            if(entity != null) {
+                prepareRequestEntity(entity, conn);
+            }
+            return runRequest(uri, responseType, conn);
         } catch (IOException e) {
             return throwLocalException(e);
         } finally {
             IOUtils.disconnectQuietly(conn);
         }
     }
 
+    private <T> JavaType getJavaType(Class<T> reqres) {
+        return objectMapper.getTypeFactory().constructType(reqres);
+    }
+
     // trick java with a dummy return
     private <T> T throwLocalException(IOException e) {
         throw new IntercomException(String.format("Local exception calling [%s]. Check connectivity and settings. [%s]", uri.toASCIIString(), e.getMessage()), e);
@@ -151,13 +141,12 @@ private HttpURLConnection initializeConnection(URI uri, String method) throws IO
         return conn;
     }
 
-    private <T> T runRequest(URI uri, Class<T> response, HttpURLConnection conn) throws IOException {
+    private <T> T runRequest(URI uri, JavaType javaType, HttpURLConnection conn) throws IOException {
         conn.connect();
         final int responseCode = conn.getResponseCode();
         if (responseCode >= 200 && responseCode < 300) {
-            return handleSuccess(response, conn, responseCode);
+            return handleSuccess(javaType, conn, responseCode);
         } else {
-            // errors are redirects for now
             return handleError(uri, conn, responseCode);
         }
     }
@@ -175,23 +164,27 @@ private <T> T handleError(URI uri, HttpURLConnection conn, int responseCode) thr
         return throwException(responseCode, errors);
     }
 
-    private <T> T handleSuccess(Class<T> response, HttpURLConnection conn, int responseCode) throws IOException {
-        if (responseCode == 202 || responseCode == 204 || Void.class.equals(response)) {
+    private <T> T handleSuccess(JavaType javaType, HttpURLConnection conn, int responseCode) throws IOException {
+        if (shouldSkipResponseEntity(javaType, conn, responseCode)) {
             return null;
         } else {
-            return readEntity(conn, responseCode, response);
+            return readEntity(conn, responseCode, javaType);
         }
     }
 
-    private <T> T readEntity(HttpURLConnection conn, int responseCode, Class<T> entityType) throws IOException {
+    private boolean shouldSkipResponseEntity(JavaType javaType, HttpURLConnection conn, int responseCode) {
+        return responseCode == 202 || responseCode == 204 || Void.class.equals(javaType.getRawClass()) || "DELETE".equals(conn.getRequestMethod());
+    }
+
+    private <T> T readEntity(HttpURLConnection conn, int responseCode, JavaType javaType) throws IOException {
         final InputStream entityStream = conn.getInputStream();
         try {
             if (logger.isDebugEnabled()) {
                 final String text = CharStreams.toString(new InputStreamReader(entityStream));
                 logger.debug("api server response status[{}] --\n{}\n-- ", responseCode, text);
-                return objectMapper.readValue(text, entityType);
+                return objectMapper.readValue(text, javaType);
             } else {
-                return objectMapper.readValue(entityStream, entityType);
+                return objectMapper.readValue(entityStream, javaType);
             }
         } finally {
             IOUtils.closeQuietly(entityStream);

intercom-java/src/main/java/io/intercom/api/Intercom.java

@@ -30,6 +30,10 @@ public class Intercom {
 
     private static volatile HttpConnectorSupplier httpConnectorSupplier = HttpConnectorSupplier.defaultSupplier;
 
+    public static long currentTimestamp() {
+        return System.currentTimeMillis()/1000;
+    }
+
     public static int getConnectionTimeout() {
         return connectionTimeout;
     }