grpc: Implement GrpcMetadata

Author: Jozott00

cinterop-c/MODULE.bazel.lock

@@ -89,6 +89,12 @@ void kgrpc_server_set_batch_method_allocator(
     kgrpc_batch_call_allocator allocator
 );
 
+/**
+ * Append an grpc_metadata entry to the given grpc_metadata_array.
+ *
+ * @return false if the array has not enough capacity, true otherwise.
+ */
+bool kgrpc_metadata_array_append(grpc_metadata_array *array, grpc_slice key, grpc_slice value);
 
 #ifdef __cplusplus
 }

cinterop-c/include/kgrpc.h

@@ -3,6 +3,7 @@
 #include <kgrpc.h>
 #include "src/core/lib/iomgr/iomgr.h"
 #include "src/core/server/server.h"
+#include "grpc/support/string_util.h"
 
 extern "C" {
 
@@ -53,6 +54,18 @@ void kgrpc_server_set_batch_method_allocator(
             });
 }
 
+bool kgrpc_metadata_array_append(grpc_metadata_array *array, grpc_slice key, grpc_slice value) {
+    if (array->capacity - array->count <= 0) {
+        return false;
+    }
+    grpc_metadata entry = {
+        .key = key,
+        .value = value
+    };
+    array->metadata[array->count++] = entry;
+    return true;
+}
+
 }
 
 

cinterop-c/src/kgrpc.cpp

@@ -17,17 +17,17 @@ import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.single
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
-import kotlinx.rpc.grpc.client.ClientCallScope
-import kotlinx.rpc.grpc.client.GrpcClient
 import kotlinx.rpc.grpc.GrpcMetadata
 import kotlinx.rpc.grpc.Status
 import kotlinx.rpc.grpc.StatusCode
 import kotlinx.rpc.grpc.StatusException
-import kotlinx.rpc.grpc.internal.CallbackFuture
+import kotlinx.rpc.grpc.client.ClientCallScope
+import kotlinx.rpc.grpc.client.GrpcClient
 import kotlinx.rpc.grpc.descriptor.MethodDescriptor
 import kotlinx.rpc.grpc.descriptor.MethodType
-import kotlinx.rpc.grpc.internal.Ready
 import kotlinx.rpc.grpc.descriptor.methodType
+import kotlinx.rpc.grpc.internal.CallbackFuture
+import kotlinx.rpc.grpc.internal.Ready
 import kotlinx.rpc.grpc.internal.singleOrStatus
 import kotlinx.rpc.grpc.statusCode
 import kotlinx.rpc.internal.utils.InternalRpcApi

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

@@ -26,9 +26,10 @@ import kotlinx.coroutines.CompletableJob
 import kotlinx.rpc.grpc.GrpcMetadata
 import kotlinx.rpc.grpc.Status
 import kotlinx.rpc.grpc.StatusCode
+import kotlinx.rpc.grpc.descriptor.MethodDescriptor
 import kotlinx.rpc.grpc.internal.BatchResult
 import kotlinx.rpc.grpc.internal.CompletionQueue
-import kotlinx.rpc.grpc.descriptor.MethodDescriptor
+import kotlinx.rpc.grpc.internal.destroyEntries
 import kotlinx.rpc.grpc.internal.internalError
 import kotlinx.rpc.grpc.internal.toByteArray
 import kotlinx.rpc.grpc.internal.toGrpcByteBuffer
