support specify customize fury serializer (#10)

Closes #6

deployment/src/main/java/io/quarkiverse/fury/deployment/FuryProcessor.java

@@ -34,10 +34,7 @@ public void findSerializableClasses(
             CombinedIndexBuildItem combinedIndex, BuildProducer<FurySerializerBuildItem> pojoProducer) {
         combinedIndex.getIndex().getAnnotations(FURY_SERIALIZATION).stream()
                 .filter(annotation -> annotation.target().kind() == AnnotationTarget.Kind.CLASS)
-                .forEach(
-                        i -> {
-                            pojoProducer.produce(new FurySerializerBuildItem(i.target().asClass()));
-                        });
+                .forEach(i -> pojoProducer.produce(new FurySerializerBuildItem(i.target().asClass())));
     }
 
     @BuildStep
@@ -52,7 +49,7 @@ void unremovableBeans(
     public void registerClasses(
             FuryBuildItem fury, List<FurySerializerBuildItem> classes, FuryRecorder recorder) {
         for (FurySerializerBuildItem item : classes) {
-            recorder.registerClass(fury.getFury(), item.getClazz(), item.getClassId());
+            recorder.registerClass(fury.getFury(), item.getClazz(), item.getClassId(), item.getSerializer());
         }
     }
 

deployment/src/main/java/io/quarkiverse/fury/deployment/FurySerializerBuildItem.java

@@ -1,5 +1,6 @@
 package io.quarkiverse.fury.deployment;
 
+import org.apache.fury.serializer.Serializer;
 import org.jboss.jandex.ClassInfo;
 import org.jboss.jandex.JandexReflection;
 
@@ -9,10 +10,17 @@
 public final class FurySerializerBuildItem extends MultiBuildItem {
     private final Class<?> clazz;
     private final int classId;
+    private final Class<? extends Serializer> serializer;
 
     public FurySerializerBuildItem(ClassInfo classInfo) {
         clazz = JandexReflection.loadClass(classInfo);
-        classId = clazz.getDeclaredAnnotation(FurySerialization.class).classId();
+        FurySerialization annotation = clazz.getDeclaredAnnotation(FurySerialization.class);
+        classId = annotation.classId();
+        if (annotation.serializer() == Serializer.class) {
+            serializer = null;
+        } else {
+            serializer = annotation.serializer();
+        }
     }
 
     public int getClassId() {
@@ -22,4 +30,8 @@ public int getClassId() {
     public Class<?> getClazz() {
         return clazz;
     }
+
+    public Class<? extends Serializer> getSerializer() {
+        return serializer;
+    }
 }

integration-tests/src/main/java/io/quarkiverse/fury/it/Bar.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+package io.quarkiverse.fury.it;
+
+import io.quarkiverse.fury.FurySerialization;
+
+@FurySerialization(serializer = BarSerializer.class)
+public record Bar(int f1, String f2) {
+}

integration-tests/src/main/java/io/quarkiverse/fury/it/BarSerializer.java

@@ -0,0 +1,22 @@
+package io.quarkiverse.fury.it;
+
+import org.apache.fury.Fury;
+import org.apache.fury.memory.MemoryBuffer;
+import org.apache.fury.serializer.Serializer;
+
+public class BarSerializer extends Serializer<Bar> {
+    public BarSerializer(Fury fury, Class<Bar> type) {
+        super(fury, type);
+    }
+
+    @Override
+    public void write(MemoryBuffer buffer, Bar value) {
+        buffer.writeVarInt32(value.f1());
+        fury.writeJavaString(buffer, value.f2());
+    }
+
+    @Override
+    public Bar read(MemoryBuffer buffer) {
+        return new Bar(buffer.readVarInt32(), fury.readJavaString(buffer));
+    }
+}

integration-tests/src/main/java/io/quarkiverse/fury/it/FuryResources.java

@@ -9,6 +9,10 @@
 import jakarta.ws.rs.Path;
 
 import org.apache.fury.BaseFury;
+import org.apache.fury.Fury;
+import org.apache.fury.ThreadSafeFury;
+import org.apache.fury.serializer.Serializer;
+import org.apache.fury.util.Preconditions;
 
 @Path("/fury")
 @ApplicationScoped
@@ -25,6 +29,21 @@ public Boolean testSerializeFooRecord() {
         return foo1.equals(foo2);
     }
 
+    @GET
+    @Path("/bar")
+    public Boolean testSerializeBarRecord() {
+        Bar bar = new Bar(10, "abc");
+        Bar bar2 = (Bar) fury.deserialize(fury.serialize(bar));
+        Serializer serializer;
+        if (fury instanceof ThreadSafeFury) {
+            serializer = ((ThreadSafeFury) fury).execute(f -> f.getClassResolver().getSerializer(Bar.class));
+        } else {
+            serializer = ((Fury) fury).getClassResolver().getSerializer(Bar.class);
+        }
+        Preconditions.checkArgument(serializer instanceof BarSerializer, serializer);
+        return bar2.equals(bar);
+    }
+
     @GET
     @Path("/pojo")
     public Boolean testSerializePOJO() {

runtime/src/main/java/io/quarkiverse/fury/FuryRecorder.java

@@ -5,6 +5,7 @@
 import org.apache.fury.ThreadSafeFury;
 import org.apache.fury.config.FuryBuilder;
 import org.apache.fury.resolver.ClassResolver;
+import org.apache.fury.serializer.Serializer;
 import org.apache.fury.util.Preconditions;
 
 import io.quarkus.arc.runtime.BeanContainer;
@@ -26,7 +27,8 @@ public RuntimeValue<BaseFury> createFury(
     }
 
     public void registerClass(
-            final RuntimeValue<BaseFury> fury, final Class<?> clazz, final int classId) {
+            final RuntimeValue<BaseFury> fury, final Class<?> clazz,
+            final int classId, Class<? extends Serializer> serializer) {
         BaseFury furyValue = fury.getValue();
         if (classId > 0) {
             Preconditions.checkArgument(
@@ -55,5 +57,8 @@ public void registerClass(
             // Generate serializer bytecode.
             furyValue.register(clazz, true);
         }
+        if (serializer != null) {
+            furyValue.registerSerializer(clazz, serializer);
+        }
     }
 }

runtime/src/main/java/io/quarkiverse/fury/FurySerialization.java

@@ -6,11 +6,20 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.apache.fury.serializer.Serializer;
+
 /** Marks a class as a Fury serializable. */
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 public @interface FurySerialization {
     /** Class id must be greater or equal to 256, and it must be different between classes. */
     int classId() default -1;
+
+    /**
+     * Specify a customized serializer for current class.
+     * This should be null to let Fury create serializer for current class. But if
+     * users want to customize serialization for this class, one can provide serializer here.
+     */
+    Class<? extends Serializer> serializer() default Serializer.class;
 }