Start with rsocket support

core/build.gradle.kts

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

core/src/commonMain/kotlin/com/powersync/PowerSyncDatabase.kt

@@ -6,6 +6,7 @@ import com.powersync.db.Queries
 import com.powersync.db.crud.CrudBatch
 import com.powersync.db.crud.CrudTransaction
 import com.powersync.db.schema.Schema
+import com.powersync.sync.SyncOptions
 import com.powersync.sync.SyncStatus
 import com.powersync.utils.JsonParam
 import kotlin.coroutines.cancellation.CancellationException
@@ -94,6 +95,7 @@ public interface PowerSyncDatabase : Queries {
         crudThrottleMs: Long = 1000L,
         retryDelayMs: Long = 5000L,
         params: Map<String, JsonParam?> = emptyMap(),
+        options: SyncOptions = SyncOptions()
     )
 
     /**

core/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt

@@ -18,6 +18,7 @@ import com.powersync.db.internal.PowerSyncVersion
 import com.powersync.db.schema.Schema
 import com.powersync.db.schema.toSerializable
 import com.powersync.sync.PriorityStatusEntry
+import com.powersync.sync.SyncOptions
 import com.powersync.sync.SyncStatus
 import com.powersync.sync.SyncStatusData
 import com.powersync.sync.SyncStream
@@ -153,6 +154,7 @@ internal class PowerSyncDatabaseImpl(
         crudThrottleMs: Long,
         retryDelayMs: Long,
         params: Map<String, JsonParam?>,
+        options: SyncOptions
     ) {
         waitReady()
         mutex.withLock {
@@ -168,13 +170,13 @@ internal class PowerSyncDatabaseImpl(
                     params = params.toJsonObject(),
                     scope = scope,
                     createClient = createClient,
+                    options = options,
                 ),
                 crudThrottleMs,
             )
         }
     }
 
-    @OptIn(FlowPreview::class)
     internal fun connectInternal(
         stream: SyncStream,
         crudThrottleMs: Long,

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

@@ -0,0 +1,37 @@
+package com.powersync.sync
+
+import io.rsocket.kotlin.keepalive.KeepAlive
+import kotlin.time.Duration.Companion.seconds
+
+public class SyncOptions(
+    public val method: ConnectionMethod = ConnectionMethod.WebSocket,
+)
+
+/**
+ * The connection method to use when the SDK connects to the sync service.
+ */
+public sealed interface ConnectionMethod {
+    /**
+     * Receive sync lines via an 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.
+     */
+    public data object Http: ConnectionMethod
+
+    /**
+     * Receive binary sync lines via RSocket over a WebSocket connection.
+     *
+     * This is the default mode, and recommended for most clients.
+     */
+    public data class WebSocket(
+        val keepAlive: KeepAlive = DefaultKeepAlive
+    ): ConnectionMethod {
+        private companion object {
+            val DefaultKeepAlive = KeepAlive(
+                interval = 20.0.seconds,
+                maxLifetime = 30.0.seconds,
+            )
+        }
+    }
+}

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

@@ -1,5 +1,6 @@
 package com.powersync.sync
 
+import BuildConfig
 import co.touchlab.kermit.Logger
 import co.touchlab.kermit.Severity
 import co.touchlab.stately.concurrency.AtomicBoolean
@@ -18,6 +19,7 @@ import io.ktor.client.call.body
 import io.ktor.client.plugins.HttpTimeout
 import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
 import io.ktor.client.plugins.timeout
+import io.ktor.client.plugins.websocket.WebSockets
 import io.ktor.client.request.get
 import io.ktor.client.request.headers
 import io.ktor.client.request.preparePost
@@ -26,9 +28,21 @@ import io.ktor.client.statement.bodyAsText
 import io.ktor.http.ContentType
 import io.ktor.http.HttpHeaders
 import io.ktor.http.HttpStatusCode
+import io.ktor.http.URLBuilder
+import io.ktor.http.URLProtocol
+import io.ktor.http.Url
 import io.ktor.http.contentType
+import io.ktor.http.takeFrom
 import io.ktor.utils.io.ByteReadChannel
 import io.ktor.utils.io.readUTF8Line
