Revert to getOrCreateFeed instead of queryFeeds

Author: gpunto

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

@@ -15,21 +15,22 @@
  */
 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.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.FeedsFilterField
-import io.getstream.feeds.android.client.api.state.query.FeedsQuery
+import io.getstream.feeds.android.client.api.state.query.FeedQuery
 import io.getstream.feeds.android.client.internal.repository.FeedsRepository
-import java.util.concurrent.ConcurrentHashMap
+import io.getstream.feeds.android.client.internal.repository.GetOrCreateInfo
 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
+import java.util.concurrent.ConcurrentHashMap
 
 /**
  * Handles re-watching feeds upon reconnection.
@@ -68,25 +69,20 @@ internal class FeedWatchHandler(
     }
 
     private suspend fun rewatchAll() {
-        val watchedCopy = watched.keys.toSet()
-
-        if (watchedCopy.isNotEmpty()) {
-            val toRewatch = watchedCopy.map(FeedId::rawValue)
-
-            retryProcessor
-                .retry(retryPolicy) {
-                    feedsRepository
-                        .queryFeeds(FeedsQuery(FeedsFilterField.feed.`in`(toRewatch)))
-                        .getOrThrow()
-                }
-                .onFailure { errorBus.emit(StreamFeedRewatchException(watchedCopy, it)) }
-        }
+        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 ids: Set<FeedId>, cause: Throwable? = null) :
-    StreamClientException("Failed to rewatch feeds", cause)
+internal class StreamFeedRewatchException(val id: FeedId, cause: Throwable? = null) :
+    StreamClientException("Failed to rewatch feed", cause)

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

@@ -16,15 +16,13 @@
 package io.getstream.feeds.android.client.internal.client.reconnect
 
 import app.cash.turbine.test
-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.model.exceptions.StreamClientException
 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
+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
@@ -58,11 +56,10 @@ internal class FeedWatchHandlerTest {
 
     @Test
     fun `on connected event, query feeds that are still being watched`() = runTest {
-        coEvery { feedsRepository.queryFeeds(any()) } returns Result.success(mockk())
+        coEvery { feedsRepository.getOrCreateFeed(any()) } returns Result.success(mockk())
         val id1 = FeedId("group", "id1")
         val id2 = FeedId("group", "id2")
         val id3 = FeedId("group", "id3")
-        val expectedQuery = FeedsQuery(FeedsFilterField.feed.`in`("group:id1", "group:id3"))
 
         handler.onStartWatching(id1)
         handler.onStartWatching(id2)
@@ -71,7 +68,10 @@ internal class FeedWatchHandlerTest {
 
         connectionEvents.emit(StreamConnectionState.Connected(mockk(), "connection-id"))
 
-        coVerify { feedsRepository.queryFeeds(expectedQuery) }
+        coVerify {
+            feedsRepository.getOrCreateFeed(FeedQuery(id1))
+            feedsRepository.getOrCreateFeed(FeedQuery(id3))
+        }
     }
 
     @Test
@@ -85,7 +85,11 @@ internal class FeedWatchHandlerTest {
         errorBus.test {
             connectionEvents.emit(StreamConnectionState.Connected(mockk(), "connection-id"))
 
-            assertEquals(setOf(id1, id2), (awaitItem() as? StreamFeedRewatchException)?.ids)
+            val erroredIds = setOf(
+                (awaitItem() as? StreamFeedRewatchException)?.id,
+                (awaitItem() as? StreamFeedRewatchException)?.id,
+            )
+            assertEquals(setOf(id1, id2), erroredIds)
         }
     }
 
@@ -95,11 +99,11 @@ internal class FeedWatchHandlerTest {
         handler.onStartWatching(id1)
 
         listOf(
-                StreamConnectionState.Idle,
-                StreamConnectionState.Connecting.Opening("user-id"),
-                StreamConnectionState.Connecting.Authenticating("user-id"),
-                StreamConnectionState.Disconnected(),
-            )
+            StreamConnectionState.Idle,
+            StreamConnectionState.Connecting.Opening("user-id"),
+            StreamConnectionState.Connecting.Authenticating("user-id"),
+            StreamConnectionState.Disconnected(),
+        )
             .forEach { connectionEvents.emit(it) }
 
         verify { feedsRepository wasNot called }