KRPC-113 Support gRPC: Ktor Integration

Mr3zee · Mr3zee · commit ce5ddbcf7ad2 · 2025-07-14T12:47:43.000+02:00
diff --git a/grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/GrpcClient.kt b/grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/GrpcClient.kt
@@ -10,6 +10,7 @@ import kotlinx.rpc.RpcClient
 import kotlinx.rpc.grpc.descriptor.GrpcClientDelegate
 import kotlinx.rpc.grpc.descriptor.GrpcServiceDescriptor
 import kotlinx.rpc.internal.utils.map.RpcInternalConcurrentHashMap
+import kotlin.time.Duration
 
 /**
  * GrpcClient manages gRPC communication by providing implementation for making asynchronous RPC calls.
@@ -19,6 +20,20 @@ import kotlinx.rpc.internal.utils.map.RpcInternalConcurrentHashMap
 public class GrpcClient internal constructor(private val channel: ManagedChannel) : RpcClient {
     private val stubs = RpcInternalConcurrentHashMap<Long, GrpcClientDelegate>()
 
+    public fun shutdown() {
+        stubs.clear()
+        channel.shutdown()
+    }
+
+    public fun shutdownNow() {
+        stubs.clear()
+        channel.shutdownNow()
+    }
+
+    public suspend fun awaitTermination(duration: Duration) {
+        channel.awaitTermination(duration)
+    }
+
     override suspend fun <T> call(call: RpcCall): T {
         return call.delegate().call(call)
     }
@@ -39,11 +54,11 @@ public class GrpcClient internal constructor(private val channel: ManagedChannel
  * Constructor function for the [GrpcClient] class.
  */
 public fun GrpcClient(
-    name: String,
+    hostname: String,
     port: Int,
     configure: ManagedChannelBuilder<*>.() -> Unit = {},
 ): GrpcClient {
-    val channel = ManagedChannelBuilder(name, port).apply(configure).buildChannel()
+    val channel = ManagedChannelBuilder(hostname, port).apply(configure).buildChannel()
     return GrpcClient(channel)
 }
 
diff --git a/grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/GrpcServer.kt b/grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/GrpcServer.kt
@@ -4,6 +4,7 @@
 
 package kotlinx.rpc.grpc
 
+import kotlinx.atomicfu.atomic
 import kotlinx.rpc.RpcServer
 import kotlinx.rpc.descriptor.serviceDescriptorOf
 import kotlinx.rpc.grpc.annotations.Grpc
@@ -65,9 +66,13 @@ public class GrpcServer internal constructor(
         return grpc.delegate.definitionFor(service)
     }
 
+    private val buildLock = atomic(false)
+
     internal fun build() {
-        internalServer = Server(serverBuilder)
-        isBuilt = true
+        if (buildLock.compareAndSet(expect = false, update = true)) {
+            internalServer = Server(serverBuilder)
+            isBuilt = true
+        }
     }
 
     override val isShutdown: Boolean
