Spotless

Author: aleksandar-apostolov

CLAUDE.md

@@ -162,9 +162,65 @@ StreamClient (Main Interface)
 │   ├── StreamNetworkMonitor (ConnectivityManager callbacks)
 │   └── StreamLifecycleMonitor (ProcessLifecycleOwner observer)
 ├── StreamConnectionRecoveryEvaluator (Reconnect heuristics)
+├── StreamCidWatcher (Watch registry & rewatch coordinator)
+│   ├── StreamSubscriptionManager<StreamCidRewatchListener> (Rewatch listener registry)
+│   └── StreamSubscriptionManager<StreamClientListener> (Connection state monitoring)
 └── StreamSubscriptionManager<StreamClientListener> (Event distribution)
 ```
 
+### Watch Management: StreamCidWatcher
+
+**Purpose:** Manages a registry of watched resources (channels/conversations) and automatically triggers re-watching when the WebSocket connection state changes.
+
+**Location:**
+- Interface: `stream-android-core/src/main/java/io/getstream/android/core/api/watcher/StreamCidWatcher.kt`
+- Implementation: `stream-android-core/src/main/java/io/getstream/android/core/internal/watcher/StreamCidWatcherImpl.kt`
+- Listener: `stream-android-core/src/main/java/io/getstream/android/core/api/watcher/StreamCidRewatchListener.kt`
+
+**Key Concept:** When the WebSocket reconnects (network recovery, app resume), all active watches must be re-established on the server. `StreamCidWatcher` maintains which `StreamCid`s (Channel IDs) are currently watched and notifies listeners on every `Connected` state transition.
+
+**Core Operations:**
+```kotlin
+// Add a CID to watch registry
+fun watch(cid: StreamCid): Result<StreamCid>
+
+// Remove a CID from watch registry
+fun stopWatching(cid: StreamCid): Result<StreamCid>
+
+// Clear all watched CIDs
+fun clear(): Result<Unit>
+
+// Subscribe to rewatch notifications
+fun subscribe(
+    listener: StreamCidRewatchListener,
+    options: StreamSubscriptionManager.Options
+): Result<StreamSubscription>
+
+// Lifecycle management (StreamStartableComponent)
+fun start(): Result<Unit>
+fun stop(): Result<Unit>
+```
+
+**Usage Flow:**
+1. Product SDK watches a channel: `watcher.watch(StreamCid.parse("messaging:general"))`
+2. Watcher adds CID to internal `ConcurrentHashMap<StreamCid, Unit>` registry
+3. Product SDK registers a rewatch listener: `watcher.subscribe(StreamCidRewatchListener { cids -> ... })`
+4. Call `watcher.start()` to begin monitoring connection state changes
+5. On `StreamConnectionState.Connected` event, watcher invokes all listeners with complete CID list
+6. Product SDK re-establishes server-side watches for each CID
+
+**Implementation Details:**
+- Thread-safe: Uses `ConcurrentHashMap` for CID registry (line 36 in `StreamCidWatcherImpl.kt`)
+- Async execution: Rewatch callbacks invoked on internal coroutine scope with `SupervisorJob + Dispatchers.Default` (line 45)
+- Error handling: Exceptions from rewatch callbacks are caught, logged, and surfaced via `StreamClientListener.onError` (lines 61-67)
+- Idempotent: Multiple `watch()` calls with same CID only maintain one entry
+- Only triggers on `Connected` state when registry is non-empty (line 51)
+
+**Test Coverage:**
+- Location: `stream-android-core/src/test/java/io/getstream/android/core/internal/watcher/StreamCidWatcherImplTest.kt`
+- 24 comprehensive test cases covering watch operations, lifecycle, state changes, error handling, and concurrency
+- 100% instruction/branch/line coverage (verified via Kover)
+
 ## Configuration Defaults
 
 ### StreamBatcher

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

@@ -290,13 +290,15 @@ public fun StreamClient(
             logger = logProvider.taggedLogger("SCConnectionRecoveryEvaluator"),
             singleFlightProcessor = singleFlight,
         ),
-    cidWatcher: StreamCidWatcher = StreamCidWatcher(
-        logProvider.taggedLogger("SCCidRewatcher"),
-        streamRewatchSubscriptionManager = StreamSubscriptionManager(
-            logger = logProvider.taggedLogger("SCRewatchSubscriptionManager")
+    cidWatcher: StreamCidWatcher =
+        StreamCidWatcher(
+            logProvider.taggedLogger("SCCidRewatcher"),
+            streamRewatchSubscriptionManager =
+                StreamSubscriptionManager(
+                    logger = logProvider.taggedLogger("SCRewatchSubscriptionManager")
+                ),
+            streamClientSubscriptionManager = clientSubscriptionManager,
         ),
-        streamClientSubscriptionManager = clientSubscriptionManager,
-    )
 ): StreamClient {
     val clientLogger = logProvider.taggedLogger(tag = "SCClient")
     val parent = scope.coroutineContext[Job]

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

@@ -1,29 +1,42 @@
+/*
+ * Copyright (c) 2014-2025 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.model
 
 import io.getstream.android.core.annotations.StreamPublishedApi
 
 @ConsistentCopyVisibility
 @StreamPublishedApi
-public data class StreamCid private constructor(
-    public val type: String,
-    public val id: String,
-) {
+public data class StreamCid private constructor(public val type: String, public val id: String) {
 
     public companion object {
 
-        public fun parse(cid: String) : StreamCid {
+        public fun parse(cid: String): StreamCid {
             require(cid.isNotEmpty())
             val split = cid.split(":")
             require(split.size == 2)
             return StreamCid(split[0], split[1])
         }
 
-        public fun fromTypeAndId(type: String, id: String) : StreamCid {
+        public fun fromTypeAndId(type: String, id: String): StreamCid {
             require(type.isNotEmpty())
             require(id.isNotEmpty())
             return StreamCid(type, id)
         }
     }
 
     public fun formatted(): String = "$type:$id"
-}
+}

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

@@ -1,3 +1,19 @@
+/*
+ * Copyright (c) 2014-2025 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.watcher
 
 import io.getstream.android.core.annotations.StreamInternalApi
@@ -8,25 +24,23 @@ import io.getstream.android.core.api.model.StreamCid
  * state change.
  *
  * This functional interface is used with [StreamCidWatcher] to receive notifications when the
- * WebSocket connection state changes (especially during reconnections), allowing the product SDK
- * to re-establish server-side watches for all actively monitored resources.
+ * WebSocket connection state changes (especially during reconnections), allowing the product SDK to
+ * re-establish server-side watches for all actively monitored resources.
  *
  * ## When This Is Called
  *
- * The callback is triggered on every [StreamConnectionState.Connected] event when the watched
- * CID registry is non-empty. This ensures that all active subscriptions are restored after:
+ * The callback is triggered on every [StreamConnectionState.Connected] event when the watched CID
+ * registry is non-empty. This ensures that all active subscriptions are restored after:
  * - Initial connection establishment
  * - Network reconnection after temporary disconnection
  * - WebSocket recovery after connection loss
  *
  * ## Threading
- *
  * - The callback is invoked asynchronously on an internal coroutine scope
  * - Implementations should be thread-safe as concurrent invocations are possible
  * - Long-running operations should be dispatched to an appropriate dispatcher
  *
  * ## Error Handling
- *
  * - Exceptions thrown by this callback are caught and logged by the watcher
  * - Errors are also surfaced via [StreamClientListener.onError] for user-level handling
  * - A failing callback will not prevent other listeners from being notified
@@ -59,24 +73,23 @@ public fun interface StreamCidRewatchListener {
     /**
      * Called when watched resources need to be re-subscribed.
      *
-     * This method is invoked with the complete list of [StreamCid]s currently in the watch
-     * registry whenever the connection state changes to [StreamConnectionState.Connected] and
-     * the registry is non-empty.
+     * This method is invoked with the complete list of [StreamCid]s currently in the watch registry
+     * whenever the connection state changes to [StreamConnectionState.Connected] and the registry
+     * is non-empty.
      *
      * Implementations should iterate over the provided list and re-establish server-side watches
      * for each CID to maintain active subscriptions across reconnection cycles.
      *
      * ## Contract
-     *
      * - The list is never empty when this method is called
      * - The list represents a snapshot of the watch registry at the time of invocation
      * - Modifications to the list do not affect the internal registry
      * - Multiple calls may occur for the same set of CIDs during reconnection scenarios
      *
      * ## Error Handling
      *
-     * Exceptions thrown by this method are caught, logged, and surfaced via error callbacks.
-     * The failure of one listener does not prevent other listeners from being notified.
+     * Exceptions thrown by this method are caught, logged, and surfaced via error callbacks. The
+     * failure of one listener does not prevent other listeners from being notified.
      *
      * @param list A non-empty list of [StreamCid]s that require re-watching
      */

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

@@ -1,3 +1,19 @@
+/*
+ * Copyright (c) 2014-2025 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.watcher
 
 import io.getstream.android.core.annotations.StreamInternalApi
@@ -6,7 +22,6 @@ import io.getstream.android.core.api.model.StreamCid
 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.StreamSubscription
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
 import io.getstream.android.core.internal.watcher.StreamCidWatcherImpl
 
@@ -15,22 +30,22 @@ import io.getstream.android.core.internal.watcher.StreamCidWatcherImpl
  * triggers re-watching when the connection state changes.
  *
  * This component is critical for maintaining active subscriptions across reconnection cycles. When
- * the WebSocket disconnects and reconnects, all active watches must be re-established on the server.
- * [StreamCidWatcher] tracks which [StreamCid]s (Channel IDs) are currently being watched and
- * invokes a callback on every connection state change, allowing the product SDK to re-subscribe.
+ * the WebSocket disconnects and reconnects, all active watches must be re-established on the
+ * server. [StreamCidWatcher] tracks which [StreamCid]s (Channel IDs) are currently being watched
+ * and invokes a callback on every connection state change, allowing the product SDK to
+ * re-subscribe.
  *
  * ## Typical Usage Flow
- *
  * 1. Product SDK watches a channel: `watcher.watch(StreamCid.parse("messaging:general"))`
  * 2. Watcher adds the CID to its internal registry
- * 3. Product SDK registers a rewatch listener: `watcher.subscribe(StreamCidRewatchListener { cids -> resubscribe(cids) })`
+ * 3. Product SDK registers a rewatch listener: `watcher.subscribe(StreamCidRewatchListener { cids
+ *    -> resubscribe(cids) })`
  * 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 CIDs: `["messaging:general", "livestream:sports"]`
  * 7. Product SDK re-establishes server-side watches for each CID
  *
  * ## Threading
- *
  * - All methods are thread-safe and can be called from any thread
  * - The rewatch callback is invoked asynchronously on an internal coroutine scope
  * - Callback invocation happens asynchronously on connection state changes
@@ -47,7 +62,8 @@ import io.getstream.android.core.internal.watcher.StreamCidWatcherImpl
  * @see StreamStartableComponent Lifecycle contract for start/stop operations
  */
 @StreamInternalApi
-public interface StreamCidWatcher : StreamObservable<StreamCidRewatchListener>, StreamStartableComponent {
+public interface StreamCidWatcher :
+    StreamObservable<StreamCidRewatchListener>, StreamStartableComponent {
 
     /**
      * Registers a channel or resource as actively watched.
@@ -56,9 +72,11 @@ public interface StreamCidWatcher : StreamObservable<StreamCidRewatchListener>,
      * changes (especially during reconnection), the rewatch callback will include this CID in the
      * list of resources that need to be re-subscribed.
      *
-     * Multiple calls with the same [cid] are idempotent—only one entry is maintained per unique CID.
+     * Multiple calls with the same [cid] are idempotent—only one entry is maintained per unique
+     * CID.
      *
      * ## Example
+     *
      * ```kotlin
      * val cid = StreamCid.parse("messaging:general")
      * watcher.watch(cid)
