Merge pull request #467 from segmentio/attribution

Support Attribution Tracking

analytics-tests/src/test/java/com/segment/analytics/AnalyticsTest.java

@@ -125,7 +125,8 @@ public static void grantPermission(final Application app, final String permissio
     analytics = new Analytics(application, networkExecutor, stats, traitsCache, analyticsContext,
         defaultOptions, Logger.with(NONE), "qaz", Collections.singletonList(factory), client,
         Cartographer.INSTANCE, projectSettingsCache, "foo", DEFAULT_FLUSH_QUEUE_SIZE,
-        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, false, new CountDownLatch(0), false, optOut);
+        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, false, new CountDownLatch(0), false, false,
+        optOut);
 
     // Used by singleton tests.
     grantPermission(RuntimeEnvironment.application, Manifest.permission.INTERNET);
@@ -596,7 +597,8 @@ protected boolean matchesSafely(Application.ActivityLifecycleCallbacks item) {
     analytics = new Analytics(application, networkExecutor, stats, traitsCache, analyticsContext,
         defaultOptions, Logger.with(NONE), "qaz", Collections.singletonList(factory), client,
         Cartographer.INSTANCE, projectSettingsCache, "foo", DEFAULT_FLUSH_QUEUE_SIZE,
-        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, true, new CountDownLatch(0), false, optOut);
+        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, true, new CountDownLatch(0), false, false,
+        optOut);
 
     callback.get().onActivityCreated(null, null);
 
@@ -654,7 +656,8 @@ protected boolean matchesSafely(Application.ActivityLifecycleCallbacks item) {
     analytics = new Analytics(application, networkExecutor, stats, traitsCache, analyticsContext,
         defaultOptions, Logger.with(NONE), "qaz", Collections.singletonList(factory), client,
         Cartographer.INSTANCE, projectSettingsCache, "foo", DEFAULT_FLUSH_QUEUE_SIZE,
-        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, true, new CountDownLatch(0), false, optOut);
+        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, true, new CountDownLatch(0), false, false,
+        optOut);
 
     callback.get().onActivityCreated(null, null);
 
@@ -694,7 +697,8 @@ protected boolean matchesSafely(Application.ActivityLifecycleCallbacks item) {
     analytics = new Analytics(application, networkExecutor, stats, traitsCache, analyticsContext,
         defaultOptions, Logger.with(NONE), "qaz", Collections.singletonList(factory), client,
         Cartographer.INSTANCE, projectSettingsCache, "foo", DEFAULT_FLUSH_QUEUE_SIZE,
-        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, false, new CountDownLatch(0), true, optOut);
+        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, false, new CountDownLatch(0), true, false,
+        optOut);
 
     Activity activity = mock(Activity.class);
     PackageManager packageManager = mock(PackageManager.class);
@@ -733,7 +737,8 @@ protected boolean matchesSafely(Application.ActivityLifecycleCallbacks item) {
     analytics = new Analytics(application, networkExecutor, stats, traitsCache, analyticsContext,
         defaultOptions, Logger.with(NONE), "qaz", Collections.singletonList(factory), client,
         Cartographer.INSTANCE, projectSettingsCache, "foo", DEFAULT_FLUSH_QUEUE_SIZE,
-        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, false, new CountDownLatch(0), false, optOut);
+        DEFAULT_FLUSH_INTERVAL, analyticsExecutor, false, new CountDownLatch(0), false, false,
+        optOut);
 
     Activity activity = mock(Activity.class);
     Bundle bundle = new Bundle();

analytics-tests/src/test/java/com/segment/analytics/ClientTest.java

