Rename and update tests

Author: aleksandar-apostolov

CLAUDE.md

@@ -163,8 +163,8 @@ StreamClient (Main Interface)
 │   └── StreamLifecycleMonitor (ProcessLifecycleOwner observer)
 ├── StreamConnectionRecoveryEvaluator (Reconnect heuristics)
 ├── StreamWatcher<T> (Watch registry & rewatch coordinator)
-│   ├── StreamSubscriptionManager<StreamRewatchListener<T>> (Rewatch listener registry)
-│   └── StreamSubscriptionManager<StreamClientListener> (Connection state monitoring)
+│   ├── StateFlow<StreamConnectionState> (Observes connection state)
+│   └── StreamSubscriptionManager<StreamRewatchListener<T>> (Rewatch listener registry)
 └── StreamSubscriptionManager<StreamClientListener> (Event distribution)
 ```
 
@@ -201,26 +201,75 @@ fun start(): Result<Unit>
 fun stop(): Result<Unit>
 ```
 
+**Factory Function:**
+```kotlin
+fun <T> StreamWatcher(
+    scope: CoroutineScope,
+    logger: StreamLogger,
+    connectionState: StateFlow<StreamConnectionState>
+): StreamWatcher<T>
+```
+
+The factory creates the watcher with an internal `StreamSubscriptionManager<StreamRewatchListener<T>>` automatically - product SDKs don't need to manage this.
+
 **Usage Flow:**
-1. Product SDK watches a channel: `watcher.watch("messaging:general")` (for `StreamWatcher<String>`)
-2. Watcher adds the identifier to an internal `ConcurrentHashMap<T, Unit>` registry
-3. Product SDK registers a rewatch listener: `watcher.subscribe(StreamRewatchListener { ids, connectionId -> ... })`
-4. Call `watcher.start()` to begin monitoring connection state changes
+1. Create watcher with connection state flow: `val watcher = StreamWatcher<String>(scope, logger, streamClient.connectionState)`
+2. Product SDK registers a rewatch listener: `watcher.subscribe(StreamRewatchListener { ids, connectionId -> ... })`
+3. Call `watcher.start()` to begin monitoring connection state changes
+4. Product SDK watches channels: `watcher.watch("messaging:general")`
 5. On `StreamConnectionState.Connected` event, watcher invokes all listeners with complete identifier list AND the current connectionId
 6. Product SDK re-establishes server-side watches for each identifier using the provided connectionId
 
+**Complete Example:**
+```kotlin
+// Create watcher for channel IDs
+val watcher = StreamWatcher<String>(
+    scope = CoroutineScope(SupervisorJob() + Dispatchers.Default),
+    logger = logger,
+    connectionState = streamClient.connectionState
+)
+
+// Register rewatch listener
+watcher.subscribe(
+    listener = StreamRewatchListener { channelIds, connectionId ->
+        // Re-establish server-side watches after reconnection
+        channelIds.forEach { channelId ->
+            channelApi.watch(channelId, connectionId)
+        }
+    },
+    options = StreamSubscriptionManager.Options()
+).getOrThrow()
+
+// Start monitoring
+watcher.start().getOrThrow()
+
+// Watch channels as users view them
+watcher.watch("messaging:general")
+watcher.watch("messaging:random")
+
+// When done
+watcher.stop()
+```
+
 **Implementation Details:**
 - **Generic type parameter**: Allows watching any type `T` - common usage is `String` for channel IDs, but can be custom data classes
-- Thread-safe: Uses `ConcurrentHashMap<T, Unit>` for the registry (line 52 in `StreamWatcherImpl.kt`)
-- Async execution: Rewatch callbacks invoked on internal coroutine scope with `SupervisorJob + Dispatchers.Default` (line 61)
-- Error handling: Exceptions from rewatch callbacks are caught, logged, and surfaced via `StreamClientListener.onError` (lines 77-82)
+- **StateFlow observation**: Watcher observes `StateFlow<StreamConnectionState>` directly, no intermediate subscription manager needed
+  - Safer: Only receives connection state, no access to full StreamClient APIs
+  - Simpler: Factory creates internal subscription manager automatically
+  - Testable: Easy to test with `MutableStateFlow`
+  - Standalone: Truly independent component with minimal dependencies
+- Thread-safe: Uses `ConcurrentHashMap<T, Unit>` for the registry (line 54 in `StreamWatcherImpl.kt`)
+- Async execution: Rewatch callbacks invoked on internal coroutine scope with `SupervisorJob + Dispatchers.Default` (line 78)
+- Error handling: Exceptions from rewatch callbacks are caught and logged (lines 88-92)
 - Idempotent: Multiple `watch()` calls with the same identifier only maintain one entry
