Fix flushing

Author: simolus3

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

@@ -27,7 +27,7 @@ import com.powersync.utils.JsonUtil
 import dev.mokkery.answering.returns
 import dev.mokkery.everySuspend
 import dev.mokkery.mock
-import kotlinx.coroutines.CompletableDeferred
+import dev.mokkery.verify
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.runBlocking
@@ -38,6 +38,7 @@ import kotlin.test.AfterTest
 import kotlin.test.BeforeTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertFalse
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
@@ -119,7 +120,7 @@ class SyncIntegrationTest {
 
     @Test
     @OptIn(DelicateCoroutinesApi::class)
-    fun closesResponseStreamOnDisconnect() = runTest {
+    fun closesResponseStreamOnDatabaseClose() = runTest {
         val syncStream = syncStream()
         database.connectInternal(syncStream, 1000L)
 
@@ -132,10 +133,48 @@ class SyncIntegrationTest {
             turbine.cancel()
         }
 
-        // Closing the database should close the channel
-        val channelClose = CompletableDeferred<Unit>()
-        syncLines.invokeOnClose { channelClose.complete(Unit) }
-        channelClose.await()
+        // Closing the database should have closed the channel
+        assertTrue { syncLines.isClosedForSend }
+    }
+
+    @Test
+    @OptIn(DelicateCoroutinesApi::class)
+    fun cleansResourcesOnDisconnect() = runTest {
+        val syncStream = syncStream()
+        database.connectInternal(syncStream, 1000L)
+
+        turbineScope(timeout = 10.0.seconds) {
+            val turbine = database.currentStatus.asFlow().testIn(this)
+            turbine.waitFor { it.connected }
+
+            database.disconnect()
+            turbine.waitFor { !it.connected }
+            turbine.cancel()
+        }
+
+        // Disconnecting should have closed the channel
+        assertTrue { syncLines.isClosedForSend }
+
+        // And called invalidateCredentials on the connector
+        verify { connector.invalidateCredentials() }
+    }
+
+    @Test
+    fun cannotUpdateSchemaWhileConnected() = runTest {
+        val syncStream = syncStream()
+        database.connectInternal(syncStream, 1000L)
+
+        turbineScope(timeout = 10.0.seconds) {
+            val turbine = database.currentStatus.asFlow().testIn(this)
+            turbine.waitFor { it.connected }
+            turbine.cancel()
+        }
+
+        assertFailsWith<PowerSyncException>("Cannot update schema while connected") {
+            database.updateSchema(Schema())
+        }
+
+        database.close()
     }
 
     @Test

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

@@ -47,10 +47,11 @@ internal class MockSyncService(
         val scope = CoroutineScope(context)
 
         return if (data.url.encodedPath == "/sync/stream") {
-            val job = scope.writer(autoFlush = true) {
+            val job = scope.writer {
                 lines.consumeEach {
                     val serializedLine = JsonUtil.json.encodeToString(it)
                     channel.writeStringUtf8("$serializedLine\n")
+                    channel.flush()
                 }
             }