@@ -112,7 +112,7 @@ public class ClientTest {
     try {
       connection.close();
       fail(">= 300 return code should throw an exception");
-    } catch (Client.UploadException e) {
+    } catch (Client.HTTPException e) {
       assertThat(e).hasMessage("HTTP 300: bar. "
           + "Response: Could not read response body for rejected message: "
           + "java.io.IOException: Underlying input stream returned zero bytes");

analytics/src/main/java/com/segment/analytics/Analytics.java

@@ -48,6 +48,10 @@
 import com.segment.analytics.internal.Private;
 import com.segment.analytics.internal.Utils;
 import com.segment.analytics.internal.Utils.AnalyticsNetworkExecutorService;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.Writer;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.LinkedHashMap;
@@ -103,6 +107,7 @@ public class Analytics {
   @Private static final Properties EMPTY_PROPERTIES = new Properties();
   private static final String VERSION_KEY = "version";
   private static final String BUILD_KEY = "build";
+  private static final String TRACKED_ATTRIBUTION_KEY = "tracked_attribution";
 
   private final Application application;
   final ExecutorService networkExecutor;
@@ -187,11 +192,12 @@ public static void setSingletonInstance(Analytics analytics) {
 
   Analytics(Application application, ExecutorService networkExecutor, Stats stats,
       Traits.Cache traitsCache, AnalyticsContext analyticsContext, Options defaultOptions,
-      Logger logger, String tag, final List<Integration.Factory> factories, Client client,
+      final Logger logger, String tag, final List<Integration.Factory> factories, Client client,
       Cartographer cartographer, ProjectSettings.Cache projectSettingsCache, String writeKey,
       int flushQueueSize, long flushIntervalInMillis, final ExecutorService analyticsExecutor,
       final boolean shouldTrackApplicationLifecycleEvents, CountDownLatch advertisingIdLatch,
-      final boolean shouldRecordScreenViews, BooleanPreference optOut) {
+      final boolean shouldRecordScreenViews, final boolean trackAttributionInformation,
+      BooleanPreference optOut) {
     this.application = application;
     this.networkExecutor = networkExecutor;
     this.stats = stats;
@@ -244,6 +250,14 @@ public static void setSingletonInstance(Analytics analytics) {
         if (!trackedApplicationLifecycleEvents.getAndSet(true)
             && shouldTrackApplicationLifecycleEvents) {
           trackApplicationLifecycleEvents();
+
+          if (trackAttributionInformation) {
+            analyticsExecutor.submit(new Runnable() {
+              @Override public void run() {
+                trackAttributionInformation();
+              }
+            });
+          }
         }
         runOnMainThread(IntegrationOperation.onActivityCreated(activity, savedInstanceState));
       }
@@ -277,6 +291,37 @@ public static void setSingletonInstance(Analytics analytics) {
     });
   }
 
+  @Private void trackAttributionInformation() {
+    SharedPreferences sharedPreferences = getSegmentSharedPreferences(application);
+    boolean trackedAttribution = sharedPreferences.getBoolean(TRACKED_ATTRIBUTION_KEY, false);
+    if (trackedAttribution) {
+      return;
+    }
+
+    waitForAdvertisingId();
+
+    Client.Connection connection = null;
+    try {
+      connection = client.attribution();
+
+      // Write the request body.
+      Writer writer = new BufferedWriter(new OutputStreamWriter(connection.os));
+      cartographer.toJson(analyticsContext, writer);
+
+      // Read the response body.
+      Map<String, Object> map =
+          cartographer.fromJson(buffer(connection.connection.getInputStream()));
+      Properties properties = new Properties(map);
+
+      track("Install Attributed", properties);
+      sharedPreferences.edit().putBoolean(TRACKED_ATTRIBUTION_KEY, true).apply();
+    } catch (IOException e) {
+      logger.error(e, "Unable to track attribution information. Retrying on next launch.");
+    } finally {
+      closeQuietly(connection);
+    }
+  }
+
   @Private void trackApplicationLifecycleEvents() {
     // Get the current version.
     PackageInfo packageInfo = getPackageInfo(application);
@@ -920,6 +965,7 @@ public static class Builder {
     private List<Integration.Factory> factories;
     private boolean trackApplicationLifecycleEvents = false;
     private boolean recordScreenViews = false;
+    private boolean trackAttributionInformation = true;
 
     /** Start building a new {@link Analytics} instance. */
     public Builder(Context context, String writeKey) {
@@ -1095,6 +1141,12 @@ public Builder recordScreenViews() {
       return this;
     }
 
+    /** Automatically track attribution information from enabled providers. */
+    public Builder trackAttributionInformation() {
+      this.trackAttributionInformation = true;
+      return this;
+    }
+
     /** Create a {@link Analytics} client. */
     public Analytics build() {
       if (isNullOrEmpty(tag)) {
@@ -1153,7 +1205,8 @@ public Analytics build() {
       return new Analytics(application, networkExecutor, stats, traitsCache, analyticsContext,
           defaultOptions, logger, tag, factories, client, cartographer, projectSettingsCache,
           writeKey, flushQueueSize, flushIntervalInMillis, Executors.newSingleThreadExecutor(),
-          trackApplicationLifecycleEvents, advertisingIdLatch, recordScreenViews, optOut);
+          trackApplicationLifecycleEvents, advertisingIdLatch, recordScreenViews,
+          trackAttributionInformation, optOut);
     }
   }
 

analytics/src/main/java/com/segment/analytics/Client.java

@@ -63,7 +63,7 @@ private static Connection createPostConnection(HttpURLConnection connection) thr
             } catch (IOException e) {
               responseBody = "Could not read response body for rejected message: " + e.toString();
             }
-            throw new UploadException(responseCode, connection.getResponseMessage(), responseBody);
+            throw new HTTPException(responseCode, connection.getResponseMessage(), responseBody);
           }
         } finally {
           super.close();
@@ -92,6 +92,11 @@ Connection upload() throws IOException {
     return createPostConnection(connection);
   }
 
+  Connection attribution() throws IOException {
+    HttpURLConnection connection = connectionFactory.attribution(writeKey);
+    return createPostConnection(connection);
+  }
+
   Connection fetchSettings() throws IOException {
     HttpURLConnection connection = connectionFactory.projectSettings(writeKey);
     int responseCode = connection.getResponseCode();
@@ -102,13 +107,13 @@ Connection fetchSettings() throws IOException {
     return createGetConnection(connection);
   }
 
-  /** Represents an exception during uploading events that should not be retried. */
-  static class UploadException extends IOException {
+  /** Represents an HTTP exception thrown for unexpected/non 2xx response codes. */
+  static class HTTPException extends IOException {
     final int responseCode;
     final String responseMessage;
     final String responseBody;
 
-    UploadException(int responseCode, String responseMessage, String responseBody) {
+    HTTPException(int responseCode, String responseMessage, String responseBody) {
       super("HTTP " + responseCode + ": " + responseMessage + ". Response: " + responseBody);
       this.responseCode = responseCode;
       this.responseMessage = responseMessage;
@@ -121,8 +126,7 @@ static class UploadException extends IOException {
    * InputStream} or write to the connection via {@link OutputStream}.
    */
   static abstract class Connection implements Closeable {
-
-    protected final HttpURLConnection connection;
+    final HttpURLConnection connection;
     final InputStream is;
     final OutputStream os;
 

analytics/src/main/java/com/segment/analytics/ConnectionFactory.java

@@ -37,8 +37,20 @@ public HttpURLConnection upload(String writeKey) throws IOException {
   }
 
   /**
-   * Configures defaults for connections opened with both {@link #upload(String)} and {@link
-   * #projectSettings(String)}.
+   * Return a {@link HttpURLConnection} that writes gets attribution information from {@code
+   * https://mobile-service.segment.com/attribution}.
+   */
+  public HttpURLConnection attribution(String writeKey) throws IOException {
+    HttpURLConnection connection = openConnection("https://mobile-service.segment.com/attribution");
+    connection.setRequestProperty("Authorization", authorizationHeader(writeKey));
+    connection.setRequestMethod("POST");
+    connection.setDoOutput(true);
+    return connection;
+  }
+
+  /**
+   * Configures defaults for connections opened with {@link #upload(String)},  {@link
+   * #attribution(String)} and {@link #projectSettings(String)}.
    */
   protected HttpURLConnection openConnection(String url) throws IOException {
     HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection();

analytics/src/main/java/com/segment/analytics/SegmentIntegration.java

@@ -282,8 +282,7 @@ private boolean shouldFlush() {
     }
 
     logger.verbose("Uploading payloads in queue to Segment.");
-    int payloadsUploaded;
-
+    int payloadsUploaded = 0;
     Client.Connection connection = null;
     try {
       // Open a connection.
@@ -296,15 +295,18 @@ private boolean shouldFlush() {
       PayloadWriter payloadWriter = new PayloadWriter(writer);
       payloadQueue.forEach(payloadWriter);
       writer.endBatchArray().endObject().close();
-      // Don't use the result of QueueFiles#forEach, since we may not read the last element.
+      // Don't use the result of QueueFiles#forEach, since we may not upload the last element.
       payloadsUploaded = payloadWriter.payloadCount;
 
-      try {
-        // Upload the payloads.
-        connection.close();
-      } catch (Client.UploadException e) {
+      // Upload the payloads.
+      connection.close();
+    } catch (Client.HTTPException e) {
+      if (e.responseCode >= 400 && e.responseCode < 500) {
         // Simply log and proceed to remove the rejected payloads from the queue.
         logger.error(e, "Payloads were rejected by server. Marked for removal.");
+      } else {
+        logger.error(e, "Error while uploading payloads");
+        return;
       }
     } catch (IOException e) {
       logger.error(e, "Error while uploading payloads");