Remove websocket transport, use binary HTTP stream when available (#232)

Author: simolus3

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 * Update SQLite to 3.50.3.
 * Android: Ensure JNI libraries are 16KB-aligned.
+* Support receiving binary sync lines over HTTP when the Rust client is enabled.
+* Remove the experimental websocket transport mode.
 
 ## 1.3.0
 

PowerSyncKotlin/src/appleMain/kotlin/com/powersync/SDK.kt

@@ -2,7 +2,6 @@
 
 package com.powersync
 
-import com.powersync.sync.ConnectionMethod
 import com.powersync.sync.SyncOptions
 
 /**
@@ -25,16 +24,9 @@ public fun throwPowerSyncException(exception: PowerSyncException): Unit = throw
 @OptIn(ExperimentalPowerSyncAPI::class)
 public fun createSyncOptions(
     newClient: Boolean,
-    webSocket: Boolean,
     userAgent: String,
 ): SyncOptions =
     SyncOptions(
         newClientImplementation = newClient,
-        method =
-            if (webSocket) {
-                ConnectionMethod.WebSocket()
-            } else {
-                ConnectionMethod.Http
-            },
         userAgent = userAgent,
     )

core/build.gradle.kts

@@ -207,8 +207,6 @@ kotlin {
                 implementation(libs.ktor.client.contentnegotiation)
                 implementation(libs.ktor.serialization.json)
                 implementation(libs.kotlinx.io)
-                implementation(libs.rsocket.core)
-                implementation(libs.rsocket.transport.websocket)
                 implementation(libs.kotlinx.coroutines.core)
                 implementation(libs.kotlinx.datetime)
                 implementation(libs.stately.concurrency)

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

@@ -8,6 +8,7 @@ import com.powersync.ExperimentalPowerSyncAPI
  */
 abstract class AbstractSyncTest(
     private val useNewSyncImplementation: Boolean,
+    protected val useBson: Boolean = false,
 ) {
     @OptIn(ExperimentalPowerSyncAPI::class)
     val options: SyncOptions get() {

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

@@ -30,6 +30,7 @@ import io.kotest.matchers.collections.shouldHaveSize
 import io.kotest.matchers.shouldBe
 import io.kotest.matchers.shouldNotBe
 import io.kotest.matchers.string.shouldContain
+import io.ktor.http.ContentType
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.Dispatchers
@@ -859,4 +860,44 @@ class NewSyncIntegrationTest : BaseSyncIntegrationTest(true) {
                 query.cancelAndIgnoreRemainingEvents()
             }
         }
+
+    @OptIn(ExperimentalStdlibApi::class)
+    @Test
+    fun bson() =
+        databaseTest {
+            // There's no up-to-date bson library for Kotlin multiplatform, so this test verifies BSON support with byte
+            // strings created with package:bson in Dart.
+            syncLinesContentType = ContentType("application", "vnd.powersync.bson-stream")
+
+            turbineScope(timeout = 10.0.seconds) {
+                val query =
+                    database
+                        .watch("SELECT name FROM users", throttleMs = 0L) {
+                            it.getString(0)!!
+                        }.testIn(this)
+                query.awaitItem() shouldBe emptyList()
+
+                database.connect(connector, options = options)
+
+                // {checkpoint: {last_op_id: 1, write_checkpoint: null, buckets: [{bucket: a, checksum: 0, priority: 3, count: null}]}}
+                syncLines.send(
+                    "8100000003636865636b706f696e740070000000026c6173745f6f705f6964000200000031000a77726974655f636865636b706f696e7400046275636b657473003e00000003300036000000026275636b65740002000000610010636865636b73756d0000000000107072696f7269747900030000000a636f756e740000000000"
+                        .hexToByteArray(),
+                )
+
+                // {data: {bucket: a, data: [{checksum: 0, data: {"name":"username"}, op: PUT, op_id: 1, object_id: u, object_type: users}]}}
+                syncLines.send(
+                    "9e00000003646174610093000000026275636b6574000200000061000464617461007a0000000330007200000010636865636b73756d0000000000026461746100140000007b226e616d65223a22757365726e616d65227d00026f70000400000050555400026f705f696400020000003100026f626a6563745f696400020000007500026f626a6563745f74797065000600000075736572730000000000"
+                        .hexToByteArray(),
+                )
+
+                // {checkpoint_complete: {last_op_id: 1}}
+                syncLines.send(
+                    "3100000003636865636b706f696e745f636f6d706c6574650017000000026c6173745f6f705f6964000200000031000000".hexToByteArray(),
+                )
+
+                query.awaitItem() shouldBe listOf("username")
+                query.cancelAndIgnoreRemainingEvents()
+            }
+        }
 }

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

@@ -16,11 +16,11 @@ import com.powersync.createPowerSyncDatabaseImpl
 import com.powersync.db.PowerSyncDatabaseImpl
 import com.powersync.db.schema.Schema
 import com.powersync.sync.LegacySyncImplementation
-import com.powersync.sync.SyncLine
 import com.powersync.utils.JsonUtil
 import io.ktor.client.HttpClient
 import io.ktor.client.HttpClientConfig
 import io.ktor.client.engine.mock.toByteArray
+import io.ktor.http.ContentType
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.runTest
@@ -84,8 +84,8 @@ internal class ActiveDatabaseTest(
             ),
         )
 