-- Only triggers on `Connected` state when registry is non-empty (line 67)
-- Connection ID extracted from `Connected` state and passed to all listeners (line 70)
+- Only triggers on `Connected` state when registry is non-empty (line 77)
+- Connection ID extracted from `Connected` state and passed to all listeners (line 80)
+- Lifecycle: `start()` launches a coroutine to collect from StateFlow, `stop()` cancels the collection job (lines 61-72)
 
 **Test Coverage:**
 - Location: `stream-android-core/src/test/java/io/getstream/android/core/internal/watcher/StreamWatcherImplTest.kt`
-- 29 comprehensive test cases covering watch operations, lifecycle, state changes, error handling, concurrency, and connectionId verification
+- 30 comprehensive test cases covering watch operations, lifecycle, state changes, error handling, concurrency, and connectionId verification
+- Tests use MutableStateFlow to directly emit connection state changes, verifying watcher responds correctly
 - 100% instruction/branch/line coverage (verified via Kover)
 
 ## Configuration Defaults

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

@@ -50,7 +50,6 @@ import io.getstream.android.core.api.socket.listeners.StreamClientListener
 import io.getstream.android.core.api.socket.monitor.StreamHealthMonitor
 import io.getstream.android.core.api.subscribe.StreamObservable
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
-import io.getstream.android.core.api.watcher.StreamWatcher
 import io.getstream.android.core.internal.client.StreamClientImpl
 import io.getstream.android.core.internal.observers.StreamNetworkAndLifeCycleMonitor
 import io.getstream.android.core.internal.serialization.StreamCompositeEventSerializationImpl

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

@@ -20,7 +20,6 @@ import io.getstream.android.core.annotations.StreamInternalApi
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.connection.StreamConnectionState
 import io.getstream.android.core.api.observers.StreamStartableComponent
-import io.getstream.android.core.api.socket.listeners.StreamClientListener
 import io.getstream.android.core.api.subscribe.StreamObservable
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
 import io.getstream.android.core.internal.watcher.StreamWatcherImpl
@@ -39,8 +38,8 @@ import kotlinx.coroutines.flow.StateFlow
  * ## Typical Usage Flow
  * 1. Product SDK watches a channel: `watcher.watch("messaging:general")`
  * 2. Watcher adds the identifier to its internal registry
- * 3. Product SDK registers a rewatch listener: `watcher.subscribe(StreamRewatchListener { ids, connectionId ->
- *    resubscribe(ids, connectionId) })`
+ * 3. Product SDK registers a rewatch listener: `watcher.subscribe(StreamRewatchListener { ids,
+ *    connectionId -> resubscribe(ids, connectionId) })`
  * 4. Call `watcher.start()` to begin monitoring connection state changes
  * 5. Connection state changes (e.g., reconnection after network loss)
  * 6. Watcher invokes listener with all watched entries: `["messaging:general",
@@ -138,24 +137,22 @@ public interface StreamWatcher<T> :
 }
 
 /**
- * Creates a new [StreamWatcher] instance with the provided dependencies.
+ * Creates a new [StreamWatcher] instance that observes connection state changes.
  *
- * This factory method instantiates the default implementation ([StreamWatcherImpl]) and wires all
- * required dependencies for monitoring connection state changes and triggering rewatch callbacks.
+ * This factory method instantiates the default implementation ([StreamWatcherImpl]) and creates the
+ * necessary internal subscription manager. The watcher observes the provided connection state flow
+ * and triggers rewatch callbacks when the connection state changes to Connected.
  *
  * ## Parameters
- * - **scope**: The [CoroutineScope] used for launching async operations (rewatch callbacks). This
- *   scope should typically use `SupervisorJob + Dispatchers.Default` to ensure callback failures
- *   don't cancel the scope and to avoid blocking the main thread. The scope is NOT cancelled when
- *   [StreamWatcher.stop] is called, allowing the component to be restarted.
+ * - **scope**: The [CoroutineScope] used for launching async operations (collecting state flow and
+ *   invoking rewatch callbacks). This scope should typically use `SupervisorJob + Dispatchers.
+ *   Default` to ensure callback failures don't cancel the scope and to avoid blocking the main
+ *   thread. The scope is NOT cancelled when [StreamWatcher.stop] is called, allowing the component
+ *   to be restarted.
  * - **logger**: A [StreamLogger] instance for diagnostic output, tagged appropriately (e.g.,
- *   "SCWatcher"). Used for logging state changes, rewatch events, and errors.
- * - **streamRewatchSubscriptionManager**: Manages subscriptions to [StreamRewatchListener]. This
- *   manager handles the list of listeners that will be invoked when connection state changes
- *   require re-watching identifiers.
- * - **streamClientSubscriptionManager**: Manages subscriptions to [StreamClientListener]. This is
- *   used to subscribe to connection state changes (via [StreamClientListener.onState]) and to
- *   surface errors (via [StreamClientListener.onError]) when rewatch callbacks fail.
+ *   "Watcher"). Used for logging state changes, rewatch events, and errors.
+ * - **connectionState**: A [StateFlow] providing connection state updates. Typically obtained from
+ *   `streamClient.connectionState`. The watcher will collect from this flow when started.
  *
  * ## Lifecycle
  *
@@ -169,18 +166,11 @@ public interface StreamWatcher<T> :
  * val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
  * val logger = StreamLogger.getLogger("MyApp.Watcher")
  *
- * val rewatchManager = StreamSubscriptionManager<StreamRewatchListener<String>>(
- *     logger = logger.withTag("RewatchSubscriptions")
- * )
- * val clientManager = StreamSubscriptionManager<StreamClientListener>(
- *     logger = logger.withTag("ClientSubscriptions")
- * )
- *
+ * // Create watcher - only needs the connection state flow
  * val watcher = StreamWatcher<String>(
  *     scope = scope,
  *     logger = logger,
- *     streamRewatchSubscriptionManager = rewatchManager,
- *     streamClientSubscriptionManager = clientManager
+ *     connectionState = streamClient.connectionState
  * )
  *
  * // Register rewatch listener
@@ -197,10 +187,10 @@ public interface StreamWatcher<T> :
  * watcher.watch("livestream:sports")
  * ```
  *
