KRPC-62 WebSocketSession KtorRPCClient

docs/pages/kotlinx-rpc/topics/transport.topic

@@ -34,12 +34,17 @@
                     }
                 }
 
-                val rpcClient: RPCClient =
+                val rpcClient: KtorRPCClient =
                     ktorClient.rpc("ws://localhost:4242/services") { // this: HttpRequestBuilder
                         rpcConfig { // this: RPCConfigBuilder.Client
                             waitForServices = false
                         }
                     }
+
+                // access WebSocketSession that created the connection
+                rpcClient.webSocketSession
+
+                // create RPC service
                 val myService: MyService = rpcClient.withService<MyService>()
             </code-block>
             <p>Note that in this example, only the latter defined <code>RPCConfig</code> will be used.</p>
@@ -56,7 +61,7 @@
                     }
 
                     routing {
-                        rpc(&quot;/services&quot;) { // this RPCRoute
+                        rpc(&quot;/services&quot;) { // this RPCRoute, inherits WebSocketSession
                             rpcConfig { // this: RPCConfigBuilder.Server
                                 waitForServices = false
                             }
@@ -85,7 +90,7 @@
                     suspend fun processImage(url: Srting): ProcessedImage
                 }
 
-// ### CLIENT CODE ###
+                // ### CLIENT CODE ###
 
                 val client = HttpClient {
                     installRPC {
@@ -99,7 +104,7 @@
 
                 service.processImage(url = &quot;https://catsanddogs.com/cats/1&quot;)
 
-// ### SERVER CODE ###
+                // ### SERVER CODE ###
 
                 class ImageServiceImpl(override val coroutineContext: CoroutineContext) : ImageService {
                     // user defined classes

transport/transport-ktor/src/jvmTest/kotlin/kotlinx/rpc/transport/ktor/KtorTransportTest.kt

@@ -7,6 +7,7 @@
 package kotlinx.rpc.transport.ktor
 
 import io.ktor.client.*
+import io.ktor.client.plugins.websocket.*
 import io.ktor.server.application.*
 import io.ktor.server.engine.*
 import io.ktor.server.netty.*
@@ -29,8 +30,12 @@ interface NewService : RPC {
     suspend fun echo(value: String): String
 }
 
-class NewServiceImpl(override val coroutineContext: CoroutineContext) : NewService {
+class NewServiceImpl(
+    override val coroutineContext: CoroutineContext,
+    private val call: ApplicationCall,
+) : NewService {
     override suspend fun echo(value: String): String {
+        assertEquals("test-header", call.request.headers["TestHeader"])
         return value
     }
 }
@@ -52,22 +57,30 @@ class KtorTransportTest {
                         waitForServices = true
                     }
 
-                    registerService<NewService> { NewServiceImpl(it) }
+                    registerService<NewService> { NewServiceImpl(it, call) }
                 }
             }
         }.start()
 
         val clientWithGlobalConfig = HttpClient {
-            installRPC {
+            install(WebSockets) {
+                maxFrameSize = Int.MAX_VALUE.toLong() - 42
+            }
+            install(kotlinx.rpc.transport.ktor.client.RPC) {
                 serialization {
                     json()
                 }
             }
         }
 
-        val serviceWithGlobalConfig = clientWithGlobalConfig
-            .rpc("ws://localhost:4242/rpc")
-            .withService<NewService>()
+        val ktorRPCClient = clientWithGlobalConfig
+            .rpc("ws://localhost:4242/rpc") {
+                headers["TestHeader"] = "test-header"
+            }
+
+        assertEquals(Int.MAX_VALUE.toLong() - 42, ktorRPCClient.webSocketSession.maxFrameSize)
+
+        val serviceWithGlobalConfig = ktorRPCClient.withService<NewService>()
 
         val firstActual = serviceWithGlobalConfig.echo("Hello, world!")
 
@@ -81,6 +94,8 @@ class KtorTransportTest {
         }
 
         val serviceWithLocalConfig = clientWithNoConfig.rpc("ws://localhost:4242/rpc") {
+            headers["TestHeader"] = "test-header"
+
             rpcConfig {
                 serialization {
                     json()

transport/transport-ktor/transport-ktor-client/api/transport-ktor-client.api

@@ -7,6 +7,10 @@ public final class kotlinx/rpc/transport/ktor/client/KtorClientDslKt {
 	public static synthetic fun rpcConfig$default (Lio/ktor/client/request/HttpRequestBuilder;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
 }
 
+public abstract interface class kotlinx/rpc/transport/ktor/client/KtorRPCClient : kotlinx/rpc/RPCClient {
+	public abstract fun getWebSocketSession ()Lio/ktor/websocket/WebSocketSession;
+}
+
 public final class kotlinx/rpc/transport/ktor/client/RPCKt {
 	public static final fun getRPC ()Lio/ktor/client/plugins/api/ClientPlugin;
 	public static final fun installRPC (Lio/ktor/client/HttpClientConfig;Lkotlin/jvm/functions/Function1;)V

transport/transport-ktor/transport-ktor-client/src/commonMain/kotlin/kotlinx/rpc/transport/ktor/client/KtorClientDsl.kt

@@ -39,7 +39,7 @@ public fun HttpRequestBuilder.rpcConfig(configBuilder: RPCConfigBuilder.Client.(
 public suspend fun HttpClient.rpc(
     urlString: String,
     block: HttpRequestBuilder.() -> Unit = {},
-): RPCClient {
+): KtorRPCClient {
     return rpc {
         url(urlString)
         block()
@@ -55,7 +55,7 @@ public suspend fun HttpClient.rpc(
  */
 public suspend fun HttpClient.rpc(
     block: HttpRequestBuilder.() -> Unit = {},
-): RPCClient {
+): KtorRPCClient {
     pluginOrNull(WebSockets)
         ?: error("RPC for client requires $WebSockets plugin to be installed firstly")
 
@@ -72,5 +72,5 @@ public suspend fun HttpClient.rpc(
     val rpcConfig = pluginConfigBuilder?.apply(requestConfigBuilder)?.build()
         ?: rpcClientConfig(requestConfigBuilder)
 
-    return KtorRPCClient(session, rpcConfig)
+    return KtorRPCClientImpl(session, rpcConfig)
 }

transport/transport-ktor/transport-ktor-client/src/commonMain/kotlin/kotlinx/rpc/transport/ktor/client/KtorRPCClient.kt

@@ -5,11 +5,21 @@
 package kotlinx.rpc.transport.ktor.client
 
 import io.ktor.websocket.*
+import kotlinx.rpc.RPCClient
 import kotlinx.rpc.RPCConfig
 import kotlinx.rpc.client.KRPCClient
 import kotlinx.rpc.transport.ktor.KtorTransport
 
-internal class KtorRPCClient(
-    webSocketSession: WebSocketSession,
+/**
+ * [RPCClient] implementation for Ktor, containing [webSocketSession] object,
+ * that is used to maintain connection.
+ */
+public interface KtorRPCClient : RPCClient {
+    public val webSocketSession: WebSocketSession
+}
+
+internal class KtorRPCClientImpl(
+    override val webSocketSession: WebSocketSession,
     config: RPCConfig.Client,
-): KRPCClient(config, KtorTransport(webSocketSession))
+): KRPCClient(config, KtorTransport(webSocketSession)), KtorRPCClient
+