Streaming support module

gthea · gthea · commit 5f5184a9d193 · 2026-03-25T16:48:24.000-03:00
diff --git a/main/build.gradle b/main/build.gradle
@@ -62,6 +62,7 @@ dependencies {
     implementation project(':http')
     implementation project(':events-domain')
     implementation project(':streaming')
+    implementation project(':streaming-support')
 
     // External dependencies
     implementation libs.roomRuntime
diff --git a/main/src/main/java/io/split/android/client/SplitFactoryHelper.java b/main/src/main/java/io/split/android/client/SplitFactoryHelper.java
@@ -22,7 +22,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantLock;
 
-import io.split.android.client.common.CompressionUtilProvider;
+import io.split.android.client.streaming.support.CompressionUtilProvider;
 import io.split.android.client.events.EventsManagerCoordinator;
 import io.split.android.client.events.SplitInternalEvent;
 import io.split.android.client.lifecycle.SplitLifecycleManager;
diff --git a/main/src/main/java/io/split/android/client/SplitFactoryImpl.java b/main/src/main/java/io/split/android/client/SplitFactoryImpl.java
@@ -24,7 +24,7 @@
 
 import io.split.android.client.main.BuildConfig;
 import io.split.android.client.api.Key;
-import io.split.android.client.common.CompressionUtilProvider;
+import io.split.android.client.streaming.support.CompressionUtilProvider;
 import io.split.android.client.events.EventsManagerCoordinator;
 import io.split.android.client.factory.FactoryMonitor;
 import io.split.android.client.factory.FactoryMonitorImpl;
diff --git a/main/src/main/java/io/split/android/client/service/sseclient/notifications/InstantUpdateChangeNotification.java b/main/src/main/java/io/split/android/client/service/sseclient/notifications/InstantUpdateChangeNotification.java
@@ -4,7 +4,7 @@
 
 import com.google.gson.annotations.SerializedName;
 
-import io.split.android.client.common.CompressionType;
+import io.split.android.client.streaming.support.CompressionType;
 
 public abstract class InstantUpdateChangeNotification extends IncomingNotification {
 
diff --git a/main/src/main/java/io/split/android/client/service/sseclient/notifications/MembershipNotification.java b/main/src/main/java/io/split/android/client/service/sseclient/notifications/MembershipNotification.java
@@ -6,7 +6,7 @@
 
 import java.util.Set;
 
-import io.split.android.client.common.CompressionType;
+import io.split.android.client.streaming.support.CompressionType;
 
 public class MembershipNotification extends IncomingNotification {
 
diff --git a/main/src/main/java/io/split/android/client/service/sseclient/notifications/MySegmentsV2PayloadDecoder.java b/main/src/main/java/io/split/android/client/service/sseclient/notifications/MySegmentsV2PayloadDecoder.java
@@ -5,7 +5,7 @@
 
 import io.split.android.client.exceptions.MySegmentsParsingException;
 import io.split.android.client.utils.Base64Util;
-import io.split.android.client.utils.CompressionUtil;
+import io.split.android.client.streaming.support.CompressionUtil;
 import io.split.android.client.utils.MurmurHash3;
 import io.split.android.client.utils.StringHelper;
 
diff --git a/main/src/main/java/io/split/android/client/service/sseclient/notifications/memberships/MembershipsNotificationProcessorImpl.java b/main/src/main/java/io/split/android/client/service/sseclient/notifications/memberships/MembershipsNotificationProcessorImpl.java
@@ -5,8 +5,8 @@
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 
-import io.split.android.client.common.CompressionType;
-import io.split.android.client.common.CompressionUtilProvider;
+import io.split.android.client.streaming.support.CompressionType;
+import io.split.android.client.streaming.support.CompressionUtilProvider;
 import io.split.android.client.service.executor.SplitTaskExecutor;
 import io.split.android.client.service.mysegments.MySegmentUpdateParams;
 import io.split.android.client.service.mysegments.MySegmentsUpdateTask;
diff --git a/main/src/main/java/io/split/android/client/service/sseclient/notifications/mysegments/MembershipsNotificationProcessorFactoryImpl.java b/main/src/main/java/io/split/android/client/service/sseclient/notifications/mysegments/MembershipsNotificationProcessorFactoryImpl.java
@@ -4,7 +4,7 @@
 
 import androidx.annotation.NonNull;
 
-import io.split.android.client.common.CompressionUtilProvider;
+import io.split.android.client.streaming.support.CompressionUtilProvider;
 import io.split.android.client.service.executor.SplitTaskExecutor;
 import io.split.android.client.service.sseclient.notifications.MySegmentsV2PayloadDecoder;
 import io.split.android.client.service.sseclient.notifications.NotificationParser;
diff --git a/main/src/main/java/io/split/android/client/service/sseclient/reactor/SplitUpdatesWorker.java b/main/src/main/java/io/split/android/client/service/sseclient/reactor/SplitUpdatesWorker.java
@@ -8,7 +8,7 @@
 
 import java.util.concurrent.BlockingQueue;
 
-import io.split.android.client.common.CompressionUtilProvider;
+import io.split.android.client.streaming.support.CompressionUtilProvider;
 import io.split.android.client.dtos.Helper;
 import io.split.android.client.dtos.RuleBasedSegment;
 import io.split.android.client.dtos.Split;
@@ -23,7 +23,7 @@
 import io.split.android.client.storage.rbs.RuleBasedSegmentStorage;
 import io.split.android.client.storage.splits.SplitsStorage;
 import io.split.android.client.utils.Base64Util;
-import io.split.android.client.utils.CompressionUtil;
+import io.split.android.client.streaming.support.CompressionUtil;
 import io.split.android.client.utils.Json;
 import io.split.android.client.utils.logger.Logger;
 
diff --git a/main/src/test/java/io/split/android/client/service/sseclient/NotificationParserTest.java b/main/src/test/java/io/split/android/client/service/sseclient/NotificationParserTest.java
@@ -11,7 +11,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import io.split.android.client.common.CompressionType;
+import io.split.android.client.streaming.support.CompressionType;
 import io.split.android.client.service.sseclient.notifications.ControlNotification;
 import io.split.android.client.service.sseclient.notifications.HashingAlgorithm;
 import io.split.android.client.service.sseclient.notifications.IncomingNotification;
diff --git a/main/src/test/java/io/split/android/client/service/sseclient/SplitUpdateWorkerTest.java b/main/src/test/java/io/split/android/client/service/sseclient/SplitUpdateWorkerTest.java
@@ -22,8 +22,8 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 
-import io.split.android.client.common.CompressionType;
-import io.split.android.client.common.CompressionUtilProvider;
+import io.split.android.client.streaming.support.CompressionType;
+import io.split.android.client.streaming.support.CompressionUtilProvider;
 import io.split.android.client.service.executor.SplitTaskExecutionInfo;
 import io.split.android.client.service.executor.SplitTaskExecutor;
 import io.split.android.client.service.executor.SplitTaskFactory;
@@ -38,7 +38,7 @@
 import io.split.android.client.service.synchronizer.Synchronizer;
 import io.split.android.client.storage.rbs.RuleBasedSegmentStorage;
 import io.split.android.client.storage.splits.SplitsStorage;
-import io.split.android.client.utils.CompressionUtil;
+import io.split.android.client.streaming.support.CompressionUtil;
 import io.split.android.fake.SplitTaskExecutorStub;
 
 public class SplitUpdateWorkerTest {
diff --git a/main/src/test/java/io/split/android/client/service/sseclient/notifications/mysegments/MySegmentsNotificationProcessorImplTest.java b/main/src/test/java/io/split/android/client/service/sseclient/notifications/mysegments/MySegmentsNotificationProcessorImplTest.java
@@ -28,7 +28,7 @@
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 
-import io.split.android.client.common.CompressionUtilProvider;
+import io.split.android.client.streaming.support.CompressionUtilProvider;
 import io.split.android.client.exceptions.MySegmentsParsingException;
 import io.split.android.client.service.executor.SplitTaskExecutor;
 import io.split.android.client.service.mysegments.MySegmentUpdateParams;
@@ -43,7 +43,7 @@
 import io.split.android.client.service.sseclient.notifications.NotificationParser;
 import io.split.android.client.service.sseclient.notifications.NotificationType;
 import io.split.android.client.service.sseclient.notifications.memberships.MembershipsNotificationProcessorImpl;
-import io.split.android.client.utils.CompressionUtil;
+import io.split.android.client.streaming.support.CompressionUtil;
 
 public class MySegmentsNotificationProcessorImplTest {
 
diff --git a/main/src/test/java/io/split/android/client/service/sseclient/sseclient/notifications/SplitsChangeNotificationTest.java b/main/src/test/java/io/split/android/client/service/sseclient/sseclient/notifications/SplitsChangeNotificationTest.java
@@ -5,7 +5,7 @@
 
 import org.junit.Test;
 
-import io.split.android.client.common.CompressionType;
+import io.split.android.client.streaming.support.CompressionType;
 import io.split.android.client.service.sseclient.notifications.SplitsChangeNotification;
 import io.split.android.client.utils.Json;
 
diff --git a/settings.gradle b/settings.gradle
@@ -9,6 +9,7 @@ include ':main'
 include ':events'
 include ':events-domain'
 include ':streaming'
+include ':streaming-support'
 include ':backoff'
 include ':tracker'
 include ':submitter'
diff --git a/streaming-support/.gitignore b/streaming-support/.gitignore
@@ -0,0 +1,6 @@
+/build
+.gradle
+*.iml
+.DS_Store
+.classpath
+.settings
diff --git a/streaming-support/build.gradle b/streaming-support/build.gradle
@@ -0,0 +1,22 @@
+plugins {
+    id 'com.android.library'
+}
+
+apply from: "$rootDir/gradle/common-android-library.gradle"
+
+android {
+    namespace 'io.split.android.client.streaming.support'
+
+    compileOptions {
+        sourceCompatibility JavaVersion.VERSION_1_8
+        targetCompatibility JavaVersion.VERSION_1_8
+    }
+}
+
+dependencies {
+    api project(':logger')
+    implementation libs.gson
+
+    testImplementation libs.junit4
+    testImplementation libs.mockitoCore
+}
diff --git a/streaming-support/consumer-rules.pro b/streaming-support/consumer-rules.pro
@@ -0,0 +1 @@
+# Add project specific ProGuard rules here.
diff --git a/streaming-support/proguard-rules.pro b/streaming-support/proguard-rules.pro
@@ -0,0 +1 @@
+# Add project specific ProGuard rules here.
diff --git a/streaming-support/src/main/java/io/split/android/client/streaming/support/CompressionType.java b/streaming-support/src/main/java/io/split/android/client/streaming/support/CompressionType.java
@@ -1,4 +1,4 @@
-package io.split.android.client.common;
+package io.split.android.client.streaming.support;
 
 import com.google.gson.annotations.SerializedName;
 
diff --git a/streaming-support/src/main/java/io/split/android/client/streaming/support/CompressionUtil.java b/streaming-support/src/main/java/io/split/android/client/streaming/support/CompressionUtil.java
@@ -1,4 +1,4 @@
-package io.split.android.client.utils;
+package io.split.android.client.streaming.support;
 
 public interface CompressionUtil {
     byte[] decompress(byte[] compressed);
diff --git a/streaming-support/src/main/java/io/split/android/client/streaming/support/CompressionUtilProvider.java b/streaming-support/src/main/java/io/split/android/client/streaming/support/CompressionUtilProvider.java
@@ -1,27 +1,20 @@
-package io.split.android.client.common;
-
-import androidx.annotation.Nullable;
+package io.split.android.client.streaming.support;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
-import io.split.android.client.utils.CompressionUtil;
-import io.split.android.client.utils.Gzip;
 import io.split.android.client.utils.logger.Logger;
-import io.split.android.client.utils.Zlib;
 
 public class CompressionUtilProvider {
     Map<CompressionType, CompressionUtil> mCompressionUtils = new ConcurrentHashMap<>();
 
-    @Nullable
     public CompressionUtil get(CompressionType type) {
         CompressionUtil util = mCompressionUtils.get(type);
         return (util != null ? util : create(type));
     }
 
     // Using a method instead of a factory to avoid
     // a complex architecture.
-    @Nullable
     private CompressionUtil create(CompressionType type) {
         switch (type) {
             case NONE:
diff --git a/streaming-support/src/main/java/io/split/android/client/streaming/support/Gzip.java b/streaming-support/src/main/java/io/split/android/client/streaming/support/Gzip.java
@@ -1,16 +1,17 @@
-package io.split.android.client.utils;
+package io.split.android.client.streaming.support;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.zip.GZIPInputStream;
 
-import io.split.android.client.service.ServiceConstants;
 import io.split.android.client.utils.logger.Logger;
 
 public class Gzip implements CompressionUtil {
 
+    private static final int BUFFER_SIZE = 256 * 1024; // 256KB buffer
+
     @Override
     public byte[] decompress(byte[] input) {
         if (input == null || input.length == 0) {
@@ -21,7 +22,7 @@ public byte[] decompress(byte[] input) {
         GZIPInputStream gzipIn = null;
         try {
             gzipIn = new GZIPInputStream(in);
-            byte[] buffer = new byte[ServiceConstants.MY_SEGMENT_V2_DATA_SIZE];
+            byte[] buffer = new byte[BUFFER_SIZE];
             int byteCount;
             while ((byteCount = gzipIn.read(buffer)) >= 0) {
                 out.write(buffer, 0, byteCount);
diff --git a/streaming-support/src/main/java/io/split/android/client/streaming/support/Zlib.java b/streaming-support/src/main/java/io/split/android/client/streaming/support/Zlib.java
@@ -1,13 +1,14 @@
-package io.split.android.client.utils;
+package io.split.android.client.streaming.support;
 
 import java.util.Arrays;
 import java.util.zip.Inflater;
 
-import io.split.android.client.service.ServiceConstants;
 import io.split.android.client.utils.logger.Logger;
 
 public class Zlib implements CompressionUtil {
 
+    private static final int BUFFER_SIZE = 256 * 1024; // 256KB buffer
+
     @Override
     public byte[] decompress(byte[] input) {
         if (input == null || input.length == 0) {
@@ -16,7 +17,7 @@ public byte[] decompress(byte[] input) {
         try {
             Inflater inflater = new Inflater();
             inflater.setInput(input);
-            byte[] result = new byte[ServiceConstants.MY_SEGMENT_V2_DATA_SIZE];
+            byte[] result = new byte[BUFFER_SIZE];
             int resultLength = inflater.inflate(result);
             inflater.end();
             return Arrays.copyOfRange(result, 0, resultLength);
diff --git a/streaming-support/src/test/java/io/split/android/client/streaming/support/GzipTest.java b/streaming-support/src/test/java/io/split/android/client/streaming/support/GzipTest.java
@@ -0,0 +1,92 @@
+package io.split.android.client.streaming.support;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.zip.GZIPOutputStream;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+public class GzipTest {
+
+    private Gzip gzip;
+
+    @Before
+    public void setUp() {
+        gzip = new Gzip();
+    }
+
+    @Test
+    public void decompress_validGzipData_returnsDecompressedBytes() throws IOException {
+        // Arrange
+        byte[] original = "Hello, World! This is a test message for gzip compression.".getBytes();
+        byte[] compressed = compressWithGzip(original);
+
+        // Act
+        byte[] decompressed = gzip.decompress(compressed);
+
+        // Assert
+        assertNotNull(decompressed);
+        assertArrayEquals(original, decompressed);
+    }
+
+    @Test
+    public void decompress_emptyArray_returnsNull() {
+        // Act
+        byte[] result = gzip.decompress(new byte[0]);
+
+        // Assert
+        assertNull(result);
+    }
+
+    @Test
+    public void decompress_nullInput_returnsNull() {
+        // Act
+        byte[] result = gzip.decompress(null);
+
+        // Assert
+        assertNull(result);
+    }
+
+    @Test
+    public void decompress_invalidGzipData_returnsNull() {
+        // Arrange
+        byte[] invalidData = "This is not gzip compressed data".getBytes();
+
+        // Act
+        byte[] result = gzip.decompress(invalidData);
+
+        // Assert
+        assertNull(result);
+    }
+
+    @Test
+    public void decompress_largeData_decompressesSuccessfully() throws IOException {
+        // Arrange
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < 10000; i++) {
+            sb.append("Line ").append(i).append(": Some test data\n");
+        }
+        byte[] original = sb.toString().getBytes();
+        byte[] compressed = compressWithGzip(original);
+
+        // Act
+        byte[] decompressed = gzip.decompress(compressed);
+
+        // Assert
+        assertNotNull(decompressed);
+        assertArrayEquals(original, decompressed);
+    }
+
+    private byte[] compressWithGzip(byte[] data) throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        GZIPOutputStream gzipOut = new GZIPOutputStream(out);
+        gzipOut.write(data);
+        gzipOut.close();
+        return out.toByteArray();
+    }
+}
diff --git a/streaming-support/src/test/java/io/split/android/client/streaming/support/ZlibTest.java b/streaming-support/src/test/java/io/split/android/client/streaming/support/ZlibTest.java
