Replace StreamBatcher with StreamEventAggregator instead in the StreamSocketSession

stream-android-core/src/main/java/io/getstream/android/core/api/StreamClient.kt

@@ -35,7 +35,8 @@
 import io.getstream.android.core.api.model.connection.network.StreamNetworkState
 import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleMonitor
 import io.getstream.android.core.api.observers.network.StreamNetworkMonitor
-import io.getstream.android.core.api.processing.StreamBatcher
+import io.getstream.android.core.api.processing.StreamEventAggregationPolicy
+import io.getstream.android.core.api.processing.StreamEventAggregator
 import io.getstream.android.core.api.processing.StreamSerialProcessingQueue
 import io.getstream.android.core.api.processing.StreamSingleFlightProcessor
 import io.getstream.android.core.api.recovery.StreamConnectionRecoveryEvaluator
@@ -249,14 +250,7 @@
         socketFactory =
             components.socketFactory
                 ?: StreamWebSocketFactory(logger = logProvider.taggedLogger("SCWebSocketFactory")),
-        batcher =
-            components.batcher
-                ?: StreamBatcher(
-                    scope = scope,
-                    batchSize = socketConfig.batchSize,
-                    initialDelayMs = socketConfig.batchInitialDelayMs,
-                    maxDelayMs = socketConfig.batchMaxDelayMs,
-                ),
+        eventAggregator = components.eventAggregator,
         healthMonitor =
             components.healthMonitor
                 ?: StreamHealthMonitor(
@@ -277,7 +271,7 @@
  */
 @Suppress("LongParameterList", "LongMethod")
 @SuppressLint("ExposeAsStateFlow")
 internal fun createStreamClientInternal(
 
     // Android
     scope: CoroutineScope,
@@ -320,13 +314,7 @@
     connectionIdHolder: StreamConnectionIdHolder = StreamConnectionIdHolder(),
     socketFactory: StreamWebSocketFactory =
         StreamWebSocketFactory(logger = logProvider.taggedLogger("SCWebSocketFactory")),
-    batcher: StreamBatcher<String> =
-        StreamBatcher(
-            scope = scope,
-            batchSize = socketConfig.batchSize,
-            initialDelayMs = socketConfig.batchInitialDelayMs,
-            maxDelayMs = socketConfig.batchMaxDelayMs,
-        ),
+    eventAggregator: StreamEventAggregator<*>? = null,
 
     // Monitoring
     healthMonitor: StreamHealthMonitor =
@@ -427,6 +415,28 @@
                 ),
         )
 