-    @OptIn(LegacySyncImplementation::class)
-    var syncLines = Channel<SyncLine>()
+    var syncLines = Channel<Any>()
+    var syncLinesContentType = ContentType("application", "x-ndjson")
     var requestedSyncStreams = mutableListOf<JsonElement>()
     var checkpointResponse: () -> WriteCheckpointResponse = {
         WriteCheckpointResponse(WriteCheckpointData("1000"))
@@ -124,6 +124,7 @@ internal class ActiveDatabaseTest(
             MockSyncService(
                 lines = syncLines,
                 generateCheckpoint = { checkpointResponse() },
+                syncLinesContentType = { syncLinesContentType },
                 trackSyncRequest = {
                     val parsed = JsonUtil.json.parseToJsonElement(it.body.toByteArray().decodeToString())
                     requestedSyncStreams.add(parsed)

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

@@ -2,9 +2,6 @@ package com.powersync.sync
 
 import com.powersync.ExperimentalPowerSyncAPI
 import com.powersync.PowerSyncDatabase
-import io.rsocket.kotlin.keepalive.KeepAlive
-import kotlin.time.Duration
-import kotlin.time.Duration.Companion.seconds
 
 /**
  * Experimental options that can be passed to [PowerSyncDatabase.connect] to specify an experimental
@@ -19,8 +16,6 @@ public class SyncOptions
     constructor(
         @property:ExperimentalPowerSyncAPI
         public val newClientImplementation: Boolean = false,
-        @property:ExperimentalPowerSyncAPI
-        public val method: ConnectionMethod = ConnectionMethod.Http,
         /**
          * The user agent to use for requests made to the PowerSync service.
          */
@@ -37,53 +32,3 @@ public class SyncOptions
             public val defaults: SyncOptions = SyncOptions()
         }
     }
-
-/**
- * The connection method to use when the SDK connects to the sync service.
- */
-@ExperimentalPowerSyncAPI
-public sealed interface ConnectionMethod {
-    /**
-     * Receive sync lines via  streamed HTTP response from the sync service.
-     *
-     * This mode is less efficient than [WebSocket] because it doesn't support backpressure
-     * properly and uses JSON instead of the more efficient BSON representation for sync lines.
-     *
-     * This is currently the default, but this will be changed once [WebSocket] support is stable.
-     */
-    @ExperimentalPowerSyncAPI
-    public data object Http : ConnectionMethod
-
-    /**
-     * Receive binary sync lines via RSocket over a WebSocket connection.
-     *
-     * This connection mode is currently experimental and requires a recent sync service to work.
-     * WebSocket support is only available when enabling the [SyncOptions.newClientImplementation].
-     */
-    @ExperimentalPowerSyncAPI
-    public data class WebSocket(
-        val keepAlive: RSocketKeepAlive = RSocketKeepAlive.default,
-    ) : ConnectionMethod
-}
-
-/**
- * Keep-alive options for long-running RSocket streams:
- *
- * The client will ping the server every [interval], and assumes the connection to be closed if it
- * hasn't received an acknowledgement in [maxLifetime].
- */
-@ExperimentalPowerSyncAPI
-public data class RSocketKeepAlive(
-    val interval: Duration,
-    val maxLifetime: Duration,
-) {
-    internal fun toRSocket(): KeepAlive = KeepAlive(interval, maxLifetime)
-
-    internal companion object {
-        val default =
-            RSocketKeepAlive(
-                interval = 20.0.seconds,
-                maxLifetime = 30.0.seconds,
-            )
-    }
-}