Streaming modules

build.gradle

@@ -145,7 +145,7 @@ dependencies {
         return candidates.find { findProject(it) != null }
     }
 
-    ['main', 'logger', 'executor', 'events', 'events-domain', 'api', 'http-api', 'http', 'fallback', 'backoff', 'tracker', 'submitter'].each { moduleName ->
+    ['main', 'logger', 'events', 'events-domain', 'api', 'http-api', 'http', 'fallback', 'backoff', 'tracker', 'submitter', 'streaming', 'streaming-support', 'executor'].each { moduleName ->
         def resolvedPath = resolveProjectPath(moduleName)
         if (resolvedPath != null) {
             include project(resolvedPath)

main/build.gradle

@@ -60,8 +60,10 @@ dependencies {
     api clientModuleProject('submitter')
 
     // Internal module dependencies
-    implementation clientModuleProject('http')
-    implementation clientModuleProject('events-domain')
+    implementation clientModuleProject(':http')
+    implementation clientModuleProject(':events-domain')
+    implementation clientModuleProject(':streaming')
+    implementation clientModuleProject(':streaming-support')
 
     // External dependencies
     implementation libs.roomRuntime

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;
@@ -54,13 +54,22 @@
 import io.split.android.client.service.sseclient.reactor.MySegmentsUpdateWorkerRegistry;
 import io.split.android.client.service.sseclient.reactor.SplitUpdatesWorker;
 import io.split.android.client.service.sseclient.sseclient.BackoffCounterTimer;
+import io.split.android.client.service.sseclient.sseclient.HttpFetcherStreamingAuthFetcher;
+import io.split.android.client.service.sseclient.sseclient.NotificationProcessorUpdateListener;
 import io.split.android.client.service.sseclient.sseclient.PushNotificationManager;
 import io.split.android.client.service.sseclient.sseclient.SseAuthenticator;
 import io.split.android.client.service.sseclient.sseclient.SseClient;
-import io.split.android.client.service.sseclient.sseclient.SseClientImpl;
+import io.split.android.client.service.sseclient.sseclient.HttpClientStreamingTransport;
+import io.split.android.client.service.sseclient.sseclient.DefaultSseClient;
+import io.split.android.client.service.sseclient.sseclient.EventSourceClientImpl;
 import io.split.android.client.service.sseclient.sseclient.SseHandler;
 import io.split.android.client.service.sseclient.sseclient.SseRefreshTokenTimer;
+import io.split.android.client.service.sseclient.sseclient.SplitTaskExecutorStreamingScheduler;
 import io.split.android.client.service.sseclient.sseclient.StreamingComponents;
+import io.split.android.client.service.sseclient.sseclient.TelemetryRuntimeProducerStreamingTelemetry;
+import io.split.android.client.service.sseclient.spi.StreamingScheduler;
+import io.split.android.client.service.sseclient.spi.StreamingTelemetry;
+import io.split.android.client.service.sseclient.spi.UpdateNotificationListener;
 import io.split.android.client.service.synchronizer.RolloutCacheManager;
 import io.split.android.client.service.synchronizer.RolloutCacheManagerImpl;
 import io.split.android.client.service.synchronizer.SyncGuardian;
@@ -288,37 +297,41 @@ SyncManager buildSyncManager(SplitClientConfig config,
     }
 
     @NonNull
-    PushNotificationManager getPushNotificationManager(SplitTaskExecutor splitTaskExecutor,
+    PushNotificationManager getPushNotificationManager(StreamingScheduler scheduler,
                                                        SseAuthenticator sseAuthenticator,
                                                        PushManagerEventBroadcaster pushManagerEventBroadcaster,
                                                        SseClient sseClient,
-                                                       TelemetryRuntimeProducer telemetryRuntimeProducer,
+                                                       StreamingTelemetry telemetry,
                                                        long defaultSseConnectionDelayInSecs,
                                                        int sseDisconnectionDelayInSecs) {
         return new PushNotificationManager(pushManagerEventBroadcaster,
                 sseAuthenticator,
                 sseClient,
-                new SseRefreshTokenTimer(splitTaskExecutor, pushManagerEventBroadcaster),
-                telemetryRuntimeProducer,
+                new SseRefreshTokenTimer(scheduler, pushManagerEventBroadcaster),
+                scheduler,
+                telemetry,
                 defaultSseConnectionDelayInSecs,
                 sseDisconnectionDelayInSecs,
                 null);
     }
 
     public SseClient getSseClient(String streamingServiceUrlString,
                                   NotificationParser notificationParser,
-                                  NotificationProcessor notificationProcessor,
-                                  TelemetryRuntimeProducer telemetryRuntimeProducer,
+                                  UpdateNotificationListener updateListener,
+                                  StreamingTelemetry telemetry,
                                   PushManagerEventBroadcaster pushManagerEventBroadcaster,
                                   HttpClient httpClient) {
         SseHandler sseHandler = new SseHandler(notificationParser,
-                notificationProcessor,
-                telemetryRuntimeProducer,
+                updateListener,
+                telemetry,
                 pushManagerEventBroadcaster);
 
-        return new SseClientImpl(URI.create(streamingServiceUrlString),
-                httpClient,
-                new EventStreamParser(),
+        EventSourceClientImpl eventSourceClient = new EventSourceClientImpl(
+                new HttpClientStreamingTransport(httpClient),
+                new EventStreamParser());
+
+        return new DefaultSseClient(URI.create(streamingServiceUrlString),
+                eventSourceClient,
                 sseHandler);
     }
 
@@ -396,22 +409,25 @@ public StreamingComponents buildStreamingComponents(@NonNull SplitTaskExecutor s
                 notificationParser, splitsUpdateNotificationQueue);
 
         PushManagerEventBroadcaster pushManagerEventBroadcaster = new PushManagerEventBroadcaster();
+        StreamingScheduler scheduler = new SplitTaskExecutorStreamingScheduler(splitTaskExecutor);
+        StreamingTelemetry streamingTelemetry = new TelemetryRuntimeProducerStreamingTelemetry(storageContainer.getTelemetryStorage());
+        UpdateNotificationListener updateListener = new NotificationProcessorUpdateListener(notificationProcessor);
 
         SseClient sseClient = getSseClient(config.streamingServiceUrl(),
                 notificationParser,
-                notificationProcessor,
-                storageContainer.getTelemetryStorage(),
+                updateListener,
+                streamingTelemetry,
                 pushManagerEventBroadcaster,
                 defaultHttpClient);
 
-        SseAuthenticator sseAuthenticator = new SseAuthenticator(splitApiFacade.getSseAuthenticationFetcher(),
+        SseAuthenticator sseAuthenticator = new SseAuthenticator(new HttpFetcherStreamingAuthFetcher(splitApiFacade.getSseAuthenticationFetcher()),
                 new SseJwtParser(), flagsSpec);
 
-        PushNotificationManager pushNotificationManager = getPushNotificationManager(splitTaskExecutor,
+        PushNotificationManager pushNotificationManager = getPushNotificationManager(scheduler,
                 sseAuthenticator,
                 pushManagerEventBroadcaster,
                 sseClient,
-                storageContainer.getTelemetryStorage(),
+                streamingTelemetry,
                 config.defaultSSEConnectionDelay(),
                 config.sseDisconnectionDelay());
 

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;

main/src/main/java/io/split/android/client/service/sseclient/StreamingConstants.java

@@ -0,0 +1,21 @@
+package io.split.android.client.service.sseclient;
+
+/**
+ * Constants used by the streaming module.
+ */
+public final class StreamingConstants {
+
+    private StreamingConstants() {
+        // Utility class
+    }
+
+    /**
+     * Buffer size for segment data decompression.
+     */
+    public static final int SEGMENT_DATA_BUFFER_SIZE = 1024 * 10; // 10KB
+
+    /**
+     * Query param for flags spec in streaming auth.
+     */
+    public static final String FLAGS_SPEC_PARAM = "s";
+}

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 {
 

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 {
 

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;
 

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;

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;

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;
 

main/src/main/java/io/split/android/client/service/sseclient/spi/StreamingAuthException.java

@@ -0,0 +1,32 @@
+package io.split.android.client.service.sseclient.spi;
+
+import androidx.annotation.Nullable;
+
+/**
+ * Exception thrown by streaming auth fetchers.
+ */
+public class StreamingAuthException extends Exception {
+
+    @Nullable
+    private final Integer mStatusCode;
+
+    public StreamingAuthException(String message) {
+        super(message);
+        mStatusCode = null;
+    }
+
+    public StreamingAuthException(String message, Throwable cause) {
+        super(message, cause);
+        mStatusCode = null;
+    }
+
+    public StreamingAuthException(String message, Throwable cause, Integer statusCode) {
+        super(message, cause);
+        mStatusCode = statusCode;
+    }
+
+    @Nullable
+    public Integer getStatusCode() {
+        return mStatusCode;
+    }
+}

main/src/main/java/io/split/android/client/service/sseclient/spi/StreamingAuthFetcher.java

@@ -0,0 +1,23 @@
+package io.split.android.client.service.sseclient.spi;
+
+import androidx.annotation.NonNull;
+
+import java.util.Map;
+
+import io.split.android.client.service.sseclient.SseAuthenticationResponse;
+
+/**
+ * Abstraction for fetching streaming authentication tokens.
+ */
+public interface StreamingAuthFetcher {
+
+    /**
+     * Executes the auth request with the provided parameters.
+     *
+     * @param params request parameters
+     * @return authentication response
+     * @throws StreamingAuthException when request fails
+     */
+    @NonNull
+    SseAuthenticationResponse execute(@NonNull Map<String, Object> params) throws StreamingAuthException;
+}

main/src/main/java/io/split/android/client/service/sseclient/spi/StreamingScheduler.java

@@ -0,0 +1,40 @@
+package io.split.android.client.service.sseclient.spi;
+
+import androidx.annotation.NonNull;
+import androidx.annotation.Nullable;
+
+/**
+ * Interface for scheduling delayed tasks within the streaming module.
+ * Implementations should provide timer/scheduling capabilities backed
+ * by the host application's task executor.
+ */
+public interface StreamingScheduler {
+
+    /**
+     * Schedules a task to run after the specified delay.
+     *
+     * @param task the runnable to execute
+     * @param delaySeconds delay before execution in seconds
+     * @param listener optional listener to be notified when task completes
+     * @return a unique task ID that can be used to cancel the task
+     */
+    @NonNull
+    String schedule(@NonNull Runnable task, long delaySeconds, @Nullable TaskExecutionListener listener);
+
+    /**
+     * Cancels a previously scheduled task.
+     *
+     * @param taskId the ID returned by schedule()
+     */
+    void cancel(@Nullable String taskId);
+
+    /**
+     * Listener interface for task completion notifications.
+     */
+    interface TaskExecutionListener {
+        /**
+         * Called when a scheduled task has completed execution.
+         */
+        void onTaskExecuted();
+    }
+}