Use ConcurrentHashMap and rename connection flow property in FeedWatchHandler

Author: gpunto

stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/Create.kt

@@ -251,7 +251,7 @@ internal fun createFeedsClient(
 
     val feedWatchHandler =
         FeedWatchHandler(
-            connectedEvents = client.connectionState,
+            connectionState = client.connectionState,
             feedsRepository = feedsRepository,
             scope = clientScope,
         )

stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandler.kt

@@ -25,47 +25,46 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.filterIsInstance
 import kotlinx.coroutines.joinAll
 import kotlinx.coroutines.launch
+import java.util.concurrent.ConcurrentHashMap
 
 /**
  * Handles re-watching feeds upon reconnection.
  *
  * Keeps track of feeds that are being watched and re-subscribes to them when the connection is
  * re-established.
  *
- * @property connectedEvents A [Flow] that emits events when the connection state changes to
+ * @property connectionState A [Flow] that emits events when the connection state changes to
  *   [StreamConnectionState.Connected].
  * @property feedsRepository The [FeedsRepository] used to rewatch feeds on connection.
  * @property scope The [CoroutineScope] in which to launch coroutines for re-watching feeds.
  */
 internal class FeedWatchHandler(
-    private val connectedEvents: Flow<StreamConnectionState>,
+    private val connectionState: Flow<StreamConnectionState>,
     private val feedsRepository: FeedsRepository,
     scope: CoroutineScope,
 ) {
-    private val watched = mutableSetOf<FeedId>()
+    private val watched = ConcurrentHashMap<FeedId, Unit>()
 
     init {
         scope.launch {
-            connectedEvents.filterIsInstance<StreamConnectionState.Connected>().collect {
+            connectionState.filterIsInstance<StreamConnectionState.Connected>().collect {
                 rewatchAll()
             }
         }
     }
 
-    @Synchronized
     fun onStartWatching(feedId: FeedId) {
-        watched += feedId
+        watched[feedId] = Unit
     }
 
-    @Synchronized
     fun onStopWatching(feedId: FeedId) {
         watched -= feedId
     }
 
     private suspend fun rewatchAll() {
         coroutineScope {
-            synchronized(this@FeedWatchHandler) { watched.toSet() }
-                .map { launch { feedsRepository.getOrCreateFeed(FeedQuery(it, watch = true)) } }
+            watched
+                .map { launch { feedsRepository.getOrCreateFeed(FeedQuery(it.key, watch = true)) } }
                 .joinAll()
         }
     }