grpc-native: Fixes after rebase

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

Author: Jozott00

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

@@ -4,15 +4,10 @@
 
 package kotlinx.rpc.grpc.internal
 
-import kotlinx.io.Source
 import kotlinx.rpc.grpc.codec.MessageCodec
 import kotlinx.rpc.internal.utils.InternalRpcApi
 import kotlinx.rpc.protobuf.input.stream.InputStream
 
-@InternalRpcApi
-internal actual val MethodDescriptor<*, *>.type: MethodType
-    get() = TODO("Not yet implemented")
-
 @InternalRpcApi
 public actual class MethodDescriptor<Request, Response> internal constructor(
     private val fullMethodName: String,
@@ -72,23 +67,21 @@ public actual fun <Request, Response> methodDescriptor(
 ): MethodDescriptor<Request, Response> {
     val requestMarshaller = object : MethodDescriptor.Marshaller<Request> {
         override fun stream(value: Request): InputStream {
-            val source = requestCodec.encode(value)
-            return object : InputStream(source) {}
+            return requestCodec.encode(value)
         }
 
         override fun parse(stream: InputStream): Request {
-            return requestCodec.decode(stream.source)
+            return requestCodec.decode(stream)
         }
     }
 
     val responseMarshaller = object : MethodDescriptor.Marshaller<Response> {
         override fun stream(value: Response): InputStream {
-            val source = responseCodec.encode(value)
-            return object : InputStream(source) {}
+            return responseCodec.encode(value)
         }
 
         override fun parse(stream: InputStream): Response {
-            return responseCodec.decode(stream.source)
+            return responseCodec.decode(stream)
         }
     }
 

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

@@ -11,10 +11,11 @@ import kotlinx.atomicfu.atomic
 import kotlinx.cinterop.*
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableJob
-import kotlinx.io.Buffer
 import kotlinx.rpc.grpc.GrpcTrailers
 import kotlinx.rpc.grpc.Status
 import kotlinx.rpc.grpc.StatusCode
+import kotlinx.rpc.protobuf.input.stream.asInputStream
+import kotlinx.rpc.protobuf.input.stream.asSource
 import libgrpcpp_c.*
 import kotlin.experimental.ExperimentalNativeApi
 import kotlin.native.ref.createCleaner
@@ -228,7 +229,7 @@ internal class NativeClientCall<Request, Response>(
 
         val arena = Arena()
         val inputStream = methodDescriptor.getRequestMarshaller().stream(message)
-        val byteBuffer = (inputStream.source as Buffer).toGrpcByteBuffer()
+        val byteBuffer = inputStream.asSource().toGrpcByteBuffer()
         val op = arena.alloc<grpc_op> {
             op = GRPC_OP_SEND_MESSAGE
             data.send_message.send_message = byteBuffer

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

@@ -7,14 +7,29 @@
 package kotlinx.rpc.grpc.internal
 
 import kotlinx.cinterop.*
-import kotlinx.io.Buffer
-import kotlinx.io.Source
-import kotlinx.io.UnsafeIoApi
+import kotlinx.io.*
 import kotlinx.io.unsafe.UnsafeBufferOperations
 import kotlinx.rpc.grpc.StatusCode
 import libgrpcpp_c.*
 import platform.posix.memcpy
 
+@OptIn(ExperimentalForeignApi::class, InternalIoApi::class, UnsafeIoApi::class)
+internal fun Sink.writeFully(buffer: CPointer<ByteVar>, offset: Long, length: Long) {
+    var consumed = 0L
+    while (consumed < length) {
+        UnsafeBufferOperations.writeToTail(this.buffer, 1) { array, start, endExclusive ->
+            val size = minOf(length - consumed, (endExclusive - start).toLong())
+
+            array.usePinned {
+                memcpy(it.addressOf(start), buffer + offset + consumed, size.convert())
+            }
+
+            consumed += size
+            size.toInt()
+        }
+    }
+}
+
 internal suspend fun withArena(block: suspend (Arena) -> Unit) =
     Arena().let { arena ->
         try {
@@ -24,8 +39,6 @@ internal suspend fun withArena(block: suspend (Arena) -> Unit) =
         }
     }
 
-internal fun Buffer.asInputStream(): InputStream = object : InputStream(this) {}
-
 internal fun CPointer<grpc_byte_buffer>.toKotlin(destroy: Boolean = true): Buffer = memScoped {
     val reader = alloc<grpc_byte_buffer_reader>()
     check(grpc_byte_buffer_reader_init(reader.ptr, this@toKotlin) == 1)