Add sync engine unit tests. Handle error in onEach to prevent flow …

…completion.
ReactiveCircus · Mar 17, 2024 · d3b3b9d · d3b3b9d
1 parent a5cf54e
commit d3b3b9d
Show file tree

Hide file tree

Showing 6 changed files with 441 additions and 18 deletions.
diff --git a/...kotlin/io/github/reactivecircus/kstreamlined/kmp/database/testing/inMemoryDatabase.ios.kt b/...kotlin/io/github/reactivecircus/kstreamlined/kmp/database/testing/inMemoryDatabase.ios.kt
@@ -7,10 +7,11 @@ import co.touchlab.sqliter.DatabaseConfiguration
 import io.github.reactivecircus.kstreamlined.kmp.database.KStreamlinedDatabase
 
 internal actual fun createInMemoryDriver(): SqlDriver {
+    index++
     val schema = KStreamlinedDatabase.Schema
     return NativeSqliteDriver(
         DatabaseConfiguration(
-            name = "kstreamlined-test.db",
+            name = "kstreamlined-test-$index.db",
             version = schema.version.toInt(),
             create = { connection ->
                 wrapConnection(connection) { schema.create(it) }
@@ -24,3 +25,5 @@ internal actual fun createInMemoryDriver(): SqlDriver {
         ),
     )
 }
+
+private var index = 0
diff --git a/...monMain/sqldelight/io/github/reactivecircus/kstreamlined/kmp/database/FeedOriginEntity.sq b/...monMain/sqldelight/io/github/reactivecircus/kstreamlined/kmp/database/FeedOriginEntity.sq
@@ -15,3 +15,6 @@ SELECT * FROM feedOriginEntity;
 upsertFeedOrigin:
 INSERT OR REPLACE INTO feedOriginEntity
 VALUES (:key, :title, :description, :selected);
+
+updateSelection:
+UPDATE feedOriginEntity SET selected = :selected WHERE key = :key;
diff --git a/...c/commonMain/kotlin/io/github/reactivecircus/kstreamlined/kmp/feed/sync/FeedSyncEngine.kt b/...c/commonMain/kotlin/io/github/reactivecircus/kstreamlined/kmp/feed/sync/FeedSyncEngine.kt
@@ -8,7 +8,8 @@ public interface FeedSyncEngine {
 }
 
 public sealed interface SyncState {
-    public data object Idle : SyncState
+    public data object Initializing : SyncState
     public data object Syncing : SyncState
+    public data object Idle : SyncState
     public data object Error : SyncState
 }
diff --git a/kmp/feed-sync/runtime/build.gradle.kts b/kmp/feed-sync/runtime/build.gradle.kts
@@ -18,6 +18,7 @@ kotlin {
                 implementation(project(":kmp:networking:testing"))
                 implementation(project(":kmp:database-testing"))
                 implementation(libs.kotlinx.coroutines.test)
+                implementation(libs.turbine)
             }
         }
     }

diff --git a/...mmonMain/kotlin/io/github/reactivecircus/kstreamlined/kmp/feed/sync/FeedSyncEngineImpl.kt b/...mmonMain/kotlin/io/github/reactivecircus/kstreamlined/kmp/feed/sync/FeedSyncEngineImpl.kt
@@ -18,13 +18,13 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.merge
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.receiveAsFlow
 import kotlinx.datetime.Clock
+import kotlin.coroutines.cancellation.CancellationException
 
 public class FeedSyncEngineImpl(
     private val feedService: FeedService,
@@ -33,7 +33,7 @@ public class FeedSyncEngineImpl(
     dbDispatcher: CoroutineDispatcher = Dispatchers.IO,
     private val clock: Clock = Clock.System,
 ) : FeedSyncEngine {
-    private val _syncState = MutableStateFlow<SyncState>(SyncState.Syncing)
+    private val _syncState = MutableStateFlow<SyncState>(SyncState.Initializing)
     override val syncState: StateFlow<SyncState> = _syncState
 
     private val manualSyncTrigger = Channel<SyncRequest>()
@@ -52,13 +52,18 @@ public class FeedSyncEngineImpl(
     init {
         merge(manualSyncTrigger.receiveAsFlow(), automaticSyncTrigger)
             .onEach { syncRequest ->
-                _syncState.value = SyncState.Syncing
-                performSync(syncRequest)
-                _syncState.value = SyncState.Idle
-            }
-            .catch {
-                Logger.w("Sync failed", it)
-                _syncState.value = SyncState.Error
+                runCatching {
+                    val decision = syncRequestEvaluator.evaluate(syncRequest)
+                    if (decision.shouldSyncFeedSources || decision.shouldSyncFeedItems) {
+                        _syncState.value = SyncState.Syncing
+                        performSync(decision)
+                    }
+                    _syncState.value = SyncState.Idle
+                }.onFailure {
+                    if (it is CancellationException) throw it
+                    Logger.w("Sync failed", it)
+                    _syncState.value = SyncState.Error
+                }
             }
             .launchIn(syncEngineScope)
     }
@@ -67,25 +72,24 @@ public class FeedSyncEngineImpl(
         manualSyncTrigger.send(SyncRequest(forceRefresh))
     }
 
-    private suspend fun performSync(syncRequest: SyncRequest) = coroutineScope {
-        val (shouldSyncSources, shouldSyncItems) = syncRequestEvaluator.evaluate(syncRequest)
+    private suspend fun performSync(syncDecision: SyncDecision) = coroutineScope {
+        val (shouldSyncSources, shouldSyncItems) = syncDecision
 
-        val feedSourcesDeferred = if (shouldSyncSources) {
+        val feedSourcesDeferred = takeIf { shouldSyncSources }?.let {
             async { feedService.fetchFeedOrigins() }
-        } else { null }
+        }
 
-        val feedEntriesDeferred = if (shouldSyncItems) {
+        val feedEntriesDeferred = takeIf { shouldSyncItems }?.let {
             async {
                 feedService.fetchFeedEntries(
                     filters = db.feedOriginEntityQueries.allFeedOrigins().executeAsList().asNetworkModels()
                 )
             }
-        } else { null }
+        }
 
         val feedSources = feedSourcesDeferred?.await()
         val feedEntries = feedEntriesDeferred?.await()
 
-        if (feedSources == null && feedEntries == null) return@coroutineScope
         db.transaction {
             val currentFeedOrigins = db.feedOriginEntityQueries.allFeedOrigins().executeAsList()