Skip to content

Commit

Permalink
#221 (comment): fix serialization/deserialization of types managed fo…
Browse files Browse the repository at this point in the history 
…rm Apache Kafka (#222)
  • Loading branch information
@masesdevelopers
masesdevelopers authored Aug 31, 2023
1 parent 48420b5 commit ab6be44
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 21 deletions.
11 changes: 10 additions & 1 deletion src/net/KNet/Specific/Serialization/KNetSerDes.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,20 @@

namespace MASES.KNet.Serialization
{
/// <summary>
/// KNet common serializer/deserializer
/// </summary>
/// <typeparam name="T">The type to serialize/deserialize</typeparam>
public interface IKNetSerDes<T> : IKNetSerializer<T>, IKNetDeserializer<T>
{

}

/// <summary>
/// Common serializer/deserializer
/// </summary>
/// <typeparam name="T">The type to serialize/deserialize</typeparam>
public class KNetSerDes<T> : IKNetSerializer<T>, IKNetDeserializer<T>
public class KNetSerDes<T> : IKNetSerDes<T>
{
readonly KNetSerialization.SerializationType _SerializationType = KNetSerialization.InternalSerDesType<T>();
Serializer<byte[]> _KafkaSerializer = new ByteArraySerializer();
Expand Down
41 changes: 21 additions & 20 deletions src/net/KNet/Specific/Serialization/KNetSerialization.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
using Java.Lang;
using Java.Nio;
using MASES.JCOBridge.C2JBridge;
using MASES.JCOBridge.C2JBridge.JVMInterop;
using Org.Apache.Kafka.Common.Errors;
using Org.Apache.Kafka.Common.Serialization;
using Org.Apache.Kafka.Common.Utils;
Expand Down Expand Up @@ -150,7 +151,7 @@ public static byte[] SerializeByteArray(string topic, byte[] data)
return data;
}

static readonly Serializer<ByteBuffer> _ByteBufferSerializer = new ByteBufferSerializer();
static readonly ByteBufferSerializer _ByteBufferSerializer = new ByteBufferSerializer();
/// <summary>
/// Serialize a <see cref="SerializationType.ByteBuffer"/>
/// </summary>
Expand All @@ -159,7 +160,7 @@ public static byte[] SerializeByteBuffer(string topic, ByteBuffer data)
return _ByteBufferSerializer.Serialize(topic, data);
}

static readonly Serializer<Bytes> _BytesSerializer = new BytesSerializer();
static readonly BytesSerializer _BytesSerializer = new BytesSerializer();
/// <summary>
/// Serialize a <see cref="SerializationType.Bytes"/>
/// </summary>
Expand All @@ -168,7 +169,7 @@ public static byte[] SerializeBytes(string topic, Bytes data)
return _BytesSerializer.Serialize(topic, data);
}

static readonly Serializer<double> _DoubleSerializer = new DoubleSerializer();
static readonly DoubleSerializer _DoubleSerializer = new DoubleSerializer();
/// <summary>
/// Serialize a <see cref="SerializationType.Double"/>
/// </summary>
Expand All @@ -177,7 +178,7 @@ public static byte[] SerializeDouble(string topic, double data)
return _DoubleSerializer.Serialize(topic, data);
}

static readonly Serializer<float> _FloatSerializer = new FloatSerializer();
static readonly FloatSerializer _FloatSerializer = new FloatSerializer();
/// <summary>
/// Serialize a <see cref="SerializationType.Float"/>
/// </summary>
Expand All @@ -186,7 +187,7 @@ public static byte[] SerializeFloat(string topic, float data)
return _FloatSerializer.Serialize(topic, data);
}

static readonly Serializer<int> _IntSerializer = new IntegerSerializer();
static readonly IntegerSerializer _IntSerializer = new IntegerSerializer();
/// <summary>
/// Serialize a <see cref="SerializationType.Int"/>
/// </summary>
Expand All @@ -208,7 +209,7 @@ public static byte[] SerializeLong(string topic, long data)
//return new byte[] { (byte)((int)(data >>> 56)), (byte)((int)(data >>> 48)), (byte)((int)(data >>> 40)), (byte)((int)(data >>> 32)), (byte)((int)(data >>> 24)), (byte)((int)(data >>> 16)), (byte)((int)(data >>> 8)), ((byte)data) };
}

