From 13a86dc0d5d3635fd78ca8c3c646e3526b82deb4 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sun, 23 Jan 2022 13:54:32 +0100 Subject: [PATCH] - use byte array special function for KxsBinary format - --- .../kafkatorio/processor/serdes/kxsBinary.kt | 53 ++++++++++++++++++- 1 file changed, 51 insertions(+), 2 deletions(-) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt index 0c0a66f..c255676 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/serdes/kxsBinary.kt @@ -6,10 +6,10 @@ import java.io.DataInput import java.io.DataInputStream import java.io.DataOutput import java.io.DataOutputStream -import java.util.zip.DeflaterOutputStream import kotlinx.serialization.BinaryFormat import kotlinx.serialization.DeserializationStrategy import kotlinx.serialization.SerializationStrategy +import kotlinx.serialization.builtins.ByteArraySerializer import kotlinx.serialization.descriptors.SerialDescriptor import kotlinx.serialization.encoding.AbstractDecoder import kotlinx.serialization.encoding.AbstractEncoder @@ -19,7 +19,7 @@ import kotlinx.serialization.modules.EmptySerializersModule import kotlinx.serialization.modules.SerializersModule -val kxsBinaryMapper: KxsBinary = KxsBinary() +val kxsBinary: KxsBinary = KxsBinary() /** [`https://github.com/Kotlin/kotlinx.serialization/blob/master/docs/formats.md#efficient-binary-format`](https://github.com/Kotlin/kotlinx.serialization/blob/master/docs/formats.md#efficient-binary-format) */ class KxsBinary( @@ -43,6 +43,8 @@ class KxsBinary( } } +private val byteArraySerializer = ByteArraySerializer() + class KxsDataOutputEncoder( private val output: DataOutput, override val serializersModule: SerializersModule = EmptySerializersModule, @@ -69,14 +71,38 @@ class KxsDataOutputEncoder( override fun encodeNull() = encodeBoolean(false) override fun encodeNotNullMark() = encodeBoolean(true) + + override fun encodeSerializableValue(serializer: SerializationStrategy, value: T) { + if (serializer.descriptor == byteArraySerializer.descriptor) + encodeByteArray(value as ByteArray) + else + super.encodeSerializableValue(serializer, value) + } + + private fun encodeByteArray(bytes: ByteArray) { + encodeCompactSize(bytes.size) + output.write(bytes) + } + + private fun encodeCompactSize(value: Int) { + if (value < 0xff) { + output.writeByte(value) + } else { + output.writeByte(0xff) + output.writeInt(value) + } + } } + class KxsDataInputDecoder( private val input: DataInput, override val serializersModule: SerializersModule = EmptySerializersModule, var elementsCount: Int = 0, ) : AbstractDecoder() { + private var elementIndex = 0 + override fun decodeBoolean(): Boolean = input.readByte().toInt() != 0 override fun decodeByte(): Byte = input.readByte() override fun decodeShort(): Short = input.readShort() @@ -102,4 +128,27 @@ class KxsDataInputDecoder( decodeInt().also { elementsCount = it } override fun decodeNotNullMark(): Boolean = decodeBoolean() + + override fun decodeSerializableValue( + deserializer: DeserializationStrategy, + previousValue: T? + ): T = + @Suppress("UNCHECKED_CAST") + when (deserializer.descriptor) { + byteArraySerializer.descriptor -> decodeByteArray() as T + else -> super.decodeSerializableValue(deserializer, previousValue) + } + + private fun decodeByteArray(): ByteArray { + val bytes = ByteArray(decodeCompactSize()) + input.readFully(bytes) + return bytes + } + + private fun decodeCompactSize(): Int { + val byte = input.readByte().toInt() and 0xff + if (byte < 0xff) return byte + return input.readInt() + } + }