grpc-native: Code cleanup

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

Author: Jozott00

cinterop-c/MODULE.bazel.lock

@@ -16,10 +16,12 @@ import kotlinx.rpc.internal.utils.InternalRpcApi
 public expect class ServerServiceDefinition {
     public fun getServiceDescriptor(): ServiceDescriptor
     public fun getMethods(): Collection<ServerMethodDefinition<*, *>>
+
+    public fun getMethod(methodName: String): ServerMethodDefinition<*, *>?
 }
 
 @InternalRpcApi
 public expect fun serverServiceDefinition(
     serviceDescriptor: ServiceDescriptor,
-    methods: Collection<ServerMethodDefinition<*, *>>
+    methods: Collection<ServerMethodDefinition<*, *>>,
 ): ServerServiceDefinition

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

@@ -6,17 +6,66 @@
 
 package kotlinx.rpc.grpc
 
+import kotlinx.rpc.grpc.internal.MethodDescriptor
+import kotlinx.rpc.grpc.internal.ServerMethodDefinition
+import kotlinx.rpc.internal.utils.map.RpcInternalConcurrentHashMap
+
 /**
  * Registry of services and their methods used by servers to dispatching incoming calls.
  */
-public actual abstract class HandlerRegistry
+public actual abstract class HandlerRegistry {
+    /**
+     * Returns the [ServerServiceDefinition]s provided by the registry, or an empty list if not
+     * supported by the implementation.
+     */
+    public abstract fun getServices(): List<ServerServiceDefinition>
+
+    /**
+     * Lookup a [ServerMethodDefinition] by its fully-qualified name.
+     *
+     * @param methodName to lookup [ServerMethodDefinition] for.
+     * @param authority the authority for the desired method (to do virtual hosting). If `null`
+     * the first matching method will be returned.
+     * @return the resolved method or `null` if no method for that name exists.
+     */
+    public abstract fun lookupMethod(
+        methodName: String, authority: String?,
+    ): ServerMethodDefinition<*, *>?
+
+    /**
+     * Lookup a [ServerMethodDefinition] by its fully-qualified name.
+     *
+     * @param methodName to lookup [ServerMethodDefinition] for.
+     * @return the resolved method or `null` if no method for that name exists.
+     */
+    public fun lookupMethod(methodName: String): ServerMethodDefinition<*, *>? {
+        return lookupMethod(methodName, null)
+    }
+
+}
 
 internal actual class MutableHandlerRegistry : HandlerRegistry() {
+
+    private val services = RpcInternalConcurrentHashMap<String, ServerServiceDefinition>()
+
     actual fun addService(service: ServerServiceDefinition): ServerServiceDefinition? {
-        error("Native target is not supported in gRPC")
+        return services.put(service.getServiceDescriptor().getName(), service)
     }
 
     actual fun removeService(service: ServerServiceDefinition): Boolean {
-        error("Native target is not supported in gRPC")
+        return services.remove(service.getServiceDescriptor().getName()) != null
+    }
+
+    override fun getServices(): List<ServerServiceDefinition> {
+        return services.values.toList()
+    }
+
+    override fun lookupMethod(
+        methodName: String,
+        authority: String?,
+    ): ServerMethodDefinition<*, *>? {
+        val serviceName = MethodDescriptor.extractFullServiceName(methodName) ?: return null
+        val service = services[serviceName] ?: return null
+        return service.getMethod(methodName)
     }
 }

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

@@ -4,8 +4,9 @@
 
 package kotlinx.rpc.grpc
 
-import kotlinx.rpc.grpc.internal.GrpcServerCredentials
+import kotlinx.rpc.grpc.internal.GrpcInsecureServerCredentials
 import kotlinx.rpc.grpc.internal.NativeServer