static readonly Serializer<short> _ShortSerializer = new ShortSerializer();
static readonly ShortSerializer _ShortSerializer = new ShortSerializer();
/// <summary>
/// Serialize a <see cref="SerializationType.Short"/>
/// </summary>
Expand Down Expand Up @@ -247,49 +248,49 @@ public static byte[] DeserializeByteArray(string topic, byte[] data)
return data;
}

static readonly Deserializer<ByteBuffer> _ByteBufferDeserializer = new ByteBufferDeserializer();
static readonly ByteBufferDeserializer _ByteBufferDeserializer = new ByteBufferDeserializer();
/// <summary>
/// Deserialize a <see cref="SerializationType.ByteBuffer"/>
/// </summary>
public static ByteBuffer DeserializeByteBuffer(string topic, byte[] data)
{
return _ByteBufferDeserializer.Deserialize(topic, data);
return JVMBridgeBase.Wraps<ByteBuffer>(_ByteBufferDeserializer.Deserialize(topic, data) as IJavaObject);
}

static readonly Deserializer<Bytes> _BytesDeserializer = new BytesDeserializer();
static readonly BytesDeserializer _BytesDeserializer = new BytesDeserializer();
/// <summary>
/// Deserialize a <see cref="SerializationType.Bytes"/>
/// </summary>
public static Bytes DeserializeBytes(string topic, byte[] data)
{
return _BytesDeserializer.Deserialize(topic, data);
return JVMBridgeBase.Wraps< Bytes>(_BytesDeserializer.Deserialize(topic, data) as IJavaObject);
}

static readonly Deserializer<double> _DoubleDeserializer = new DoubleDeserializer();
static readonly DoubleDeserializer _DoubleDeserializer = new DoubleDeserializer();
/// <summary>
/// Deserialize a <see cref="SerializationType.Double"/>
/// </summary>
public static double DeserializeDouble(string topic, byte[] data)
{
return _DoubleDeserializer.Deserialize(topic, data);
return (double)_DoubleDeserializer.Deserialize(topic, data);
}

static readonly Deserializer<float> _FloatDeserializer = new FloatDeserializer();
static readonly FloatDeserializer _FloatDeserializer = new FloatDeserializer();
/// <summary>
/// Deserialize a <see cref="SerializationType.Float"/>
/// </summary>
public static float DeserializeFloat(string topic, byte[] data)
{
return _FloatDeserializer.Deserialize(topic, data);
return (float)_FloatDeserializer.Deserialize(topic, data);
}

static readonly Deserializer<int> _IntDeserializer = new IntegerDeserializer();
static readonly IntegerDeserializer _IntDeserializer = new IntegerDeserializer();
/// <summary>
/// Deserialize a <see cref="SerializationType.Int"/>
/// </summary>
public static int DeserializeInt(string topic, byte[] data)
{
return _IntDeserializer.Deserialize(topic, data);
return (int)_IntDeserializer.Deserialize(topic, data);

//if (data == null)
//{
Expand Down Expand Up @@ -317,13 +318,13 @@ public static int DeserializeInt(string topic, byte[] data)
//}
}

static readonly Deserializer<long> _LongDeserializer = new LongDeserializer();
static readonly LongDeserializer _LongDeserializer = new LongDeserializer();
/// <summary>
/// Deserialize a <see cref="SerializationType.Long"/>
/// </summary>
public static long DeserializeLong(string topic, byte[] data)
{
return _LongDeserializer.Deserialize(topic, data);
return (long)_LongDeserializer.Deserialize(topic, data);

//if (data == null)
//{
Expand Down Expand Up @@ -351,13 +352,13 @@ public static long DeserializeLong(string topic, byte[] data)
//}
}

static readonly Deserializer<short> _ShortDeserializer = new ShortDeserializer();
static readonly ShortDeserializer _ShortDeserializer = new ShortDeserializer();
/// <summary>
/// Deserialize a <see cref="SerializationType.Short"/>
/// </summary>
public static short DeserializeShort(string topic, byte[] data)
{
return _ShortDeserializer.Deserialize(topic, data);
return (short)_ShortDeserializer.Deserialize(topic, data);

//if (data == null)
//{
Expand Down

0 comments on commit ab6be44

Please sign in to comment.