- * @param scope Coroutine scope for async operations (rewatch callback invocations)
+ * @param scope Coroutine scope for async operations (state collection and rewatch callback
+ *   invocations)
  * @param logger Logger for diagnostic output and error reporting
- * @param streamRewatchSubscriptionManager Subscription manager for rewatch listeners
- * @param streamClientSubscriptionManager Subscription manager for client state listeners
+ * @param connectionState StateFlow providing connection state updates
  * @return A new [StreamWatcher] instance (implementation: [StreamWatcherImpl])
  * @see StreamWatcher The interface contract
  * @see StreamWatcherImpl The concrete implementation
@@ -210,12 +200,13 @@ public interface StreamWatcher<T> :
 public fun <T> StreamWatcher(
     scope: CoroutineScope,
     logger: StreamLogger,
-    streamRewatchSubscriptionManager: StreamSubscriptionManager<StreamRewatchListener<T>>,
-    streamClientSubscriptionManager: StreamSubscriptionManager<StreamClientListener>,
-): StreamWatcher<T> =
-    StreamWatcherImpl(
+    connectionState: StateFlow<StreamConnectionState>,
+): StreamWatcher<T> {
+    val rewatchSubscriptions = StreamSubscriptionManager<StreamRewatchListener<T>>(logger)
+    return StreamWatcherImpl(
         scope = scope,
+        connectionState = connectionState,
+        rewatchSubscriptions = rewatchSubscriptions,
         logger = logger,
-        rewatchSubscriptions = streamRewatchSubscriptionManager,
-        clientSubscriptions = streamClientSubscriptionManager,
     )
+}

stream-android-core/src/main/java/io/getstream/android/core/internal/watcher/StreamWatcherImpl.kt

