Emit new connection events

Author: simolus3

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

@@ -913,4 +913,18 @@ class NewSyncIntegrationTest : BaseSyncIntegrationTest(true) {
                 query.cancelAndIgnoreRemainingEvents()
             }
         }
+
+    @Test
+    fun `ends iteration on http close`() = databaseTest {
+        turbineScope(timeout = 10.0.seconds) {
+            val turbine = database.currentStatus.asFlow().testIn(this)
+            database.connect(TestConnector(), options = getOptions())
+            turbine.waitFor { it.connected }
+
+            syncLines.close()
+            turbine.waitFor { !it.connected }
+
+            turbine.cancelAndIgnoreRemainingEvents()
+        }
+    }
 }

core/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt

@@ -8,6 +8,7 @@ import com.powersync.sync.Instruction
 import com.powersync.sync.LegacySyncImplementation
 import com.powersync.sync.SyncDataBatch
 import com.powersync.sync.SyncLocalDatabaseResult
+import com.powersync.utils.JsonUtil
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.json.JsonObject
 
@@ -55,25 +56,49 @@ internal interface BucketStorage {
 }
 
 internal sealed interface PowerSyncControlArguments {
+    /**
+     * Returns the arguments for the `powersync_control` SQL invocation.
+     */
+    val sqlArguments: Pair<String, Any?>
+
     @Serializable
     class Start(
         val parameters: JsonObject,
         val schema: SerializableSchema,
-    ) : PowerSyncControlArguments
+    ) : PowerSyncControlArguments {
+        override val sqlArguments: Pair<String, Any?>
+            get() = "start" to JsonUtil.json.encodeToString(this)
+    }
 
-    data object Stop : PowerSyncControlArguments
+    data object Stop : PowerSyncControlArguments {
+        override val sqlArguments: Pair<String, Any?> = "stop" to null
+    }
 
     data class TextLine(
         val line: String,
-    ) : PowerSyncControlArguments
+    ) : PowerSyncControlArguments {
+        override val sqlArguments: Pair<String, Any?> = "line_text" to line
+    }
 
     class BinaryLine(
-        val line: ByteArray,
+        line: ByteArray,
     ) : PowerSyncControlArguments {
         override fun toString(): String = "BinaryLine"
+
+        override val sqlArguments: Pair<String, Any?> = "line_binary" to line
+    }
+
+    data object CompletedUpload : PowerSyncControlArguments {
+        override val sqlArguments: Pair<String, Any?> = "completed_upload" to null
     }
 
-    data object CompletedUpload : PowerSyncControlArguments
+    data object ConnectionEstablished : PowerSyncControlArguments {
+        override val sqlArguments: Pair<String, Any?> = "connection" to "established"
+    }
+
+    data object ResponseStreamEnd : PowerSyncControlArguments {
+        override val sqlArguments: Pair<String, Any?> = "connection" to "end"
+    }
 }
 
 @Serializable

core/src/commonMain/kotlin/com/powersync/bucket/BucketStorageImpl.kt

@@ -358,17 +358,7 @@ internal class BucketStorageImpl(
         db.writeTransaction { tx ->
             logger.v { "powersync_control: $args" }
 
-            val (op: String, data: Any?) =
-                when (args) {
-                    is PowerSyncControlArguments.Start -> "start" to JsonUtil.json.encodeToString(args)
-                    PowerSyncControlArguments.Stop -> "stop" to null
-
-                    PowerSyncControlArguments.CompletedUpload -> "completed_upload" to null
-
-                    is PowerSyncControlArguments.BinaryLine -> "line_binary" to args.line
-                    is PowerSyncControlArguments.TextLine -> "line_text" to args.line
-                }
-
+            val (op: String, data: Any?) = args.sqlArguments
             tx.get("SELECT powersync_control(?, ?) AS r", listOf(op, data), ::handleControlResult)
         }
 }

core/src/commonMain/kotlin/com/powersync/sync/SyncStream.kt

@@ -299,14 +299,14 @@ internal class SyncStream(
                 throw RuntimeException("Received error when connecting to sync stream: ${httpResponse.bodyAsText()}")
             }
 
-            status.update { copy(connected = true, connecting = false) }
             block(isBson, httpResponse)
         }
     }
 
     private fun receiveTextLines(req: JsonElement): Flow<String> =
         flow {
             connectToSyncEndpoint(req, supportBson = false) { isBson, response ->
+                status.update { copy(connected = true, connecting = false) }
                 check(!isBson)
 
                 emitAll(response.body<ByteReadChannel>().lines())
@@ -316,13 +316,16 @@ internal class SyncStream(
     private fun receiveTextOrBinaryLines(req: JsonElement): Flow<PowerSyncControlArguments> =
         flow {
             connectToSyncEndpoint(req, supportBson = false) { isBson, response ->
+                emit(PowerSyncControlArguments.ConnectionEstablished)
                 val body = response.body<ByteReadChannel>()
 
                 if (isBson) {
                     emitAll(body.bsonObjects().map { PowerSyncControlArguments.BinaryLine(it) })
                 } else {
                     emitAll(body.lines().map { PowerSyncControlArguments.TextLine(it) })
                 }
+
+                emit(PowerSyncControlArguments.ResponseStreamEnd)
             }
         }