+import io.rsocket.kotlin.keepalive.KeepAlive
+import io.rsocket.kotlin.ktor.client.RSocketSupport
+import io.rsocket.kotlin.ktor.client.rSocket
+import io.rsocket.kotlin.payload.Payload
+import io.rsocket.kotlin.payload.PayloadMimeType
+import io.rsocket.kotlin.payload.buildPayload
+import io.rsocket.kotlin.payload.data
+import io.rsocket.kotlin.payload.metadata
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
@@ -44,8 +58,11 @@ import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import kotlinx.datetime.Clock
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.JsonObject
+import kotlin.time.Duration.Companion.seconds
 
 internal class SyncStream(
     private val bucketStorage: BucketStorage,
@@ -55,6 +72,7 @@ internal class SyncStream(
     private val logger: Logger,
     private val params: JsonObject,
     private val scope: CoroutineScope,
+    private val options: SyncOptions,
     createClient: (HttpClientConfig<*>.() -> Unit) -> HttpClient,
 ) {
     private var isUploadingCrud = AtomicBoolean(false)
@@ -71,6 +89,37 @@ internal class SyncStream(
         createClient {
             install(HttpTimeout)
             install(ContentNegotiation)
+
+            (options.method as? ConnectionMethod.WebSocket)?.let {
+                install(WebSockets)
+                install(RSocketSupport) {
+                    connector {
+                        connectionConfig {
+                            payloadMimeType = PayloadMimeType(
+                                metadata = "application/json",
+                                data = "application/json"
+                            )
+
+                            setupPayload {
+                                buildPayload {
+                                    @Serializable
+                                    class ConnectionSetupMetadata(
+                                        // Kind of annoying to specify this here, https://github.com/rsocket/rsocket-kotlin/issues/311
+                                        val token: String = "TODO: token",
+                                        @SerialName("user_agent")
+                                        val userAgent: String = "Kotlin SDK"
+                                    )
+
+                                    metadata(JsonUtil.json.encodeToString(ConnectionSetupMetadata()))
+                                }
+                            }
+
+                            keepAlive = it.keepAlive
+                        }
+                    }
+}
+            }
+
         }
 
     fun invalidateCredentials() {
@@ -184,7 +233,7 @@ internal class SyncStream(
         return body.data.writeCheckpoint
     }
 
-    private fun streamingSyncRequest(req: JsonObject): Flow<String> =
+    private fun connectViaHttp(req: JsonObject): Flow<String> =
         flow {
             val credentials = connector.getCredentialsCached()
             require(credentials != null) { "Not logged in" }
@@ -225,6 +274,22 @@ internal class SyncStream(
             }
         }
 
+    private fun connectViaWebSocket(req: JsonObject): Flow<ByteArray> = flow {
+        val credentials = connector.getCredentialsCached()
+        require(credentials != null) { "Not logged in" }
+        val uri = URLBuilder(credentials.endpointUri("sync/stream")).apply {
+            protocol = when (protocolOrNull) {
+                URLProtocol.HTTP -> URLProtocol.WS
+                else -> URLProtocol.WSS
+            }
+        }
+
+        val rSocket = httpClient.rSocket { url.takeFrom(uri) }
+        rSocket.requestStream(buildPayload {
+            metadata(JsonUtil.json.encodeToString(""))
+        })
+    }
+
     private suspend fun streamingSyncIteration() {
         val iteration = ActiveIteration()
 
@@ -308,7 +373,7 @@ internal class SyncStream(
         }
 
         private suspend fun connect(start: Instruction.EstablishSyncStream) {
-            streamingSyncRequest(start.request).collect { rawLine ->
+            connectViaHttp(start.request).collect { rawLine ->
                 control("line_text", rawLine)
             }
         }

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

@@ -100,6 +100,7 @@ class SyncStreamTest {
                     logger = logger,
                     params = JsonObject(emptyMap()),
                     scope = this,
+                    options = SyncOptions(),
                 )
 
             syncStream.invalidateCredentials()
@@ -137,6 +138,7 @@ class SyncStreamTest {
                     logger = logger,
                     params = JsonObject(emptyMap()),
                     scope = this,
+                    options = SyncOptions(),
                 )
 
             syncStream.status.update { copy(connected = true) }
@@ -176,6 +178,7 @@ class SyncStreamTest {
                     logger = logger,
                     params = JsonObject(emptyMap()),
                     scope = this,
+                    options = SyncOptions()
                 )
 
             // Launch streaming sync in a coroutine that we'll cancel after verification

gradle/libs.versions.toml

@@ -14,7 +14,8 @@ kotlin = "2.1.10"
 coroutines = "1.8.1"
 kotlinx-datetime = "0.6.2"
 kotlinx-io = "0.5.4"
-ktor = "3.0.1"
+ktor = "3.1.0"
+rsocket = "0.20.0"
 uuid = "0.8.2"
 powersync-core = "0.3.12"
 sqlite-jdbc = "3.49.1.0"
@@ -85,6 +86,7 @@ ktor-client-contentnegotiation = { module = "io.ktor:ktor-client-content-negotia
 ktor-client-mock = { module = "io.ktor:ktor-client-mock", version.ref = "ktor" }
 ktor-serialization-json = { module = "io.ktor:ktor-serialization-kotlinx-json", version.ref = "ktor" }
 kotlinx-coroutines-core = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "coroutines" }
+rsocket-client = { module = "io.rsocket.kotlin:ktor-client-rsocket", version.ref = "rsocket" }
 
 sqldelight-driver-native = { module = "app.cash.sqldelight:native-driver", version.ref = "sqlDelight" }
 sqliter = { module = "co.touchlab:sqliter-driver", version.ref = "sqliter" }