grpc: Add client interceptor support

Signed-off-by: Johannes Zottele <[email protected]>

Author: Jozott00

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/ClientInterceptor.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
+
+import kotlinx.coroutines.flow.Flow
+import kotlinx.rpc.grpc.internal.GrpcCallOptions
+import kotlinx.rpc.grpc.internal.MethodDescriptor
+
+/**
+ * Represents a client call scope within a coroutine context, providing access to properties and
+ * functions required to manage the lifecycle and behavior of a client-side remote procedure call
+ * (RPC) in a coroutine-based environment.
+ *
+ * @param Request the type of the request message sent to the gRPC server.
+ * @param Response the type of the response message received from the gRPC server.
+ */
+public interface ClientCallScope<Request, Response> {
+    public val method: MethodDescriptor<Request, Response>
+    public val metadata: GrpcTrailers
+    public val callOptions: GrpcCallOptions
+    public fun onHeaders(block: (GrpcTrailers) -> Unit)
+    public fun onClose(block: (Status, GrpcTrailers) -> Unit)
+    public fun cancel(message: String, cause: Throwable? = null)
+    public fun proceed(request: Flow<Request>): Flow<Response>
+}
+
+public interface ClientInterceptor {
+
+    /**
+     * Intercepts and transforms the flow of requests and responses in a client call.
+     * An interceptor can throw an exception at any time to cancel the call.
+     *
+     * @param scope The scope of the client call, providing context and methods for managing
+     * the call lifecycle and metadata.
+     * @param request A flow of requests to be sent to the server.
+     * @return A flow of responses received from the server.
+     */
+    public fun <Request, Response> intercept(
+        scope: ClientCallScope<Request, Response>,
+        request: Flow<Request>,
+    ): Flow<Response>
+
+}

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/GrpcClient.kt

@@ -32,8 +32,9 @@ private typealias RequestClient = Any
  * @field channel The [ManagedChannel] used to communicate with remote gRPC services.
  */
 public class GrpcClient internal constructor(
-    private val channel: ManagedChannel,
+    internal val channel: ManagedChannel,
     messageCodecResolver: MessageCodecResolver = EmptyMessageCodecResolver,
+    internal val interceptors: List<ClientInterceptor>,
 ) : RpcClient {
     private val delegates = RpcInternalConcurrentHashMap<String, GrpcServiceDelegate>()
     private val messageCodecResolver = messageCodecResolver + ThrowingMessageCodecResolver
@@ -58,15 +59,13 @@ public class GrpcClient internal constructor(
 
         return when (methodDescriptor.type) {
             MethodType.UNARY -> unaryRpc(
-                channel = channel.platformApi,
                 descriptor = methodDescriptor,
                 request = request,
                 callOptions = callOptions,
                 trailers = trailers,
             )
 
             MethodType.CLIENT_STREAMING -> @Suppress("UNCHECKED_CAST") clientStreamingRpc(
-                channel = channel.platformApi,
                 descriptor = methodDescriptor,
                 requests = request as Flow<RequestClient>,
                 callOptions = callOptions,
@@ -83,15 +82,13 @@ public class GrpcClient internal constructor(
 
         when (methodDescriptor.type) {
             MethodType.SERVER_STREAMING -> serverStreamingRpc(
-                channel = channel.platformApi,
                 descriptor = methodDescriptor,
                 request = request,
                 callOptions = callOptions,
                 trailers = trailers,
             )
 
             MethodType.BIDI_STREAMING -> @Suppress("UNCHECKED_CAST") bidirectionalStreamingRpc(
-                channel = channel.platformApi,
                 descriptor = methodDescriptor,
                 requests = request as Flow<RequestClient>,
                 callOptions = callOptions,
@@ -131,23 +128,57 @@ public class GrpcClient internal constructor(
 public fun GrpcClient(
     hostname: String,
     port: Int,
-    credentials: ClientCredentials? = null,
-    messageCodecResolver: MessageCodecResolver = EmptyMessageCodecResolver,
-    configure: ManagedChannelBuilder<*>.() -> Unit = {},
+    configure: GrpcClientConfiguration.() -> Unit = {},
 ): GrpcClient {
-    val channel = ManagedChannelBuilder(hostname, port, credentials).apply(configure).buildChannel()
-    return GrpcClient(channel, messageCodecResolver)
+    val config = GrpcClientConfiguration().apply(configure)
+    return GrpcClient(ManagedChannelBuilder(hostname, port, config.credentials), config)
 }
 
 /**
  * Constructor function for the [GrpcClient] class.
  */
 public fun GrpcClient(
     target: String,
-    credentials: ClientCredentials? = null,
-    messageCodecResolver: MessageCodecResolver = EmptyMessageCodecResolver,
-    configure: ManagedChannelBuilder<*>.() -> Unit = {},
+    configure: GrpcClientConfiguration.() -> Unit = {},
+): GrpcClient {
+    val config = GrpcClientConfiguration().apply(configure)
+    return GrpcClient(ManagedChannelBuilder(target, config.credentials), config)
+}
+
+private fun GrpcClient(
+    builder: ManagedChannelBuilder<*>,
+    config: GrpcClientConfiguration,
 ): GrpcClient {
-    val channel = ManagedChannelBuilder(target, credentials).apply(configure).buildChannel()
-    return GrpcClient(channel, messageCodecResolver)
+    val channel = builder.apply {
+        config.overrideAuthority?.let { overrideAuthority(it) }
+    }.buildChannel()
+    return GrpcClient(channel, config.messageCodecResolver, config.interceptors)
 }
+
+public class GrpcClientConfiguration internal constructor() {
+    internal var messageCodecResolver: MessageCodecResolver = EmptyMessageCodecResolver
+    internal var credentials: ClientCredentials? = null
+    internal var overrideAuthority: String? = null
+    internal val interceptors: MutableList<ClientInterceptor> = mutableListOf()
+
+    public fun usePlaintext() {
+        credentials = createInsecureClientCredentials()
+    }
+
+    public fun useCredentials(credentials: ClientCredentials) {
+        this@GrpcClientConfiguration.credentials = credentials
+    }
+
+    public fun overrideAuthority(authority: String) {
+        overrideAuthority = authority
+    }
+
+    public fun useMessageCodecResolver(messageCodecResolver: MessageCodecResolver) {
+        this.messageCodecResolver = messageCodecResolver
+    }
+
+    public fun intercept(vararg interceptors: ClientInterceptor) {
+        this.interceptors.addAll(interceptors)
+    }
+
+}

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.kt

@@ -68,6 +68,8 @@ public interface ManagedChannel {
  * Builder class for [ManagedChannel].
  */
 public expect abstract class ManagedChannelBuilder<T : ManagedChannelBuilder<T>> {
+
+    // TODO: Not used anymore
     public fun usePlaintext(): T
 
     public abstract fun overrideAuthority(authority: String): T

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/credentials.kt

@@ -10,6 +10,8 @@ public expect abstract class ServerCredentials
 public expect class InsecureClientCredentials : ClientCredentials
 public expect class InsecureServerCredentials : ServerCredentials
 
+internal expect fun createInsecureClientCredentials(): ClientCredentials
+
 public expect class TlsClientCredentials : ClientCredentials
 public expect class TlsServerCredentials : ServerCredentials