+    val eventParser =
+        StreamCompositeEventSerializationImpl(
+            internal =
+                serializationConfig.eventParser ?: StreamEventSerialization(compositeSerialization),
+            external = serializationConfig.productEventSerializers,
+        )
+
+    val resolvedAggregator =
+        eventAggregator
+            ?: StreamEventAggregator(
+                scope = clientScope,
+                policy =
+                    StreamEventAggregationPolicy.from(
+                        typeExtractor = { raw -> eventParser.peekType(raw) },
+                        deserializer = { raw -> eventParser.deserialize(raw) },
+                        aggregationThreshold = socketConfig.aggregationThreshold,
+                        maxWindowMs = socketConfig.aggregationMaxWindowMs,
+                        dispatchQueueCapacity = socketConfig.aggregationDispatchQueueCapacity,
+                    ),
+                logger = logProvider.taggedLogger("SCEventAggregator"),
+            )
+
     val mutableConnectionState = MutableStateFlow<StreamConnectionState>(StreamConnectionState.Idle)
     return StreamClientImpl(
         user = user,
@@ -446,15 +456,9 @@
                 products = products,
                 config = socketConfig,
                 jsonSerialization = compositeSerialization,
-                eventParser =
-                    StreamCompositeEventSerializationImpl(
-                        internal =
-                            serializationConfig.eventParser
-                                ?: StreamEventSerialization(compositeSerialization),
-                        external = serializationConfig.productEventSerializers,
-                    ),
+                eventParser = eventParser,
                 healthMonitor = healthMonitor,
-                batcher = batcher,
+                aggregator = resolvedAggregator,
                 internalSocket = socket,
                 subscriptionManager =
                     StreamSubscriptionManager(

stream-android-core/src/main/java/io/getstream/android/core/api/model/config/StreamComponentProvider.kt

@@ -22,7 +22,7 @@ import io.getstream.android.core.api.components.StreamAndroidComponentsProvider
 import io.getstream.android.core.api.log.StreamLoggerProvider
 import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleMonitor
 import io.getstream.android.core.api.observers.network.StreamNetworkMonitor
-import io.getstream.android.core.api.processing.StreamBatcher
+import io.getstream.android.core.api.processing.StreamEventAggregator
 import io.getstream.android.core.api.processing.StreamSerialProcessingQueue
 import io.getstream.android.core.api.processing.StreamSingleFlightProcessor
 import io.getstream.android.core.api.recovery.StreamConnectionRecoveryEvaluator
@@ -63,7 +63,7 @@ import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
  * @param tokenManager Token lifecycle manager.
  * @param connectionIdHolder Connection ID storage.
  * @param socketFactory WebSocket factory.
- * @param batcher WebSocket message batcher.
+ * @param eventAggregator WebSocket event aggregator.
  * @param healthMonitor Connection health monitor.
  * @param networkMonitor Network connectivity monitor.
  * @param lifecycleMonitor App lifecycle monitor.
@@ -80,7 +80,7 @@ public data class StreamComponentProvider(
     val tokenManager: StreamTokenManager? = null,
     val connectionIdHolder: StreamConnectionIdHolder? = null,
     val socketFactory: StreamWebSocketFactory? = null,
-    val batcher: StreamBatcher<String>? = null,
+    val eventAggregator: StreamEventAggregator<*>? = null,
     val healthMonitor: StreamHealthMonitor? = null,
     val networkMonitor: StreamNetworkMonitor? = null,
     val lifecycleMonitor: StreamLifecycleMonitor? = null,

stream-android-core/src/main/java/io/getstream/android/core/api/model/config/StreamSocketConfig.kt

@@ -25,8 +25,8 @@
  * Configuration for the Stream WebSocket connection.
  *
  * Holds both **identity** (URL, API key, auth type) and **operational** tunables (health check
- * timing, batching, connection timeout). Products pass this to the [StreamClient] factory to
- * describe their socket.
+ * timing, event aggregation, connection timeout). Products pass this to the [StreamClient] factory
+ * to describe their socket.
  *
  * ### Usage
  *
@@ -38,15 +38,16 @@
  *     clientInfoHeader = clientInfo,
  * )
  *
- * // SFU socket — aggressive timing, no batching
+ * // SFU socket — aggressive timing, low aggregation threshold
  * val sfuSocket = StreamSocketConfig.jwt(
  *     url = StreamWsUrl.fromString("wss://sfu.stream-io-api.com"),
  *     apiKey = apiKey,
  *     clientInfoHeader = clientInfo,
  *     healthCheckIntervalMs = 5_000,
  *     livenessThresholdMs = 15_000,
  *     connectionTimeoutMs = 2_000,
- *     batchSize = 1,
+ *     aggregationThreshold = 10,
+ *     aggregationMaxWindowMs = 200,
  * )
  * ```
  *
@@ -58,9 +59,12 @@
  * @param livenessThresholdMs Time without a health check ack before the connection is considered
  *   unhealthy in milliseconds.
  * @param connectionTimeoutMs WebSocket connection timeout in milliseconds.
- * @param batchSize Maximum number of WebSocket messages to batch before flushing.
- * @param batchInitialDelayMs Initial debounce window for batching in milliseconds.
- * @param batchMaxDelayMs Maximum debounce window for batching in milliseconds.
+ * @param aggregationThreshold Number of accumulated events that triggers aggregated delivery
+ *   instead of individual dispatch.
+ * @param aggregationMaxWindowMs Maximum time the aggregator collects events before delivering. This
+ *   is the latency ceiling in milliseconds.
+ * @param aggregationDispatchQueueCapacity Bounded capacity of the dispatch queue between the
+ *   aggregator's collector and dispatcher coroutines.
  */
 @Suppress("LongParameterList")
 @StreamInternalApi
@@ -74,9 +78,9 @@
     val healthCheckIntervalMs: Long = DEFAULT_HEALTH_INTERVAL_MS,
     val livenessThresholdMs: Long = DEFAULT_LIVENESS_MS,
     val connectionTimeoutMs: Long = DEFAULT_CONNECTION_TIMEOUT_MS,
-    val batchSize: Int = DEFAULT_BATCH_SIZE,
-    val batchInitialDelayMs: Long = DEFAULT_BATCH_INIT_DELAY_MS,
-    val batchMaxDelayMs: Long = DEFAULT_BATCH_MAX_DELAY_MS,
+    val aggregationThreshold: Int = DEFAULT_AGGREGATION_THRESHOLD,
+    val aggregationMaxWindowMs: Long = DEFAULT_AGGREGATION_MAX_WINDOW_MS,
+    val aggregationDispatchQueueCapacity: Int = DEFAULT_AGGREGATION_DISPATCH_QUEUE_CAPACITY,
 ) {
     /** Default values for [StreamSocketConfig] fields. */
     public companion object {
@@ -92,14 +96,14 @@
         /** Default connection timeout: 10 seconds. */
         public const val DEFAULT_CONNECTION_TIMEOUT_MS: Long = 10_000L
 
-        /** Default batch size: 10 messages. */
-        public const val DEFAULT_BATCH_SIZE: Int = 10
+        /** Default aggregation threshold: 50 events trigger aggregated delivery. */
+        public const val DEFAULT_AGGREGATION_THRESHOLD: Int = 50
 
-        /** Default initial batch delay: 100ms. */
-        public const val DEFAULT_BATCH_INIT_DELAY_MS: Long = 100L
+        /** Default aggregation max window: 500ms latency ceiling. */
+        public const val DEFAULT_AGGREGATION_MAX_WINDOW_MS: Long = 500L
 
-        /** Default max batch delay: 1 second. */
-        public const val DEFAULT_BATCH_MAX_DELAY_MS: Long = 1_000L
+        /** Default dispatch queue capacity: 16 items. */
+        public const val DEFAULT_AGGREGATION_DISPATCH_QUEUE_CAPACITY: Int = 16
 
         /**
          * Creates a JWT-based [StreamSocketConfig].
@@ -110,22 +114,22 @@
          * @param healthCheckIntervalMs Interval between health check pings in milliseconds.
          * @param livenessThresholdMs Liveness threshold in milliseconds.
          * @param connectionTimeoutMs WebSocket connection timeout in milliseconds.
-         * @param batchSize Maximum batch size before flush.
-         * @param batchInitialDelayMs Initial debounce window in milliseconds.
-         * @param batchMaxDelayMs Maximum debounce window in milliseconds.
+         * @param aggregationThreshold Events before aggregated delivery triggers.
+         * @param aggregationMaxWindowMs Maximum collection window in milliseconds.
+         * @param aggregationDispatchQueueCapacity Dispatch queue capacity.
          * @return A JWT-based [StreamSocketConfig].
          */
         @Suppress("LongParameterList")
         public fun jwt(
             url: StreamWsUrl,
             apiKey: StreamApiKey,
             clientInfoHeader: StreamHttpClientInfoHeader,
             healthCheckIntervalMs: Long = DEFAULT_HEALTH_INTERVAL_MS,
             livenessThresholdMs: Long = DEFAULT_LIVENESS_MS,
             connectionTimeoutMs: Long = DEFAULT_CONNECTION_TIMEOUT_MS,
-            batchSize: Int = DEFAULT_BATCH_SIZE,
-            batchInitialDelayMs: Long = DEFAULT_BATCH_INIT_DELAY_MS,
-            batchMaxDelayMs: Long = DEFAULT_BATCH_MAX_DELAY_MS,
+            aggregationThreshold: Int = DEFAULT_AGGREGATION_THRESHOLD,
+            aggregationMaxWindowMs: Long = DEFAULT_AGGREGATION_MAX_WINDOW_MS,
+            aggregationDispatchQueueCapacity: Int = DEFAULT_AGGREGATION_DISPATCH_QUEUE_CAPACITY,
         ): StreamSocketConfig =
             StreamSocketConfig(
                 url = url,
@@ -135,9 +139,9 @@
                 healthCheckIntervalMs = healthCheckIntervalMs,
                 livenessThresholdMs = livenessThresholdMs,
                 connectionTimeoutMs = connectionTimeoutMs,
-                batchSize = batchSize,
-                batchInitialDelayMs = batchInitialDelayMs,
-                batchMaxDelayMs = batchMaxDelayMs,
+                aggregationThreshold = aggregationThreshold,
+                aggregationMaxWindowMs = aggregationMaxWindowMs,
+                aggregationDispatchQueueCapacity = aggregationDispatchQueueCapacity,
             )
 
         /**
@@ -149,22 +153,22 @@
          * @param healthCheckIntervalMs Interval between health check pings in milliseconds.
          * @param livenessThresholdMs Liveness threshold in milliseconds.
          * @param connectionTimeoutMs WebSocket connection timeout in milliseconds.
-         * @param batchSize Maximum batch size before flush.
-         * @param batchInitialDelayMs Initial debounce window in milliseconds.
-         * @param batchMaxDelayMs Maximum debounce window in milliseconds.
+         * @param aggregationThreshold Events before aggregated delivery triggers.
+         * @param aggregationMaxWindowMs Maximum collection window in milliseconds.
+         * @param aggregationDispatchQueueCapacity Dispatch queue capacity.
          * @return An anonymous [StreamSocketConfig].
          */
         @Suppress("LongParameterList")
         public fun anonymous(
             url: StreamWsUrl,
             apiKey: StreamApiKey,
             clientInfoHeader: StreamHttpClientInfoHeader,
             healthCheckIntervalMs: Long = DEFAULT_HEALTH_INTERVAL_MS,
             livenessThresholdMs: Long = DEFAULT_LIVENESS_MS,
             connectionTimeoutMs: Long = DEFAULT_CONNECTION_TIMEOUT_MS,
-            batchSize: Int = DEFAULT_BATCH_SIZE,
-            batchInitialDelayMs: Long = DEFAULT_BATCH_INIT_DELAY_MS,
-            batchMaxDelayMs: Long = DEFAULT_BATCH_MAX_DELAY_MS,
+            aggregationThreshold: Int = DEFAULT_AGGREGATION_THRESHOLD,
+            aggregationMaxWindowMs: Long = DEFAULT_AGGREGATION_MAX_WINDOW_MS,
+            aggregationDispatchQueueCapacity: Int = DEFAULT_AGGREGATION_DISPATCH_QUEUE_CAPACITY,
         ): StreamSocketConfig =
             StreamSocketConfig(
                 url = url,
@@ -174,9 +178,9 @@
                 healthCheckIntervalMs = healthCheckIntervalMs,
                 livenessThresholdMs = livenessThresholdMs,
                 connectionTimeoutMs = connectionTimeoutMs,
-                batchSize = batchSize,
-                batchInitialDelayMs = batchInitialDelayMs,
-                batchMaxDelayMs = batchMaxDelayMs,
+                aggregationThreshold = aggregationThreshold,
+                aggregationMaxWindowMs = aggregationMaxWindowMs,
+                aggregationDispatchQueueCapacity = aggregationDispatchQueueCapacity,
             )
 
         /**
@@ -189,13 +193,13 @@
          * @param healthCheckIntervalMs Interval between health check pings in milliseconds.
          * @param livenessThresholdMs Liveness threshold in milliseconds.
          * @param connectionTimeoutMs WebSocket connection timeout in milliseconds.
-         * @param batchSize Maximum batch size before flush.
-         * @param batchInitialDelayMs Initial debounce window in milliseconds.
-         * @param batchMaxDelayMs Maximum debounce window in milliseconds.
+         * @param aggregationThreshold Events before aggregated delivery triggers.
+         * @param aggregationMaxWindowMs Maximum collection window in milliseconds.
+         * @param aggregationDispatchQueueCapacity Dispatch queue capacity.
          * @return A custom [StreamSocketConfig].
          */
         @Suppress("LongParameterList")
         public fun custom(
             url: StreamWsUrl,
             apiKey: StreamApiKey,
             authType: String,
@@ -203,9 +207,9 @@
             healthCheckIntervalMs: Long = DEFAULT_HEALTH_INTERVAL_MS,
             livenessThresholdMs: Long = DEFAULT_LIVENESS_MS,
             connectionTimeoutMs: Long = DEFAULT_CONNECTION_TIMEOUT_MS,
-            batchSize: Int = DEFAULT_BATCH_SIZE,
-            batchInitialDelayMs: Long = DEFAULT_BATCH_INIT_DELAY_MS,
-            batchMaxDelayMs: Long = DEFAULT_BATCH_MAX_DELAY_MS,
+            aggregationThreshold: Int = DEFAULT_AGGREGATION_THRESHOLD,
+            aggregationMaxWindowMs: Long = DEFAULT_AGGREGATION_MAX_WINDOW_MS,
+            aggregationDispatchQueueCapacity: Int = DEFAULT_AGGREGATION_DISPATCH_QUEUE_CAPACITY,
         ): StreamSocketConfig {
             require(authType.isNotBlank()) { "Auth type must not be blank" }
             return StreamSocketConfig(
@@ -216,9 +220,9 @@
                 healthCheckIntervalMs = healthCheckIntervalMs,
                 livenessThresholdMs = livenessThresholdMs,
                 connectionTimeoutMs = connectionTimeoutMs,
-                batchSize = batchSize,
-                batchInitialDelayMs = batchInitialDelayMs,
-                batchMaxDelayMs = batchMaxDelayMs,
+                aggregationThreshold = aggregationThreshold,
+                aggregationMaxWindowMs = aggregationMaxWindowMs,
+                aggregationDispatchQueueCapacity = aggregationDispatchQueueCapacity,
             )
         }
     }

stream-android-core/src/main/java/io/getstream/android/core/api/processing/StreamAggregatedEvent.kt

@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-core-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.android.core.api.processing
+
+import io.getstream.android.core.annotations.StreamInternalApi
+
+/**
+ * A batch of events produced by [StreamEventAggregator] during a traffic spike.
+ *
+ * When the aggregator detects high event throughput, it collects events within a time window and
+ * delivers them as a single [StreamAggregatedEvent] instead of dispatching each individually. This
+ * allows product SDKs to apply all updates sequentially in one atomic state update and one UI
+ * recomposition per window instead of one per event.
+ *
+ * Events are stored in arrival order. Product SDKs should process them sequentially to maintain
+ * correctness (e.g. "reaction added" must be processed before "reaction removed" for the same
+ * entity).
+ *
+ * During normal (low) traffic, events flow through individually — this class is only used during
+ * spikes.
+ *
+ * ### Usage
+ *
+ * ```kotlin
+ * when (event) {
+ *     is StreamAggregatedEvent<*> -> {
+ *         event.events.forEach { singleEvent ->
+ *             handleSingleEvent(singleEvent)
+ *         }
+ *     }
+ *     else -> handleSingleEvent(event)
+ * }
+ * ```
+ *
+ * @param T The type of the individual events.
+ * @property events Events in arrival order. Process sequentially to maintain correctness.
+ */
+@StreamInternalApi
+public class StreamAggregatedEvent<T>(events: List<T>) {
+    public val events: List<T> = events.toList()
+}