@@ -70,7 +88,7 @@ public interface StreamCidWatcher : StreamObservable<StreamCidRewatchListener>,
      * @return [Result.success] with the [cid] if registration succeeded, or [Result.failure] if an
      *   error occurred (e.g., internal registry corruption)
      */
-    public fun watch(cid: StreamCid) : Result<StreamCid>
+    public fun watch(cid: StreamCid): Result<StreamCid>
 
     /**
      * Unregisters a channel or resource from the watch registry.
@@ -81,6 +99,7 @@ public interface StreamCidWatcher : StreamObservable<StreamCidRewatchListener>,
      * Calling this method with a CID that is not being watched is a no-op (not an error).
      *
      * ## Example
+     *
      * ```kotlin
      * val cid = StreamCid.parse("messaging:general")
      * watcher.stopWatching(cid)
@@ -92,17 +111,18 @@ public interface StreamCidWatcher : StreamObservable<StreamCidRewatchListener>,
      * @return [Result.success] with the [cid] if removal succeeded, or [Result.failure] if an error
      *   occurred (e.g., internal registry corruption)
      */
-    public fun stopWatching(cid: StreamCid) : Result<StreamCid>
+    public fun stopWatching(cid: StreamCid): Result<StreamCid>
 
     /**
      * Removes all entries from the watch registry.
      *
-     * After calling this method, the rewatch callback will NOT be invoked on subsequent
-     * connection state changes (since the registry is empty) until new resources are watched.
+     * After calling this method, the rewatch callback will NOT be invoked on subsequent connection
+     * state changes (since the registry is empty) until new resources are watched.
      *
      * This method is typically called during cleanup or logout to ensure no stale watches remain.
      *
      * ## Example
+     *
      * ```kotlin
      * // During logout
      * watcher.clear()
@@ -120,10 +140,10 @@ public interface StreamCidWatcher : StreamObservable<StreamCidRewatchListener>,
 public fun StreamCidWatcher(
     logger: StreamLogger,
     streamRewatchSubscriptionManager: StreamSubscriptionManager<StreamCidRewatchListener>,
-    streamClientSubscriptionManager: StreamSubscriptionManager<StreamClientListener>
-) : StreamCidWatcher = StreamCidWatcherImpl(
-    logger = logger,
-    rewatchSubscriptions = streamRewatchSubscriptionManager,
-    clientSubscriptions = streamClientSubscriptionManager
-)
-
+    streamClientSubscriptionManager: StreamSubscriptionManager<StreamClientListener>,
+): StreamCidWatcher =
+    StreamCidWatcherImpl(
+        logger = logger,
+        rewatchSubscriptions = streamRewatchSubscriptionManager,
+        clientSubscriptions = streamClientSubscriptionManager,
+    )