Rewatch feeds on socket reconnection (#61)

gpunto · aleksandar-apostolov · web-flow · commit 22b99623329e · 2025-09-10T15:18:36.000+02:00
* Rewatch feeds on socket reconnection

* Use ConcurrentHashMap and rename connection flow property in FeedWatchHandler

* Use queryFeeds instead of getOrCreateFeed

* Retry rewatching feeds on failures

* Post rewatch errors in a bus and log them

* Revert to getOrCreateFeed instead of queryFeeds

---------

Co-authored-by: Aleksandar Apostolov &lt;apostolov.alexandar@gmail.com&gt;
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
@@ -26,6 +26,7 @@ import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.log.StreamLoggerProvider
 import io.getstream.android.core.api.model.config.StreamClientSerializationConfig
 import io.getstream.android.core.api.model.config.StreamHttpConfig
+import io.getstream.android.core.api.model.exceptions.StreamClientException
 import io.getstream.android.core.api.model.value.StreamApiKey
 import io.getstream.android.core.api.model.value.StreamHttpClientInfoHeader
 import io.getstream.android.core.api.model.value.StreamUserId
@@ -48,6 +49,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
@@ -68,6 +70,7 @@ import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.plus
 import okhttp3.OkHttpClient
 import okhttp3.logging.HttpLoggingInterceptor
@@ -247,6 +250,16 @@ internal fun createFeedsClient(
     val pollsRepository = PollsRepositoryImpl(feedsApi)
 
     val moderation = ModerationImpl(moderationRepository)
+    val errorBus = MutableSharedFlow<StreamClientException>(extraBufferCapacity = 100)
+
+    val feedWatchHandler =
+        FeedWatchHandler(
+            connectionState = client.connectionState,
+            feedsRepository = feedsRepository,
+            retryProcessor = StreamRetryProcessor(logProvider.taggedLogger("WatchHandler")),
+            errorBus = errorBus,
+            scope = clientScope,
+        )
 
     // Build client
     return FeedsClientImpl(
@@ -271,6 +284,9 @@ internal fun createFeedsClient(
                 maxStrongSubscriptions = Integer.MAX_VALUE,
                 maxWeakSubscriptions = Integer.MAX_VALUE,
             ),
+        feedWatchHandler = feedWatchHandler,
+        errorBus = errorBus,
+        scope = clientScope,
         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
@@ -19,6 +19,7 @@ import io.getstream.android.core.api.StreamClient
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.connection.StreamConnectedUser
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.model.exceptions.StreamClientException
 import io.getstream.android.core.api.model.value.StreamApiKey
 import io.getstream.android.core.api.socket.listeners.StreamClientListener
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
@@ -63,6 +64,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
@@ -95,11 +97,13 @@ import io.getstream.feeds.android.network.models.DeleteActivitiesRequest
 import io.getstream.feeds.android.network.models.DeleteActivitiesResponse
 import io.getstream.feeds.android.network.models.ListDevicesResponse
 import io.getstream.feeds.android.network.models.WSEvent
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
+import kotlinx.coroutines.launch
 
 internal class FeedsClientImpl(
     private val coreClient: StreamClient,
@@ -118,7 +122,10 @@ internal class FeedsClientImpl(
     private val pollsRepository: PollsRepository,
     override val uploader: FeedUploader,
     override val moderation: Moderation,
+    private val feedWatchHandler: FeedWatchHandler,
     private val logger: StreamLogger,
+    scope: CoroutineScope,
+    errorBus: Flow<StreamClientException>,
 ) : FeedsClient {
 
     override val state: StateFlow<StreamConnectionState>
@@ -147,12 +154,19 @@ internal class FeedsClientImpl(
             }
         }
 
+    init {
+        scope.launch {
+            errorBus.collect { logger.e(it) { "[FeedsClient] Received error from bus" } }
+        }
+    }
+
     override suspend fun connect(): Result<StreamConnectedUser> {
         if (user.type == UserAuthType.ANONYMOUS) {
             logger.e { "[connect] Attempting to connect an anonymous user, returning an error." }
             return Result.failure(IllegalArgumentException("Anonymous users cannot connect."))
         }
         coreClient.subscribe(clientSubscription)
+        connectionRecoveryHandler.start()
         return coreClient.connect()
     }
 
@@ -175,6 +189,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,88 @@
+/*
+ * 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.StreamRetryPolicy
+import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.model.exceptions.StreamClientException
+import io.getstream.android.core.api.processing.StreamRetryProcessor
+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.getstream.feeds.android.client.internal.repository.GetOrCreateInfo
+import java.util.concurrent.ConcurrentHashMap
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.async
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.filterIsInstance
+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 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 connectionState: Flow<StreamConnectionState>,
+    private val feedsRepository: FeedsRepository,
+    private val retryProcessor: StreamRetryProcessor,
+    private val errorBus: MutableSharedFlow<StreamClientException>,
+    private val scope: CoroutineScope,
+) {
+    private val watched = ConcurrentHashMap<FeedId, Unit>()
+
+    init {
+        scope.launch {
+            connectionState.filterIsInstance<StreamConnectionState.Connected>().collect {
+                rewatchAll()
+            }
+        }
+    }
+
+    fun onStartWatching(feedId: FeedId) {
+        watched[feedId] = Unit
+    }
+
+    fun onStopWatching(feedId: FeedId) {
+        watched -= feedId
+    }
+
+    private suspend fun rewatchAll() {
+        coroutineScope { watched.keys.map { id -> async { rewatch(id) } } }
+    }
+
+    private suspend fun rewatch(id: FeedId): Result<GetOrCreateInfo> =
+        retryProcessor
+            .retry(retryPolicy) {
+                feedsRepository.getOrCreateFeed(FeedQuery(id, watch = true)).getOrThrow()
+            }
+            .onFailure { errorBus.emit(StreamFeedRewatchException(id)) }
+
+    companion object {
+        private val retryPolicy = StreamRetryPolicy.exponential(maxRetries = 3)
+    }
+}
+
+internal class StreamFeedRewatchException(val id: FeedId, cause: Throwable? = null) :
+    StreamClientException("Failed to rewatch feed", cause)
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
@@ -19,6 +19,7 @@ import io.getstream.android.core.api.StreamClient
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.connection.StreamConnectedUser
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.model.exceptions.StreamClientException
 import io.getstream.android.core.api.model.value.StreamApiKey
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
 import io.getstream.feeds.android.client.api.Moderation
@@ -27,6 +28,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
@@ -47,13 +49,19 @@ import io.getstream.feeds.android.network.models.ListDevicesResponse
 import io.mockk.coEvery
 import io.mockk.every
 import io.mockk.mockk
+import io.mockk.verify
 import java.util.Date
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.runTest
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
 import org.junit.Test
 
+@OptIn(ExperimentalCoroutinesApi::class)
 internal class FeedsClientImplTest {
     private val coreClient: StreamClient = mockk(relaxed = true)
     private val feedsEventsSubscriptionManager: StreamSubscriptionManager<FeedsEventListener> =
@@ -72,7 +80,10 @@ 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 scope = TestScope(UnconfinedTestDispatcher())
+    private val errorBus = MutableSharedFlow<StreamClientException>()
 
     private val feedsClient: FeedsClientImpl =
         FeedsClientImpl(
@@ -92,9 +103,21 @@ internal class FeedsClientImplTest {
             pollsRepository = pollsRepository,
             uploader = uploader,
             moderation = moderation,
+            feedWatchHandler = feedWatchHandler,
             logger = logger,
+            scope = scope,
+            errorBus = errorBus,
         )
 
+    @Test
+    fun `on error bus event, then log it`() = runTest {
+        val exception = StreamClientException("error!")
+
+        errorBus.emit(exception)
+
+        verify { logger.e(exception, any()) }
+    }
+
     @Test
     fun `connect when user is regular, then subscribes to client and connects successfully`() =
         runTest {
@@ -136,7 +159,10 @@ internal class FeedsClientImplTest {
                 pollsRepository = pollsRepository,
                 uploader = uploader,
                 moderation = moderation,
+                feedWatchHandler = feedWatchHandler,
                 logger = logger,
+                scope = scope,
+                errorBus = errorBus,
             )
 
         val result = anonymousClient.connect()
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
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
