From ad9cb07e4c71d02bc38493af0151749f0053a50d Mon Sep 17 00:00:00 2001 From: MASES Public Developers Team <94312179+masesdevelopers@users.noreply.github.com> Date: Sat, 25 May 2024 21:25:40 +0200 Subject: [PATCH] Added and managed `ISerDesSelector` (breaking change) (#466) * Added and managed ISerDesSelector to retrieve serialization classes based on the JVM type, all serialization extension where updated to reports the classes extending ISerDesSelector without any reference to the data transfer mechanism * Doc update * Added new properties and methods to ISerDes and ISerDesSelector * Update description --- src/documentation/articles/usageSerDes.md | 17 +- src/net/KNet.Serialization.Avro/AvroSerDes.cs | 1111 ++++++++++------- src/net/KNet.Serialization.Json/JsonSerDes.cs | 614 +++++---- .../MessagePackSerDes.cs | 410 +++--- .../ProtobufSerDes.cs | 378 +++--- .../KNet/Specific/Consumer/KNetConsumer.cs | 18 +- src/net/KNet/Specific/GenericConfigBuilder.cs | 75 +- .../KNet/Specific/Producer/KNetProducer.cs | 18 +- .../Replicator/KNetCompactedReplicator.cs | 90 +- .../Serialization/IGenericSerDesFactory.cs | 16 +- src/net/KNet/Specific/Serialization/SerDes.cs | 179 ++- tests/net/KNetBenchmark/ProgramKafka.cs | 8 +- .../KNetCompactedReplicatorTest/Program.cs | 6 +- tests/net/KNetTest/Program.cs | 16 +- 14 files changed, 1767 insertions(+), 1189 deletions(-) diff --git a/src/documentation/articles/usageSerDes.md b/src/documentation/articles/usageSerDes.md index 375843dd57..0556e8a5da 100644 --- a/src/documentation/articles/usageSerDes.md +++ b/src/documentation/articles/usageSerDes.md @@ -70,28 +70,29 @@ SerDesRaw deserializer = new SerDesRaw() Otherwise the user can use a ready made class like in the following snippet: ```c# -ISerDesRaw serdes = new JsonSerDes.ValueRaw(); +ISerDesRaw serdes = JsonSerDes.Value.NewByteArraySerDes(); ``` A single `JsonSerDes.ValueRaw` can be used in serialization and deserialization, and produce Json serialized data. ## Key and Value versions -The reader noticed that in the example was used `JsonSerDes.ValueRaw`. It is a serializer/deserializer, based on `byte` array, generally used for values because it stores, within the record `Headers` information related to the value itself. +The reader noticed that in the example was used `JsonSerDes.Value().NewByteArraySerDes()`. It is a serializer/deserializer, based on `byte` array, generally used for values because it stores, within the record `Headers` information related to the value itself. + All packages listed above have multiple types based on the scope and data exchange mechanism: -- [Serialization Format].KeyRaw: key serializer/deserializer based on `byte` array -- [Serialization Format].KeyBuffered: key serializer/deserializer based on `ByteBuffer` -- [Serialization Format].ValueRaw: value serializer/deserializer based on `byte` array -- [Serialization Format].ValueBuffered: value serializer/deserializer based on `ByteBuffer` +- [Serialization Format].Key: key serializer/deserializer can manages data transfer using both `byte` array and `ByteBuffer` +- [Serialization Format].Value: value serializer/deserializer can manages data transfer using both `byte` array and `ByteBuffer` -where [Serialization format] depends on the serializatin package in use. +where [Serialization format] depends on the serializatin package in use and the selection of the data transfer can be made from underlying code or can be requested from the user: +- `[Serialization Format].[Key or Value].NewByteArraySerDes()`: returns an `ISerDesRaw` +- `[Serialization Format].[Key or Value].NewByteBufferSerDes()`: returns an `ISerDesBuffered` > [!TIP] > As specified above, each serializer stores info within the `Headers` and this behavior is controlled from a property named `UseHeaders`. > If the user writes a code like: > >```c# -> ISerDesRaw serdes = new JsonSerDes.ValueRaw(); +> ISerDesRaw serdes = JsonSerDes.Value.NewByteArraySerDes(); > serdes.UseHeader = false; >``` > The `ISerDesRaw` instance does not writes the `Headers` and can be used both for key and value. diff --git a/src/net/KNet.Serialization.Avro/AvroSerDes.cs b/src/net/KNet.Serialization.Avro/AvroSerDes.cs index a59b2fa45b..f5bd73ffb4 100644 --- a/src/net/KNet.Serialization.Avro/AvroSerDes.cs +++ b/src/net/KNet.Serialization.Avro/AvroSerDes.cs @@ -28,7 +28,7 @@ namespace MASES.KNet.Serialization.Avro { /// - /// Base class to define extensions of for Avro, for example + /// Base class to define extensions of for Avro, for example /// public static class AvroSerDes { @@ -71,563 +71,740 @@ public static void BuildSchemaClassesFromFiles(string outputFolder, params strin } /// - /// Base class to define Key extensions of , for example + /// Base class to define Key extensions of , for example /// public static class Key { /// - /// Avro Key extension of for Binary encoding, for example + /// Avro Key extension of for Binary encoding, for example /// /// - public class BinaryRaw : SerDesRaw where T : global::Avro.Specific.ISpecificRecord, new() + public class Binary : ISerDesSelector where T : global::Avro.Specific.ISpecificRecord, new() { - global::Avro.Schema _schema; /// - /// Use this property to get the of + /// Returns a new instance of /// - public global::Avro.Schema Schema - { - get { return _schema; } - private set - { - _schema = value; - SpecificWriter = new SpecificDefaultWriter(_schema); - SpecificReader = new SpecificDefaultReader(_schema, _schema); - } - } + /// The of + public static ISerDesSelector NewInstance() => new Binary(); + /// + public static string SelectorTypeName => typeof(Binary<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(BinaryRaw); + /// + public static Type ByteBufferSerDes => typeof(BinaryBuffered); + /// + public static ISerDes NewSerDes() + { + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; + } + /// + public static ISerDesRaw NewByteArraySerDes() { return new BinaryRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new BinaryBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); - SpecificDefaultWriter SpecificWriter = null; - SpecificDefaultReader SpecificReader = null; - - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(BinaryRaw<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; /// - /// Default initializer + /// Avro Key extension of for Binary encoding, for example /// - public BinaryRaw() - { - var tRecord = new T(); - Schema = tRecord.Schema; - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + /// + sealed class BinaryRaw : SerDesRaw where TData : global::Avro.Specific.ISpecificRecord, new() + { + global::Avro.Schema _schema; + /// + /// Use this property to get the of + /// + public global::Avro.Schema Schema + { + get { return _schema; } + private set + { + _schema = value; + SpecificWriter = new SpecificDefaultWriter(_schema); + SpecificReader = new SpecificDefaultReader(_schema, _schema); + } + } - using MemoryStream memStream = new(); - BinaryEncoder encoder = new(memStream); - SpecificWriter.Write(data, encoder); - return memStream.ToArray(); - } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (data == null) return default; + SpecificDefaultWriter SpecificWriter = null; + SpecificDefaultReader SpecificReader = null; + + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public BinaryRaw(string selectorName) + { + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + var tRecord = new TData(); + Schema = tRecord.Schema; + } + /// + public override byte[] Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + + using MemoryStream memStream = new(); + BinaryEncoder encoder = new(memStream); + SpecificWriter.Write(data, encoder); + return memStream.ToArray(); + } + /// + public override TData Deserialize(string topic, byte[] data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (data == null) return default; - using MemoryStream memStream = new(data); - BinaryDecoder decoder = new(memStream); - T t = new T(); - t = SpecificReader.Read(t!, decoder); - return t; + using MemoryStream memStream = new(data); + BinaryDecoder decoder = new(memStream); + TData t = new TData(); + t = SpecificReader.Read(t!, decoder); + return t; + } } - } - /// - /// Avro Key extension of for Binary encoding, for example - /// - /// - public class BinaryBuffered : SerDesBuffered where T : global::Avro.Specific.ISpecificRecord, new() - { - global::Avro.Schema _schema; /// - /// Use this property to get the of + /// Avro Key extension of for Binary encoding, for example /// - public global::Avro.Schema Schema - { - get { return _schema; } - private set + /// + sealed class BinaryBuffered : SerDesBuffered where TData : global::Avro.Specific.ISpecificRecord, new() + { + global::Avro.Schema _schema; + /// + /// Use this property to get the of + /// + public global::Avro.Schema Schema { - _schema = value; - SpecificWriter = new SpecificDefaultWriter(_schema); - SpecificReader = new SpecificDefaultReader(_schema, _schema); + get { return _schema; } + private set + { + _schema = value; + SpecificWriter = new SpecificDefaultWriter(_schema); + SpecificReader = new SpecificDefaultReader(_schema, _schema); + } } - } - - SpecificDefaultWriter SpecificWriter = null; - SpecificDefaultReader SpecificReader = null; - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(BinaryBuffered<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; - /// - /// Default initializer - /// - public BinaryBuffered() - { - var tRecord = new T(); - Schema = tRecord.Schema; - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + SpecificDefaultWriter SpecificWriter = null; + SpecificDefaultReader SpecificReader = null; + + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public BinaryBuffered(string selectorName) + { + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + var tRecord = new T(); + Schema = tRecord.Schema; + } + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); - MemoryStream memStream = new(); - BinaryEncoder encoder = new(memStream); - SpecificWriter.Write(data, encoder); - return ByteBuffer.From(memStream); - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (data == null) return default; + MemoryStream memStream = new(); + BinaryEncoder encoder = new(memStream); + SpecificWriter.Write(data, encoder); + return ByteBuffer.From(memStream); + } + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (data == null) return default; - BinaryDecoder decoder = new(data.ToStream()); - T t = new T(); - t = SpecificReader.Read(t!, decoder); - return t; + BinaryDecoder decoder = new(data.ToStream()); + TData t = new TData(); + t = SpecificReader.Read(t!, decoder); + return t; + } } } /// - /// Avro Key extension of for Json encoding, for example + /// Avro Key extension of for Json encoding, for example /// /// - public class JsonRaw : SerDesRaw where T : global::Avro.Specific.ISpecificRecord, new() + public class Json : ISerDesSelector where T : global::Avro.Specific.ISpecificRecord, new() { - global::Avro.Schema _schema; /// - /// Use this property to get the of + /// Returns a new instance of /// - public global::Avro.Schema Schema - { - get { return _schema; } - private set - { - _schema = value; - SpecificWriter = new SpecificDefaultWriter(_schema); - SpecificReader = new SpecificDefaultReader(_schema, _schema); - } - } - - SpecificDefaultWriter SpecificWriter = null; - SpecificDefaultReader SpecificReader = null; + /// The of + public static ISerDesSelector NewInstance() => new Json(); + /// + public static string SelectorTypeName => typeof(Json<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(JsonRaw); + /// + public static Type ByteBufferSerDes => typeof(JsonBuffered); + /// + public static ISerDes NewSerDes() + { + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; + } + /// + public static ISerDesRaw NewByteArraySerDes() { return new JsonRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new JsonBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(JsonRaw<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; /// - /// Default initializer + /// Avro Key extension of for Json encoding, for example /// - public JsonRaw() - { - var tRecord = new T(); - Schema = tRecord.Schema; - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + /// + sealed class JsonRaw : SerDesRaw where TData : global::Avro.Specific.ISpecificRecord, new() + { + global::Avro.Schema _schema; + /// + /// Use this property to get the of + /// + public global::Avro.Schema Schema + { + get { return _schema; } + private set + { + _schema = value; + SpecificWriter = new SpecificDefaultWriter(_schema); + SpecificReader = new SpecificDefaultReader(_schema, _schema); + } + } - using MemoryStream memStream = new(); - JsonEncoder encoder = new(Schema, memStream); - SpecificWriter.Write(data, encoder); - return memStream.ToArray(); - } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (data == null) return default; + SpecificDefaultWriter SpecificWriter = null; + SpecificDefaultReader SpecificReader = null; + + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public JsonRaw(string selectorName) + { + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + var tRecord = new TData(); + Schema = tRecord.Schema; + } + /// + public override byte[] Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + + using MemoryStream memStream = new(); + JsonEncoder encoder = new(Schema, memStream); + SpecificWriter.Write(data, encoder); + return memStream.ToArray(); + } + /// + public override TData Deserialize(string topic, byte[] data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (data == null) return default; - using MemoryStream memStream = new(data); - JsonDecoder decoder = new(Schema, memStream); - T t = new T(); - t = SpecificReader.Read(t!, decoder); - return t; + using MemoryStream memStream = new(data); + JsonDecoder decoder = new(Schema, memStream); + TData t = new TData(); + t = SpecificReader.Read(t!, decoder); + return t; + } } - } - /// - /// Avro Key extension of for Json encoding, for example - /// - /// - public class JsonBuffered : SerDesBuffered where T : global::Avro.Specific.ISpecificRecord, new() - { - global::Avro.Schema _schema; /// - /// Use this property to get the of + /// Avro Key extension of for Json encoding, for example /// - public global::Avro.Schema Schema - { - get { return _schema; } - private set + /// + sealed class JsonBuffered : SerDesBuffered where TData : global::Avro.Specific.ISpecificRecord, new() + { + global::Avro.Schema _schema; + /// + /// Use this property to get the of + /// + public global::Avro.Schema Schema { - _schema = value; - SpecificWriter = new SpecificDefaultWriter(_schema); - SpecificReader = new SpecificDefaultReader(_schema, _schema); + get { return _schema; } + private set + { + _schema = value; + SpecificWriter = new SpecificDefaultWriter(_schema); + SpecificReader = new SpecificDefaultReader(_schema, _schema); + } } - } - - SpecificDefaultWriter SpecificWriter = null; - SpecificDefaultReader SpecificReader = null; - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(JsonBuffered<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; - /// - /// Default initializer - /// - public JsonBuffered() - { - var tRecord = new T(); - Schema = tRecord.Schema; - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + SpecificDefaultWriter SpecificWriter = null; + SpecificDefaultReader SpecificReader = null; + + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public JsonBuffered(string selectorName) + { + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + var tRecord = new TData(); + Schema = tRecord.Schema; + } + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); - MemoryStream memStream = new(); - JsonEncoder encoder = new(Schema, memStream); - SpecificWriter.Write(data, encoder); - return ByteBuffer.From(memStream); - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (data == null) return default; + MemoryStream memStream = new(); + JsonEncoder encoder = new(Schema, memStream); + SpecificWriter.Write(data, encoder); + return ByteBuffer.From(memStream); + } + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (data == null) return default; - JsonDecoder decoder = new(Schema, data.ToStream()); - T t = new T(); - t = SpecificReader.Read(t!, decoder); - return t; + JsonDecoder decoder = new(Schema, data.ToStream()); + TData t = new TData(); + t = SpecificReader.Read(t!, decoder); + return t; + } } } } /// - /// Base class to define Value extensions of , for example + /// Base class to define Value extensions of , for example /// public static class Value { /// - /// Avro Value extension of for Binary encoding, for example + /// Avro Value extension of for Binary encoding, for example /// /// - public class BinaryRaw : SerDesRaw where T : global::Avro.Specific.ISpecificRecord, new() + public class Binary : ISerDesSelector where T : global::Avro.Specific.ISpecificRecord, new() { - global::Avro.Schema _schema; /// - /// Use this property to get the of + /// Returns a new instance of /// - public global::Avro.Schema Schema - { - get { return _schema; } - private set - { - _schema = value; - SpecificWriter = new SpecificDefaultWriter(_schema); - SpecificReader = new SpecificDefaultReader(_schema, _schema); - } - } - - SpecificDefaultWriter SpecificWriter = null; - SpecificDefaultReader SpecificReader = null; + /// The of + public static ISerDesSelector NewInstance() => new Binary(); + /// + public static string SelectorTypeName => typeof(Binary<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(BinaryRaw); + /// + public static Type ByteBufferSerDes => typeof(BinaryBuffered); + /// + public static ISerDes NewSerDes() + { + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; + } + /// + public static ISerDesRaw NewByteArraySerDes() { return new BinaryRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new BinaryBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(BinaryRaw<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; /// - /// Default initializer + /// Avro Value extension of for Binary encoding, for example /// - public BinaryRaw() - { - var tRecord = new T(); - Schema = tRecord.Schema; - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + /// + sealed class BinaryRaw : SerDesRaw where TData : global::Avro.Specific.ISpecificRecord, new() + { + global::Avro.Schema _schema; + /// + /// Use this property to get the of + /// + public global::Avro.Schema Schema + { + get { return _schema; } + private set + { + _schema = value; + SpecificWriter = new SpecificDefaultWriter(_schema); + SpecificReader = new SpecificDefaultReader(_schema, _schema); + } + } - MemoryStream memStream = new(); - BinaryEncoder encoder = new(memStream); - SpecificWriter.Write(data, encoder); - return memStream.ToArray(); - } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (data == null) return default; + SpecificDefaultWriter SpecificWriter = null; + SpecificDefaultReader SpecificReader = null; + + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public BinaryRaw(string selectorName) + { + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + var tRecord = new TData(); + Schema = tRecord.Schema; + } + /// + public override byte[] Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + + MemoryStream memStream = new(); + BinaryEncoder encoder = new(memStream); + SpecificWriter.Write(data, encoder); + return memStream.ToArray(); + } + /// + public override TData Deserialize(string topic, byte[] data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (data == null) return default; - using MemoryStream memStream = new(data); - BinaryDecoder decoder = new(memStream); - T t = new T(); - t = SpecificReader.Read(t!, decoder); - return t; + using MemoryStream memStream = new(data); + BinaryDecoder decoder = new(memStream); + TData t = new TData(); + t = SpecificReader.Read(t!, decoder); + return t; + } } - } - /// - /// Avro Value extension of for Binary encoding, for example - /// - /// - public class BinaryBuffered : SerDesBuffered where T : global::Avro.Specific.ISpecificRecord, new() - { - global::Avro.Schema _schema; /// - /// Use this property to get the of + /// Avro Value extension of for Binary encoding, for example /// - public global::Avro.Schema Schema - { - get { return _schema; } - private set + /// + sealed class BinaryBuffered : SerDesBuffered where TData : global::Avro.Specific.ISpecificRecord, new() + { + global::Avro.Schema _schema; + /// + /// Use this property to get the of + /// + public global::Avro.Schema Schema { - _schema = value; - SpecificWriter = new SpecificDefaultWriter(_schema); - SpecificReader = new SpecificDefaultReader(_schema, _schema); + get { return _schema; } + private set + { + _schema = value; + SpecificWriter = new SpecificDefaultWriter(_schema); + SpecificReader = new SpecificDefaultReader(_schema, _schema); + } } - } - - SpecificDefaultWriter SpecificWriter = null; - SpecificDefaultReader SpecificReader = null; - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(BinaryBuffered<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; - /// - /// Default initializer - /// - public BinaryBuffered() - { - var tRecord = new T(); - Schema = tRecord.Schema; - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + SpecificDefaultWriter SpecificWriter = null; + SpecificDefaultReader SpecificReader = null; + + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public BinaryBuffered(string selectorName) + { + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + var tRecord = new TData(); + Schema = tRecord.Schema; + } + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); - MemoryStream memStream = new(); - BinaryEncoder encoder = new(memStream); - SpecificWriter.Write(data, encoder); - return ByteBuffer.From(memStream); - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (data == null) return default; + MemoryStream memStream = new(); + BinaryEncoder encoder = new(memStream); + SpecificWriter.Write(data, encoder); + return ByteBuffer.From(memStream); + } + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (data == null) return default; - BinaryDecoder decoder = new(data.ToStream()); - T t = new T(); - t = SpecificReader.Read(t!, decoder); - return t; + BinaryDecoder decoder = new(data.ToStream()); + TData t = new TData(); + t = SpecificReader.Read(t!, decoder); + return t; + } } } + /// - /// Avro Value extension of for Json encoding, for example + /// Avro Value extension of for Json encoding, for example /// /// - public class JsonRaw : SerDesRaw where T : global::Avro.Specific.ISpecificRecord, new() + public class Json : ISerDesSelector where T : global::Avro.Specific.ISpecificRecord, new() { - global::Avro.Schema _schema; /// - /// Use this property to get the of + /// Returns a new instance of /// - public global::Avro.Schema Schema - { - get { return _schema; } - private set - { - _schema = value; - SpecificWriter = new SpecificDefaultWriter(_schema); - SpecificReader = new SpecificDefaultReader(_schema, _schema); - } - } + /// The of + public static ISerDesSelector NewInstance() => new Json(); + /// + public static string SelectorTypeName => typeof(Json<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(JsonRaw); + /// + public static Type ByteBufferSerDes => typeof(JsonBuffered); + /// + public static ISerDes NewSerDes() + { + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; + } + /// + public static ISerDesRaw NewByteArraySerDes() { return new JsonRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new JsonBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); - SpecificDefaultWriter SpecificWriter = null; - SpecificDefaultReader SpecificReader = null; - - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(JsonRaw<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; /// - /// Default initializer + /// Avro Value extension of for Json encoding, for example /// - public JsonRaw() - { - var tRecord = new T(); - Schema = tRecord.Schema; - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + /// + sealed class JsonRaw : SerDesRaw where TData : global::Avro.Specific.ISpecificRecord, new() + { + global::Avro.Schema _schema; + /// + /// Use this property to get the of + /// + public global::Avro.Schema Schema + { + get { return _schema; } + private set + { + _schema = value; + SpecificWriter = new SpecificDefaultWriter(_schema); + SpecificReader = new SpecificDefaultReader(_schema, _schema); + } + } - using MemoryStream memStream = new(); - JsonEncoder encoder = new(Schema, memStream); - SpecificWriter.Write(data, encoder); - return memStream.ToArray(); - } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (data == null) return default; + SpecificDefaultWriter SpecificWriter = null; + SpecificDefaultReader SpecificReader = null; + + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public JsonRaw(string selectorName) + { + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + var tRecord = new TData(); + Schema = tRecord.Schema; + } + /// + public override byte[] Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); - using MemoryStream memStream = new(data); - JsonDecoder decoder = new(Schema, memStream); - T t = new T(); - t = SpecificReader.Read(t!, decoder); - return t; + using MemoryStream memStream = new(); + JsonEncoder encoder = new(Schema, memStream); + SpecificWriter.Write(data, encoder); + return memStream.ToArray(); + } + /// + public override TData Deserialize(string topic, byte[] data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (data == null) return default; + + using MemoryStream memStream = new(data); + JsonDecoder decoder = new(Schema, memStream); + TData t = new TData(); + t = SpecificReader.Read(t!, decoder); + return t; + } } - } - /// - /// Avro Value extension of for Json encoding, for example - /// - /// - public class JsonBuffered : SerDesBuffered where T : global::Avro.Specific.ISpecificRecord, new() - { - global::Avro.Schema _schema; /// - /// Use this property to get the of + /// Avro Value extension of for Json encoding, for example /// - public global::Avro.Schema Schema - { - get { return _schema; } - private set + /// + sealed class JsonBuffered : SerDesBuffered where TData : global::Avro.Specific.ISpecificRecord, new() + { + global::Avro.Schema _schema; + /// + /// Use this property to get the of + /// + public global::Avro.Schema Schema { - _schema = value; - SpecificWriter = new SpecificDefaultWriter(_schema); - SpecificReader = new SpecificDefaultReader(_schema, _schema); + get { return _schema; } + private set + { + _schema = value; + SpecificWriter = new SpecificDefaultWriter(_schema); + SpecificReader = new SpecificDefaultReader(_schema, _schema); + } } - } - - SpecificDefaultWriter SpecificWriter = null; - SpecificDefaultReader SpecificReader = null; - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(JsonBuffered<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; - /// - /// Default initializer - /// - public JsonBuffered() - { - var tRecord = new T(); - Schema = tRecord.Schema; - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + SpecificDefaultWriter SpecificWriter = null; + SpecificDefaultReader SpecificReader = null; + + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public JsonBuffered(string selectorName) + { + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + var tRecord = new T(); + Schema = tRecord.Schema; + } + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); - MemoryStream memStream = new(); - JsonEncoder encoder = new(Schema, memStream); - SpecificWriter.Write(data, encoder); - return ByteBuffer.From(memStream); - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (data == null) return default; + MemoryStream memStream = new(); + JsonEncoder encoder = new(Schema, memStream); + SpecificWriter.Write(data, encoder); + return ByteBuffer.From(memStream); + } + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (data == null) return default; - JsonDecoder decoder = new(Schema, data.ToStream()); - T t = new T(); - t = SpecificReader.Read(t!, decoder); - return t; + JsonDecoder decoder = new(Schema, data.ToStream()); + TData t = new TData(); + t = SpecificReader.Read(t!, decoder); + return t; + } } } } diff --git a/src/net/KNet.Serialization.Json/JsonSerDes.cs b/src/net/KNet.Serialization.Json/JsonSerDes.cs index 025fe0ede9..c90cd41a2b 100644 --- a/src/net/KNet.Serialization.Json/JsonSerDes.cs +++ b/src/net/KNet.Serialization.Json/JsonSerDes.cs @@ -31,148 +31,191 @@ namespace MASES.KNet.Serialization.Json { /// - /// Base class to define extensions of for Json, for example + /// Base class to define extensions of for Json, for example /// public static class JsonSerDes { /// - /// Json extension of for Key, for example + /// Json extension of for Key, for example /// /// - public class KeyRaw : SerDesRaw + public class Key : ISerDesSelector { - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(KeyRaw<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = null; - readonly ISerDesRaw _defaultSerDes = default!; -#if NET462_OR_GREATER - /// - /// Settings used from and - /// - public JsonSerializerSettings Options { get; set; } = new JsonSerializerSettings(); -#else /// - /// Settings used from and + /// Returns a new instance of /// - public JsonSerializerOptions Options { get; set; } = new JsonSerializerOptions(); -#endif - /// - public override bool UseHeaders => true; + /// The of + public static ISerDesSelector NewInstance() => new Key(); + /// + public static string SelectorTypeName => typeof(Key<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(KeyRaw); + /// + public static Type ByteBufferSerDes => typeof(KeyBuffered); + /// + public static ISerDes NewSerDes() + { + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; + } + /// + public static ISerDesRaw NewByteArraySerDes() { return new KeyRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new KeyBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); + /// - /// Default initializer + /// Json extension of for Key, for example /// - public KeyRaw() + /// + sealed class KeyRaw : SerDesRaw { - if (KNetSerialization.IsInternalManaged()) + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = null; + readonly ISerDesRaw _defaultSerDes = default!; +#if NET462_OR_GREATER + /// + /// Settings used from and + /// + public JsonSerializerSettings Options { get; set; } = new JsonSerializerSettings(); +#else + /// + /// Settings used from and + /// + public JsonSerializerOptions Options { get; set; } = new JsonSerializerOptions(); +#endif + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public KeyRaw(string selectorName) { - _defaultSerDes = new SerDesRaw(); - keyTypeName = Encoding.UTF8.GetBytes(typeof(T).FullName!); + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + if (KNetSerialization.IsInternalManaged()) + { + _defaultSerDes = new SerDesRaw(); + keyTypeName = Encoding.UTF8.GetBytes(typeof(TData).FullName!); + } + else + { + keyTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + } } - else + /// + public override byte[] Serialize(string topic, TData data) { - keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); + return SerializeWithHeaders(topic, null, data); } - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); - if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data); + if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data); #if NET462_OR_GREATER - var jsonStr = Newtonsoft.Json.JsonConvert.SerializeObject(data, Options); - return Encoding.UTF8.GetBytes(jsonStr); + var jsonStr = Newtonsoft.Json.JsonConvert.SerializeObject(data, Options); + return Encoding.UTF8.GetBytes(jsonStr); #else - var jsonStr = System.Text.Json.JsonSerializer.Serialize(data, Options); - return Encoding.UTF8.GetBytes(jsonStr); + var jsonStr = System.Text.Json.JsonSerializer.Serialize(data, Options); + return Encoding.UTF8.GetBytes(jsonStr); #endif - } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (_defaultSerDes != null) return _defaultSerDes.DeserializeWithHeaders(topic, headers, data); - if (data == null) return default; + } + /// + public override TData Deserialize(string topic, byte[] data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (_defaultSerDes != null) return _defaultSerDes.DeserializeWithHeaders(topic, headers, data); + if (data == null) return default; #if NET462_OR_GREATER - var jsonStr = Encoding.UTF8.GetString(data); - return Newtonsoft.Json.JsonConvert.DeserializeObject(jsonStr, Options); + var jsonStr = Encoding.UTF8.GetString(data); + return Newtonsoft.Json.JsonConvert.DeserializeObject(jsonStr, Options); #else - return System.Text.Json.JsonSerializer.Deserialize(data, Options)!; + return System.Text.Json.JsonSerializer.Deserialize(data, Options)!; #endif + } } - } - /// - /// Json extension of for Key, for example - /// - /// - public class KeyBuffered : SerDesBuffered - { + /// + /// Json extension of for Key, for example + /// + /// + sealed class KeyBuffered : SerDesBuffered + { #if NET462_OR_GREATER - readonly Newtonsoft.Json.JsonSerializer _serializer; + readonly Newtonsoft.Json.JsonSerializer _serializer; #endif - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(KeyBuffered<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = null; - readonly ISerDesBuffered _defaultSerDes = default!; + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = null; + readonly ISerDesBuffered _defaultSerDes = default!; #if NET462_OR_GREATER - /// - /// When set to the oprion forces to use with to reduce memory copy - /// - /// Added specifically to .NET Framework because NewtonSoft JSon seems to have problems with large streams - public bool UseStreamWithByteBuffer { get; set; } = true; - /// - /// Settings used from and - /// - public JsonSerializerSettings Options { get; set; } = new JsonSerializerSettings(); + /// + /// When set to the oprion forces to use with to reduce memory copy + /// + /// Added specifically to .NET Framework because NewtonSoft JSon seems to have problems with large streams + public bool UseStreamWithByteBuffer { get; set; } = true; + /// + /// Settings used from and + /// + public JsonSerializerSettings Options { get; set; } = new JsonSerializerSettings(); #else - /// - /// Settings used from and - /// - public JsonSerializerOptions Options { get; set; } = new JsonSerializerOptions(); + /// + /// Settings used from and + /// + public JsonSerializerOptions Options { get; set; } = new JsonSerializerOptions(); #endif - /// - public override bool UseHeaders => true; - /// - /// Default initializer - /// - public KeyBuffered() - { - if (KNetSerialization.IsInternalManaged()) - { - _defaultSerDes = new SerDesBuffered(); - keyTypeName = Encoding.UTF8.GetBytes(typeof(T).FullName!); - } - else + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public KeyBuffered(string selectorName) { + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + if (KNetSerialization.IsInternalManaged()) + { + _defaultSerDes = new SerDesBuffered(); + keyTypeName = Encoding.UTF8.GetBytes(typeof(TData).FullName!); + } + else + { #if NET462_OR_GREATER _serializer = new Newtonsoft.Json.JsonSerializer(); _serializer.Formatting = Newtonsoft.Json.Formatting.None; #endif - keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); + keyTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + } } - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); - if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data); + if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data); #if NET462_OR_GREATER if (UseStreamWithByteBuffer) @@ -191,231 +234,276 @@ public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers h return ByteBuffer.From(Encoding.UTF8.GetBytes(jsonStr)); } #else - var ms = new MemoryStream(); - System.Text.Json.JsonSerializer.Serialize(ms, data, Options); - return ByteBuffer.From(ms); + var ms = new MemoryStream(); + System.Text.Json.JsonSerializer.Serialize(ms, data, Options); + return ByteBuffer.From(ms); #endif - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (_defaultSerDes != null) return _defaultSerDes.DeserializeWithHeaders(topic, headers, data); - if (data == null) return default; + } + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (_defaultSerDes != null) return _defaultSerDes.DeserializeWithHeaders(topic, headers, data); + if (data == null) return default; #if NET462_OR_GREATER if (UseStreamWithByteBuffer) { using (StreamReader sw = new StreamReader(data.ToStream())) using (Newtonsoft.Json.JsonReader reader = new Newtonsoft.Json.JsonTextReader(sw)) { - return _serializer.Deserialize(reader); + return _serializer.Deserialize(reader); } } else { var jsonStr = Encoding.UTF8.GetString((byte[])data); - return Newtonsoft.Json.JsonConvert.DeserializeObject(jsonStr, Options); + return Newtonsoft.Json.JsonConvert.DeserializeObject(jsonStr, Options); } #else - return System.Text.Json.JsonSerializer.Deserialize(data.ToStream(), Options)!; + return System.Text.Json.JsonSerializer.Deserialize(data.ToStream(), Options)!; #endif + } } } /// - /// Json extension of for Value, for example + /// Json extension of for Value, for example /// /// - public class ValueRaw : SerDesRaw + public class Value : ISerDesSelector { - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(ValueRaw<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = null!; - readonly ISerDesRaw _defaultSerDes = default!; -#if NET462_OR_GREATER /// - /// Settings used from and + /// Returns a new instance of /// - public JsonSerializerSettings Options { get; set; } = new JsonSerializerSettings(); -#else - /// - /// Settings used from and - /// - public JsonSerializerOptions Options { get; set; } = new JsonSerializerOptions(); -#endif - /// - public override bool UseHeaders => true; + /// The of + public static ISerDesSelector NewInstance() => new Value(); + /// + public static string SelectorTypeName => typeof(Value<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(ValueRaw); + /// + public static Type ByteBufferSerDes => typeof(ValueBuffered); + /// + public static ISerDes NewSerDes() + { + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; + } + /// + public static ISerDesRaw NewByteArraySerDes() { return new ValueRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new ValueBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); + /// - /// Default initializer + /// Json extension of for Value, for example /// - public ValueRaw() + /// + sealed class ValueRaw : SerDesRaw { - if (KNetSerialization.IsInternalManaged()) + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = null!; + readonly ISerDesRaw _defaultSerDes = default!; +#if NET462_OR_GREATER + /// + /// Settings used from and + /// + public JsonSerializerSettings Options { get; set; } = new JsonSerializerSettings(); +#else + /// + /// Settings used from and + /// + public JsonSerializerOptions Options { get; set; } = new JsonSerializerOptions(); +#endif + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public ValueRaw(string selectorName) { - _defaultSerDes = new SerDesRaw(); - valueTypeName = Encoding.UTF8.GetBytes(typeof(T).FullName!); + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + if (KNetSerialization.IsInternalManaged()) + { + _defaultSerDes = new SerDesRaw(); + valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).FullName!); + } + else + { + valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + } } - else + /// + public override byte[] Serialize(string topic, TData data) { - valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); + return SerializeWithHeaders(topic, null, data); } - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); - if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data); + if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data); #if NET462_OR_GREATER - var jsonStr = Newtonsoft.Json.JsonConvert.SerializeObject(data, Options); - return Encoding.UTF8.GetBytes(jsonStr); + var jsonStr = Newtonsoft.Json.JsonConvert.SerializeObject(data, Options); + return Encoding.UTF8.GetBytes(jsonStr); #else - var jsonStr = System.Text.Json.JsonSerializer.Serialize(data, Options); - return Encoding.UTF8.GetBytes(jsonStr); + var jsonStr = System.Text.Json.JsonSerializer.Serialize(data, Options); + return Encoding.UTF8.GetBytes(jsonStr); #endif - } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (_defaultSerDes != null) return _defaultSerDes.DeserializeWithHeaders(topic, headers, data); + } + /// + public override TData Deserialize(string topic, byte[] data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (_defaultSerDes != null) return _defaultSerDes.DeserializeWithHeaders(topic, headers, data); - if (data == null) return default; + if (data == null) return default; #if NET462_OR_GREATER - var jsonStr = Encoding.UTF8.GetString(data); - return Newtonsoft.Json.JsonConvert.DeserializeObject(jsonStr, Options); + var jsonStr = Encoding.UTF8.GetString(data); + return Newtonsoft.Json.JsonConvert.DeserializeObject(jsonStr, Options); #else - return System.Text.Json.JsonSerializer.Deserialize(data, Options)!; + return System.Text.Json.JsonSerializer.Deserialize(data, Options)!; #endif + } } - } - /// - /// Json extension of for Value, for example - /// - /// - public class ValueBuffered : SerDesBuffered - { + /// + /// Json extension of for Value, for example + /// + /// + sealed class ValueBuffered : SerDesBuffered + { #if NET462_OR_GREATER - readonly Newtonsoft.Json.JsonSerializer _serializer; + readonly Newtonsoft.Json.JsonSerializer _serializer; #endif - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(ValueBuffered<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = null!; - readonly ISerDesBuffered _defaultSerDes = default!; + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = null!; + readonly ISerDesBuffered _defaultSerDes = default!; #if NET462_OR_GREATER - /// - /// When set to the oprion forces to use with to reduce memory copy - /// - /// Added specifically to .NET Framework because NewtonSoft JSon seems to have problems with large streams - public bool UseByteBufferWithStream { get; set; } = true; - /// - /// Settings used from and - /// - public JsonSerializerSettings Options { get; set; } = new JsonSerializerSettings(); + /// + /// When set to the oprion forces to use with to reduce memory copy + /// + /// Added specifically to .NET Framework because NewtonSoft JSon seems to have problems with large streams + public bool UseByteBufferWithStream { get; set; } = true; + /// + /// Settings used from and + /// + public JsonSerializerSettings Options { get; set; } = new JsonSerializerSettings(); #else - /// - /// Settings used from and - /// - public JsonSerializerOptions Options { get; set; } = new JsonSerializerOptions(); + /// + /// Settings used from and + /// + public JsonSerializerOptions Options { get; set; } = new JsonSerializerOptions(); #endif - /// - public override bool UseHeaders => true; - /// - /// Default initializer - /// - public ValueBuffered() - { - if (KNetSerialization.IsInternalManaged()) - { - _defaultSerDes = new SerDesBuffered(); - valueTypeName = Encoding.UTF8.GetBytes(typeof(T).FullName!); - } - else + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public ValueBuffered(string selectorName) { + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + if (KNetSerialization.IsInternalManaged()) + { + _defaultSerDes = new SerDesBuffered(); + valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).FullName!); + } + else + { #if NET462_OR_GREATER - _serializer = new Newtonsoft.Json.JsonSerializer(); - _serializer.Formatting = Newtonsoft.Json.Formatting.None; + _serializer = new Newtonsoft.Json.JsonSerializer(); + _serializer.Formatting = Newtonsoft.Json.Formatting.None; #endif - valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); + valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + } } - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); - if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data); + if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data); #if NET462_OR_GREATER - if (UseByteBufferWithStream) - { - var ms = new MemoryStream(); - using (StreamWriter sw = new StreamWriter(ms, new UTF8Encoding(false), 128, true)) - using (Newtonsoft.Json.JsonWriter writer = new Newtonsoft.Json.JsonTextWriter(sw)) + if (UseByteBufferWithStream) { - _serializer.Serialize(writer, data); + var ms = new MemoryStream(); + using (StreamWriter sw = new StreamWriter(ms, new UTF8Encoding(false), 128, true)) + using (Newtonsoft.Json.JsonWriter writer = new Newtonsoft.Json.JsonTextWriter(sw)) + { + _serializer.Serialize(writer, data); + } + return ByteBuffer.From(ms); } + else + { + var jsonStr = Newtonsoft.Json.JsonConvert.SerializeObject(data, Options); + return ByteBuffer.From(Encoding.UTF8.GetBytes(jsonStr)); + } +#else + var ms = new MemoryStream(); + System.Text.Json.JsonSerializer.Serialize(ms, data, Options); return ByteBuffer.From(ms); +#endif } - else + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) { - var jsonStr = Newtonsoft.Json.JsonConvert.SerializeObject(data, Options); - return ByteBuffer.From(Encoding.UTF8.GetBytes(jsonStr)); + return DeserializeWithHeaders(topic, null, data); } -#else - var ms = new MemoryStream(); - System.Text.Json.JsonSerializer.Serialize(ms, data, Options); - return ByteBuffer.From(ms); -#endif - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (_defaultSerDes != null) return _defaultSerDes.DeserializeWithHeaders(topic, headers, data); + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (_defaultSerDes != null) return _defaultSerDes.DeserializeWithHeaders(topic, headers, data); - if (data == null) return default; + if (data == null) return default; #if NET462_OR_GREATER - if (UseByteBufferWithStream) - { - using (StreamReader sw = new StreamReader(data.ToStream())) - using (Newtonsoft.Json.JsonReader reader = new Newtonsoft.Json.JsonTextReader(sw)) + if (UseByteBufferWithStream) { - return _serializer.Deserialize(reader); + using (StreamReader sw = new StreamReader(data.ToStream())) + using (Newtonsoft.Json.JsonReader reader = new Newtonsoft.Json.JsonTextReader(sw)) + { + return _serializer.Deserialize(reader); + } + } + else + { + var jsonStr = Encoding.UTF8.GetString((byte[])data); + return Newtonsoft.Json.JsonConvert.DeserializeObject(jsonStr, Options); } - } - else - { - var jsonStr = Encoding.UTF8.GetString((byte[])data); - return Newtonsoft.Json.JsonConvert.DeserializeObject(jsonStr, Options); - } #else - return System.Text.Json.JsonSerializer.Deserialize(data.ToStream(), Options)!; + return System.Text.Json.JsonSerializer.Deserialize(data.ToStream(), Options)!; #endif + } } } } diff --git a/src/net/KNet.Serialization.MessagePack/MessagePackSerDes.cs b/src/net/KNet.Serialization.MessagePack/MessagePackSerDes.cs index 49e507e528..29010b0aa6 100644 --- a/src/net/KNet.Serialization.MessagePack/MessagePackSerDes.cs +++ b/src/net/KNet.Serialization.MessagePack/MessagePackSerDes.cs @@ -26,217 +26,305 @@ namespace MASES.KNet.Serialization.MessagePack { /// - /// Base class to define extensions of for MessagePack, for example + /// Base class to define extensions of for MessagePack, for example /// public static class MessagePackSerDes { /// - /// MessagePack extension of for Key, for example + /// MessagePack extension of for Key, for example /// /// - public class KeyRaw : SerDesRaw + public class Key : ISerDesSelector { - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(KeyRaw<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); /// - /// Get or set the to be used, default is + /// Returns a new instance of /// - public MessagePackSerializerOptions MessagePackSerializerOptions { get; set; } = null; - /// - public override bool UseHeaders => true; + /// The of + public static ISerDesSelector NewInstance() => new Key(); + /// + public static string SelectorTypeName => typeof(Key<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(KeyRaw); + /// + public static Type ByteBufferSerDes => typeof(KeyBuffered); + /// + public static ISerDes NewSerDes() + { + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; + } + /// + public static ISerDesRaw NewByteArraySerDes() { return new KeyRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new KeyBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); + /// - /// Default initializer + /// MessagePack extension of for Key, for example /// - public KeyRaw() + /// + sealed class KeyRaw : SerDesRaw { - if (KNetSerialization.IsInternalManaged()) + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + /// Get or set the to be used, default is + /// + public MessagePackSerializerOptions MessagePackSerializerOptions { get; set; } = null; + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public KeyRaw(string selectorName) { - throw new InvalidOperationException($"{typeof(T).Name} is a type managed from basic serializer, do not use {typeof(KeyRaw).FullName}"); + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + if (KNetSerialization.IsInternalManaged()) + { + throw new InvalidOperationException($"{typeof(TData).Name} is a type managed from basic serializer, do not use {typeof(KeyRaw).FullName}"); + } } - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + /// + public override byte[] Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); - return MessagePackSerializer.Serialize(data, MessagePackSerializerOptions); - } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (data == null) return default; - using (MemoryStream stream = new MemoryStream(data)) + return MessagePackSerializer.Serialize(data, MessagePackSerializerOptions); + } + /// + public override TData Deserialize(string topic, byte[] data) { - return MessagePackSerializer.Deserialize(stream, MessagePackSerializerOptions); + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (data == null) return default; + using (MemoryStream stream = new MemoryStream(data)) + { + return MessagePackSerializer.Deserialize(stream, MessagePackSerializerOptions); + } } } - } - /// - /// MessagePack extension of for Key, for example - /// - /// - public class KeyBuffered : SerDesBuffered - { - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(KeyBuffered<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); /// - /// Get or set the to be used, default is + /// MessagePack extension of for Key, for example /// - public MessagePackSerializerOptions MessagePackSerializerOptions { get; set; } = null; - /// - public override bool UseHeaders => true; - /// - /// Default initializer - /// - public KeyBuffered() + /// + sealed class KeyBuffered : SerDesBuffered { - if (KNetSerialization.IsInternalManaged()) + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + /// Get or set the to be used, default is + /// + public MessagePackSerializerOptions MessagePackSerializerOptions { get; set; } = null; + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public KeyBuffered(string selectorName) { - throw new InvalidOperationException($"{typeof(T).Name} is a type managed from basic serializer, do not use {typeof(KeyBuffered).FullName}"); + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + if (KNetSerialization.IsInternalManaged()) + { + throw new InvalidOperationException($"{typeof(TData).Name} is a type managed from basic serializer, do not use {typeof(KeyBuffered).FullName}"); + } } - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); - var ms = new MemoryStream(); - MessagePackSerializer.Serialize(ms, data, MessagePackSerializerOptions); - return ByteBuffer.From(ms); - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (data == null) return default; - return MessagePackSerializer.Deserialize(data.ToStream(), MessagePackSerializerOptions); + var ms = new MemoryStream(); + MessagePackSerializer.Serialize(ms, data, MessagePackSerializerOptions); + return ByteBuffer.From(ms); + } + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (data == null) return default; + return MessagePackSerializer.Deserialize(data.ToStream(), MessagePackSerializerOptions); + } } } /// - /// MessagePack extension of for Value, for example + /// MessagePack extension of for Value, for example /// /// - public class ValueRaw : SerDesRaw + public class Value : ISerDesSelector { - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(ValueRaw<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); /// - /// Get or set the to be used, default is + /// Returns a new instance of /// - public MessagePackSerializerOptions MessagePackSerializerOptions { get; set; } = null; - /// - public override bool UseHeaders => true; + /// The of + public static ISerDesSelector NewInstance() => new Value(); + /// + public static string SelectorTypeName => typeof(Value<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(ValueRaw); + /// + public static Type ByteBufferSerDes => typeof(ValueBuffered); + /// + public static ISerDes NewSerDes() + { + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; + } + /// + public static ISerDesRaw NewByteArraySerDes() { return new ValueRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new ValueBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); + /// - /// Default initializer + /// MessagePack extension of for Value, for example /// - public ValueRaw() + /// + sealed class ValueRaw : SerDesRaw { - if (KNetSerialization.IsInternalManaged()) + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + /// Get or set the to be used, default is + /// + public MessagePackSerializerOptions MessagePackSerializerOptions { get; set; } = null; + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public ValueRaw(string selectorName) { - throw new InvalidOperationException($"{typeof(T).Name} is a type managed from basic serializer, do not use {typeof(ValueRaw).FullName}"); + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + if (KNetSerialization.IsInternalManaged()) + { + throw new InvalidOperationException($"{typeof(TData).Name} is a type managed from basic serializer, do not use {typeof(ValueRaw).FullName}"); + } } - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + /// + public override byte[] Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); - return MessagePackSerializer.Serialize(data, MessagePackSerializerOptions); - } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (data == null) return default; - using (MemoryStream stream = new MemoryStream(data)) + return MessagePackSerializer.Serialize(data, MessagePackSerializerOptions); + } + /// + public override TData Deserialize(string topic, byte[] data) { - return MessagePackSerializer.Deserialize(stream, MessagePackSerializerOptions); + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (data == null) return default; + using (MemoryStream stream = new MemoryStream(data)) + { + return MessagePackSerializer.Deserialize(stream, MessagePackSerializerOptions); + } } } - } - /// - /// MessagePack extension of for Value, for example - /// - /// - public class ValueBuffered : SerDesBuffered - { - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(ValueBuffered<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); /// - /// Get or set the to be used, default is + /// MessagePack extension of for Value, for example /// - public MessagePackSerializerOptions MessagePackSerializerOptions { get; set; } = null; - /// - public override bool UseHeaders => true; - /// - /// Default initializer - /// - public ValueBuffered() + /// + sealed class ValueBuffered : SerDesBuffered { - if (KNetSerialization.IsInternalManaged()) + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + /// Get or set the to be used, default is + /// + public MessagePackSerializerOptions MessagePackSerializerOptions { get; set; } = null; + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public ValueBuffered(string selectorName) { - throw new InvalidOperationException($"{typeof(T).Name} is a type managed from basic serializer, do not use {typeof(ValueBuffered).FullName}"); + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + if (KNetSerialization.IsInternalManaged()) + { + throw new InvalidOperationException($"{typeof(TData).Name} is a type managed from basic serializer, do not use {typeof(ValueBuffered).FullName}"); + } } - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) - { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); - var ms = new MemoryStream(); - MessagePackSerializer.Serialize(ms, data, MessagePackSerializerOptions); - return ByteBuffer.From(ms); - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (data == null) return default; - return MessagePackSerializer.Deserialize(data.ToStream(), MessagePackSerializerOptions); + var ms = new MemoryStream(); + MessagePackSerializer.Serialize(ms, data, MessagePackSerializerOptions); + return ByteBuffer.From(ms); + } + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) + { + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (data == null) return default; + return MessagePackSerializer.Deserialize(data.ToStream(), MessagePackSerializerOptions); + } } } } diff --git a/src/net/KNet.Serialization.Protobuf/ProtobufSerDes.cs b/src/net/KNet.Serialization.Protobuf/ProtobufSerDes.cs index fcc8c87a6b..a5a2cf6c15 100644 --- a/src/net/KNet.Serialization.Protobuf/ProtobufSerDes.cs +++ b/src/net/KNet.Serialization.Protobuf/ProtobufSerDes.cs @@ -27,199 +27,283 @@ namespace MASES.KNet.Serialization.Protobuf { /// - /// Base class to define extensions of for Protobuf, for example + /// Base class to define extensions of for Protobuf, for example /// public static class ProtobufSerDes { /// - /// Protobuf extension of for Key, for example + /// Protobuf extension of for Key, for example /// /// - public class KeyRaw : SerDesRaw where T : IMessage, new() + public class Key : ISerDesSelector where T : IMessage, new() { - readonly MessageParser _parser = new MessageParser(() => new T()); - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(KeyRaw<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; /// - /// Default initializer + /// Returns a new instance of /// - public KeyRaw() + /// The of + public static ISerDesSelector NewInstance() => new Key(); + /// + public static string SelectorTypeName => typeof(Key<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(KeyRaw); + /// + public static Type ByteBufferSerDes => typeof(KeyBuffered); + /// + public static ISerDes NewSerDes() { - - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) + /// + public static ISerDesRaw NewByteArraySerDes() { return new KeyRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new KeyBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); + + /// + /// Protobuf extension of for Key, for example + /// + /// + sealed class KeyRaw : SerDesRaw where TData : IMessage, new() { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + readonly MessageParser _parser = new MessageParser(() => new TData()); + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public KeyRaw(string selectorName) + { + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + } + /// + public override byte[] Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); - using (MemoryStream stream = new MemoryStream()) + using (MemoryStream stream = new MemoryStream()) + { + data.WriteTo(stream); + return stream.ToArray(); + } + } + /// + public override TData Deserialize(string topic, byte[] data) { - data.WriteTo(stream); - return stream.ToArray(); + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (data == null) return default; + return _parser.ParseFrom(data); } } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (data == null) return default; - return _parser.ParseFrom(data); - } - } - /// - /// Protobuf extension of for Key, for example - /// - /// - public class KeyBuffered : SerDesBuffered where T : IMessage, new() - { - readonly MessageParser _parser = new MessageParser(() => new T()); - readonly byte[] keySerDesName = Encoding.UTF8.GetBytes(typeof(KeyBuffered<>).ToAssemblyQualified()); - readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; /// - /// Default initializer + /// Protobuf extension of for Key, for example /// - public KeyBuffered() - { - - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) + /// + sealed class KeyBuffered : SerDesBuffered where TData : IMessage, new() { - headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); - headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); + readonly MessageParser _parser = new MessageParser(() => new TData()); + readonly byte[] keySerDesName; + readonly byte[] keyTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public KeyBuffered(string selectorName) + { + keySerDesName = Encoding.UTF8.GetBytes(selectorName); + } + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName); + headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName); - MemoryStream stream = new MemoryStream(); + MemoryStream stream = new MemoryStream(); + { + data.WriteTo(stream); + return ByteBuffer.From(stream); + } + } + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) { - data.WriteTo(stream); - return ByteBuffer.From(stream); + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (data == null) return default; + return _parser.ParseFrom(data.ToStream()); } - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (data == null) return default; - return _parser.ParseFrom(data.ToStream()); } } /// - /// Protobuf extension of for Value, for example + /// MessagePack extension of for Value, for example /// /// - public class ValueRaw : SerDesRaw where T : IMessage, new() + public class Value : ISerDesSelector where T : IMessage, new() { - readonly MessageParser _parser = new MessageParser(() => new T()); - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(ValueRaw<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; /// - /// Default initializer + /// Returns a new instance of /// - public ValueRaw() + /// The of + public static ISerDesSelector NewInstance() => new Value(); + /// + public static string SelectorTypeName => typeof(Value<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(ValueRaw); + /// + public static Type ByteBufferSerDes => typeof(ValueBuffered); + /// + public static ISerDes NewSerDes() { - - } - /// - public override byte[] Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; } - /// - public override byte[] SerializeWithHeaders(string topic, Headers headers, T data) + /// + public static ISerDesRaw NewByteArraySerDes() { return new ValueRaw(SelectorTypeName); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new ValueBuffered(SelectorTypeName); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); + + /// + /// Protobuf extension of for Value, for example + /// + /// + sealed class ValueRaw : SerDesRaw where TData : IMessage, new() { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + readonly MessageParser _parser = new MessageParser(() => new TData()); + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public ValueRaw(string selectorName) + { + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + } + /// + public override byte[] Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); - using (MemoryStream stream = new MemoryStream()) + using (MemoryStream stream = new MemoryStream()) + { + data.WriteTo(stream); + return stream.ToArray(); + } + } + /// + public override TData Deserialize(string topic, byte[] data) { - data.WriteTo(stream); - return stream.ToArray(); + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, byte[] data) + { + if (data == null) return default; + return _parser.ParseFrom(data); } } - /// - public override T Deserialize(string topic, byte[] data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, byte[] data) - { - if (data == null) return default; - return _parser.ParseFrom(data); - } - } - /// - /// Protobuf extension of for Value, for example - /// - /// - public class ValueBuffered : SerDesBuffered where T : IMessage, new() - { - readonly MessageParser _parser = new MessageParser(() => new T()); - readonly byte[] valueSerDesName = Encoding.UTF8.GetBytes(typeof(ValueBuffered<>).ToAssemblyQualified()); - readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(T).ToAssemblyQualified()); - /// - public override bool UseHeaders => true; /// - /// Default initializer + /// Protobuf extension of for Value, for example /// - public ValueBuffered() - { - - } - /// - public override Java.Nio.ByteBuffer Serialize(string topic, T data) - { - return SerializeWithHeaders(topic, null, data); - } - /// - public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, T data) + /// + sealed class ValueBuffered : SerDesBuffered where TData : IMessage, new() { - headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); - headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); + readonly MessageParser _parser = new MessageParser(() => new TData()); + readonly byte[] valueSerDesName; + readonly byte[] valueTypeName = Encoding.UTF8.GetBytes(typeof(TData).ToAssemblyQualified()); + /// + public override bool UseHeaders => true; + /// + /// Default initializer + /// + public ValueBuffered(string selectorName) + { + valueSerDesName = Encoding.UTF8.GetBytes(selectorName); + } + /// + public override Java.Nio.ByteBuffer Serialize(string topic, TData data) + { + return SerializeWithHeaders(topic, null, data); + } + /// + public override Java.Nio.ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data) + { + headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueSerDesName); + headers?.Add(KNetSerialization.ValueTypeIdentifier, valueTypeName); - MemoryStream stream = new MemoryStream(); + MemoryStream stream = new MemoryStream(); + { + data.WriteTo(stream); + return ByteBuffer.From(stream); + } + } + /// + public override TData Deserialize(string topic, Java.Nio.ByteBuffer data) { - data.WriteTo(stream); - return ByteBuffer.From(stream); + return DeserializeWithHeaders(topic, null, data); + } + /// + public override TData DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) + { + if (data == null) return default; + return _parser.ParseFrom(data.ToStream()); } - } - /// - public override T Deserialize(string topic, Java.Nio.ByteBuffer data) - { - return DeserializeWithHeaders(topic, null, data); - } - /// - public override T DeserializeWithHeaders(string topic, Headers headers, Java.Nio.ByteBuffer data) - { - if (data == null) return default; - return _parser.ParseFrom(data.ToStream()); } } } diff --git a/src/net/KNet/Specific/Consumer/KNetConsumer.cs b/src/net/KNet/Specific/Consumer/KNetConsumer.cs index 8e58e35a13..f0d58163a7 100644 --- a/src/net/KNet/Specific/Consumer/KNetConsumer.cs +++ b/src/net/KNet/Specific/Consumer/KNetConsumer.cs @@ -168,27 +168,13 @@ static Properties CheckProperties(Properties props, ISerDes keyDeserializer, ISe { if (!props.ContainsKey(Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)) { - if (!keyDeserializer.IsDirectBuffered) - { - props.Put(Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ClassNameOf()); - } - else - { - props.Put(Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ClassNameOf()); - } + props.Put(Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.JVMDeserializerClassName); } else throw new InvalidOperationException($"KNetConsumer auto manages configuration property {Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG}, remove from configuration."); if (!props.ContainsKey(Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)) { - if (!valueDeserializer.IsDirectBuffered) - { - props.Put(Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ClassNameOf()); - } - else - { - props.Put(Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ClassNameOf()); - } + props.Put(Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.JVMDeserializerClassName); } else throw new InvalidOperationException($"KNetConsumer auto manages configuration property {Org.Apache.Kafka.Clients.Consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG}, remove from configuration."); diff --git a/src/net/KNet/Specific/GenericConfigBuilder.cs b/src/net/KNet/Specific/GenericConfigBuilder.cs index 7d3a68ddf9..fa1f135bf2 100644 --- a/src/net/KNet/Specific/GenericConfigBuilder.cs +++ b/src/net/KNet/Specific/GenericConfigBuilder.cs @@ -49,8 +49,8 @@ public static T CreateFrom(T origin) var newT = new T { _options = new System.Collections.Generic.Dictionary(origin._options), - _KNetKeySerDes = origin._KNetKeySerDes, - _KNetValueSerDes = origin._KNetValueSerDes, + _KeySerDesSelector = origin._KeySerDesSelector, + _ValueSerDesSelector = origin._ValueSerDesSelector, }; return newT; } @@ -112,8 +112,8 @@ protected virtual T Clone() var clone = new T { _options = new System.Collections.Generic.Dictionary(_options), - _KNetKeySerDes = _KNetKeySerDes, - _KNetValueSerDes = _KNetValueSerDes + _KeySerDesSelector = _KeySerDesSelector, + _ValueSerDesSelector = _ValueSerDesSelector }; return clone; } @@ -150,60 +150,61 @@ public Properties ToProperties() /// public bool AutoSelectBuffered { get; set; } - Type _KNetKeySerDes = null; - /// - public Type KNetKeySerDes + Type _KeySerDesSelector = null; + /// + public Type KeySerDesSelector { - get { return _KNetKeySerDes; } + get { return _KeySerDesSelector; } set { if (value.GetConstructors().Single(ci => ci.GetParameters().Length == 0) == null) { - throw new ArgumentException($"{value.Name} does not contains a default constructor and cannot be used because it is not a valid Serializer type"); + throw new ArgumentException($"{value.Name} does not contains a default constructor and cannot be used because it is not a valid ISerDesSelector type"); } if (value.IsGenericType) { var keyT = value.GetGenericArguments(); - if (keyT.Length != 1) { throw new ArgumentException($"{value.Name} does not contains a single generic argument and cannot be used because it is not a valid Serializer type"); } + if (keyT.Length != 1) { throw new ArgumentException($"{value.Name} does not contains a single generic argument and cannot be used because it is not a valid ISerDesSelector type"); } var t = value.GetGenericTypeDefinition(); - if (t.GetInterface(typeof(ISerDes<,>).Name) == null) + if (t.GetInterface(typeof(ISerDesSelector<>).Name) == null) { - throw new ArgumentException($"{value.Name} does not implement IKNetSerDes<> and cannot be used because it is not a valid Serializer type"); + throw new ArgumentException($"{value.Name} does not implement ISerDesSelector<> and cannot be used because it is not a valid ISerDesSelector type"); } - _KNetKeySerDes = value; + _KeySerDesSelector = value; } - else throw new ArgumentException($"{value.Name} is not a generic type and cannot be used as a valid ValueContainer type"); + else throw new ArgumentException($"{value.Name} is not a generic type and cannot be used as a valid ISerDesSelector type"); } } - Type _KNetValueSerDes = null; - /// - public Type KNetValueSerDes + Type _ValueSerDesSelector = null; + /// + public Type ValueSerDesSelector { - get { return _KNetValueSerDes; } + get { return _ValueSerDesSelector; } set { if (value.GetConstructors().Single(ci => ci.GetParameters().Length == 0) == null) { - throw new ArgumentException($"{value.Name} does not contains a default constructor and cannot be used because it is not a valid Serializer type"); + throw new ArgumentException($"{value.Name} does not contains a default constructor and cannot be used because it is not a valid ISerDesSelector type"); } if (value.IsGenericType) { var keyT = value.GetGenericArguments(); - if (keyT.Length != 1) { throw new ArgumentException($"{value.Name} does not contains a single generic argument and cannot be used because it is not a valid Serializer type"); } + if (keyT.Length != 1) { throw new ArgumentException($"{value.Name} does not contains a single generic argument and cannot be used because it is not a valid ISerDesSelector type"); } var t = value.GetGenericTypeDefinition(); - if (t.GetInterface(typeof(ISerDes<,>).Name) == null) + if (t.GetInterface(typeof(ISerDesSelector<>).Name) == null) { - throw new ArgumentException($"{value.Name} does not implement IKNetSerDes<> and cannot be used because it is not a valid Serializer type"); + throw new ArgumentException($"{value.Name} does not implement ISerDesSelector<> and cannot be used because it is not a valid ISerDesSelector type"); } - _KNetValueSerDes = value; + _ValueSerDesSelector = value; } - else throw new ArgumentException($"{value.Name} is not a generic type and cannot be used as a valid Serializer type"); + else throw new ArgumentException($"{value.Name} is not a generic type and cannot be used as a valid ISerDesSelector type"); } } + readonly ConcurrentDictionary<(Type, Type), ISerDesSelector> _keySerDesSelectorComplete = new(); readonly ConcurrentDictionary<(Type, Type), ISerDes> _keySerDesComplete = new(); /// @@ -226,9 +227,15 @@ public ISerDes BuildKeySerDes() } else { - if (KNetKeySerDes == null) throw new InvalidOperationException($"No default serializer available for {typeof(TKey)}, property {nameof(KNetKeySerDes)} shall be set."); - var tmp = KNetKeySerDes.MakeGenericType(typeof(TKey)); - serDes = Activator.CreateInstance(tmp) as ISerDes; + if (KeySerDesSelector == null) throw new InvalidOperationException($"No default serializer available for {typeof(TKey)}, property {nameof(KeySerDesSelector)} shall be set."); + + var selector = _keySerDesSelectorComplete.GetOrAdd((KeySerDesSelector, typeof(TKey)), (o) => + { + var selectorForValue = o.Item1.MakeGenericType(o.Item2); + return Activator.CreateInstance(selectorForValue) as ISerDesSelector; + }) as ISerDesSelector; + + serDes = selector.NewSerDes(); } _keySerDesComplete[(typeof(TKey), typeof(TJVMTKey))] = serDes; } @@ -236,7 +243,7 @@ public ISerDes BuildKeySerDes() } } - + readonly ConcurrentDictionary<(Type, Type), ISerDesSelector> _valueSerDesSelectorComplete = new(); readonly ConcurrentDictionary<(Type, Type), ISerDes> _valueSerDesComplete = new(); /// @@ -259,9 +266,15 @@ public ISerDes BuildValueSerDes() } else { - if (KNetValueSerDes == null) throw new InvalidOperationException($"No default serializer available for {typeof(TValue)}, property {nameof(KNetValueSerDes)} shall be set."); - var tmp = KNetValueSerDes.MakeGenericType(typeof(TValue)); - serDes = Activator.CreateInstance(tmp) as ISerDes; + if (ValueSerDesSelector == null) throw new InvalidOperationException($"No default serializer available for {typeof(TValue)}, property {nameof(ValueSerDesSelector)} shall be set."); + + var selector = _valueSerDesSelectorComplete.GetOrAdd((ValueSerDesSelector, typeof(TValue)), (o) => + { + var selectorForValue = o.Item1.MakeGenericType(o.Item2); + return Activator.CreateInstance(selectorForValue) as ISerDesSelector; + }) as ISerDesSelector; + + serDes = selector.NewSerDes(); } _valueSerDesComplete[(typeof(TValue), typeof(TJVMTValue))] = serDes; } diff --git a/src/net/KNet/Specific/Producer/KNetProducer.cs b/src/net/KNet/Specific/Producer/KNetProducer.cs index 79e81bd006..b84b078101 100644 --- a/src/net/KNet/Specific/Producer/KNetProducer.cs +++ b/src/net/KNet/Specific/Producer/KNetProducer.cs @@ -243,27 +243,13 @@ static Properties CheckProperties(Properties props, ISerDes keySerializer, ISerD { if (!props.ContainsKey(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)) { - if (!keySerializer.IsDirectBuffered) - { - props.Put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ClassNameOf()); - } - else - { - props.Put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ClassNameOf()); - } + props.Put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer.JVMSerializerClassName); } else throw new InvalidOperationException($"KNetProducer auto manages configuration property {ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG}, remove from configuration."); if (!props.ContainsKey(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)) { - if (!valueSerializer.IsDirectBuffered) - { - props.Put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ClassNameOf()); - } - else - { - props.Put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ClassNameOf()); - } + props.Put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer.JVMSerializerClassName); } else throw new InvalidOperationException($"KNetProducer auto manages configuration property {ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG}, remove from configuration."); diff --git a/src/net/KNet/Specific/Replicator/KNetCompactedReplicator.cs b/src/net/KNet/Specific/Replicator/KNetCompactedReplicator.cs index 48f4c6b8f0..e54fe3aa09 100644 --- a/src/net/KNet/Specific/Replicator/KNetCompactedReplicator.cs +++ b/src/net/KNet/Specific/Replicator/KNetCompactedReplicator.cs @@ -187,7 +187,7 @@ public interface IKNetCompactedReplicator : IDictionary /// The used to create an instance of "/> /// - Type KNetKeySerDes { get; } + Type KeySerDesSelector { get; } /// /// Get or set an instance of to use in , by default it creates a default one based on /// @@ -195,7 +195,7 @@ public interface IKNetCompactedReplicator : IDictionary /// The used to create an instance of "/> /// - Type KNetValueSerDes { get; } + Type ValueSerDesSelector { get; } /// /// Get or set an instance of to use in , by default it creates a default one based on /// @@ -448,7 +448,7 @@ public void CopyTo(KeyValuePair[] array, int arrayIndex) static void OnDemandRetrieve(IConsumer consumer, string topic, K key, ILocalDataStorage data) { var topicPartition = new Org.Apache.Kafka.Common.TopicPartition(topic, data.Partition); - var topics = Java.Util.Collections.SingletonList(topicPartition); + var topics = Java.Util.Collections.Singleton(topicPartition); try { consumer.Assign(topics); @@ -540,10 +540,10 @@ public override void OnPartitionsLost(Java.Util.Collection _keySerDes = null; private bool _disposeKeySerDes = false; - private Type _KNetValueSerDes = null; + private Type _ValueSerDesSelector = null; private ISerDes _valueSerDes = null; private bool _disposeValueSerDes = false; @@ -617,60 +617,60 @@ public Func, bool, KeyValuePair public ProducerConfigBuilder ProducerConfig { get { return _producerConfig; } set { CheckStarted(); _producerConfig = value; } } - /// - public Type KNetKeySerDes + /// + public Type KeySerDesSelector { - get { return _KNetKeySerDes; } + get { return _KeySerDesSelector; } set { CheckStarted(); if (value.GetConstructors().Single(ci => ci.GetParameters().Length == 0) == null) { - throw new ArgumentException($"{value.Name} does not contains a default constructor and cannot be used because it is not a valid Serializer type"); + throw new ArgumentException($"{value.Name} does not contains a default constructor and cannot be used because it is not a valid ISerDesSelector type"); } if (value.IsGenericType) { var keyT = value.GetGenericArguments(); - if (keyT.Length != 1) { throw new ArgumentException($"{value.Name} does not contains a single generic argument and cannot be used because it is not a valid Serializer type"); } + if (keyT.Length != 1) { throw new ArgumentException($"{value.Name} does not contains a single generic argument and cannot be used because it is not a valid ISerDesSelector type"); } var t = value.GetGenericTypeDefinition(); - if (t.GetInterface(typeof(ISerDes<,>).Name) == null) + if (t.GetInterface(typeof(ISerDesSelector<>).Name) == null) { - throw new ArgumentException($"{value.Name} does not implement IKNetSerDes<> and cannot be used because it is not a valid Serializer type"); + throw new ArgumentException($"{value.Name} does not implement ISerDesSelector<> and cannot be used because it is not a valid ISerDesSelector type"); } - _KNetKeySerDes = value; + _KeySerDesSelector = value; } - else throw new ArgumentException($"{value.Name} is not a generic type and cannot be used as a valid ValueContainer type"); + else throw new ArgumentException($"{value.Name} is not a generic type and cannot be used as a valid ISerDesSelector type"); } } /// public ISerDes KeySerDes { get { return _keySerDes; } set { CheckStarted(); _keySerDes = value; } } - /// - public Type KNetValueSerDes + /// + public Type ValueSerDesSelector { - get { return _KNetValueSerDes; } + get { return _ValueSerDesSelector; } set { CheckStarted(); if (value.GetConstructors().Single(ci => ci.GetParameters().Length == 0) == null) { - throw new ArgumentException($"{value.Name} does not contains a default constructor and cannot be used because it is not a valid Serializer type"); + throw new ArgumentException($"{value.Name} does not contains a default constructor and cannot be used because it is not a valid ISerDesSelector type"); } if (value.IsGenericType) { var keyT = value.GetGenericArguments(); - if (keyT.Length != 1) { throw new ArgumentException($"{value.Name} does not contains a single generic argument and cannot be used because it is not a valid Serializer type"); } + if (keyT.Length != 1) { throw new ArgumentException($"{value.Name} does not contains a single generic argument and cannot be used because it is not a valid ISerDesSelector type"); } var t = value.GetGenericTypeDefinition(); - if (t.GetInterface(typeof(ISerDes<,>).Name) == null) + if (t.GetInterface(typeof(ISerDesSelector<>).Name) == null) { - throw new ArgumentException($"{value.Name} does not implement IKNetSerDes<> and cannot be used because it is not a valid Serializer type"); + throw new ArgumentException($"{value.Name} does not implement ISerDesSelector<> and cannot be used because it is not a valid ISerDesSelector type"); } - _KNetValueSerDes = value; + _ValueSerDesSelector = value; } - else throw new ArgumentException($"{value.Name} is not a generic type and cannot be used as a valid Serializer type"); + else throw new ArgumentException($"{value.Name} is not a generic type and cannot be used as a valid ISerDesSelector type"); } } @@ -962,17 +962,17 @@ void BuildConsumers() ConsumerConfig.GroupId = GroupId; } - if (KNetKeySerDes != null) + if (KeySerDesSelector != null) { - ConsumerConfig.KNetKeySerDes = KNetKeySerDes; + ConsumerConfig.KeySerDesSelector = KeySerDesSelector; } - else if (KeySerDes == null) throw new InvalidOperationException($"{typeof(K)} needs an external deserializer, set KNetKeySerDes or KeySerDes."); + else if (KeySerDes == null) throw new InvalidOperationException($"{typeof(K)} needs an external deserializer, set {nameof(KeySerDesSelector)} or {nameof(KeySerDes)}."); - if (KNetValueSerDes != null) + if (ValueSerDesSelector != null) { - ConsumerConfig.KNetValueSerDes = KNetValueSerDes; + ConsumerConfig.ValueSerDesSelector = ValueSerDesSelector; } - else if (ValueSerDes == null) throw new InvalidOperationException($"{typeof(V)} needs an external deserializer, set KNetValueSerDes or ValueSerDes."); + else if (ValueSerDes == null) throw new InvalidOperationException($"{typeof(V)} needs an external deserializer, set {nameof(ValueSerDesSelector)} or {nameof(ValueSerDes)}."); _assignmentWaiters = new ManualResetEvent[Partitions]; _assignmentWaitersStatus = new bool[Partitions]; @@ -990,8 +990,8 @@ void BuildConsumers() for (int i = 0; i < ConsumersToAllocate(); i++) { _consumerAssociatedPartition.Add(i, new System.Collections.Generic.List()); - _consumers[i] = (KNetKeySerDes != null || KNetValueSerDes != null) ? new KNetConsumer(ConsumerConfig) - : new KNetConsumer(ConsumerConfig, KeySerDes, ValueSerDes); + _consumers[i] = (KeySerDesSelector != null || ValueSerDesSelector != null) ? new KNetConsumer(ConsumerConfig) + : new KNetConsumer(ConsumerConfig, KeySerDes, ValueSerDes); #if NET7_0_OR_GREATER _consumers[i].ApplyPrefetch(IsPrefecth, PrefetchThreshold); #endif @@ -1013,8 +1013,8 @@ void BuildOnTheFlyConsumer() ConsumerConfigBuilder consumerConfigBuilder = ConsumerConfigBuilder.CreateFrom(_consumerConfig); consumerConfigBuilder.WithEnableAutoCommit(false).WithGroupId(Guid.NewGuid().ToString()); - _onTheFlyConsumer = (KNetKeySerDes != null || KNetValueSerDes != null) ? new KNetConsumer(consumerConfigBuilder) - : new KNetConsumer(consumerConfigBuilder, KeySerDes, ValueSerDes); + _onTheFlyConsumer = (KeySerDesSelector != null || ValueSerDesSelector != null) ? new KNetConsumer(consumerConfigBuilder) + : new KNetConsumer(consumerConfigBuilder, KeySerDes, ValueSerDes); } } [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.AggressiveInlining)] @@ -1026,20 +1026,20 @@ void BuildProducer() ProducerConfig.BootstrapServers = BootstrapServers; - if (KNetKeySerDes != null) + if (KeySerDesSelector != null) { - ProducerConfig.KNetKeySerDes = KNetKeySerDes; + ProducerConfig.KeySerDesSelector = KeySerDesSelector; } - else if (KeySerDes == null) throw new InvalidOperationException($"{typeof(K)} needs an external serializer, set KNetKeySerDes or KeySerDes."); + else if (KeySerDes == null) throw new InvalidOperationException($"{typeof(K)} needs an external serializer, set {nameof(KeySerDesSelector)} or {nameof(KeySerDes)}."); - if (KNetValueSerDes != null) + if (ValueSerDesSelector != null) { - ProducerConfig.KNetValueSerDes = KNetValueSerDes; + ProducerConfig.ValueSerDesSelector = ValueSerDesSelector; } - else if (ValueSerDes == null) throw new InvalidOperationException($"{typeof(V)} needs an external serializer, set KNetValueSerDes or ValueSerDes."); + else if (ValueSerDes == null) throw new InvalidOperationException($"{typeof(V)} needs an external serializer, set {nameof(ValueSerDesSelector)} or {nameof(ValueSerDes)}."); - _producer = (KNetKeySerDes != null || KNetValueSerDes != null) ? new KNetProducer(ProducerConfig) - : new KNetProducer(ProducerConfig, KeySerDes, ValueSerDes); + _producer = (KeySerDesSelector != null || ValueSerDesSelector != null) ? new KNetProducer(ProducerConfig) + : new KNetProducer(ProducerConfig, KeySerDes, ValueSerDes); } void ConsumerPollHandler(object o) @@ -1119,8 +1119,8 @@ public void Start() if (string.IsNullOrWhiteSpace(BootstrapServers)) throw new InvalidOperationException("BootstrapServers must be set before start."); if (string.IsNullOrWhiteSpace(StateName)) throw new InvalidOperationException("StateName must be set before start."); - if (KNetKeySerDes != null && KeySerDes != null) { throw new InvalidOperationException($"Set only one of {nameof(KNetKeySerDes)} or {nameof(KeySerDes)}."); } - if (KNetValueSerDes != null && ValueSerDes != null) { throw new InvalidOperationException($"Set only one of {nameof(KNetValueSerDes)} or {nameof(ValueSerDes)}."); } + if (KeySerDesSelector != null && KeySerDes != null) { throw new InvalidOperationException($"Set only one of {nameof(KeySerDesSelector)} or {nameof(KeySerDes)}."); } + if (ValueSerDesSelector != null && ValueSerDes != null) { throw new InvalidOperationException($"Set only one of {nameof(ValueSerDesSelector)} or {nameof(ValueSerDes)}."); } if (ConsumerInstances > Partitions) throw new InvalidOperationException("ConsumerInstances cannot be high than Partitions"); @@ -1164,13 +1164,13 @@ public void Start() finally { topic?.Dispose(); } } _disposeKeySerDes = false; - if (KNetKeySerDes == null && KeySerDes == null && KNetSerialization.IsInternalManaged()) + if (KeySerDesSelector == null && KeySerDes == null && KNetSerialization.IsInternalManaged()) { KeySerDes = new SerDes(); _disposeKeySerDes = true; } _disposeValueSerDes = false; - if (KNetValueSerDes == null && ValueSerDes == null && KNetSerialization.IsInternalManaged()) + if (ValueSerDesSelector == null && ValueSerDes == null && KNetSerialization.IsInternalManaged()) { ValueSerDes = new SerDes(); _disposeValueSerDes = true; diff --git a/src/net/KNet/Specific/Serialization/IGenericSerDesFactory.cs b/src/net/KNet/Specific/Serialization/IGenericSerDesFactory.cs index 78000d4d04..3bfd79ff91 100644 --- a/src/net/KNet/Specific/Serialization/IGenericSerDesFactory.cs +++ b/src/net/KNet/Specific/Serialization/IGenericSerDesFactory.cs @@ -31,28 +31,28 @@ public interface IGenericSerDesFactory /// The property is by default to optimized data flow between CLR and JVM bool AutoSelectBuffered { get; } /// - /// The used to create an instance of for keys with + /// The used to create an instance of for keys with /// - Type KNetKeySerDes { get; } + Type KeySerDesSelector { get; } /// - /// The used to create an instance of for values with + /// The used to create an instance of for values with /// - Type KNetValueSerDes { get; } + Type ValueSerDesSelector { get; } /// - /// Builds an instance of using the defined in + /// Builds an instance of using the defined in /// /// The type of the key /// The JVM type of the key /// An instance of - /// If is + /// If is ISerDes BuildKeySerDes(); /// - /// Builds an instance of using the defined in + /// Builds an instance of using the defined in /// /// The type of the key /// The JVM type of the key /// An instance of - /// If is + /// If is ISerDes BuildValueSerDes(); /// /// Clear the current factory diff --git a/src/net/KNet/Specific/Serialization/SerDes.cs b/src/net/KNet/Specific/Serialization/SerDes.cs index 80fd76cfda..42db34b8ad 100644 --- a/src/net/KNet/Specific/Serialization/SerDes.cs +++ b/src/net/KNet/Specific/Serialization/SerDes.cs @@ -16,6 +16,7 @@ * Refer to LICENSE for more information. */ +using Java.Lang; using Java.Nio; using Javax.Xml.Crypto; using MASES.JCOBridge.C2JBridge; @@ -25,6 +26,50 @@ namespace MASES.KNet.Serialization { + /// + /// KNet common interface to select serializer/deserializer + /// + public interface ISerDesSelector + { + /// + /// Returns the name of the implementing + /// + string SelectorTypeName { get; } + /// + /// Returns the generic implementing based on array data exchange + /// + Type ByteArraySerDes { get; } + /// + /// Returns the generic implementing based on data exchange + /// + /// Returns if data exchange is not supported + Type ByteBufferSerDes { get; } + } + + /// + /// KNet common interface to select serializer/deserializer + /// + /// The to be serialized + public interface ISerDesSelector : ISerDesSelector + { + /// + /// Returns an instance of + /// + /// The JVM type to use for + /// The of a new instance of or based on + ISerDes NewSerDes(); + /// + /// Returns an instance of + /// + /// The of a new instance of based on + ISerDesRaw NewByteArraySerDes(); + /// + /// Returns an instance of + /// + /// The of a new instance of based on + ISerDesBuffered NewByteBufferSerDes(); + } + /// /// KNet common serializer/deserializer /// @@ -39,6 +84,30 @@ public interface ISerDes : IDisposable /// Type JVMType { get; } /// + /// Returns the serializer class name to be used + /// + string JVMSerializerClassName { get; } + /// + /// Returns the deserializer class name to be used + /// + string JVMDeserializerClassName { get; } + /// + /// Returns the serdes class name to be used + /// + string JVMSerDesClassName { get; } + /// + /// Returns the JVM serializer associated to this instance + /// + Java.Lang.Class JVMSerializerClass { get; } + /// + /// Returns the JVM deserializer associated to this instance + /// + Java.Lang.Class JVMDeserializerClass { get; } + /// + /// Returns the JVM serdes associated to this instance + /// + Java.Lang.Class JVMSerDesClass { get; } + /// /// if are used /// bool UseHeaders { get; set; } @@ -67,15 +136,6 @@ public interface ISerDes : ISerDes, ISerializer, IDeserializ Serde KafkaSerde { get; } } - ///// - ///// KNet common serializer/deserializer based on array JVM type - ///// - ///// The type to serialize/deserialize - //public interface ISerDes : ISerDes, ISerializer, IDeserializer - //{ - - //} - /// /// Common serializer/deserializer /// @@ -148,6 +208,9 @@ public SerDes() if (IsDirectBuffered) { kSerde = new Serdes.ByteBufferSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); } else { @@ -155,45 +218,84 @@ public SerDes() { case KNetSerialization.SerializationType.Boolean: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.BooleanSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.ByteArray: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.ByteArraySerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.ByteBuffer: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.ByteBufferSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.Bytes: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.BytesSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.Double: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.DoubleSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.Float: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.FloatSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.Integer: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.IntegerSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.Long: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.LongSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.Short: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.ShortSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.String: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.StringSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.Guid: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.UUIDSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; case KNetSerialization.SerializationType.Void: kSerde = new Org.Apache.Kafka.Common.Serialization.Serdes.VoidSerde().CastTo>(); + JVMSerDesClassName = Class.ClassNameOf(); + JVMSerializerClassName = Class.ClassNameOf(); + JVMDeserializerClassName = Class.ClassNameOf(); break; - case KNetSerialization.SerializationType.External: - default: - throw new InvalidOperationException($"{typeof(T)} needs an external serializer: set {nameof(OnSerialize)} or {nameof(OnSerializeWithHeaders)}."); + case KNetSerialization.SerializationType.External: + default: + throw new InvalidOperationException($"{typeof(T)} needs an external serializer: set {nameof(OnSerialize)} or {nameof(OnSerializeWithHeaders)}."); } } + JVMSerDesClass = Java.Lang.Class.ForName(JVMSerDesClassName, true, Java.Lang.ClassLoader.SystemClassLoader); + JVMSerializerClass = Java.Lang.Class.ForName(JVMSerializerClassName, true, Java.Lang.ClassLoader.SystemClassLoader); + JVMDeserializerClass = Java.Lang.Class.ForName(JVMDeserializerClassName, true, Java.Lang.ClassLoader.SystemClassLoader); _KafkaSerde = kSerde; _KafkaSerializer = kSerde.SerializerDirect(); @@ -222,6 +324,18 @@ public void Dispose() public Type Type => typeof(T); /// public Type JVMType => typeof(TJVMT); + /// + public string JVMSerializerClassName { get; protected set; } + /// + public string JVMDeserializerClassName { get; protected set; } + /// + public string JVMSerDesClassName { get; protected set; } + /// + public Java.Lang.Class JVMSerializerClass { get; protected set; } + /// + public Java.Lang.Class JVMDeserializerClass { get; protected set; } + /// + public Java.Lang.Class JVMSerDesClass { get; protected set; } /// public virtual bool UseHeaders { get; set; } = false; /// @@ -366,6 +480,47 @@ public class SerDesBuffered : SerDes, ISerDesBuffered public override bool IsDirectBuffered => true; } + /// + /// Default implementation of + /// + public class DefaultSerDes : ISerDesSelector + { + /// + /// Returns a new instance of + /// + /// The of + public static ISerDesSelector NewInstance() => new DefaultSerDes(); + /// + public static string SelectorTypeName => typeof(DefaultSerDes<>).ToAssemblyQualified(); + /// + public static Type ByteArraySerDes => typeof(SerDesRaw); + /// + public static Type ByteBufferSerDes => typeof(SerDesBuffered); + /// + public static ISerDes NewSerDes() + { + if (typeof(TJVM) == typeof(Java.Nio.ByteBuffer)) return NewByteBufferSerDes() as ISerDes; + return NewByteArraySerDes() as ISerDes; + } + /// + public static ISerDesRaw NewByteArraySerDes() { return new SerDesRaw(); } + /// + public static ISerDesBuffered NewByteBufferSerDes() { return new SerDesBuffered(); } + + /// + string ISerDesSelector.SelectorTypeName => SelectorTypeName; + /// + Type ISerDesSelector.ByteArraySerDes => ByteArraySerDes; + /// + Type ISerDesSelector.ByteBufferSerDes => ByteBufferSerDes; + /// + ISerDes ISerDesSelector.NewSerDes() => NewSerDes(); + /// + ISerDesRaw ISerDesSelector.NewByteArraySerDes() => NewByteArraySerDes(); + /// + ISerDesBuffered ISerDesSelector.NewByteBufferSerDes() => NewByteBufferSerDes(); + } + /// /// Sets of predefined between specific types /// diff --git a/tests/net/KNetBenchmark/ProgramKafka.cs b/tests/net/KNetBenchmark/ProgramKafka.cs index 1966cbf59a..8f29b1389c 100644 --- a/tests/net/KNetBenchmark/ProgramKafka.cs +++ b/tests/net/KNetBenchmark/ProgramKafka.cs @@ -53,8 +53,8 @@ static Org.Apache.Kafka.Clients.Producer.IProducer KafkaProducer() .WithSendBuffer(SocketSendBufferBytes) .WithReceiveBuffer(SocketReceiveBufferBytes) .WithBufferMemory(128 * 1024 * 1024) - .WithKeySerializerClass("org.apache.kafka.common.serialization.LongSerializer") - .WithValueSerializerClass("org.apache.kafka.common.serialization.ByteArraySerializer") + .WithKeySerializerClass(Java.Lang.Class.ClassNameOf()) + .WithValueSerializerClass(Java.Lang.Class.ClassNameOf()) .WithPartitionerIgnoreKeys(true) .ToProperties(); if (UseSerdes) @@ -242,8 +242,8 @@ static Org.Apache.Kafka.Clients.Consumer.IConsumer KafkaConsumer() .WithSendBuffer(SocketSendBufferBytes) .WithReceiveBuffer(SocketReceiveBufferBytes) .WithFetchMinBytes(FetchMinBytes) - .WithKeyDeserializerClass("org.apache.kafka.common.serialization.LongDeserializer") - .WithValueDeserializerClass("org.apache.kafka.common.serialization.ByteArrayDeserializer") + .WithKeyDeserializerClass(Java.Lang.Class.ClassNameOf()) + .WithValueDeserializerClass(Java.Lang.Class.ClassNameOf()) .WithAutoOffsetReset(ConsumerConfigBuilder.AutoOffsetResetTypes.EARLIEST) .ToProperties(); if (UseSerdes) diff --git a/tests/net/KNetCompactedReplicatorTest/Program.cs b/tests/net/KNetCompactedReplicatorTest/Program.cs index 55162e43c1..156b15e8f2 100644 --- a/tests/net/KNetCompactedReplicatorTest/Program.cs +++ b/tests/net/KNetCompactedReplicatorTest/Program.cs @@ -102,7 +102,7 @@ private static void TestValues(string topicName, int length, UpdateModeTypes typ UpdateMode = type, BootstrapServers = serverToUse, StateName = topicName, - KNetValueSerDes = typeof(JsonSerDes.ValueRaw<>), + ValueSerDesSelector = typeof(JsonSerDes.Value<>), //ValueSerDes = new JsonSerDes.Value(), }) { @@ -131,7 +131,7 @@ private static void Test(string topicName, int length, UpdateModeTypes type, int UpdateMode = type, BootstrapServers = serverToUse, StateName = topicName, - KNetValueSerDes = typeof(JsonSerDes.ValueRaw<>), + ValueSerDesSelector = typeof(JsonSerDes.Value<>), // ValueSerDes = new JsonSerDes.Value(), }) { @@ -160,7 +160,7 @@ private static void TestOnlyRead(string topicName, int length, UpdateModeTypes t UpdateMode = type, BootstrapServers = serverToUse, StateName = topicName, - ValueSerDes = new JsonSerDes.ValueRaw(), + ValueSerDes = JsonSerDes.Value.NewByteArraySerDes() }) { replicator.StartAndWait(); diff --git a/tests/net/KNetTest/Program.cs b/tests/net/KNetTest/Program.cs index 0d43e3f453..445bbc423a 100644 --- a/tests/net/KNetTest/Program.cs +++ b/tests/net/KNetTest/Program.cs @@ -274,8 +274,8 @@ static void ProduceSomething() .WithRetries(0) .WithLingerMs(1); - var keySerializer = new SerDesRaw(); - var valueSerializer = new JsonSerDes.ValueRaw(); + var keySerializer = DefaultSerDes.NewByteArraySerDes(); + var valueSerializer = JsonSerDes.Value.NewByteArraySerDes(); Stopwatch watcher = new Stopwatch(); try { @@ -366,8 +366,8 @@ static void ConsumeSomething() .WithEnableAutoCommit(true) .WithAutoCommitIntervalMs(1000); - SerDesRaw keyDeserializer = new SerDesRaw(); - var valueDeserializer = new JsonSerDes.ValueRaw(); + ISerDesRaw keyDeserializer = DefaultSerDes.NewByteArraySerDes(); + var valueDeserializer = JsonSerDes.Value.NewByteArraySerDes(); ConsumerRebalanceListener rebalanceListener = null; KNetConsumer consumer = null; @@ -472,8 +472,8 @@ static void ProduceSomethingBuffered() .WithRetries(0) .WithLingerMs(1); - var keySerializer = new SerDesRaw(); // standard serDes for string - var valueSerializer = new JsonSerDes.ValueBuffered(); + var keySerializer = DefaultSerDes.NewByteArraySerDes(); // standard serDes for string + var valueSerializer = JsonSerDes.Value.NewByteBufferSerDes(); Stopwatch watcher = new Stopwatch(); try { @@ -564,8 +564,8 @@ static void ConsumeSomethingBuffered() .WithEnableAutoCommit(true) .WithAutoCommitIntervalMs(1000); - var keyDeserializer = new SerDesRaw(); - var valueDeserializer = new JsonSerDes.ValueBuffered(); + var keyDeserializer = DefaultSerDes.NewByteArraySerDes(); + var valueDeserializer = JsonSerDes.Value.NewByteBufferSerDes(); ConsumerRebalanceListener rebalanceListener = null; KNetConsumerValueBuffered consumer = null;