Rewatch feeds on socket reconnection

gpunto · gpunto · commit f85d0ef8914b · 2025-09-01T14:40:26.000+02:00
diff --git a/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/Create.kt b/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/Create.kt
@@ -48,6 +48,7 @@ import io.getstream.feeds.android.client.api.model.FeedsConfig
 import io.getstream.feeds.android.client.api.model.User
 import io.getstream.feeds.android.client.internal.client.reconnect.ConnectionRecoveryHandler
 import io.getstream.feeds.android.client.internal.client.reconnect.DefaultRetryStrategy
+import io.getstream.feeds.android.client.internal.client.reconnect.FeedWatchHandler
 import io.getstream.feeds.android.client.internal.client.reconnect.lifecycle.StreamLifecycleObserver
 import io.getstream.feeds.android.client.internal.file.StreamFeedUploader
 import io.getstream.feeds.android.client.internal.http.FeedsSingleFlightApi
@@ -248,6 +249,13 @@ internal fun createFeedsClient(
 
     val moderation = ModerationImpl(moderationRepository)
 
+    val feedWatchHandler =
+        FeedWatchHandler(
+            connectedEvents = client.connectionState,
+            feedsRepository = feedsRepository,
+            scope = clientScope,
+        )
+
     // Build client
     return FeedsClientImpl(
         apiKey = apiKey,
@@ -271,6 +279,7 @@ internal fun createFeedsClient(
                 maxStrongSubscriptions = Integer.MAX_VALUE,
                 maxWeakSubscriptions = Integer.MAX_VALUE,
             ),
+        feedWatchHandler = feedWatchHandler,
         logger = logger,
     )
 }
diff --git a/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/FeedsClientImpl.kt b/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/FeedsClientImpl.kt
@@ -63,6 +63,7 @@ import io.getstream.feeds.android.client.api.state.query.ModerationConfigsQuery
 import io.getstream.feeds.android.client.api.state.query.PollVotesQuery
 import io.getstream.feeds.android.client.api.state.query.PollsQuery
 import io.getstream.feeds.android.client.internal.client.reconnect.ConnectionRecoveryHandler
+import io.getstream.feeds.android.client.internal.client.reconnect.FeedWatchHandler
 import io.getstream.feeds.android.client.internal.repository.ActivitiesRepository
 import io.getstream.feeds.android.client.internal.repository.AppRepository
 import io.getstream.feeds.android.client.internal.repository.BookmarksRepository
@@ -118,6 +119,7 @@ internal class FeedsClientImpl(
     private val pollsRepository: PollsRepository,
     override val uploader: FeedUploader,
     override val moderation: Moderation,
+    private val feedWatchHandler: FeedWatchHandler,
     private val logger: StreamLogger,
 ) : FeedsClient {
 
@@ -153,6 +155,7 @@ internal class FeedsClientImpl(
             return Result.failure(IllegalArgumentException("Anonymous users cannot connect."))
         }
         coreClient.subscribe(clientSubscription)
+        connectionRecoveryHandler.start()
         return coreClient.connect()
     }
 
@@ -175,6 +178,7 @@ internal class FeedsClientImpl(
             feedsRepository = feedsRepository,
             pollsRepository = pollsRepository,
             subscriptionManager = feedsEventsSubscriptionManager,
+            feedWatchHandler = feedWatchHandler,
         )
 
     override fun feedList(query: FeedsQuery): FeedList =
