grpc: Address PR comments

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

Author: Jozott00

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

@@ -77,6 +77,9 @@ public interface ClientCallScope<Request, Response> {
     /**
      * Cancel the call locally, providing a human-readable [message] and an optional [cause].
      * This method won't return and abort all further processing.
+     *
+     * We made cancel throw a [StatusException] instead of returning, so control flow is explicit and
+     * race conditions between interceptors and the transport layer are avoided.
      */
     public fun cancel(message: String, cause: Throwable? = null): Nothing
 

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

@@ -68,10 +68,6 @@ 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/ServerInterceptor.kt

@@ -51,7 +51,7 @@ public interface ServerCallScope<Request, Response> {
      * It can be used by the interceptor to provide call-scoped information about
      * the current call, such as the identity of the caller or the current authentication state.
      */
-    public val grpcContext: GrpcContext
+    public val context: GrpcContext
 
     /**
      * Register a callback invoked when the call is closed (successfully or exceptionally).
@@ -64,6 +64,9 @@ public interface ServerCallScope<Request, Response> {
      *
      * This method does not return (declared as [Nothing]). After calling it, no further messages will be processed
      * or sent. Prefer setting [responseHeaders]/[responseTrailers] before closing if you need to include metadata.
+     *
+     * We made close throw a [StatusException] instead of returning, so control flow is explicit and race conditions
+     * between interceptors and the service implementation are avoided.
      */
     public fun close(status: Status, trailers: GrpcMetadata = GrpcMetadata()): Nothing
 
@@ -83,8 +86,22 @@ public interface ServerCallScope<Request, Response> {
     /**
      * Convenience for flow builders: proceeds with [request] and emits the resulting response elements into this
      * [FlowCollector]. Useful inside `flow {}` blocks within interceptors.
+     *
+     * ```
+     * val myAuthInterceptor = object : ServerInterceptor {
+     *     override fun <Request, Response> ServerCallScope<Request, Response>.intercept(request: Flow<Request>): Flow<Response> =
+     *         flow {
+     *             val authorized = mySuspendAuth(requestHeaders)
+     *             if (!authorized) {
+     *                 close(Status(StatusCode.UNAUTHENTICATED, "Not authorized"))
+     *             }
+     *
+     *             proceedUnmodified(request)
+     *         }
+     *      }
+     * ```
      */
-    public suspend fun FlowCollector<Response>.proceedFlow(request: Flow<Request>) {
+    public suspend fun FlowCollector<Response>.proceedUnmodified(request: Flow<Request>) {
         proceed(request).collect {
             emit(it)
         }
@@ -115,7 +132,7 @@ public interface ServerInterceptor {
      * - Read [ServerCallScope.requestHeaders] and populate [ServerCallScope.responseHeaders]/[ServerCallScope.responseTrailers].
      * - Register [ServerCallScope.onClose] callbacks.
      * - Transform the [request] flow or wrap the resulting response flow.
-     * - Append information to the [ServerCallScope.grpcContext].
+     * - Append information to the [ServerCallScope.context].
      *
      * IMPORTANT: You must eventually call [ServerCallScope.proceed] to actually invoke the service logic and produce
      * the response [Flow]. If [ServerCallScope.proceed] is omitted, the call will never reach the service.

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/internal/suspendServerCalls.kt

@@ -164,7 +164,7 @@ private fun <Request, Response> CoroutineScope.serverCallListenerImpl(
         implementation = implementation,
         requestHeaders = requestHeaders,
         serverCall = handler,
-        grpcContext = context.grpcContext,
+        context = context.grpcContext,
     )
 
     val rpcJob = launch(context) {
@@ -275,7 +275,7 @@ private class ServerCallScopeImpl<Request, Response>(
     val implementation: (Flow<Request>) -> Flow<Response>,
     override val requestHeaders: GrpcMetadata,
     val serverCall: ServerCall<Request, Response>,
-    override val grpcContext: GrpcContext,
+    override val context: GrpcContext,
 ) : ServerCallScope<Request, Response> {
 
     override val responseHeaders: GrpcMetadata = GrpcMetadata()

grpc/grpc-core/src/commonTest/kotlin/kotlinx/rpc/grpc/test/CoreClientTest.kt

@@ -15,6 +15,7 @@ import kotlinx.rpc.grpc.ManagedChannelBuilder
 import kotlinx.rpc.grpc.Status
 import kotlinx.rpc.grpc.StatusCode
 import kotlinx.rpc.grpc.buildChannel
+import kotlinx.rpc.grpc.createInsecureClientCredentials
 import kotlinx.rpc.grpc.internal.ClientCall
 import kotlinx.rpc.grpc.internal.GrpcDefaultCallOptions
 import kotlinx.rpc.grpc.internal.MethodDescriptor
@@ -52,9 +53,10 @@ class GrpcCoreClientTest {
     private fun ManagedChannel.newHelloCall(fullName: String = "kotlinx.rpc.grpc.test.GreeterService/SayHello"): ClientCall<HelloRequest, HelloReply> =
         platformApi.newCall(descriptorFor(fullName), GrpcDefaultCallOptions)
 
-    private fun createChannel(): ManagedChannel = ManagedChannelBuilder("localhost:$PORT")
-        .usePlaintext()
-        .buildChannel()
+    private fun createChannel(): ManagedChannel = ManagedChannelBuilder(
+        target = "localhost:$PORT",
+        credentials = createInsecureClientCredentials()
+    ).buildChannel()
 
 
     private fun helloReq(timeout: UInt = 0u): HelloRequest = HelloRequest {

grpc/grpc-core/src/commonTest/kotlin/kotlinx/rpc/grpc/test/proto/ServerInterceptorTest.kt

@@ -5,8 +5,10 @@
 package kotlinx.rpc.grpc.test.proto
 
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.toList
 import kotlinx.rpc.RpcServer
 import kotlinx.rpc.grpc.GrpcClient
 import kotlinx.rpc.grpc.GrpcMetadata
@@ -184,13 +186,61 @@ class ServerInterceptorTest : GrpcProtoTest() {
     @Test
     fun `proceedFlow - should succeed on client`() {
         val interceptor = interceptor {
-            kotlinx.coroutines.flow.flow {
-                proceedFlow(it)
+            flow {
+                proceedUnmodified(it)
             }
         }
         runGrpcTest(serverInterceptors = interceptor, test = ::unaryCall)
     }
 
+    @Test
+    fun `test exact order of interceptor execution`() {
+        val order = mutableListOf<Int>()
+        val interceptor1 = interceptor { request ->
+            flow {
+                order.add(1)
+                var i1 = 0
+                val ids = listOf(3, 7)
+                val req = request.map { order.add(ids[i1++]); it }
+
+                var i2 = 0
+                val respIds = listOf(6, 10)
+                proceed(req).collect {
+                    order.add(respIds[i2++])
+                    emit(it)
+                }
+
+                order.add(12)
+            }
+        }
+
+        val interceptor2 = interceptor { request ->
+            flow {
+                order.add(2)
+                var i1 = 0
+                val reqIds = listOf(4, 8)
+                val req = request.map { order.add(reqIds[i1++]); it }
+
+                var i2 = 0
+                val respIds = listOf(5, 9)
+                proceed(req).collect {
+                    order.add(respIds[i2++])
+                    emit(it)
+                }
+
+                order.add(11)
+            }
+        }
+        val both = interceptor1 + interceptor2
+
+        runGrpcTest(serverInterceptors = both) { bidiStream(it, 2) }
+
+        assertEquals(
+            listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12),
+            order
+        )
+    }
+
     @Test
     fun `method descriptor - full method name is exposed`() {
         var methodName: String? = null
@@ -207,6 +257,20 @@ class ServerInterceptorTest : GrpcProtoTest() {
         val response = service.UnaryEcho(EchoRequest { message = "Hello" })
         assertEquals("Hello", response.message)
     }
+
+    private suspend fun bidiStream(grpcClient: GrpcClient, count: Int = 5) {
+        val service = grpcClient.withService<EchoService>()
+        val responses = service.BidirectionalStreamingEcho(flow {
+            repeat(count) {
+                emit(EchoRequest { message = "Echo-$it" })
+            }
+        }).toList()
+        assertEquals(count, responses.size)
+        repeat(count) {
+            assertEquals("Echo-$it", responses[it].message)
+        }
+    }
+
 }
 
 

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.native.kt

@@ -19,10 +19,6 @@ public actual abstract class ManagedChannelPlatform : GrpcChannel()
  * Builder class for [ManagedChannel].
  */
 public actual abstract class ManagedChannelBuilder<T : ManagedChannelBuilder<T>> {
-    public actual open fun usePlaintext(): T {
-        error("Builder does not support usePlaintext()")
-    }
-
     public actual abstract fun overrideAuthority(authority: String): T
 }
 
@@ -33,11 +29,6 @@ internal class NativeManagedChannelBuilder(
 
     private var authority: String? = null
 
-    override fun usePlaintext(): NativeManagedChannelBuilder {
-        credentials = lazy { createInsecureClientCredentials() }
-        return this
-    }
-
     override fun overrideAuthority(authority: String): NativeManagedChannelBuilder {
         this.authority = authority
         return this