Added management of data transfer on JVM-CLR boundary based on ByteBuffer

src/jvm/knet/src/main/java/org/mases/knet/common/serialization/ByteBufferDeserializer.java

@@ -0,0 +1,41 @@
+/*
+ *  Copyright 2024 MASES s.r.l.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  Refer to LICENSE for more information.
+ */
+
+package org.mases.knet.common.serialization;
+
+import org.apache.kafka.common.serialization.Deserializer;
+import org.mases.jcobridge.JCSharedBuffer;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class ByteBufferDeserializer implements Deserializer<ByteBuffer> {
+
+    @Override
+    public ByteBuffer deserialize(String topic, byte[] data) {
+        if (data == null)
+            return null;
+
+        try {
+            return JCSharedBuffer.Create(data);
+        } catch (IOException e) {
+            // fallback to non buffered version
+            return ByteBuffer.wrap(data);
+        }
+    }
+}

src/jvm/knet/src/main/java/org/mases/knet/common/serialization/ByteBufferSerializer.java

@@ -0,0 +1,74 @@
+/*
+ *  Copyright 2024 MASES s.r.l.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  Refer to LICENSE for more information.
+ */
+
+package org.mases.knet.common.serialization;
+
+import org.apache.kafka.common.serialization.Serializer;
+import org.mases.jcobridge.JCSharedBuffer;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * {@code ByteBufferSerializer} always {@link ByteBuffer#rewind() rewinds} the position of the input buffer to zero for
+ * serialization. A manual rewind is not necessary.
+ * <p>
+ * Note: any existing buffer position is ignored.
+ * <p>
+ * The position is also rewound back to zero before {@link #serialize(String, ByteBuffer)}
+ * returns.
+ */
+public class ByteBufferSerializer implements Serializer<ByteBuffer> {
+    public byte[] serialize(String topic, ByteBuffer data) {
+        if (data == null)
+            return null;
+
+        try {
+            try (JCSharedBuffer sb = JCSharedBuffer.Create(data)) {
+                sb.rewind();
+
+                if (sb.hasArray()) {
+                    byte[] arr = sb.array();
+                    if (sb.arrayOffset() == 0 && arr.length == sb.remaining()) {
+                        return arr;
+                    }
+                }
+
+                byte[] ret = new byte[data.remaining()];
+                sb.get(ret, 0, ret.length);
+                sb.rewind();
+                return ret;
+            }
+        } catch (IOException iae) {
+            // fallback to non buffered version
+            data.rewind();
+
+            if (data.hasArray()) {
+                byte[] arr = data.array();
+                if (data.arrayOffset() == 0 && arr.length == data.remaining()) {
+                    return arr;
+                }
+            }
+
+            byte[] ret = new byte[data.remaining()];
+            data.get(ret, 0, ret.length);
+            data.rewind();
+            return ret;
+        }
+    }
+}

src/jvm/knet/src/main/java/org/mases/knet/common/serialization/Serdes.java

@@ -0,0 +1,40 @@
+/*
+ *  Copyright 2024 MASES s.r.l.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  Refer to LICENSE for more information.
+ */
+
+package org.mases.knet.common.serialization;
+
+import org.apache.kafka.common.serialization.Serde;
+
+import java.nio.ByteBuffer;
+
+public class Serdes {
+    static public final class ByteBufferSerde extends org.apache.kafka.common.serialization.Serdes.WrapperSerde<ByteBuffer> {
+        public ByteBufferSerde() {
+            super(new ByteBufferSerializer(), new ByteBufferDeserializer());
+        }
+    }
+
+    /**
+     * A serde for nullable {@code ByteBuffer} type.
+     *
+     * @return {@link Serde} of {@link ByteBuffer}
+     */
+    static public Serde<ByteBuffer> ByteBuffer() {
+        return new ByteBufferSerde();
+    }
+}

src/net/Common/Common.props

@@ -22,7 +22,7 @@
 	</PropertyGroup>
 	<PropertyGroup Condition="'$(KNET_DOCKER_BUILD_ACTIONS)' != 'true'">
 		<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-		<TargetFrameworks>net462;net6.0;net7.0;net8.0</TargetFrameworks>
+		<TargetFrameworks>net462;net6.0;net8.0</TargetFrameworks>
 	</PropertyGroup>
 	<PropertyGroup Condition="'$(KNET_DOCKER_BUILD_ACTIONS)' == 'true'">
 		<GeneratePackageOnBuild>false</GeneratePackageOnBuild>