grpc-native: Implement bridge to common

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

Author: Jozott00

cinterop-c/include/grpcpp_c.h

@@ -77,9 +77,6 @@ typedef struct grpc_channel_credentials grpc_channel_credentials_t;
 bool kgrpc_iomgr_run_in_background();
 
 
-/////// UTILS ///////
-
-
 #ifdef __cplusplus
     }
 #endif

grpc/grpc-core/build.gradle.kts

@@ -61,6 +61,13 @@ kotlin {
                 implementation(libs.grpc.netty)
             }
         }
+
+        nativeMain {
+            dependencies {
+                // required for status.proto
+                implementation(projects.protobuf.protobufCore)
+            }
+        }
     }
 
     configureCLibCInterop(project, ":grpcpp_c_static") { cinteropCLib ->

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

@@ -8,6 +8,15 @@ import kotlinx.rpc.grpc.GrpcTrailers
 import kotlinx.rpc.grpc.Status
 import kotlinx.rpc.internal.utils.InternalRpcApi
 
+/**
+ * This class represents a client-side call to a server.
+ * It provides the interface of the gRPC-Java ClientCall class; however, semantics are slightly different
+ * on JVM and Native platforms.
+ *
+ * Callback execution:
+ * - On JVM it is guaranteed that callbacks aren't executed concurrently.
+ * - On Native, it is only guaranteed that `onClose` is called after all other callbacks finished.
+ */
 @InternalRpcApi
 public expect abstract class ClientCall<Request, Response> {
     @InternalRpcApi

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

@@ -37,6 +37,12 @@ public enum class MethodType {
     UNKNOWN,
 }
 
+/**
+ * Creates a new [MethodDescriptor] instance.
+ *
+ * @param fullMethodName the full name of the method, consisting of the service name followed by a forward slash
+ *      and the method name. It does not include a leading slash.
+ */
 @InternalRpcApi
 public expect fun <Request, Response> methodDescriptor(
     fullMethodName: String,

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

@@ -4,23 +4,13 @@
 
 package kotlinx.rpc.grpc.internal
 
-import kotlinx.coroutines.CancellationException
-import kotlinx.coroutines.CoroutineName
-import kotlinx.coroutines.NonCancellable
-import kotlinx.coroutines.cancel
+import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.onFailure
-import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.single
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.withContext
-import kotlinx.rpc.grpc.GrpcTrailers
-import kotlinx.rpc.grpc.Status
-import kotlinx.rpc.grpc.StatusCode
-import kotlinx.rpc.grpc.StatusException
-import kotlinx.rpc.grpc.code
+import kotlinx.rpc.grpc.*
 import kotlinx.rpc.internal.utils.InternalRpcApi
 
 // heavily inspired by
@@ -249,7 +239,7 @@ internal fun <T> Flow<T>.singleOrStatusFlow(
 
 internal suspend fun <T> Flow<T>.singleOrStatus(
     expected: String,
-    descriptor: Any
+    descriptor: Any,
 ): T = singleOrStatusFlow(expected, descriptor).single()
 
 internal class Ready(private val isReallyReady: () -> Boolean) {

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

@@ -0,0 +1,88 @@
+/*
+ * 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.test
+
+import grpc.examples.echo.*
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.test.runTest
+import kotlinx.rpc.grpc.ManagedChannelBuilder
+import kotlinx.rpc.grpc.buildChannel
+import kotlinx.rpc.grpc.internal.*
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class CancellationClientTest {
+
+    @Test
+    fun unaryEchoTest() = runTest(
+        methodName = "UnaryEcho",
+        type = MethodType.UNARY,
+    ) { channel, descriptor ->
+        val response = unaryRpc(channel, descriptor, EchoRequest { message = "Eccchhooo" })
+        assertEquals("Eccchhooo", response.message)
+    }
+
+    @Test
+    fun serverStreamingEchoTest() = runTest(
+        methodName = "ServerStreamingEcho",
+        type = MethodType.SERVER_STREAMING,
+    ) { channel, descriptor ->
+        val response = serverStreamingRpc(channel, descriptor, EchoRequest { message = "Eccchhooo" })
+        var i = 0
+        response.collect {
+            println("Received: ${i++}")
+            assertEquals("Eccchhooo", it.message)
+        }
+    }
+
+    @Test
+    fun clientStreamingTest() = runTest(
+        methodName = "ServerStreamingEcho",
+        type = MethodType.CLIENT_STREAMING,
+    ) { channel, descriptor ->
+        val response = clientStreamingRpc(channel, descriptor, flow {
+            repeat(5) {
+                delay(100)
+                println("Sending: ${it + 1}")
+                emit(EchoRequest { message = "Eccchhooo" })
+            }
+        })
+        val expected = "Eccchhooo,Eccchhooo,Eccchhooo,Eccchhooo,Eccchhooo"
+        assertEquals(expected, response.message)
+    }
+
+    fun runTest(
+        methodName: String,
+        type: MethodType,
+        block: suspend (GrpcChannel, MethodDescriptor<EchoRequest, EchoResponse>) -> Unit,
+    ) = runTest {
+        val channel = ManagedChannelBuilder("localhost:50051")
+            .usePlaintext()
+            .buildChannel()
+
+        val methodDescriptor = methodDescriptor(
+            fullMethodName = "grpc.examples.echo.Echo/$methodName",
+            requestCodec = EchoRequestInternal.CODEC,
+            responseCodec = EchoResponseInternal.CODEC,
+            type = type,
+            schemaDescriptor = Unit,
+            idempotent = true,
+            safe = true,
+            sampledToLocalTracing = true,
+        )
+
+        try {
+            block(channel.platformApi, methodDescriptor)
+        } finally {
+            channel.shutdown()
+            channel.awaitTermination()
+        }
+
+
+    }
+
+
+}

grpc/grpc-core/src/commonTest/proto/echo_grpc.proto

@@ -0,0 +1,25 @@
+syntax = "proto3";
+
+package grpc.examples.echo;
+
+// EchoRequest is the request for echo.
+message EchoRequest {
+  string message = 1;
+}
+
+// EchoResponse is the response for echo.
+message EchoResponse {
+  string message = 1;
+}
+
+// Echo is the echo service.
+service Echo {
+  // UnaryEcho is unary echo.
+  rpc UnaryEcho(EchoRequest) returns (EchoResponse) {}
+  // ServerStreamingEcho is server side streaming.
+  rpc ServerStreamingEcho(EchoRequest) returns (stream EchoResponse) {}
+  // ClientStreamingEcho is client side streaming.
+  rpc ClientStreamingEcho(stream EchoRequest) returns (EchoResponse) {}
+  // BidirectionalStreamingEcho is bidi streaming.
+  rpc BidirectionalStreamingEcho(stream EchoRequest) returns (stream EchoResponse) {}
+}

grpc/grpc-core/src/nativeInterop/cinterop/libgrpcpp_c.def

@@ -1,6 +1,9 @@
-headers = grpcpp_c.h grpc/grpc.h grpc/credentials.h grpc/byte_buffer_reader.h
+headers = grpcpp_c.h grpc/grpc.h grpc/credentials.h grpc/byte_buffer_reader.h \
+    grpc/support/alloc.h
+
 headerFilter= grpcpp_c.h grpc/slice.h grpc/byte_buffer.h grpc/grpc.h \
-    grpc/impl/grpc_types.h grpc/credentials.h grpc/support/time.h grpc/byte_buffer_reader.h
+    grpc/impl/grpc_types.h grpc/credentials.h grpc/support/time.h grpc/byte_buffer_reader.h \
+    grpc/support/alloc.h
 
 noStringConversion = grpc_slice_from_copied_buffer my_grpc_slice_from_copied_buffer
 strictEnums = grpc_status_code grpc_connectivity_state grpc_call_error