grpc-native: Implement encoding/decoding of bytes

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

Author: Jozott00

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/internal/WireDecoder.kt

@@ -43,6 +43,7 @@ internal interface WireDecoder: AutoCloseable {
     fun readSFixed64(): Long?
     fun readEnum(): Int?
     fun readString(): String?
+    fun readBytes(): ByteArray?
 }
 
 /**

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/internal/WireEncoder.kt

@@ -29,6 +29,7 @@ internal interface WireEncoder {
     fun writeEnum(fieldNr: Int, value: Int): Boolean
     fun writeString(fieldNr: Int, value: String): Boolean
     fun flush()
+    fun writeBytes(fieldNr: Int, value: ByteArray): Boolean
 }
 
 

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

@@ -152,6 +152,7 @@ internal class WireDecoderNative(private val source: Buffer): WireDecoder {
         return null
     }
 
+    // TODO: Is it possible to avoid copying the c_str, by directly allocating a K/N String (as in readBytes)?
     override fun readString(): String? = memScoped {
         val str = alloc<CPointerVar<pw_string_t>>()
         val ok = pw_decoder_read_string(raw, str.ptr)
@@ -162,6 +163,16 @@ internal class WireDecoderNative(private val source: Buffer): WireDecoder {
             pw_string_delete(str.value)
         }
     }
+
+    override fun readBytes(): ByteArray? {
+        val length = readInt32() ?: return null
+        if (length == 0) return ByteArray(0)
+        val bytes = ByteArray(length)
+        bytes.usePinned {
+            pw_decoder_read_raw_bytes(raw, it.addressOf(0), length)
+        }
+        return bytes
+    }
 }
 
 internal actual fun WireDecoder(source: Buffer): WireDecoder = WireDecoderNative(source)

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

@@ -9,6 +9,7 @@ import kotlinx.io.Sink
 import libprotowire.*
 import kotlin.experimental.ExperimentalNativeApi
 import kotlin.native.ref.createCleaner
+import kotlin.text.isEmpty
 
 
 // TODO: Evaluate if we should implement a ZeroCopyOutputSink (similar to the ZeroCopyInputSource)
@@ -104,6 +105,15 @@ internal class WireEncoderNative(private val sink: Sink): WireEncoder {
         }
     }
 
+    override fun writeBytes(fieldNr: Int, value: ByteArray): Boolean {
+        if (value.isEmpty()) {
+            return pw_encoder_write_bytes(raw, fieldNr, null, 0)
+        }
+        return value.usePinned {
+            pw_encoder_write_bytes(raw, fieldNr, it.addressOf(0), value.size)
+        }
+    }
+
     override fun flush() {
         pw_encoder_flush(raw)
     }

grpc/grpc-core/src/nativeTest/kotlin/kotlinx/rpc/grpc/internal/WireCodecTest.kt

@@ -573,4 +573,41 @@ class WireCodecTest {
         assertNotNull(str)
         assertEquals("", str)
     }
+
+    @Test
+    fun testEmptyByteArray() {
+        val buffer = Buffer()
+
+        val encoder = WireEncoder(buffer)
+        assertTrue(encoder.writeBytes(1, ByteArray(0)))
+        encoder.flush()
+
+        val decoder = WireDecoder(buffer)
+
+        val tag = decoder.readTag()
+        assertNotNull(tag)
+        assertEquals(1, tag.fieldNr)
+        assertEquals(WireType.LENGTH_DELIMITED, tag.wireType)
+
+        val bytes = decoder.readBytes()
+        assertNotNull(bytes)
+        assertEquals(0, bytes.size)
+    }
+
+    @Test
+    fun testBytesEncodeDecode() {
+        val buffer = Buffer()
+        val encoder = WireEncoder(buffer)
+
+        val bytes = ByteArray(1000000) { it.toByte() }
+
+        assertTrue(encoder.writeBytes(1, bytes))
+        encoder.flush()
+
+        val decoder = WireDecoder(buffer)
+        val tag = decoder.readTag()
+        assertNotNull(tag)
+        assertEquals(1, tag.fieldNr)
+        assertEquals(WireType.LENGTH_DELIMITED, tag.wireType)
+    }
 }

grpc/grpcpp-c/include/protowire.h

@@ -49,7 +49,7 @@ extern "C" {
     bool pw_encoder_write_double(pw_encoder_t *self, int field_no, double value);
     bool pw_encoder_write_enum(pw_encoder_t *self, int field_no, int value);
     bool pw_encoder_write_string(pw_encoder_t *self, int field_no, const char *data, int size);
-    bool pw_encoder_write_bytes(pw_encoder_t *self, int field_no, pw_string_t *value);
+    bool pw_encoder_write_bytes(pw_encoder_t *self, int field_no, const void *data, int size);
 
 
     //// WIRE DECODER ////
@@ -95,6 +95,8 @@ extern "C" {
     bool pw_decoder_read_double(pw_decoder_t *self, double *value);
     bool pw_decoder_read_enum(pw_decoder_t *self, int *value);
     bool pw_decoder_read_string(pw_decoder_t *self, pw_string_t **opaque_string);
+    // To read an actual bytes field, you must combine read_int32 and this function
+    bool pw_decoder_read_raw_bytes(pw_decoder_t *self, void* buffer, int size);
 
 #ifdef __cplusplus
     }

grpc/grpcpp-c/src/protowire.cpp

@@ -169,16 +169,14 @@ extern "C" {
     WRITE_FIELD_FUNC( enum, Enum, int)
 
     bool pw_encoder_write_string(pw_encoder_t *self, int field_no, const char *data, int size) {
+        return pw_encoder_write_bytes(self, field_no, data, size);
+    }
+    bool pw_encoder_write_bytes(pw_encoder_t *self, int field_no, const void *data, int size) {
         WireFormatLite::WriteTag(field_no, WireFormatLite::WIRETYPE_LENGTH_DELIMITED, &self->cos);
         self->cos.WriteVarint32(size);
         self->cos.WriteRawMaybeAliased(data, size);
         return check(self);
     }
-    bool pw_encoder_write_bytes(pw_encoder_t *self, int field_no, pw_string_t *value) {
-        WireFormatLite::WriteBytes(field_no, value->str, &self->cos);
-        return check(self);
-    }
-
 
     pw_decoder_t *pw_decoder_new(pw_zero_copy_input_t zero_copy_input) {
         return new pw_decoder_t(zero_copy_input);
@@ -214,4 +212,8 @@ extern "C" {
         *string_ref = new pw_string_t;
         return WireFormatLite::ReadString(&self->cis, &(*string_ref)->str);
     }
+
+    bool pw_decoder_read_raw_bytes(pw_decoder_t *self, void* buffer, int size) {
+        return self->cis.ReadRaw(buffer, size);
+    }
 }