@@ -19,91 +19,86 @@ package io.getstream.android.core.internal.watcher
 import io.getstream.android.core.annotations.StreamInternalApi
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.connection.StreamConnectionState
-import io.getstream.android.core.api.socket.listeners.StreamClientListener
 import io.getstream.android.core.api.subscribe.StreamSubscription
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
 import io.getstream.android.core.api.watcher.StreamRewatchListener
 import io.getstream.android.core.api.watcher.StreamWatcher
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentMap
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.launch
 
 /**
- * Implementation of [StreamWatcher] that uses [StreamSubscriptionManager] to monitor connection
- * state changes and trigger rewatch callbacks.
+ * Implementation of [StreamWatcher] that observes connection state changes via [StateFlow] and
+ * triggers rewatch callbacks.
  *
- * This implementation subscribes to connection state changes via [StreamClientListener] and uses
- * the provided [CoroutineScope] for invoking rewatch callbacks asynchronously.
+ * This implementation collects from the connection state flow and uses the provided
+ * [CoroutineScope] for invoking rewatch callbacks asynchronously.
  *
- * Exceptions thrown by the rewatch callback are caught, logged, and surfaced via the error callback
- * to prevent crashes while still allowing error handling by the product SDK.
+ * Exceptions thrown by rewatch callbacks are caught and logged to prevent crashes while still
+ * allowing error handling by the product SDK.
  *
  * @param scope Coroutine scope for async operations (should use SupervisorJob to prevent callback
  *   failures from cancelling the scope)
+ * @param connectionState StateFlow providing connection state updates
  * @param watched Concurrent map storing watched entries (defaults to empty [ConcurrentHashMap])
  * @param rewatchSubscriptions Manager for rewatch listener subscriptions
- * @param clientSubscriptions Manager for subscribing to connection state change notifications
  * @param logger Logger for diagnostic output and error reporting
  */
 @StreamInternalApi
 internal class StreamWatcherImpl<T>(
     private val scope: CoroutineScope,
+    private val connectionState: StateFlow<StreamConnectionState>,
     private val watched: ConcurrentMap<T, Unit> = ConcurrentHashMap(),
     private val rewatchSubscriptions: StreamSubscriptionManager<StreamRewatchListener<T>>,
-    private val clientSubscriptions: StreamSubscriptionManager<StreamClientListener>,
     private val logger: StreamLogger,
 ) : StreamWatcher<T> {
 
-    private var subscription: StreamSubscription? = null
+    private var collectionJob: Job? = null
 
-    private val listener =
-        object : StreamClientListener {
-            override fun onState(state: StreamConnectionState) {
-                // Invoke rewatch callback on every connection state change
-                if (state is StreamConnectionState.Connected && watched.isNotEmpty()) {
-                    scope.launch {
-                        val cids = watched.keys.toSet()
-                        val connectionId = state.connectionId
-                        logger.v {
-                            "[onState] Triggering rewatch for ${cids.size} items on connection $connectionId: ${cids.joinToString()}"
-                        }
+    override fun start(): Result<Unit> {
+        if (collectionJob != null) {
+            return Result.success(Unit) // Already started
+        }
 
-                        if (cids.isNotEmpty()) {
-                            rewatchSubscriptions
-                                .forEach { it.onRewatch(cids, connectionId) }
-                                .onFailure { error ->
-                                    logger.e(error) {
-                                        "[onState] Rewatch callback failed for ${cids.size} items. Error: ${error.message}"
-                                    }
-                                    clientSubscriptions.forEach { it.onError(error) }
-                                }
-                        }
-                    }
-                } else {
-                    logger.v { "[onState] State: $state, items count: ${watched.size}" }
+        return runCatching {
+            collectionJob =
+                scope.launch {
+                    connectionState.collect { state -> handleConnectionStateChange(state) }
                 }
-            }
         }
+    }
 
-    override fun start(): Result<Unit> {
-        if (subscription != null) {
-            return Result.success(Unit) // Already started
-        }
+    private fun handleConnectionStateChange(state: StreamConnectionState) {
+        // Invoke rewatch callback when connected and have watched items
+        if (state is StreamConnectionState.Connected && watched.isNotEmpty()) {
+            scope.launch {
+                val items = watched.keys.toSet()
+                val connectionId = state.connectionId
+                logger.v {
+                    "[handleConnectionStateChange] Triggering rewatch for ${items.size} items on connection $connectionId: ${items.joinToString()}"
+                }
 
-        return clientSubscriptions
-            .subscribe(
-                listener,
-                StreamSubscriptionManager.Options(
-                    retention = StreamSubscriptionManager.Options.Retention.KEEP_UNTIL_CANCELLED
-                ),
-            )
-            .map { subscription = it }
+                if (items.isNotEmpty()) {
+                    rewatchSubscriptions
+                        .forEach { it.onRewatch(items, connectionId) }
+                        .onFailure { error ->
+                            logger.e(error) {
+                                "[handleConnectionStateChange] Rewatch callback failed for ${items.size} items. Error: ${error.message}"
+                            }
+                        }
+                }
+            }
+        } else {
+            logger.v { "[handleConnectionStateChange] State: $state, items count: ${watched.size}" }
+        }
     }
 
     override fun stop(): Result<Unit> = runCatching {
-        subscription?.cancel()
-        subscription = null
+        collectionJob?.cancel()
+        collectionJob = null
         // Don't cancel scope - allows restart like other StreamStartableComponent implementations
     }