@@ -182,7 +183,7 @@ internal class NativeClientCall<Request, Response>(
         if (!success) return
 
         // send and receive initial headers to/from the server
-        sendAndReceiveInitialMetadata()
+        sendAndReceiveInitialMetadata(headers)
     }
 
     /**
@@ -251,13 +252,16 @@ internal class NativeClientCall<Request, Response>(
         val statusCode = arena.alloc<grpc_status_code.Var>()
         val statusDetails = arena.alloc<grpc_slice>()
         val errorStr = arena.alloc<CPointerVar<ByteVar>>()
+
+        val trailingMetadata = arena.alloc<grpc_metadata_array>()
+        grpc_metadata_array_init(trailingMetadata.ptr)
+
         val op = arena.alloc<grpc_op> {
             op = GRPC_OP_RECV_STATUS_ON_CLIENT
             data.recv_status_on_client.status = statusCode.ptr
             data.recv_status_on_client.status_details = statusDetails.ptr
             data.recv_status_on_client.error_string = errorStr.ptr
-            // TODO: trailing metadata
-            data.recv_status_on_client.trailing_metadata = null
+            data.recv_status_on_client.trailing_metadata = trailingMetadata.ptr
         }
 
         when (val callResult = cq.runBatch(this@NativeClientCall.raw, op.ptr, 1u)) {
@@ -266,11 +270,13 @@ internal class NativeClientCall<Request, Response>(
                     val details = statusDetails.toByteArray().toKString()
                     val kStatusCode = statusCode.value.toKotlin()
                     val status = Status(kStatusCode, details, null)
-                    val trailers = GrpcMetadata()
+                    val trailers = GrpcMetadata(trailingMetadata)
 
                     // cleanup
                     grpc_slice_unref(statusDetails.readValue())
                     if (errorStr.value != null) gpr_free(errorStr.value)
+                    // the entries are owned by the call object, so we must only destroy the array
+                    grpc_metadata_array_destroy(trailingMetadata.readValue())
                     arena.clear()
 
                     // set close info and try to close the call.
@@ -296,29 +302,37 @@ internal class NativeClientCall<Request, Response>(
         }
     }
 
-    private fun sendAndReceiveInitialMetadata() {
+    private fun sendAndReceiveInitialMetadata(headers: GrpcMetadata) {
         // sending and receiving initial metadata
         val arena = Arena()
         val opsNum = 2uL
         val ops = arena.allocArray<grpc_op>(opsNum.convert())
 
+        // turn given headers into a grpc_metadata_array.
+        val sendInitialMetadata: grpc_metadata_array = with(headers) {
+            arena.allocRawGrpcMetadata()
+        }
+
         // send initial meta data to server
-        // TODO: initial metadata
         ops[0].op = GRPC_OP_SEND_INITIAL_METADATA
-        ops[0].data.send_initial_metadata.count = 0u
+        ops[0].data.send_initial_metadata.count = sendInitialMetadata.count
+        ops[0].data.send_initial_metadata.metadata = sendInitialMetadata.metadata
 
-        val meta = arena.alloc<grpc_metadata_array>()
-        // TODO: make metadata array an object (for lifecycle management)
-        grpc_metadata_array_init(meta.ptr)
+        val recvInitialMetadata = arena.alloc<grpc_metadata_array>()
+        grpc_metadata_array_init(recvInitialMetadata.ptr)
         ops[1].op = GRPC_OP_RECV_INITIAL_METADATA
-        ops[1].data.recv_initial_metadata.recv_initial_metadata = meta.ptr
+        ops[1].data.recv_initial_metadata.recv_initial_metadata = recvInitialMetadata.ptr
 
         runBatch(ops, opsNum, cleanup = {
-            grpc_metadata_array_destroy(meta.ptr)
+            // we must not destroy the array itself, as it is cleared when clearing the arena.
+            sendInitialMetadata.destroyEntries()
+            // the entries are owned by the call object, so we must only destroy the array
+            grpc_metadata_array_destroy(recvInitialMetadata.readValue())
             arena.clear()
         }) {
+            val headers = GrpcMetadata(recvInitialMetadata)
             safeUserCode("Failed to call onHeaders.") {
-                listener?.onHeaders(GrpcMetadata())
+                listener?.onHeaders(headers)
             }
         }
     }
@@ -447,4 +461,3 @@ internal class NativeClientCall<Request, Response>(
     }
 }
 
-

grpc/grpc-client/src/nativeMain/kotlin/kotlinx/rpc/grpc/client/internal/NativeClientCall.kt

@@ -5,6 +5,26 @@
 package kotlinx.rpc.grpc
 
 @Suppress("RedundantConstructorKeyword")
-public expect class GrpcMetadata constructor() {
-    public fun merge(trailers: GrpcMetadata)
-}
+public expect class GrpcMetadata constructor()
+
+public expect operator fun GrpcMetadata.get(key: String): String?
+public expect fun GrpcMetadata.getBinary(key: String): ByteArray?
+public expect fun GrpcMetadata.getAll(key: String): List<String>
+public expect fun GrpcMetadata.getAllBinary(key: String): List<ByteArray>
+
+public expect fun GrpcMetadata.keys(): Set<String>
+public expect operator fun GrpcMetadata.contains(key: String): Boolean
+
+public expect fun GrpcMetadata.append(key: String, value: String)
+public expect fun GrpcMetadata.appendBinary(key: String, value: ByteArray)
+
+public expect fun GrpcMetadata.remove(key: String, value: String): Boolean
+public expect fun GrpcMetadata.removeBinary(key: String, value: ByteArray): Boolean
+public expect fun GrpcMetadata.removeAll(key: String): List<String>
+public expect fun GrpcMetadata.removeAllBinary(key: String): List<ByteArray>
+
+public expect fun GrpcMetadata.merge(other: GrpcMetadata)
+public operator fun GrpcMetadata.plusAssign(other: GrpcMetadata): Unit = merge(other)
+
+public fun GrpcMetadata.copy(): GrpcMetadata = GrpcMetadata().also { it.merge(this) }
+public operator fun GrpcMetadata.plus(other: GrpcMetadata): GrpcMetadata = copy().apply { merge(other) }

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

@@ -4,14 +4,17 @@
 
 package kotlinx.rpc.grpc.test.proto
 
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.test.runTest
 import kotlinx.rpc.RpcServer
+import kotlinx.rpc.grpc.client.ClientCallScope
 import kotlinx.rpc.grpc.client.ClientCredentials
 import kotlinx.rpc.grpc.client.ClientInterceptor
 import kotlinx.rpc.grpc.client.GrpcClient
 import kotlinx.rpc.grpc.server.GrpcServer
+import kotlinx.rpc.grpc.server.ServerCallScope
 import kotlinx.rpc.grpc.server.ServerCredentials
 import kotlinx.rpc.grpc.server.ServerInterceptor
 
@@ -58,4 +61,34 @@ abstract class GrpcProtoTest {
     companion object {
         const val PORT = 8080
     }
+
+    internal fun serverInterceptor(
+        block: ServerCallScope<Any, Any>.(Flow<Any>) -> Flow<Any>,
+    ): List<ServerInterceptor> {
+        return listOf(object : ServerInterceptor {
+            @Suppress("UNCHECKED_CAST")
+            override fun <Req, Resp> ServerCallScope<Req, Resp>.intercept(
+                request: Flow<Req>,
+            ): Flow<Resp> {
+                with(this as ServerCallScope<Any, Any>) {
+                    return block(request as Flow<Any>) as Flow<Resp>
+                }
+            }
+        })
+    }
+
+    internal fun clientInterceptor(
+        block: ClientCallScope<Any, Any>.(Flow<Any>) -> Flow<Any>,
+    ): List<ClientInterceptor> {
+        return listOf(object : ClientInterceptor {
+            @Suppress("UNCHECKED_CAST")
+            override fun <Req, Resp> ClientCallScope<Req, Resp>.intercept(
+                request: Flow<Req>,
+            ): Flow<Resp> {
+                with(this as ClientCallScope<Any, Any>) {
+                    return block(request as Flow<Any>) as Flow<Resp>
+                }
+            }
+        })
+    }
 }