diff --git a/grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.kt b/grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.kt
@@ -68,7 +68,7 @@ public interface ManagedChannel {
  */
 public expect abstract class ManagedChannelBuilder<T : ManagedChannelBuilder<T>>
 
-internal expect fun ManagedChannelBuilder(name: String, port: Int): ManagedChannelBuilder<*>
+internal expect fun ManagedChannelBuilder(hostname: String, port: Int): ManagedChannelBuilder<*>
 internal expect fun ManagedChannelBuilder(target: String): ManagedChannelBuilder<*>
 
 internal expect fun ManagedChannelBuilder<*>.buildChannel(): ManagedChannel
diff --git a/grpc/grpc-core/src/jsMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.js.kt b/grpc/grpc-core/src/jsMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.js.kt
@@ -20,7 +20,7 @@ internal actual fun ManagedChannelBuilder<*>.buildChannel(): ManagedChannel {
     error("JS target is not supported in gRPC")
 }
 
-internal actual fun ManagedChannelBuilder(name: String, port: Int): ManagedChannelBuilder<*> {
+internal actual fun ManagedChannelBuilder(hostname: String, port: Int): ManagedChannelBuilder<*> {
     error("JS target is not supported in gRPC")
 }
 
diff --git a/grpc/grpc-core/src/jvmMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.jvm.kt b/grpc/grpc-core/src/jvmMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.jvm.kt
@@ -24,8 +24,8 @@ internal actual fun ManagedChannelBuilder<*>.buildChannel(): ManagedChannel {
     return build().toKotlin()
 }
 
-internal actual fun ManagedChannelBuilder(name: String, port: Int): ManagedChannelBuilder<*> {
-    return io.grpc.ManagedChannelBuilder.forAddress(name, port)
+internal actual fun ManagedChannelBuilder(hostname: String, port: Int): ManagedChannelBuilder<*> {
+    return io.grpc.ManagedChannelBuilder.forAddress(hostname, port)
 }
 
 internal actual fun ManagedChannelBuilder(target: String): ManagedChannelBuilder<*> {
diff --git a/grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.native.kt b/grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.native.kt
@@ -20,7 +20,7 @@ internal actual fun ManagedChannelBuilder<*>.buildChannel(): ManagedChannel {
     error("Native target is not supported in gRPC")
 }
 
-internal actual fun ManagedChannelBuilder(name: String, port: Int): ManagedChannelBuilder<*> {
+internal actual fun ManagedChannelBuilder(hostname: String, port: Int): ManagedChannelBuilder<*> {
     error("Native target is not supported in gRPC")
 }
 
diff --git a/grpc/grpc-core/src/wasmJsMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.wasmJs.kt b/grpc/grpc-core/src/wasmJsMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.wasmJs.kt
@@ -20,7 +20,7 @@ internal actual fun ManagedChannelBuilder<*>.buildChannel(): ManagedChannel {
     error("WasmJS target is not supported in gRPC")
 }
 
-internal actual fun ManagedChannelBuilder(name: String, port: Int): ManagedChannelBuilder<*> {
+internal actual fun ManagedChannelBuilder(hostname: String, port: Int): ManagedChannelBuilder<*> {
     error("WasmJS target is not supported in gRPC")
 }
 
diff --git a/grpc/grpc-ktor-server-test/build.gradle.kts b/grpc/grpc-ktor-server-test/build.gradle.kts
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+plugins {
+    alias(libs.plugins.conventions.jvm)
+    alias(libs.plugins.kotlinx.rpc)
+    alias(libs.plugins.protobuf)
+}
+
+dependencies {
+    // for the jar dependency
+    testImplementation(kotlin("test"))
+    testImplementation(projects.grpc.grpcCore)
+    testImplementation(projects.grpc.grpcKtorServer)
+
+    testImplementation(libs.grpc.kotlin.stub)
+    testImplementation(libs.grpc.netty)
+
+    testImplementation(libs.ktor.server.core)
+    testImplementation(libs.ktor.server.test.host)
+    testRuntimeOnly(libs.logback.classic)
+}
+
+rpc {
+    grpc {
+        enabled = true
+
+        val globalRootDir: String by extra
+
+        plugin {
+            locator {
+                path = "$globalRootDir/protobuf-plugin/build/libs/protobuf-plugin-$version-all.jar"
+            }
+        }
+
+        tasksMatching { it.isTest }.all {
+            dependsOn(project(":protobuf-plugin").tasks.jar)
+        }
+    }
+}
diff --git a/grpc/grpc-ktor-server-test/gradle.properties b/grpc/grpc-ktor-server-test/gradle.properties
@@ -0,0 +1,5 @@
+#
+# Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+#
+
+kotlinx.rpc.exclude.wasmWasi=true
diff --git a/grpc/grpc-ktor-server-test/src/test/kotlin/kotlinx/rpc/grpc/ktor/server/test/TestServer.kt b/grpc/grpc-ktor-server-test/src/test/kotlin/kotlinx/rpc/grpc/ktor/server/test/TestServer.kt
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.rpc.grpc.ktor.server.test
+
+import io.ktor.server.testing.testApplication
+import kotlinx.rpc.grpc.GrpcClient
+import kotlin.test.Test
+import kotlinx.rpc.grpc.ktor.server.grpc
+import kotlinx.rpc.registerService
+import kotlinx.rpc.withService
+import kotlin.test.assertEquals
+import kotlin.time.Duration.Companion.minutes
+
+class KtorTestServiceImpl : KtorTestService {
+    override suspend fun sayHello(message: Hello): Hello {
+        return message
+    }
+}
+
+const val PORT = 8085
+
+class TestServer {
+    @Test
+    fun testPlainRequests() = testApplication {
+        application {
+            grpc(PORT) {
+                registerService<KtorTestService> { KtorTestServiceImpl() }
+            }
+        }
+
+        startApplication()
+
+        val client = GrpcClient("localhost", PORT) {
+            usePlaintext()
+        }
+
+        val response = client.withService<KtorTestService>().sayHello(Hello { message = "Hello" })
+        assertEquals("Hello", response.message, "Wrong response message")
+
+        client.shutdown()
+        client.awaitTermination(1.minutes)
+    }
+}
diff --git a/grpc/grpc-ktor-server-test/src/test/proto/ktor-test-service.proto b/grpc/grpc-ktor-server-test/src/test/proto/ktor-test-service.proto
@@ -0,0 +1,11 @@
+syntax = "proto3";
+
+package kotlinx.rpc.grpc.ktor.server.test;
+
+message Hello {
+  string message = 1;
+}
+
+service KtorTestService {
+  rpc sayHello(Hello) returns (Hello);
+}
diff --git a/grpc/grpc-ktor-server-test/src/test/resources/logback.xml b/grpc/grpc-ktor-server-test/src/test/resources/logback.xml
@@ -0,0 +1,16 @@
+<!--
+  ~ Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+  -->
+
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{YYYY-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+    <root level="trace">
+        <appender-ref ref="STDOUT"/>
+    </root>
+    <logger name="org.eclipse.jetty" level="INFO"/>
+    <logger name="io.netty" level="TRACE"/>
+</configuration>
diff --git a/grpc/grpc-ktor-server/build.gradle.kts b/grpc/grpc-ktor-server/build.gradle.kts
@@ -0,0 +1,19 @@
+/*
+ * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+plugins {
+    alias(libs.plugins.conventions.kmp)
+    alias(libs.plugins.kotlinx.rpc)
+}
+
+kotlin {
+    sourceSets {
+        commonMain {
+            dependencies {
+                api(projects.grpc.grpcCore)
+                implementation(libs.ktor.server.core)
+            }
+        }
+    }
+}
diff --git a/grpc/grpc-ktor-server/gradle.properties b/grpc/grpc-ktor-server/gradle.properties
@@ -0,0 +1,5 @@
+#
+# Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+#
+
+kotlinx.rpc.exclude.wasmWasi=true
diff --git a/grpc/grpc-ktor-server/src/commonMain/kotlin/kotlinx/rpc/grpc/ktor/server/Server.kt b/grpc/grpc-ktor-server/src/commonMain/kotlin/kotlinx/rpc/grpc/ktor/server/Server.kt
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.rpc.grpc.ktor.server
+
+import io.ktor.server.application.Application
+import io.ktor.server.application.ApplicationStopped
+import io.ktor.server.application.ApplicationStopping
+import io.ktor.server.application.log
+import io.ktor.server.config.getAs
+import io.ktor.util.AttributeKey
+import kotlinx.rpc.RpcServer
+import kotlinx.rpc.grpc.GrpcServer
+import kotlinx.rpc.grpc.ServerBuilder
+
+@Suppress("ConstPropertyName")
+public object GrpcConfigKeys {
+    public const val grpcHostPortPath: String = "ktor.deployment.grpcPort"
+}
+
+/**
+ * Key used to store and retrieve the [GrpcServer] instance within the application's attributes.
+ */
+public val GrpcServerKey: AttributeKey<GrpcServer> = AttributeKey<GrpcServer>("GrpcServerPluginAttributesKey")
+
+/**
+ * Configures and starts a gRPC server within the Ktor application.
+ * This function integrates with the Ktor lifecycle and manages the lifecycle of the gRPC server
+ * by subscribing to [ApplicationStopping] and [ApplicationStopped] events.
+ * It ensures that a gRPC server is properly initialized, started, and shutdown when the application stops.
+ *
+ * Usage:
+ * ```kotlin
+ * fun Application.module() {
+ *     grpc(port = PORT, configure= { /* ... */ }) {
+ *         registerService<MyService> { MyServiceImpl() }
+ *     }
+ * }
+ * ```
+ *
+ * @param port The port on which the gRPC server will listen for incoming connections.
+ * Defaults to the value specified in the `ktor.deployment.grpcPort` configuration, or 8001 if not configured.
+ * @param configure Allows additional configuration of the gRPC server using a platform-specific [ServerBuilder].
+ * @param builder A block used to define and register gRPC services for the gRPC server.
+ * @return The instance of the initialized and running [GrpcServer].
+ * @throws IllegalStateException if a gRPC server is already installed or the specified port conflicts with
+ * an existing HTTP/HTTPS server port.
+ */
+public fun Application.grpc(
+    port: Int = environment.config.propertyOrNull(GrpcConfigKeys.grpcHostPortPath)?.getAs<Int>() ?: 8001,
+    configure: ServerBuilder<*>.() -> Unit = {},
+    builder: RpcServer.() -> Unit,
+): GrpcServer {
+    if (attributes.contains(GrpcServerKey)) {
+        error("gRPC Server is already installed, second call to grpc() is not allowed")
+    }
+
+    var newServer = false
+    val server = attributes.computeIfAbsent(GrpcServerKey) {
+        newServer = true
+        GrpcServer(port, configure, builder)
+    }
+
+    if (!newServer) {
+        error("A race detected while installing gRPC Server, second call to grpc() is not allowed")
+    }
+
+    server.start()
+    log.debug("Started gRPC server on port $port")
+
+    val stoppingHandle = monitor.subscribe(ApplicationStopping) {
+        log.debug("Stopping gRPC server")
+        attributes.getOrNull(GrpcServerKey)?.shutdown()
+    }
+
+    monitor.subscribe(ApplicationStopped) {
+        log.debug("gRPC server complete shutdown")
+        attributes.getOrNull(GrpcServerKey)?.shutdownNow()
+
+        stoppingHandle.dispose()
+    }
+
+    return server
+}
diff --git a/settings.gradle.kts b/settings.gradle.kts
@@ -31,6 +31,9 @@ includePublic(":protobuf-plugin")
 
 include(":grpc")
 includePublic(":grpc:grpc-core")
+includePublic(":grpc:grpc-ktor-server")
+// temporary module until KMP project structure support in Protobuf plugin
+include(":grpc:grpc-ktor-server-test")
 
 includePublic(":bom")
 

Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,7 @@ internal actual fun ManagedChannelBuilder<*>.buildChannel(): ManagedChannel {`
`20`	`20`	`error("JS target is not supported in gRPC")`
`21`	`21`	`}`
`22`	`22`
`23`		`-internal actual fun ManagedChannelBuilder(name: String, port: Int): ManagedChannelBuilder<*> {`
	`23`	`+internal actual fun ManagedChannelBuilder(hostname: String, port: Int): ManagedChannelBuilder<*> {`
`24`	`24`	`error("JS target is not supported in gRPC")`
`25`	`25`	`}`
`26`	`26`
Original file line number	Diff line number	Diff line change
`@@ -24,8 +24,8 @@ internal actual fun ManagedChannelBuilder<*>.buildChannel(): ManagedChannel {`
`24`	`24`	`return build().toKotlin()`
`25`	`25`	`}`
`26`	`26`
`27`		`-internal actual fun ManagedChannelBuilder(name: String, port: Int): ManagedChannelBuilder<*> {`
`28`		`- return io.grpc.ManagedChannelBuilder.forAddress(name, port)`
	`27`	`+internal actual fun ManagedChannelBuilder(hostname: String, port: Int): ManagedChannelBuilder<*> {`
	`28`	`+ return io.grpc.ManagedChannelBuilder.forAddress(hostname, port)`
`29`	`29`	`}`
`30`	`30`
`31`	`31`	`internal actual fun ManagedChannelBuilder(target: String): ManagedChannelBuilder<*> {`
-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +#
 +# Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
 +#
++
 +kotlinx.rpc.exclude.wasmWasi=true