fix: support arbitrarily long list of user attributes in batch events

event-handler/build.gradle

@@ -51,11 +51,14 @@ dependencies {
     api project(':shared')
     implementation "androidx.annotation:annotation:$annotations_ver"
     implementation "androidx.work:work-runtime:$work_runtime"
+    // Base64
+    implementation "commons-codec:commons-codec:1.15"
 
     compileOnly "com.noveogroup.android:android-logger:$android_logger_ver"
 
     testImplementation "junit:junit:$junit_ver"
     testImplementation "org.mockito:mockito-core:$mockito_ver"
+    testImplementation "org.powermock:powermock-mockito-release-full:$powermock_ver"
     testImplementation "com.noveogroup.android:android-logger:$android_logger_ver"
 
     androidTestImplementation "androidx.test.ext:junit:$androidx_test"

event-handler/src/main/java/com/optimizely/ab/android/event_handler/EventHandlerUtils.java

@@ -0,0 +1,80 @@
+package com.optimizely.ab.android.event_handler;
+
+import static java.util.zip.Deflater.BEST_COMPRESSION;
+
+import android.os.Build;
+
+import androidx.annotation.NonNull;
+import androidx.annotation.RequiresApi;
+import androidx.annotation.VisibleForTesting;
+
+import org.apache.commons.codec.binary.Base64;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.zip.Deflater;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+import java.util.zip.Inflater;
+
+public class EventHandlerUtils {
+
+    private static final int BUFFER_SIZE = 32*1024;
+
+    public static String compress(@NonNull String uncompressed) throws IOException {
+        byte[] data = uncompressed.getBytes();
+
+        final Deflater deflater = new Deflater();
+        //deflater.setLevel(BEST_COMPRESSION);
+        deflater.setInput(data);
+
+        try (final ByteArrayOutputStream outputStream = new ByteArrayOutputStream(data.length)) {
+            deflater.finish();
+            final byte[] buffer = new byte[BUFFER_SIZE];
+            while (!deflater.finished()) {
+                final int count = deflater.deflate(buffer);
+                outputStream.write(buffer, 0, count);
+            }
+
+            byte[] bytes = outputStream.toByteArray();
+            // encoded to Base64 (instead of byte[] since WorkManager.Data size is unexpectedly expanded with byte[]).
+            return encodeToBase64(bytes);
+        }
+    }
+
+    public static String decompress(@NonNull String base64) throws Exception {
+        byte[] data = decodeFromBase64(base64);
+
+        final Inflater inflater = new Inflater();
+        inflater.setInput(data);
+
+        try (final ByteArrayOutputStream outputStream = new ByteArrayOutputStream(data.length)) {
+            byte[] buffer = new byte[BUFFER_SIZE];
+            while (!inflater.finished()) {
+                final int count = inflater.inflate(buffer);
+                outputStream.write(buffer, 0, count);
+            }
+
+            return outputStream.toString();
+        }
+    }
+
+    static String encodeToBase64(byte[] bytes) {
+        // - org.apache.commons.Base64 is used (instead of android.util.Base64) for unit testing
+        // - encodeBase64() for backward compatibility (instead of encodeBase64String()).
+        String base64 = "";
+        if (bytes != null) {
+            byte[] encoded = Base64.encodeBase64(bytes);
+            base64= new String(encoded);
+        }
+        return base64;
+    }
+
+    static byte[] decodeFromBase64(String base64) {
+        return Base64.decodeBase64(base64.getBytes());
+    }
+
+}

event-handler/src/main/java/com/optimizely/ab/android/event_handler/EventWorker.java

@@ -18,6 +18,8 @@
 import android.content.Context;
 
 import androidx.annotation.NonNull;
+import androidx.annotation.Nullable;
+import androidx.annotation.VisibleForTesting;
 import androidx.work.Data;
 import androidx.work.Worker;
 import androidx.work.WorkerParameters;
@@ -32,7 +34,8 @@
 public class EventWorker extends Worker {
     public static final String workerId = "EventWorker";
 
-    EventDispatcher eventDispatcher;
+    @VisibleForTesting
+    public EventDispatcher eventDispatcher;
 
     public EventWorker(@NonNull Context context, @NonNull WorkerParameters workerParams) {
         super(context, workerParams);
@@ -46,30 +49,88 @@ public EventWorker(@NonNull Context context, @NonNull WorkerParameters workerPar
                 new ServiceScheduler.PendingIntentFactory(context),
                 LoggerFactory.getLogger(ServiceScheduler.class));
         eventDispatcher = new EventDispatcher(context, optlyStorage, eventDAO, eventClient, serviceScheduler, LoggerFactory.getLogger(EventDispatcher.class));
-
-    }
-
-    public static Data getData(LogEvent event) {
-        return new Data.Builder()
-                .putString("url", event.getEndpointUrl())
-                .putString("body", event.getBody())
-                .build();
     }
 
     @NonNull
     @Override
     public Result doWork() {
-        String url = getInputData().getString("url");
-        String body = getInputData().getString("body");
+        Data inputData = getInputData();
+        String url = inputData.getString("url");
+        String body = getEventBodyFromInputData(inputData);
         boolean dispatched = true;
 
-        if (url != null && !url.isEmpty() && body != null && !body.isEmpty()) {
+        if (isEventValid(url, body)) {
             dispatched = eventDispatcher.dispatch(url, body);
-        }
-        else {
+        } else {
             dispatched = eventDispatcher.dispatch();
         }
 
         return dispatched ? Result.success() : Result.retry();
     }
+
+    public static Data getData(LogEvent event) {
+        String url = event.getEndpointUrl();
+        String body = event.getBody();
+
+        // androidx.work.Data throws IllegalStateException if total data length is more than MAX_DATA_BYTES
+        // compress larger body and uncompress it before dispatching. The compress rate is very high because of repeated data (20KB -> 1KB, 45KB -> 1.5KB).
+
+        int maxSizeBeforeCompress = Data.MAX_DATA_BYTES - 1000;  // 1000 reserved for other meta data
+
+        if (body.length() < maxSizeBeforeCompress) {
+            return dataForEvent(url, body);
+        } else {
+            return compressEvent(url, body);
+        }
+    }
+
+    @VisibleForTesting
+    public static Data compressEvent(String url, String body) {
+        try {
+            String compressed = EventHandlerUtils.compress(body);
+            return dataForCompressedEvent(url, compressed);
+        } catch (Exception e) {
+            return dataForEvent(url, body);
+        }
+    }
+
+    @VisibleForTesting
+    public static Data dataForEvent(String url, String body) {
+        return new Data.Builder()
+                .putString("url", url)
+                .putString("body", body)
+                .build();
+    }
+
+    @VisibleForTesting
+    public static Data dataForCompressedEvent(String url, String compressed) {
+        return new Data.Builder()
+                .putString("url", url)
+                .putString("bodyCompressed", compressed)
+                .build();
+    }
+
+    @VisibleForTesting
+    @Nullable
+    public String getEventBodyFromInputData(Data inputData) {
+        // check non-compressed data first
+
+        String body = inputData.getString("body");
+        if (body != null) return body;
+
+        // check if data compressed
+
+        String compressed = inputData.getString("bodyCompressed");
+        try {
+            return EventHandlerUtils.decompress(compressed);
+        } catch (Exception e) {
+            return null;
+        }
+    }
+
+    @VisibleForTesting
+    public boolean isEventValid(String url, String body) {
+        return url != null && !url.isEmpty() && body != null && !body.isEmpty();
+    }
+
 }

event-handler/src/test/java/com/optimizely/ab/android/event_handler/EventHandlerUtilsTest.java

@@ -0,0 +1,82 @@
+package com.optimizely.ab.android.event_handler;
+
+import static org.junit.Assert.assertEquals;
+
+import androidx.work.Data;
+
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class EventHandlerUtilsTest {
+
+    @Test
+    public void compressAndUncompress() throws Exception {
+        String str = makeRandomString(1000);
+
+        String compressed = EventHandlerUtils.compress(str);
+        assert(compressed.length() < (str.length() * 0.5));
+
+        String uncompressed = EventHandlerUtils.decompress(compressed);
+        assertEquals(str, uncompressed);
+    }
+
+    @Test(timeout=30000)
+    public void measureCompressionDelay() throws Exception {
+        int maxEventSize = 100000;  // 100KB (~100 attributes)
+        int count = 3000;
+
+        String body = EventHandlerUtilsTest.makeRandomString(maxEventSize);
+
+        long start = System.currentTimeMillis();
+        for (int i = 0; i < count; i++) {
+            EventHandlerUtils.compress(body);
+        }
+        long end = System.currentTimeMillis();
+        float delayCompress = ((float)(end - start))/count;
+        System.out.println("Compression Delay: " + String.valueOf(delayCompress) + " millisecs");
+        assert(delayCompress < 10);   // less than 1ms for 100KB (set 10ms upperbound)
+
+        start = System.currentTimeMillis();
+        for (int i = 0; i < count; i++) {
+            String compressed = EventHandlerUtils.compress(body);
+            EventHandlerUtils.decompress(compressed);
+        }
+        end = System.currentTimeMillis();
+        float delayUncompress = ((float)(end - start))/count - delayCompress;
+        System.out.println("Uncompression Delay: " + String.valueOf(delayUncompress) + " millisecs");
+        assert(delayUncompress < 10);  // less than 1ms for 100KB (set 10ms upperbound)
+    }
+
+    public static String makeRandomString(int maxSize) {
+        StringBuilder builder = new StringBuilder();
+
+        // for high compression rate, shift repeated string window.
+        int window = 100;
+        int shift = 3;  // adjust (1...10) this for compression rate. smaller for higher rates.
+
+        int start = 0;
+        int end = start + window;
+        int i = 0;
+
+        int size = 0;
+        while (true) {
+            String str = String.valueOf(i);
+            size += str.length();
+            if (size > maxSize) {
+                break;
+            }
+            builder.append(str);
+
+            i++;
+            if (i > end) {
+                start = start + shift;
+                end = start + window;
+                i = start;
+            }
+        }
+
+        return builder.toString();
+    }
+
+}