Fix reconnect race in test

Author: simolus3

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Add `PowerSyncDatabase.inMemory` to create an in-memory SQLite database with PowerSync.
   This may be useful for testing.
 - The Supabase connector can now be subclassed to customize how rows are uploaded and how errors are handled.
+- Experimental support for sync streams.
 
 ## 1.6.1
 

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

@@ -163,9 +163,10 @@ class SyncStreamTest : AbstractSyncTest(true) {
                 requestedSyncStreams.clear()
 
                 val subscription = database.syncStream("a").subscribe()
+                waitForSyncLinesChannelClosed()
 
                 // Adding the subscription should reconnect
-                turbine.waitFor { it.connected && !it.downloading }
+                turbine.waitFor { it.connected }
                 requestedSyncStreams shouldHaveSingleElement {
                     val streams = it.jsonObject["streams"]!!.jsonObject
                     val subscriptions = streams["subscriptions"]!!.jsonArray

core/src/commonIntegrationTest/kotlin/com/powersync/testutils/TestUtils.kt

@@ -23,10 +23,12 @@ import io.ktor.client.HttpClient
 import io.ktor.client.engine.mock.toByteArray
 import io.ktor.http.ContentType
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.runTest
 import kotlinx.io.files.Path
 import kotlinx.serialization.json.JsonElement
+import kotlin.coroutines.resume
 
 expect val factory: DatabaseDriverFactory
 
@@ -102,6 +104,23 @@ internal class ActiveDatabaseTest(
 
     var connector = TestConnector()
 
+    suspend fun waitForSyncLinesChannelClosed() {
+        suspendCancellableCoroutine { continuation ->
+            var cancelled = false
+            continuation.invokeOnCancellation {
+                cancelled = true
+            }
+
+            syncLines.invokeOnClose {
+                if (!cancelled) {
+                    continuation.resume(Unit)
+                }
+
+                syncLines = Channel()
+            }
+        }
+    }
+
     fun openDatabase(schema: Schema = Schema(UserRow.table)): PowerSyncDatabaseImpl {
         logger.d { "Opening database $databaseName in directory $testDirectory" }
         val db =
@@ -123,7 +142,7 @@ internal class ActiveDatabaseTest(
     fun createSyncClient(): HttpClient {
         val engine =
             MockSyncService(
-                lines = syncLines,
+                lines = { syncLines },
                 generateCheckpoint = { checkpointResponse() },
                 syncLinesContentType = { syncLinesContentType },
                 trackSyncRequest = {

core/src/commonTest/kotlin/com/powersync/testutils/MockSyncService.kt

@@ -38,7 +38,7 @@ import kotlinx.serialization.json.JsonElement
  */
 @OptIn(LegacySyncImplementation::class)
 internal class MockSyncService(
-    private val lines: ReceiveChannel<Any>,
+    private val lines: () -> ReceiveChannel<Any>,
     private val syncLinesContentType: () -> ContentType,
     private val generateCheckpoint: () -> WriteCheckpointResponse,
     private val trackSyncRequest: suspend (HttpRequestData) -> Unit,
@@ -60,12 +60,11 @@ internal class MockSyncService(
             trackSyncRequest(data)
             val job =
                 scope.writer {
-                    lines.consume {
+                    lines().consume {
                         while (true) {
                             // Wait for a downstream listener being ready before requesting a sync line
                             channel.awaitFreeSpace()
-                            val line = receive()
-                            when (line) {
+                            when (val line = receive()) {
                                 is SyncLine -> {
                                     val serializedLine = JsonUtil.json.encodeToString(line)
                                     channel.writeStringUtf8("$serializedLine\n")

gradle.properties

@@ -19,7 +19,7 @@ development=true
 RELEASE_SIGNING_ENABLED=true
 # Library config
 GROUP=com.powersync
-LIBRARY_VERSION=1.6.1
+LIBRARY_VERSION=1.7.0
 GITHUB_REPO=https://github.com/powersync-ja/powersync-kotlin.git
 # POM
 POM_URL=https://github.com/powersync-ja/powersync-kotlin/

integrations/room/src/commonIntegrationTest/kotlin/com/powersync/integrations/room/TestDatabase.kt

@@ -37,7 +37,7 @@ interface UserDao {
     suspend fun delete(user: User)
 }
 
-@Database(entities = [User::class], version = 1)
+@Database(entities = [User::class], version = 1, exportSchema = false)
 @ConstructedBy(TestDatabaseConstructor::class)
 abstract class TestDatabase : RoomDatabase() {
     abstract fun userDao(): UserDao