+import kotlinx.rpc.grpc.internal.ServerMethodDefinition
 
 /**
  * Platform-specific gRPC server builder.
@@ -23,26 +24,22 @@ private class NativeServerBuilder(
 ) : ServerBuilder<NativeServerBuilder>() {
 
     // TODO: Add actual credentials
-    private val credentials = GrpcServerCredentials.createInsecure()
+    private val credentials = GrpcInsecureServerCredentials()
     private val services = mutableListOf<ServerServiceDefinition>()
+    private var fallbackRegistry: HandlerRegistry = DefaultFallbackRegistry
 
     override fun addService(service: ServerServiceDefinition): NativeServerBuilder {
         services.add(service)
         return this
     }
 
     override fun fallbackHandlerRegistry(registry: HandlerRegistry?): NativeServerBuilder {
-        TODO("Not yet implemented")
+        fallbackRegistry = registry ?: DefaultFallbackRegistry
+        return this
     }
 
     override fun build(): Server {
-        val server = NativeServer(port, credentials)
-
-        for (service in services) {
-            server.addService(service)
-        }
-
-        return server
+        return NativeServer(port, credentials, services, fallbackRegistry)
     }
 
 }
@@ -54,3 +51,17 @@ internal actual fun ServerBuilder(port: Int): ServerBuilder<*> {
 internal actual fun Server(builder: ServerBuilder<*>): Server {
     return builder.build()
 }
+
+private object DefaultFallbackRegistry : HandlerRegistry() {
+    override fun getServices(): List<ServerServiceDefinition> {
+        return listOf()
+    }
+
+    override fun lookupMethod(
+        methodName: String,
+        authority: String?,
+    ): ServerMethodDefinition<*, *>? {
+        return null
+    }
+
+}

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

@@ -10,11 +10,18 @@ import kotlinx.rpc.internal.utils.InternalRpcApi
 
 public actual class ServerServiceDefinition internal constructor(
     private val serviceDescriptor: ServiceDescriptor,
-    private val methods: Collection<ServerMethodDefinition<*, *>>,
+    private val methods: Map<String, ServerMethodDefinition<*, *>>,
 ) {
+
+    internal constructor(serviceDescriptor: ServiceDescriptor, methods: Collection<ServerMethodDefinition<*, *>>) :
+            this(serviceDescriptor, methods.associateBy { it.getMethodDescriptor().getFullMethodName() })
+
     public actual fun getServiceDescriptor(): ServiceDescriptor = serviceDescriptor
 
-    public actual fun getMethods(): Collection<ServerMethodDefinition<*, *>> = methods
+    public actual fun getMethods(): Collection<ServerMethodDefinition<*, *>> = methods.values
+
+    public actual fun getMethod(methodName: String): ServerMethodDefinition<*, *>? = methods[methodName]
+
 }
 
 @InternalRpcApi

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

@@ -219,19 +219,42 @@ private fun deleteCbTag(tag: CPointer<kgrpc_cb_tag>) {
     nativeHeap.free(tag)
 }
 
+/**
+ * Represents a callback tag for completion queues constructed with
+ * [grpc_completion_queue_create_for_callback].
+ *
+ * The [run] method will be invoked onces the completion queue signals the completion of the operation.
+ * It is guaranteed that the [run] method will be only invoked once.
+ *
+ * `this` object is guaranteed to be not garbage collected until the [run] method was executed.
+ */
 internal interface CallbackTag {
     fun run(ok: Boolean)
 
+    /**
+     * Creates a pointer to a gRPC callback tag that encapsulates the given `run` function.
+     * It can be passed to callback-based grpc_completion_queue.
+     */
     fun toCbTag(): CPointer<kgrpc_cb_tag> {
         return newCbTag(this, staticCFunction { functor, ok ->
             val tag = functor!!.reinterpret<kgrpc_cb_tag>()
             val callbackTag = tag.pointed.user_data!!.asStableRef<CallbackTag>().get()
+            // free the tag memory and release the stable reference to `this`
             deleteCbTag(tag)
             callbackTag.run(ok != 0)
         })
     }
 
     companion object {
+        /**
+         * Creates a pointer to a gRPC callback tag that encapsulates the given `run` function.
+         *
+         * The `run` function will be invoked when the callback is triggered with a boolean
+         * value that indicates the success or failure of the operation.
+         *
+         * @return A pointer to the newly created gRPC callback tag.
+         *         It can be passed to callback-based grpc_completion_queue.
+         */
         fun anonymous(run: (ok: Boolean) -> Unit): CPointer<kgrpc_cb_tag> {
             return object : CallbackTag {
                 override fun run(ok: Boolean) {

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

@@ -22,12 +22,7 @@ public actual class MethodDescriptor<Request, Response> internal constructor(
     public actual fun getFullMethodName(): String = fullMethodName
 
     private val serviceName: String? by lazy {
-        val index = fullMethodName.lastIndexOf('/')
-        if (index == -1) {
-            null
-        } else {
-            fullMethodName.substring(0, index)
-        }
+        extractFullServiceName(fullMethodName)
     }
 
     public actual fun getServiceName(): String? = serviceName
@@ -48,6 +43,17 @@ public actual class MethodDescriptor<Request, Response> internal constructor(
         public actual fun stream(value: T): InputStream
         public actual fun parse(stream: InputStream): T
     }
+
+    public companion object {
+        public fun extractFullServiceName(fullMethodName: String): String? {
+            val index = fullMethodName.lastIndexOf('/')
+            return if (index == -1) {
+                null
+            } else {
+                fullMethodName.take(index)
+            }
+        }
+    }
 }
 
 @InternalRpcApi

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

@@ -26,6 +26,7 @@ import libkgrpc.grpc_channel_create_call
 import libkgrpc.grpc_channel_credentials_release
 import libkgrpc.grpc_channel_destroy
 import libkgrpc.grpc_insecure_credentials_create
+import libkgrpc.grpc_slice_unref
 import kotlin.coroutines.cancellation.CancellationException
 import kotlin.experimental.ExperimentalNativeApi
 import kotlin.native.ref.createCleaner
@@ -151,6 +152,8 @@ internal class NativeManagedChannel(
             reserved = null
         ) ?: error("Failed to create call")
 
+        grpc_slice_unref(methodNameSlice)
+
         return NativeClientCall(
             cq, rawCall, methodDescriptor, callJob
         )