Sync streams (#257)

Author: simolus3

core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt

@@ -7,10 +7,10 @@ import com.powersync.PowerSyncDatabase
 import com.powersync.PowerSyncException
 import com.powersync.TestConnector
 import com.powersync.bucket.BucketChecksum
-import com.powersync.bucket.BucketPriority
 import com.powersync.bucket.Checkpoint
 import com.powersync.bucket.OpType
 import com.powersync.bucket.OplogEntry
+import com.powersync.bucket.StreamPriority
 import com.powersync.bucket.WriteCheckpointData
 import com.powersync.bucket.WriteCheckpointResponse
 import com.powersync.connectors.PowerSyncBackendConnector
@@ -165,7 +165,7 @@ abstract class BaseSyncIntegrationTest(
                         add(
                             BucketChecksum(
                                 bucket = "bucket$prio",
-                                priority = BucketPriority(prio),
+                                priority = StreamPriority(prio),
                                 checksum = 10 + prio,
                             ),
                         )
@@ -218,7 +218,7 @@ abstract class BaseSyncIntegrationTest(
 
                 // Emit a partial sync complete for each priority but the last.
                 for (priorityNo in 0..<3) {
-                    val priority = BucketPriority(priorityNo)
+                    val priority = StreamPriority(priorityNo)
                     pushData(priorityNo)
                     syncLines.send(
                         SyncLine.CheckpointPartiallyComplete(
@@ -258,7 +258,7 @@ abstract class BaseSyncIntegrationTest(
                             listOf(
                                 BucketChecksum(
                                     bucket = "bkt",
-                                    priority = BucketPriority(1),
+                                    priority = StreamPriority(1),
                                     checksum = 0,
                                 ),
                             ),
@@ -268,17 +268,17 @@ abstract class BaseSyncIntegrationTest(
             syncLines.send(
                 SyncLine.CheckpointPartiallyComplete(
                     lastOpId = "0",
-                    priority = BucketPriority(1),
+                    priority = StreamPriority(1),
                 ),
             )
 
-            database.waitForFirstSync(BucketPriority(1))
+            database.waitForFirstSync(StreamPriority(1))
             database.close()
 
             // Connect to the same database again
             database = openDatabaseAndInitialize()
             database.currentStatus.hasSynced shouldBe false
-            database.currentStatus.statusForPriority(BucketPriority(1)).hasSynced shouldBe true
+            database.currentStatus.statusForPriority(StreamPriority(1)).hasSynced shouldBe true
         }
 
     @Test

core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncProgressTest.kt

@@ -3,10 +3,10 @@ package com.powersync.sync
 import app.cash.turbine.ReceiveTurbine
 import app.cash.turbine.turbineScope
 import com.powersync.bucket.BucketChecksum
-import com.powersync.bucket.BucketPriority
 import com.powersync.bucket.Checkpoint
 import com.powersync.bucket.OpType
 import com.powersync.bucket.OplogEntry
+import com.powersync.bucket.StreamPriority
 import com.powersync.testutils.ActiveDatabaseTest
 import com.powersync.testutils.databaseTest
 import com.powersync.testutils.waitFor
@@ -35,7 +35,7 @@ abstract class BaseSyncProgressTest(
     private fun bucket(
         name: String,
         count: Int,
-        priority: BucketPriority = BucketPriority(3),
+        priority: StreamPriority = StreamPriority(3),
     ): BucketChecksum =
         BucketChecksum(
             bucket = name,
@@ -68,7 +68,7 @@ abstract class BaseSyncProgressTest(
         )
     }
 
-    private suspend fun ActiveDatabaseTest.addCheckpointComplete(priority: BucketPriority? = null) {
+    private suspend fun ActiveDatabaseTest.addCheckpointComplete(priority: StreamPriority? = null) {
         if (priority != null) {
             syncLines.send(
                 SyncLine.CheckpointPartiallyComplete(
@@ -93,7 +93,7 @@ abstract class BaseSyncProgressTest(
 
     private suspend fun ReceiveTurbine<SyncStatusData>.expectProgress(
         total: Pair<Int, Int>,
-        priorities: Map<BucketPriority, Pair<Int, Int>> = emptyMap(),
+        priorities: Map<StreamPriority, Pair<Int, Int>> = emptyMap(),
     ) {
         val item = awaitItem()
         val progress = item.downloadProgress ?: error("Expected download progress on $item")
@@ -357,7 +357,7 @@ abstract class BaseSyncProgressTest(
                 ) {
                     turbine.expectProgress(
                         prio2,
-                        mapOf(BucketPriority(0) to prio0, BucketPriority(2) to prio2),
+                        mapOf(StreamPriority(0) to prio0, StreamPriority(2) to prio2),
                     )
                 }
 
@@ -367,8 +367,8 @@ abstract class BaseSyncProgressTest(
                             lastOpId = "10",
                             checksums =
                                 listOf(
-                                    bucket("a", 5, BucketPriority(0)),
-                                    bucket("b", 5, BucketPriority(2)),
+                                    bucket("a", 5, StreamPriority(0)),
+                                    bucket("b", 5, StreamPriority(2)),
                                 ),
                         ),
                     ),
@@ -378,7 +378,7 @@ abstract class BaseSyncProgressTest(
                 addDataLine("a", 5)
                 expectProgress(5 to 5, 5 to 10)
 
-                addCheckpointComplete(BucketPriority(0))
+                addCheckpointComplete(StreamPriority(0))
                 expectProgress(5 to 5, 5 to 10)
 
                 addDataLine("b", 2)
@@ -390,8 +390,8 @@ abstract class BaseSyncProgressTest(
                         lastOpId = "14",
                         updatedBuckets =
                             listOf(
-                                bucket("a", 8, BucketPriority(0)),
-                                bucket("b", 6, BucketPriority(2)),
+                                bucket("a", 8, StreamPriority(0)),
+                                bucket("b", 6, StreamPriority(2)),
                             ),
                         removedBuckets = emptyList(),
                     ),