From 2a2b55a26921dad9865ef234752593bc77501e79 Mon Sep 17 00:00:00 2001
From: masesdevelopers <94312179+masesdevelopers@users.noreply.github.com>
Date: Thu, 16 May 2024 04:37:41 +0200
Subject: [PATCH] Updates on serializers retrieve

---
 .../Kafka/Common/Serialization/Serde.cs       | 11 ++--
 src/net/KNet/Specific/Serialization/SerDes.cs | 58 ++++++-------------
 2 files changed, 26 insertions(+), 43 deletions(-)

diff --git a/src/net/KNet/Developed/Org/Apache/Kafka/Common/Serialization/Serde.cs b/src/net/KNet/Developed/Org/Apache/Kafka/Common/Serialization/Serde.cs
index 50d260a0be..9ccd15e5a3 100644
--- a/src/net/KNet/Developed/Org/Apache/Kafka/Common/Serialization/Serde.cs
+++ b/src/net/KNet/Developed/Org/Apache/Kafka/Common/Serialization/Serde.cs
@@ -17,14 +17,15 @@
 */
 
 using MASES.JCOBridge.C2JBridge;
+using MASES.JCOBridge.C2JBridge.JVMInterop;
 
 namespace Org.Apache.Kafka.Common.Serialization
 { 
-        /// <summary>
+    /// <summary>
     /// Listener for Kafka Serializer. Extends <see cref="JVMBridgeListener"/>. Implements <see cref="ISerializer{T}"/>
     /// </summary>
     /// <remarks>Dispose the object to avoid a resource leak, the object contains a reference to the corresponding JVM object</remarks>
-    public partial class Serde<T> : ISerde<T>
+    public partial class SerdeDirect<T> : ISerde<T>
     {
         /// <summary>
         /// <see href="https://www.javadoc.io/doc/org.apache.kafka/kafka-clients/3.6.2/org/apache/kafka/common/serialization/Serde.html#deserializer--"/>
@@ -32,7 +33,8 @@ public partial class Serde<T> : ISerde<T>
         /// <returns><see cref="Org.Apache.Kafka.Common.Serialization.Deserializer"/></returns>
         public Org.Apache.Kafka.Common.Serialization.Deserializer<T> DeserializerDirect()
         {
-            return new DeserializerDirect<T>();
+            var res = this.IExecute("deserializer") as IJavaObject;
+            return WrapsDirect<DeserializerDirect<T>>(res);
         }
         /// <summary>
         /// <see href="https://www.javadoc.io/doc/org.apache.kafka/kafka-clients/3.6.2/org/apache/kafka/common/serialization/Serde.html#serializer--"/>
@@ -40,7 +42,8 @@ public Org.Apache.Kafka.Common.Serialization.Deserializer<T> DeserializerDirect(
         /// <returns><see cref="Org.Apache.Kafka.Common.Serialization.Serializer"/></returns>
         public Org.Apache.Kafka.Common.Serialization.Serializer<T> SerializerDirect()
         {
-            return new SerializerDirect<T>();
+            var res = this.IExecute("serializer") as IJavaObject;
+            return WrapsDirect<SerializerDirect<T>>(res);
         }
     }
 }
diff --git a/src/net/KNet/Specific/Serialization/SerDes.cs b/src/net/KNet/Specific/Serialization/SerDes.cs
index 129df1889a..0f6752d987 100644
--- a/src/net/KNet/Specific/Serialization/SerDes.cs
+++ b/src/net/KNet/Specific/Serialization/SerDes.cs
@@ -114,81 +114,61 @@ public SerDes()
                     throw new InvalidOperationException($"{_SerializationType} is not valid.");
             }
 
+            SerdeDirect<TJVMT> kSerde = null;
+
             if (IsDirectBuffered)
             {
-                _KafkaSerializer = new KNetByteBufferSerializer().CastDirect<Serializer<TJVMT>>();
-                _KafkaDeserializer = new KNetByteBufferDeserializer().CastDirect<Deserializer<TJVMT>>();
-                _KafkaSerde = new KNetSerdes.ByteBufferSerde().CastDirect<Serde<TJVMT>>();
+                kSerde = new KNetSerdes.ByteBufferSerde().CastTo<SerdeDirect<TJVMT>>();
             }
             else
             {
                 switch (_JVMSerializationType)
                 {
                     case KNetSerialization.SerializationType.Boolean:
-                        _KafkaSerializer = new BooleanSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new BooleanDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.BooleanSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.BooleanSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.ByteArray:
-                        _KafkaSerializer = new ByteArraySerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new ByteArrayDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.ByteArraySerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.ByteArraySerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.ByteBuffer:
-                        _KafkaSerializer = new ByteBufferSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new ByteBufferDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.ByteBufferSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.ByteBufferSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.Bytes:
-                        _KafkaSerializer = new BytesSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new BytesDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.BytesSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.BytesSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.Double:
-                        _KafkaSerializer = new DoubleSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new DoubleDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.DoubleSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.DoubleSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.Float:
-                        _KafkaSerializer = new FloatSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new FloatDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.FloatSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.FloatSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.Integer:
-                        _KafkaSerializer = new IntegerSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new IntegerDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.IntegerSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.IntegerSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.Long:
-                        _KafkaSerializer = new LongSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new LongDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.LongSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.LongSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.Short:
-                        _KafkaSerializer = new ShortSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new ShortDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.ShortSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.ShortSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.String:
-                        _KafkaSerializer = new StringSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new StringDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.StringSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.StringSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.Guid:
-                        _KafkaSerializer = new UUIDSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new UUIDDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.UUIDSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.UUIDSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                     case KNetSerialization.SerializationType.Void:
-                        _KafkaSerializer = new VoidSerializer().CastDirect<Serializer<TJVMT>>();
-                        _KafkaDeserializer = new VoidDeserializer().CastDirect<Deserializer<TJVMT>>();
-                        _KafkaSerde = new Serdes.VoidSerde().CastDirect<Serde<TJVMT>>();
+                        kSerde = new Serdes.VoidSerde().CastTo<SerdeDirect<TJVMT>>();
                         break;
                 case KNetSerialization.SerializationType.External:
                 default:
                     throw new InvalidOperationException($"{typeof(T)} needs an external serializer: set {nameof(OnSerialize)} or {nameof(OnSerializeWithHeaders)}.");
                 }
             }
+
+            _KafkaSerde = kSerde;
+            _KafkaSerializer = kSerde.SerializerDirect();
+            _KafkaDeserializer = kSerde.DeserializerDirect();
         }
         /// <summary>
         /// Finalizer