Platform networking

CHANGELOG.md

@@ -16,6 +16,7 @@
 * [Sync] If calling a function on App Services that resulted in a redirect, it would only redirect for 
 GET requests. (Issue [#1517](https://github.com/realm/realm-kotlin/pull/1517))
 * [Sync] If calling a function on App Services that resulted in a redirect, it would only redirect for GET requests. (Issue [#1517](https://github.com/realm/realm-kotlin/pull/1517))
+* [Sync] Added option to use managed WebSockets via Ktor instead of Realm's built-in WebSocket client for Sync traffic. Managed WebSockets offer improved support for proxies and firewalls that require authentication. This feature is currently opt-in and can be enabled by using `AppConfiguration.usePlatformNetworking()`. Managed WebSockets will become the default in a future version. (PR [#1528](https://github.com/realm/realm-kotlin/pull/1528)). Currently only JVM and Android platforms are supported.
 
 ### Compatibility
 * File format: Generates Realms with file format v23.
@@ -32,7 +33,9 @@ GET requests. (Issue [#1517](https://github.com/realm/realm-kotlin/pull/1517))
 * Minimum Android SDK: 16.
 
 ### Internal
-* None.
+* Updated to Realm Core 13.22.0, commit 9fe0653fef672f14c771019ba27d54d568f622d0.
+* Update to Ktor 2.3.4.
+* Switched Ktor engine to CIO for Android and JVM to work around https://youtrack.jetbrains.com/issue/KTOR-6266. Revert to OkHttp when the issue is fixed.
 
 
 ## 1.11.1 (2023-09-07)

Jenkinsfile

@@ -188,7 +188,7 @@ pipeline {
                                     "integrationtest",
                                     {
                                         forwardAdbPorts()
-                                        testAndCollect("packages", "cleanAllTests -PincludeSdkModules=false connectedAndroidTest")
+                                        testAndCollect("packages", "cleanAllTests -PREALM_USE_PLATFORM_NETWORKING=true -PincludeSdkModules=false connectedAndroidTest")
                                     }
                                 )
                             }
@@ -212,7 +212,7 @@ pipeline {
                     steps {
                         testWithServer([
                             {
-                                testAndCollect("packages", 'cleanAllTests jvmTest -PincludeSdkModules=false ')
+                                testAndCollect("packages", 'cleanAllTests jvmTest -PREALM_USE_PLATFORM_NETWORKING=true -PincludeSdkModules=false ')
                             }
                         ])
                     }
@@ -232,7 +232,7 @@ pipeline {
                     steps {
                         testWithServer([
                             {
-                                testAndCollect("packages", 'cleanAllTests :test-sync:connectedAndroidtest -PincludeSdkModules=false -PtestBuildType=debugMinified')
+                                testAndCollect("packages", 'cleanAllTests :test-sync:connectedAndroidtest -PREALM_USE_PLATFORM_NETWORKING=true -PincludeSdkModules=false -PtestBuildType=debugMinified')
                             }
                         ])
                         sh 'rm mapping.zip || true'

buildSrc/src/main/kotlin/Config.kt

@@ -131,7 +131,7 @@ object Versions {
     const val latestKotlin = "1.9.20-Beta" // https://kotlinlang.org/docs/eap.html#build-details
     const val kotlinCompileTesting = "1.5.0" // https://github.com/tschuchortdev/kotlin-compile-testing
     const val ktlint = "0.45.2" // https://github.com/pinterest/ktlint
-    const val ktor = "2.1.2" // https://github.com/ktorio/ktor
+    const val ktor = "2.3.4" // https://github.com/ktorio/ktor
     const val multidex = "2.0.1" // https://developer.android.com/jetpack/androidx/releases/multidex
     const val nexusPublishPlugin = "1.1.0" // https://github.com/gradle-nexus/publish-plugin
     const val okio = "3.2.0" // https://square.github.io/okio/#releases

packages/cinterop/src/androidInstrumentedTest/kotlin/io/realm/kotlin/test/sync/SyncEnumTests.kt

@@ -28,6 +28,7 @@ import io.realm.kotlin.internal.interop.realm_sync_errno_connection_e
 import io.realm.kotlin.internal.interop.realm_sync_errno_session_e
 import io.realm.kotlin.internal.interop.realm_sync_session_resync_mode_e
 import io.realm.kotlin.internal.interop.realm_sync_session_state_e
+import io.realm.kotlin.internal.interop.realm_sync_socket_callback_result_e
 import io.realm.kotlin.internal.interop.realm_user_state_e
 import io.realm.kotlin.internal.interop.realm_web_socket_errno_e
 import io.realm.kotlin.internal.interop.sync.AuthProvider
@@ -38,6 +39,7 @@ import io.realm.kotlin.internal.interop.sync.MetadataMode
 import io.realm.kotlin.internal.interop.sync.SyncConnectionErrorCode
 import io.realm.kotlin.internal.interop.sync.SyncSessionErrorCode
 import io.realm.kotlin.internal.interop.sync.SyncSessionResyncMode
+import io.realm.kotlin.internal.interop.sync.WebsocketCallbackResult
 import io.realm.kotlin.internal.interop.sync.WebsocketErrorCode
 import org.junit.Test
 import kotlin.reflect.KClass
@@ -135,6 +137,13 @@ class SyncEnumTests {
         }
     }
 
+    @Test
+    fun websocketResultCode() {
+        checkEnum(realm_sync_socket_callback_result_e::class) { nativeValue ->
+            WebsocketCallbackResult.of(nativeValue)
+        }
+    }
+
     private inline fun <T : Any> checkEnum(
         enumClass: KClass<out Any>,
         mapNativeValue: (Int) -> T?

packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt

@@ -29,6 +29,9 @@ import io.realm.kotlin.internal.interop.sync.NetworkTransport
 import io.realm.kotlin.internal.interop.sync.ProgressDirection
 import io.realm.kotlin.internal.interop.sync.SyncSessionResyncMode
 import io.realm.kotlin.internal.interop.sync.SyncUserIdentity
+import io.realm.kotlin.internal.interop.sync.WebSocketTransport
+import io.realm.kotlin.internal.interop.sync.WebsocketCallbackResult
+import io.realm.kotlin.internal.interop.sync.WebsocketErrorCode
 import kotlinx.coroutines.CoroutineDispatcher
 import org.mongodb.kbson.ObjectId
 import kotlin.jvm.JvmInline
@@ -101,9 +104,15 @@ interface RealmUserT : CapiT
 interface RealmNetworkTransportT : CapiT
 interface RealmSyncSessionT : CapiT
 interface RealmSubscriptionT : CapiT
+interface RealmSyncSocketObserverPointerT : CapiT
+interface RealmSyncSocketCallbackPointerT : CapiT
+
 interface RealmBaseSubscriptionSet : CapiT
+interface RealmSyncSocket : CapiT
 interface RealmSubscriptionSetT : RealmBaseSubscriptionSet
 interface RealmMutableSubscriptionSetT : RealmBaseSubscriptionSet
+interface RealmSyncSocketT : RealmSyncSocket
+
 // Public type aliases binding to internal verbose type safe type definitions. This should allow us
 // to easily change implementation details later on.
 typealias RealmAsyncOpenTaskPointer = NativePointer<RealmAsyncOpenTaskT>
@@ -119,7 +128,11 @@ typealias RealmSubscriptionPointer = NativePointer<RealmSubscriptionT>
 typealias RealmBaseSubscriptionSetPointer = NativePointer<out RealmBaseSubscriptionSet>
 typealias RealmSubscriptionSetPointer = NativePointer<RealmSubscriptionSetT>
 typealias RealmMutableSubscriptionSetPointer = NativePointer<RealmMutableSubscriptionSetT>
-
+typealias RealmSyncSocketPointer = NativePointer<RealmSyncSocketT>
+typealias RealmSyncSocketObserverPointer = NativePointer<RealmSyncSocketObserverPointerT>
+typealias RealmSyncSocketCallbackPointer = NativePointer<RealmSyncSocketCallbackPointerT>
+typealias RealmWebsocketHandlerCallbackPointer = NativePointer<CapiT>
+typealias RealmWebsocketProviderPointer = NativePointer<CapiT>
 /**
  * Class for grouping and normalizing values we want to send as part of
  * logging in Sync Users.
@@ -486,7 +499,6 @@ expect object RealmInterop {
     fun realm_app_link_credentials(app: RealmAppPointer, user: RealmUserPointer, credentials: RealmCredentialsPointer, callback: AppCallback<RealmUserPointer>)
     fun realm_clear_cached_apps()
     fun realm_app_sync_client_get_default_file_path_for_realm(
-        app: RealmAppPointer,
         syncConfig: RealmSyncConfigurationPointer,
         overriddenName: String?
     ): String
@@ -786,4 +798,22 @@ expect object RealmInterop {
     fun realm_sync_subscriptionset_commit(
         mutableSubscriptionSet: RealmMutableSubscriptionSetPointer
     ): RealmSubscriptionSetPointer
+
+    fun realm_sync_set_websocket_transport(
+        syncClientConfig: RealmSyncClientConfigurationPointer,
+        webSocketTransport: WebSocketTransport
+    )
+
+    fun realm_sync_socket_callback_complete(nativePointer: RealmWebsocketHandlerCallbackPointer, cancelled: Boolean = false, status: WebsocketCallbackResult = WebsocketCallbackResult.RLM_ERR_SYNC_SOCKET_SUCCESS, reason: String = "")
+
+    fun realm_sync_socket_websocket_connected(nativePointer: RealmWebsocketProviderPointer, protocol: String)
+
+    fun realm_sync_socket_websocket_error(nativePointer: RealmWebsocketProviderPointer)
+
+    fun realm_sync_socket_websocket_message(
+        nativePointer: RealmWebsocketProviderPointer,
+        data: ByteArray
+    ): Boolean
+
+    fun realm_sync_socket_websocket_closed(nativePointer: RealmWebsocketProviderPointer, wasClean: Boolean, errorCode: WebsocketErrorCode, reason: String = "")
 }

packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/sync/ProtocolErrorCode.kt

@@ -121,6 +121,21 @@ expect enum class WebsocketErrorCode : CodeDescription {
     RLM_ERR_WEBSOCKET_FATAL_ERROR;
 
     companion object {
-        internal fun of(nativeValue: Int): WebsocketErrorCode?
+        fun of(nativeValue: Int): WebsocketErrorCode?
+    }
+}
+
+expect enum class WebsocketCallbackResult : CodeDescription {
+    RLM_ERR_SYNC_SOCKET_SUCCESS,
+    RLM_ERR_SYNC_SOCKET_OPERATION_ABORTED,
+    RLM_ERR_SYNC_SOCKET_RUNTIME,
+    RLM_ERR_SYNC_SOCKET_OUT_OF_MEMORY,
+    RLM_ERR_SYNC_SOCKET_ADDRESS_SPACE_EXHAUSTED,
+    RLM_ERR_SYNC_SOCKET_CONNECTION_CLOSED,
+    RLM_ERR_SYNC_SOCKET_NOT_SUPPORTED,
+    RLM_ERR_SYNC_SOCKET_INVALID_ARGUMENT;
+
+    companion object {
+        fun of(nativeValue: Int): WebsocketCallbackResult?
     }
 }

packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/sync/WebSocketTransport.kt

@@ -0,0 +1,87 @@
+package io.realm.kotlin.internal.interop.sync
+
+import io.realm.kotlin.internal.interop.RealmInterop
+import io.realm.kotlin.internal.interop.RealmWebsocketHandlerCallbackPointer
+import io.realm.kotlin.internal.interop.RealmWebsocketProviderPointer
+import kotlinx.coroutines.Job
+
+interface WebSocketTransport {
+    fun post(handlerCallback: RealmWebsocketHandlerCallbackPointer)
+
+    fun createTimer(
+        delayInMilliseconds: Long,
+        handlerCallback: RealmWebsocketHandlerCallbackPointer,
+    ): CancellableTimer
+
+    @Suppress("LongParameterList")
+    fun connect(
+        observer: WebSocketObserver,
+        path: String,
+        address: String,
+        port: Long,
+        isSsl: Boolean,
+        numProtocols: Long,
+        supportedProtocols: String
+    ): WebSocketClient
+
+    fun write(
+        webSocketClient: WebSocketClient,
+        data: ByteArray,
+        length: Long,
+        handlerCallback: RealmWebsocketHandlerCallbackPointer
+    )
+
+    fun runCallback(
+        handlerCallback: RealmWebsocketHandlerCallbackPointer,
+        cancelled: Boolean = false,
+        status: WebsocketCallbackResult = WebsocketCallbackResult.RLM_ERR_SYNC_SOCKET_SUCCESS,
+        reason: String = ""
+    ) {
+        RealmInterop.realm_sync_socket_callback_complete(
+            handlerCallback,
+            cancelled,
+            status,
+            reason
+        )
+    }
+
+    fun close()
+}
+
+class CancellableTimer(
+    private val job: Job,
+    private val cancelCallback: () -> Unit
+) {
+    fun cancel() {
+        job.cancel()
+        cancelCallback()
+    }
+}
+
+interface WebSocketClient {
+    fun send(message: ByteArray, handlerCallback: RealmWebsocketHandlerCallbackPointer)
+    fun closeWebsocket()
+}
+
+class WebSocketObserver(private val webSocketObserverPointer: RealmWebsocketProviderPointer) {
+    fun onConnected(protocol: String) {
+        RealmInterop.realm_sync_socket_websocket_connected(webSocketObserverPointer, protocol)
+    }
+
+    fun onError() {
+        RealmInterop.realm_sync_socket_websocket_error(webSocketObserverPointer)
+    }
+
+    fun onNewMessage(data: ByteArray): Boolean {
+        return RealmInterop.realm_sync_socket_websocket_message(webSocketObserverPointer, data)
+    }
+
+    fun onClose(wasClean: Boolean, errorCode: WebsocketErrorCode, reason: String) {
+        RealmInterop.realm_sync_socket_websocket_closed(
+            webSocketObserverPointer,
+            wasClean,
+            errorCode,
+            reason
+        )
+    }
+}

packages/cinterop/src/jvm/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt

@@ -29,6 +29,9 @@ import io.realm.kotlin.internal.interop.sync.NetworkTransport
 import io.realm.kotlin.internal.interop.sync.ProgressDirection
 import io.realm.kotlin.internal.interop.sync.SyncSessionResyncMode
 import io.realm.kotlin.internal.interop.sync.SyncUserIdentity
+import io.realm.kotlin.internal.interop.sync.WebSocketTransport
+import io.realm.kotlin.internal.interop.sync.WebsocketCallbackResult
+import io.realm.kotlin.internal.interop.sync.WebsocketErrorCode
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.launch
@@ -1115,7 +1118,8 @@ actual object RealmInterop {
     }
 
     actual fun realm_user_get_auth_provider(user: RealmUserPointer): AuthProvider {
-        return AuthProvider.of(realmc.realm_user_get_auth_provider(user.cptr()))
+        TODO("No longer valid")
+//        return AuthProvider.of(realmc.realm_user_get_auth_provider(user.cptr()))
     }
 
     actual fun realm_user_get_access_token(user: RealmUserPointer): String {
@@ -1157,7 +1161,6 @@ actual object RealmInterop {
     }
 
     actual fun realm_app_sync_client_get_default_file_path_for_realm(
-        app: RealmAppPointer,
         syncConfig: RealmSyncConfigurationPointer,
         overriddenName: String?
     ): String {
@@ -1977,6 +1980,36 @@ actual object RealmInterop {
         return LongPointerWrapper(realmc.realm_sync_subscription_set_commit(mutableSubscriptionSet.cptr()))
     }
 
+    actual fun realm_sync_set_websocket_transport(
+        syncClientConfig: RealmSyncClientConfigurationPointer,
+        webSocketTransport: WebSocketTransport
+    ) {
+        realmc.realm_sync_websocket_new(syncClientConfig.cptr(), webSocketTransport)
+    }
+
+    actual fun realm_sync_socket_callback_complete(nativePointer: RealmWebsocketHandlerCallbackPointer, cancelled: Boolean, status: WebsocketCallbackResult, reason: String) {
+        realmc.realm_sync_websocket_callback_complete(cancelled, nativePointer.cptr(), status.nativeValue, reason)
+    }
+
+    actual fun realm_sync_socket_websocket_connected(nativePointer: RealmWebsocketProviderPointer, protocol: String) {
+        realmc.realm_sync_websocket_connected(nativePointer.cptr(), protocol)
+    }
+
+    actual fun realm_sync_socket_websocket_error(nativePointer: RealmWebsocketProviderPointer) {
+        realmc.realm_sync_websocket_error(nativePointer.cptr())
+    }
+
+    actual fun realm_sync_socket_websocket_message(
+        nativePointer: RealmWebsocketProviderPointer,
+        data: ByteArray
+    ): Boolean {
+        return realmc.realm_sync_websocket_message(nativePointer.cptr(), data, data.size.toLong())
+    }
+
+    actual fun realm_sync_socket_websocket_closed(nativePointer: RealmWebsocketProviderPointer, wasClean: Boolean, errorCode: WebsocketErrorCode, reason: String) {
+        realmc.realm_sync_websocket_closed(nativePointer.cptr(), wasClean, errorCode.nativeValue, reason)
+    }
+
     fun <T : CapiT> NativePointer<T>.cptr(): Long {
         return (this as LongPointerWrapper).ptr
     }

packages/cinterop/src/jvm/kotlin/io/realm/kotlin/internal/interop/sync/ProtocolErrorCode.kt

@@ -19,6 +19,7 @@ package io.realm.kotlin.internal.interop.sync
 import io.realm.kotlin.internal.interop.CodeDescription
 import io.realm.kotlin.internal.interop.realm_sync_errno_connection_e
 import io.realm.kotlin.internal.interop.realm_sync_errno_session_e
+import io.realm.kotlin.internal.interop.realm_sync_socket_callback_result_e
 import io.realm.kotlin.internal.interop.realm_web_socket_errno_e
 
 actual enum class SyncConnectionErrorCode(
@@ -129,7 +130,50 @@ actual enum class WebsocketErrorCode(
     RLM_ERR_WEBSOCKET_FATAL_ERROR("FatalError", realm_web_socket_errno_e.RLM_ERR_WEBSOCKET_FATAL_ERROR);
 
     actual companion object {
-        internal actual fun of(nativeValue: Int): WebsocketErrorCode? =
+        actual fun of(nativeValue: Int): WebsocketErrorCode? =
+            values().firstOrNull { value ->
+                value.nativeValue == nativeValue
+            }
+    }
+}
+
+actual enum class WebsocketCallbackResult(override val description: String, override val nativeValue: Int) : CodeDescription {
+
+    RLM_ERR_SYNC_SOCKET_SUCCESS(
+        "Websocket callback success",
+        realm_sync_socket_callback_result_e.RLM_ERR_SYNC_SOCKET_SUCCESS
+    ),
+    RLM_ERR_SYNC_SOCKET_OPERATION_ABORTED(
+        "Websocket callback aborted",
+        realm_sync_socket_callback_result_e.RLM_ERR_SYNC_SOCKET_OPERATION_ABORTED
+    ),
+    RLM_ERR_SYNC_SOCKET_RUNTIME(
+        "Websocket Runtime error",
+        realm_sync_socket_callback_result_e.RLM_ERR_SYNC_SOCKET_RUNTIME
+    ),
+    RLM_ERR_SYNC_SOCKET_OUT_OF_MEMORY(
+        "Websocket out of memory ",
+        realm_sync_socket_callback_result_e.RLM_ERR_SYNC_SOCKET_OUT_OF_MEMORY
+    ),
+    RLM_ERR_SYNC_SOCKET_ADDRESS_SPACE_EXHAUSTED(
+        "Websocket address space exhausted",
+        realm_sync_socket_callback_result_e.RLM_ERR_SYNC_SOCKET_ADDRESS_SPACE_EXHAUSTED
+    ),
+    RLM_ERR_SYNC_SOCKET_CONNECTION_CLOSED(
+        "Websocket connection closed",
+        realm_sync_socket_callback_result_e.RLM_ERR_SYNC_SOCKET_CONNECTION_CLOSED
+    ),
+    RLM_ERR_SYNC_SOCKET_NOT_SUPPORTED(
+        "Websocket not supported",
+        realm_sync_socket_callback_result_e.RLM_ERR_SYNC_SOCKET_NOT_SUPPORTED
+    ),
+    RLM_ERR_SYNC_SOCKET_INVALID_ARGUMENT(
+        "Websocket invalid argument",
+        realm_sync_socket_callback_result_e.RLM_ERR_SYNC_SOCKET_INVALID_ARGUMENT
+    );
+
+    actual companion object {
+        actual fun of(nativeValue: Int): WebsocketCallbackResult? =
             values().firstOrNull { value ->
                 value.nativeValue == nativeValue
             }