Retry rewatching feeds on failures

Author: gpunto

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

@@ -253,6 +253,7 @@ internal fun createFeedsClient(
         FeedWatchHandler(
             connectionState = client.connectionState,
             feedsRepository = feedsRepository,
+            retryProcessor = StreamRetryProcessor(logProvider.taggedLogger("WatchHandler")),
             scope = clientScope,
         )
 

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

@@ -16,7 +16,9 @@
 package io.getstream.feeds.android.client.internal.client.reconnect
 
 import io.getstream.android.core.api.filter.`in`
+import io.getstream.android.core.api.model.StreamRetryPolicy
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+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.FeedsFilterField
 import io.getstream.feeds.android.client.api.state.query.FeedsQuery
@@ -41,6 +43,7 @@ import kotlinx.coroutines.launch
 internal class FeedWatchHandler(
     private val connectionState: Flow<StreamConnectionState>,
     private val feedsRepository: FeedsRepository,
+    private val retryProcessor: StreamRetryProcessor,
     scope: CoroutineScope,
 ) {
     private val watched = ConcurrentHashMap<FeedId, Unit>()
@@ -64,7 +67,15 @@ internal class FeedWatchHandler(
     private suspend fun rewatchAll() {
         val toRewatch = watched.keys.mapTo(mutableListOf(), FeedId::rawValue)
         if (toRewatch.isNotEmpty()) {
-            feedsRepository.queryFeeds(FeedsQuery(FeedsFilterField.feed.`in`(toRewatch)))
+            retryProcessor.retry(retryPolicy) {
+                feedsRepository
+                    .queryFeeds(FeedsQuery(FeedsFilterField.feed.`in`(toRewatch)))
+                    .getOrThrow()
+            }
         }
     }
+
+    companion object {
+        private val retryPolicy = StreamRetryPolicy.exponential(maxRetries = 3)
+    }
 }

stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandlerTest.kt

@@ -16,7 +16,10 @@
 package io.getstream.feeds.android.client.internal.client.reconnect
 
 import io.getstream.android.core.api.filter.`in`
+import io.getstream.android.core.api.model.StreamRetryPolicy
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.processing.StreamRetryProcessor
+import io.getstream.android.core.result.runSafely
 import io.getstream.feeds.android.client.api.model.FeedId
 import io.getstream.feeds.android.client.api.state.query.FeedsFilterField
 import io.getstream.feeds.android.client.api.state.query.FeedsQuery
@@ -38,8 +41,9 @@ internal class FeedWatchHandlerTest {
     private val connectionEvents = MutableSharedFlow<StreamConnectionState>()
     private val feedsRepository: FeedsRepository = mockk()
     private val scope = TestScope(UnconfinedTestDispatcher())
+    private val retryProcessor: StreamRetryProcessor = TestRetryProcessor()
 
-    private val handler = FeedWatchHandler(connectionEvents, feedsRepository, scope)
+    private val handler = FeedWatchHandler(connectionEvents, feedsRepository, retryProcessor, scope)
 
     @Test
     fun `on connected event, get feeds that are still being watched`() = runTest {
@@ -74,4 +78,11 @@ internal class FeedWatchHandlerTest {
 
         verify { feedsRepository wasNot called }
     }
+
+    private class TestRetryProcessor : StreamRetryProcessor {
+        override suspend fun <T> retry(
+            policy: StreamRetryPolicy,
+            block: suspend () -> T,
+        ): Result<T> = runSafely { block() }
+    }
 }