|
| 1 | +/* |
| 2 | + * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license. |
| 3 | + */ |
| 4 | + |
| 5 | +package kotlinx.rpc.grpc.bridge |
| 6 | + |
| 7 | +import kotlinx.cinterop.* |
| 8 | +import kotlinx.coroutines.suspendCancellableCoroutine |
| 9 | +import libgrpcpp_c.* |
| 10 | +import kotlin.coroutines.resume |
| 11 | +import kotlin.coroutines.resumeWithException |
| 12 | + |
| 13 | +@OptIn(ExperimentalForeignApi::class) |
| 14 | +internal class GrpcClient(target: String) : AutoCloseable { |
| 15 | + private var clientPtr: CPointer<grpc_client_t> = |
| 16 | + grpc_client_create_insecure(target) ?: error("Failed to create client") |
| 17 | + |
| 18 | + fun callUnaryBlocking(method: String, req: GrpcSlice): GrpcSlice { |
| 19 | + memScoped { |
| 20 | + val result = alloc<grpc_slice>() |
| 21 | + grpc_client_call_unary_blocking(clientPtr, method, req.cSlice, result.ptr) |
| 22 | + return GrpcSlice(result.readValue()) |
| 23 | + } |
| 24 | + } |
| 25 | + |
| 26 | + suspend fun callUnary(method: String, req: GrpcByteBuffer): GrpcByteBuffer = |
| 27 | + suspendCancellableCoroutine { continuation -> |
| 28 | + val context = grpc_context_create() |
| 29 | + val method = grpc_method_create(method) |
| 30 | + |
| 31 | + val reqRawBuf = nativeHeap.alloc<CPointerVar<grpc_byte_buffer>>() |
| 32 | + reqRawBuf.value = req.cByteBuffer |
| 33 | + |
| 34 | + val respRawBuf: CPointerVar<grpc_byte_buffer> = nativeHeap.alloc() |
| 35 | + |
| 36 | + val continueCb = { st: grpc_status_code_t -> |
| 37 | + // cleanup allocations owned by this method (this runs always) |
| 38 | + grpc_method_delete(method) |
| 39 | + grpc_context_delete(context) |
| 40 | + nativeHeap.free(reqRawBuf) |
| 41 | + |
| 42 | + if (st != GRPC_C_STATUS_OK) { |
| 43 | + continuation.resumeWithException(RuntimeException("Call failed with code: $st")) |
| 44 | + } else { |
| 45 | + val result = respRawBuf.value |
| 46 | + if (result == null) { |
| 47 | + continuation.resumeWithException(RuntimeException("No response received")) |
| 48 | + } else { |
| 49 | + continuation.resume(GrpcByteBuffer(result)) |
| 50 | + } |
| 51 | + } |
| 52 | + |
| 53 | + nativeHeap.free(respRawBuf) |
| 54 | + } |
| 55 | + val cbCtxStable = StableRef.create(continueCb) |
| 56 | + |
| 57 | + grpc_client_call_unary_callback( |
| 58 | + clientPtr, method, context, reqRawBuf.ptr, respRawBuf.ptr, |
| 59 | + cbCtxStable.asCPointer(), staticCFunction { st, ctx -> |
| 60 | + val cbCtxStable = ctx!!.asStableRef<(grpc_status_code_t) -> Unit>() |
| 61 | + cbCtxStable.get()(st) |
| 62 | + cbCtxStable.dispose() |
| 63 | + }) |
| 64 | + } |
| 65 | + |
| 66 | + override fun close() { |
| 67 | + grpc_client_delete(clientPtr) |
| 68 | + } |
| 69 | + |
| 70 | +} |
0 commit comments