grpc-native: Address PR comments

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

Author: Jozott00

cinterop-c/include/kgrpc.h

@@ -57,6 +57,7 @@ bool kgrpc_iomgr_run_in_background();
  *
  * Wraps the internal C++ API `Server::SetRegisteredMethodAllocator()` to enable
  * callback-driven method dispatch via the Core C API.
+ * If the C++ API is exposed to the C API, this can be removed (https://github.com/grpc/grpc/issues/40632).
  *
  * When the gRPC Core needs to accept a new call for the specified method, it invokes:
  *   kgrpc_registered_call_allocation alloc = allocator();

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

@@ -6,6 +6,7 @@ package kotlinx.rpc.grpc.test
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.test.runTest
 import kotlinx.coroutines.withTimeout
 import kotlinx.rpc.grpc.GrpcServer
 import kotlinx.rpc.grpc.GrpcTrailers
@@ -270,21 +271,20 @@ class GreeterServiceImpl : GreeterService {
      * Run this on JVM before executing tests.
      */
     @Test
-    fun runServer() {
-        runBlocking {
-            val server = GrpcServer(
-                port = PORT,
-                builder = { registerService<GreeterService> { GreeterServiceImpl() } }
-            )
+    fun runServer() = runTest {
+        val server = GrpcServer(
+            port = PORT,
+            builder = { registerService<GreeterService> { GreeterServiceImpl() } }
+        )
+
+        try {
+            server.start()
+            println("Server started")
+            server.awaitTermination()
+        } finally {
+            server.shutdown()
+            server.awaitTermination()
 
-            try {
-                server.start()
-                println("Server started")
-                server.awaitTermination()
-            } finally {
-                server.shutdown()
-                server.awaitTermination()
-            }
         }
     }
 

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/internal/NativeServer.kt

@@ -70,7 +70,8 @@ internal class NativeServer(
 
     private val cq = CompletionQueue()
 
-    val raw: CPointer<grpc_server> = grpc_server_create(null, null)!!
+    val raw: CPointer<grpc_server> = grpc_server_create(null, null)
+        ?: error("Failed to create server")
 
     // holds all stable references to MethodAllocationCtx objects.
     // the stable references must eventually be disposed.

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/internal/NativeServerCall.kt

@@ -59,12 +59,12 @@ internal class NativeServerCall<Request, Response>(
         grpc_call_unref(it)
     }
 
-    private var listener = DeferredCallListener<Request>()
+    private val listener = DeferredCallListener<Request>()
     private var methodDescriptor: MethodDescriptor<Request, Response>? = null
-    private var callbackMutex = ReentrantLock()
+    private val callbackMutex = ReentrantLock()
     private var initialized = false
     private var cancelled = false
-    private var finalized = atomic(false)
+    private val finalized = atomic(false)
 
     // Tracks whether at least one request message has been received on this call.
     private var receivedFirstMessage = false
@@ -190,7 +190,7 @@ internal class NativeServerCall<Request, Response>(
 
     override fun request(numMessages: Int) {
         check(initialized) { internalError("Call not initialized") }
-        // TODO: Remove the num constraint
+        // TODO: Remove the num constraint (KRPC-213)
         require(numMessages == 1) { internalError("numMessages must be 1") }
         val methodDescriptor = checkNotNull(methodDescriptor) { internalError("Method descriptor not set") }
 
@@ -320,34 +320,34 @@ private class DeferredCallListener<T> : ServerCall.Listener<T>() {
     @Volatile
     private var delegate: ServerCall.Listener<T>? = null
     private val mutex = ReentrantLock()
-    private val q = ArrayDeque<(ServerCall.Listener<T>) -> Unit>()
+    private val queue = ArrayDeque<(ServerCall.Listener<T>) -> Unit>()
 
     fun setDelegate(d: ServerCall.Listener<T>) {
         mutex.withLock {
             if (delegate != null) return
             delegate = d
         }
         // drain the queue
-        q.forEach { it(d) }
-        q.clear()
+        queue.forEach { it(d) }
+        queue.clear()
     }
 
-    private inline fun deliver(crossinline f: (ServerCall.Listener<T>) -> Unit) {
-        val d = delegate
-        if (d != null) {
+    private inline fun deliver(crossinline invokeListener: (ServerCall.Listener<T>) -> Unit) {
+        val currentDelegate = delegate
+        if (currentDelegate != null) {
             // fast path (delegate is already set)
-            f(d); return
+            invokeListener(currentDelegate); return
         }
         // slow path: re-check under lock
-        val dd = mutex.withLock {
+        val safeCurrentDelegate = mutex.withLock {
             val cur = delegate
             if (cur == null) {
-                q.addLast { f(it) }
+                queue.addLast { invokeListener(it) }
                 null
             } else cur
         }
         // if the delegate was already set, call it
-        if (dd != null) f(dd)
+        if (safeCurrentDelegate != null) invokeListener(safeCurrentDelegate)
     }
 
     override fun onMessage(message: T) = deliver { it.onMessage(message) }