diff --git a/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandler.kt b/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandler.kt
@@ -0,0 +1,71 @@
+/*
+ * 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-feeds-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.feeds.android.client.internal.client.reconnect
+
+import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.feeds.android.client.api.model.FeedId
+import io.getstream.feeds.android.client.api.state.query.FeedQuery
+import io.getstream.feeds.android.client.internal.repository.FeedsRepository
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.filterIsInstance
+import kotlinx.coroutines.joinAll
+import kotlinx.coroutines.launch
+
+/**
+ * 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 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 feedsRepository: FeedsRepository,
+    scope: CoroutineScope,
+) {
+    private val watched = mutableSetOf<FeedId>()
+
+    init {
+        scope.launch {
+            connectedEvents.filterIsInstance<StreamConnectionState.Connected>().collect {
+                rewatchAll()
+            }
+        }
+    }
+
+    @Synchronized
+    fun onStartWatching(feedId: FeedId) {
+        watched += feedId
+    }
+
+    @Synchronized
+    fun onStopWatching(feedId: FeedId) {
+        watched -= feedId
+    }
+
+    private suspend fun rewatchAll() {
+        coroutineScope {
+            synchronized(this) { watched.toSet() }
+                .map { launch { feedsRepository.getOrCreateFeed(FeedQuery(it, watch = true)) } }
+                .joinAll()
+        }
+    }
+}
diff --git a/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/state/FeedImpl.kt b/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/state/FeedImpl.kt
@@ -32,6 +32,7 @@ import io.getstream.feeds.android.client.api.state.Feed
 import io.getstream.feeds.android.client.api.state.FeedState
 import io.getstream.feeds.android.client.api.state.query.FeedQuery
 import io.getstream.feeds.android.client.api.state.query.MembersQuery
+import io.getstream.feeds.android.client.internal.client.reconnect.FeedWatchHandler
 import io.getstream.feeds.android.client.internal.repository.ActivitiesRepository
 import io.getstream.feeds.android.client.internal.repository.BookmarksRepository
 import io.getstream.feeds.android.client.internal.repository.CommentsRepository
@@ -92,6 +93,7 @@ internal class FeedImpl(
     private val feedsRepository: FeedsRepository,
     private val pollsRepository: PollsRepository,
     private val subscriptionManager: StreamSubscriptionManager<FeedsEventListener>,
+    private val feedWatchHandler: FeedWatchHandler,
 ) : Feed {
 
     private val memberList: MemberListImpl =
@@ -127,13 +129,17 @@ internal class FeedImpl(
         get() = _state
 
     override suspend fun getOrCreate(): Result<FeedData> {
+        if (query.watch) {
+            feedWatchHandler.onStartWatching(query.fid)
+        }
         return feedsRepository
             .getOrCreateFeed(query)
             .onSuccess { _state.onQueryFeed(it) }
             .map { it.feed }
     }
 
     override suspend fun stopWatching(): Result<Unit> {
+        feedWatchHandler.onStopWatching(query.fid)
         return feedsRepository.stopWatching(groupId = group, feedId = id)
     }
 
diff --git a/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/client/FeedsClientImplTest.kt b/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/client/FeedsClientImplTest.kt
@@ -27,6 +27,7 @@ import io.getstream.feeds.android.client.api.model.PushNotificationsProvider
 import io.getstream.feeds.android.client.api.model.User
 import io.getstream.feeds.android.client.api.model.UserAuthType
 import io.getstream.feeds.android.client.internal.client.reconnect.ConnectionRecoveryHandler
+import io.getstream.feeds.android.client.internal.client.reconnect.FeedWatchHandler
 import io.getstream.feeds.android.client.internal.repository.ActivitiesRepository
 import io.getstream.feeds.android.client.internal.repository.AppRepository
 import io.getstream.feeds.android.client.internal.repository.BookmarksRepository
@@ -72,6 +73,7 @@ internal class FeedsClientImplTest {
     private val pollsRepository: PollsRepository = mockk(relaxed = true)
     private val uploader: FeedUploader = mockk(relaxed = true)
     private val moderation: Moderation = mockk(relaxed = true)
+    private val feedWatchHandler: FeedWatchHandler = mockk(relaxed = true)
     private val logger: StreamLogger = mockk(relaxed = true)
 
     private val feedsClient: FeedsClientImpl =
@@ -92,6 +94,7 @@ internal class FeedsClientImplTest {
             pollsRepository = pollsRepository,
             uploader = uploader,
             moderation = moderation,
+            feedWatchHandler = feedWatchHandler,
             logger = logger,
         )
 
@@ -136,6 +139,7 @@ internal class FeedsClientImplTest {
                 pollsRepository = pollsRepository,
                 uploader = uploader,
                 moderation = moderation,
+                feedWatchHandler = feedWatchHandler,
                 logger = logger,
             )
 
diff --git a/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandlerTest.kt b/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandlerTest.kt
@@ -0,0 +1,77 @@
+/*
+ * 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-feeds-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.feeds.android.client.internal.client.reconnect
+
+import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.feeds.android.client.api.model.FeedId
+import io.getstream.feeds.android.client.api.state.query.FeedQuery
+import io.getstream.feeds.android.client.internal.repository.FeedsRepository
+import io.mockk.called
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+
+@OptIn(ExperimentalCoroutinesApi::class)
+internal class FeedWatchHandlerTest {
+    private val connectionEvents = MutableSharedFlow<StreamConnectionState>()
+    private val feedsRepository: FeedsRepository = mockk()
+    private val scope = TestScope(UnconfinedTestDispatcher())
+
+    private val handler = FeedWatchHandler(connectionEvents, feedsRepository, scope)
+
+    @Test
+    fun `on connected event, get feeds that are still being watched`() = runTest {
+        coEvery { feedsRepository.getOrCreateFeed(any()) } returns Result.failure(Exception())
+        val id1 = FeedId("group", "id1")
+        val id2 = FeedId("group", "id2")
+        val id3 = FeedId("group", "id3")
+
+        handler.onStartWatching(id1)
+        handler.onStartWatching(id2)
+        handler.onStartWatching(id3)
+        handler.onStopWatching(id2)
+
+        connectionEvents.emit(StreamConnectionState.Connected(mockk(), "connection-id"))
+
+        coVerify {
+            feedsRepository.getOrCreateFeed(FeedQuery(id1))
+            feedsRepository.getOrCreateFeed(FeedQuery(id3))
+        }
+    }
+
+    @Test
+    fun `on non-connected event, do nothing`() = runTest {
+        val id1 = FeedId("group", "id1")
+        handler.onStartWatching(id1)
+
+        listOf(
+                StreamConnectionState.Idle,
+                StreamConnectionState.Connecting.Opening("user-id"),
+                StreamConnectionState.Connecting.Authenticating("user-id"),
+                StreamConnectionState.Disconnected(),
+            )
+            .forEach { connectionEvents.emit(it) }
+
+        verify { feedsRepository wasNot called }
+    }
+}
diff --git a/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/state/FeedImplTest.kt b/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/state/FeedImplTest.kt
